Skip to content

chatomon.schemas

Data contracts shared by the engine, the orbitals and the connectors.

This module imports nothing from chatomon, so every other module can depend on it without creating import cycles.

Classes:

Name Description
Message

One message of the conversation history.
Verdict

The result of an orbital's check.
OrbitalContext

What an orbital sees when it checks a message.
InputDecision

What the connector must do after the input chain.
OutputDecision

What the connector must do after the output chain.
OrbitalResult

Audit record of one orbital execution.
OutputAttempt

Audit record of one reply of the nucleus and of the output chain that checked it.
TurnReport

Audit record of one conversation turn, delivered to on_turn.

Attributes:

Name Type Description
Side

Where an orbital runs: on the user's message ("input") or on the nucleus reply ("output").
VerdictAction

What an orbital decided about the text it checked.
ErrorPolicy

What the engine does when an orbital fails or times out.
CaptureContent

How much message content chatomon records in turn reports; logs and traces carry
FinalAction

How a turn ended, as recorded in the TurnReport.
OutputDecisionKind

What the engine decided about a reply of the nucleus: accept sends it, replace
DEFAULT_REFUSAL

Reply used when a message is blocked without a specific reply.
DEFAULT_BLOCKED_PLACEHOLDER

Text that replaces a blocked user message in the conversation history.
DEFAULT_BLOCKED_GUIDANCE

Guidance the nucleus receives first when it writes the refusal of a blocked message.

Side module-attribute 

Side = Literal['input', 'output']

Where an orbital runs: on the user's message ("input") or on the nucleus reply ("output").

VerdictAction module-attribute 

VerdictAction = Literal['allow', 'block', 'respond']

What an orbital decided about the text it checked.

ErrorPolicy module-attribute 

ErrorPolicy = Literal['block', 'allow']

What the engine does when an orbital fails or times out.

CaptureContent module-attribute 

CaptureContent = Literal['none', 'hash', 'full']

How much message content chatomon records in turn reports; logs and traces carry labels only.

FinalAction module-attribute 

FinalAction = Literal[
    "answered",
    "orbital_response",
    "blocked",
    "guided_refusal",
    "fallback_refusal",
    "aborted",
]

How a turn ended, as recorded in the TurnReport.

OutputDecisionKind module-attribute 

OutputDecisionKind = Literal[
    "accept", "replace", "regenerate"
]

What the engine decided about a reply of the nucleus: accept sends it, replace sends another text instead, regenerate asks the nucleus for a new one.

DEFAULT_REFUSAL module-attribute 

DEFAULT_REFUSAL = "I'm sorry, I can't help with that."

Reply used when a message is blocked without a specific reply.

DEFAULT_BLOCKED_PLACEHOLDER module-attribute 

DEFAULT_BLOCKED_PLACEHOLDER = (
    "[message removed by chatomon]"
)

Text that replaces a blocked user message in the conversation history.

DEFAULT_BLOCKED_GUIDANCE module-attribute 

DEFAULT_BLOCKED_GUIDANCE = "The user's last message is withheld from you and a placeholder stands in its place. Do not mention the placeholder or that a message was removed: reply as the guidance says."

Guidance the nucleus receives first when it writes the refusal of a blocked message.

The nucleus is then called with the placeholder in place of the message. A model that is not told what the placeholder is can answer about it instead of following the orbital's guidance.

Message pydantic-model

Bases: BaseModel

One message of the conversation history.

Config:

  • frozen: True

Fields:

role pydantic-field 

role: Literal['user', 'assistant']

Who wrote the message.

content pydantic-field 

content: str

The message text.

Verdict pydantic-model

Bases: BaseModel

The result of an orbital's check.

Build verdicts with allow, block and respond rather than the constructor.

Examples:

from chatomon import Verdict

ok = Verdict.allow(guidance="The user sounds frustrated: be empathetic.")
refusal = Verdict.block(reply="I can only help with parcel tracking.")
rewrite = Verdict.block(guidance="Do not mention competitors.")
faq = Verdict.respond("We are open Monday to Friday, 9 to 18.")

assert ok.action == "allow"
assert refusal.reply and rewrite.guidance and faq.reply

Config:

  • frozen: True
  • arbitrary_types_allowed: True

Fields:

Validators:

  • _check_consistency

action pydantic-field 

action: VerdictAction

"allow", "block" or "respond".

reply pydantic-field 

reply: str | None = None

Text shown to the user (block with a reply, or respond).

guidance pydantic-field 

guidance: str | None = None

Instruction for the nucleus (allow on input, or block).

reason pydantic-field 

reason: str | None = None

Short label for the audit report. Must never contain user text.

escalate pydantic-field 

escalate: bool = False

Whether the conversation needs a human.

annotations pydantic-field 

annotations: Any = None

Data for later orbitals. Treated as content by capture_content.

allow classmethod 

allow(
    guidance: str | None = None,
    *,
    reason: str | None = None,
    annotations: Any = None
) -> Verdict

Let the text through.

Parameters:

Name Type Description Default
guidance str | None

On the input side, an instruction the nucleus follows for this turn. Ignored (with a warning) on the output side.

 None
reason str | None

Short label for the audit report.

 None
annotations Any

Data for later orbitals.

 None

Returns:

Type Description
Verdict

An allow verdict.
Source code in src/chatomon/schemas.py 
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
@classmethod
def allow(
    cls, guidance: str | None = None, *, reason: str | None = None, annotations: Any = None
) -> Verdict:
    """Let the text through.

    Args:
        guidance: On the input side, an instruction the nucleus follows for this
            turn. Ignored (with a warning) on the output side.
        reason: Short label for the audit report.
        annotations: Data for later orbitals.

    Returns:
        An `allow` verdict.
    """
    return cls(action="allow", guidance=guidance, reason=reason, annotations=annotations)

block classmethod 

block(
    reply: str | None = None,
    *,
    guidance: str | None = None,
    reason: str | None = None,
    escalate: bool = False,
    annotations: Any = None
) -> Verdict

Stop the text.

With reply, the user receives it. With guidance, the nucleus writes the answer following it (on the input side without seeing the blocked message; on the output side by regenerating its reply). With neither, the atom's refusal is used.

Parameters:

Name Type Description Default
reply str | None

Text shown to the user.

 None
guidance str | None

Instruction for the nucleus.

 None
reason str | None

Short label for the audit report.

 None
escalate bool

Whether the conversation needs a human.

 False
annotations Any

Data for later orbitals.

 None

Returns:

Type Description
Verdict

A block verdict.
Source code in src/chatomon/schemas.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
@classmethod
def block(
    cls,
    reply: str | None = None,
    *,
    guidance: str | None = None,
    reason: str | None = None,
    escalate: bool = False,
    annotations: Any = None,
) -> Verdict:
    """Stop the text.

    With `reply`, the user receives it. With `guidance`, the nucleus writes the
    answer following it (on the input side without seeing the blocked message;
    on the output side by regenerating its reply). With neither, the atom's
    refusal is used.

    Args:
        reply: Text shown to the user.
        guidance: Instruction for the nucleus.
        reason: Short label for the audit report.
        escalate: Whether the conversation needs a human.
        annotations: Data for later orbitals.

    Returns:
        A `block` verdict.
    """
    return cls(
        action="block",
        reply=reply,
        guidance=guidance,
        reason=reason,
        escalate=escalate,
        annotations=annotations,
    )

respond classmethod 

respond(
    text: str,
    *,
    reason: str | None = None,
    escalate: bool = False,
    annotations: Any = None
) -> Verdict

Answer the user directly instead of the nucleus.

Parameters:

Name Type Description Default
text str

The reply shown to the user.

 required
reason str | None

Short label for the audit report.

 None
escalate bool

Whether the conversation needs a human.

 False
annotations Any

Data for later orbitals.

 None

Returns:

Type Description
Verdict

A respond verdict.
Source code in src/chatomon/schemas.py 
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
@classmethod
def respond(
    cls,
    text: str,
    *,
    reason: str | None = None,
    escalate: bool = False,
    annotations: Any = None,
) -> Verdict:
    """Answer the user directly instead of the nucleus.

    Args:
        text: The reply shown to the user.
        reason: Short label for the audit report.
        escalate: Whether the conversation needs a human.
        annotations: Data for later orbitals.

    Returns:
        A `respond` verdict.
    """
    return cls(
        action="respond",
        reply=text,
        reason=reason,
        escalate=escalate,
        annotations=annotations,
    )

OrbitalContext pydantic-model

Bases: BaseModel

What an orbital sees when it checks a message.

Config:

  • frozen: True
  • arbitrary_types_allowed: True

Fields:

side pydantic-field 

side: Side

Whether the text is the user's message ("input") or the nucleus reply ("output").

message pydantic-field 

message: str

The text under check.

history pydantic-field 

history: tuple[Message, ...] = ()

Earlier messages, with blocked user messages already replaced by the placeholder. On the output side it includes the current user message.

annotations pydantic-field 

annotations: Mapping[str, Any]

Annotations of the orbitals that already ran in this turn, by orbital name.

The values are the objects the producing orbitals returned, shared with later orbitals and the report: treat them as read-only.

attempt pydantic-field 

attempt: int = 1

On the output side, the attempt number (1 is the first reply).

turn_id pydantic-field 

turn_id: str

Identifier of the turn.

conversation_id pydantic-field 

conversation_id: str | None = None

Identifier of the conversation, when the framework provides one.

InputDecision pydantic-model

Bases: BaseModel

What the connector must do after the input chain.

Config:

  • frozen: True

Fields:

Validators:

  • _check_payload

kind pydantic-field 

kind: Literal['proceed', 'reply', 'guided_refusal']

proceed: call the nucleus with the message; reply: send reply without calling the nucleus; guided_refusal: call the nucleus without the message, with guidance only.

guidance pydantic-field 

guidance: tuple[str, ...] = ()

Guidance to add to this turn's model call. In a guided_refusal it is the atom's blocked_guidance, when it has one, followed by the guidance of the orbital that blocked.

reply pydantic-field 

reply: str | None = None

The reply to send when kind is reply.

redact_user_message pydantic-field 

redact_user_message: bool = False

Whether the user message must be replaced by the placeholder in the history.

OutputDecision pydantic-model

Bases: BaseModel

What the connector must do after the output chain.

Config:

  • frozen: True

Fields:

Validators:

  • _check_payload

kind pydantic-field 

kind: OutputDecisionKind

accept: send the reply; replace: send text instead; regenerate: call the nucleus again with guidance, then run the output chain again.

text pydantic-field 

text: str | None = None

The replacement reply when kind is replace.

guidance pydantic-field 

guidance: tuple[str, ...] = ()

Guidance for the regeneration when kind is regenerate.

OrbitalResult pydantic-model

Bases: BaseModel

Audit record of one orbital execution.

Config:

  • frozen: True

Fields:

name pydantic-field 

name: str

Orbital name.

side pydantic-field 

side: Side

Side it ran on.

kind pydantic-field 

kind: str

Orbital class name.

fingerprint pydantic-field 

fingerprint: str

Hash of the orbital's configuration at build time.

action pydantic-field 

action: VerdictAction

Action of the verdict that was applied (after on_error).

reason pydantic-field 

reason: str | None = None

Verdict reason label.

has_guidance pydantic-field 

has_guidance: bool = False

Whether the verdict carried guidance.

escalate pydantic-field 

escalate: bool = False

Whether the verdict asked for a human.

attempt pydantic-field 

attempt: int = 1

Attempt number (output side).

duration_ms pydantic-field 

duration_ms: float

Execution time in milliseconds.

error_type pydantic-field 

error_type: str | None = None

Exception class name when the orbital failed or timed out.

error_message pydantic-field 

error_message: str | None = None

Exception message; only with capture_content="full".

error_policy pydantic-field 

error_policy: ErrorPolicy | None = None

The on_error policy applied when the orbital failed.

guidance pydantic-field 

guidance: str | None = None

Guidance text, according to capture_content.

reply pydantic-field 

reply: str | None = None

Reply text, according to capture_content.

annotations pydantic-field 

annotations: str | None = None

Annotations as JSON, according to capture_content.

OutputAttempt pydantic-model

Bases: BaseModel

Audit record of one reply of the nucleus and of the output chain that checked it.

Config:

  • frozen: True

Fields:

nucleus_reply pydantic-field 

nucleus_reply: str | None = None

The reply the nucleus wrote, according to capture_content. It is what the nucleus wrote whatever the decision: a text that took its place is in the result of the orbital that gave it and, when it ended the turn, in final_reply.

results pydantic-field 

results: tuple[OrbitalResult, ...]

Results of the output chain on this reply, in order.

decision pydantic-field 

decision: OutputDecisionKind | None = None

What the engine decided about the reply; None when the turn was aborted while the output chain was running.

TurnReport pydantic-model

Bases: BaseModel

Audit record of one conversation turn, delivered to on_turn.

Config:

  • frozen: True

Fields:

turn_id pydantic-field 

turn_id: str

Identifier of the turn.

conversation_id pydantic-field 

conversation_id: str | None

Identifier of the conversation, when available.

atom_name pydantic-field 

atom_name: str

Name of the atom.

started_at pydantic-field 

started_at: datetime

When the turn started (UTC).

duration_ms pydantic-field 

duration_ms: float

Time from the start of the turn to its report, in milliseconds.

final_action pydantic-field 

final_action: FinalAction

How the turn ended.

escalated pydantic-field 

escalated: bool

Whether any orbital asked for a human.

input pydantic-field 

input: tuple[OrbitalResult, ...]

Results of the input chain.

output pydantic-field 

output: tuple[OutputAttempt, ...]

The attempts of the nucleus, in order: one per reply the output chain checked.

user_message pydantic-field 

user_message: str | None = None

The user message, according to capture_content.

final_reply pydantic-field 

final_reply: str | None = None

The reply sent to the user, according to capture_content.

error_type pydantic-field 

error_type: str | None = None

Exception class name when the turn was aborted.

error_message pydantic-field 

error_message: str | None = None

Exception message of an aborted turn; only with capture_content="full".