Skip to content

chatomon.orbitals.limits

Ready-made orbitals that cap the length of messages and the number of turns.

Classes:

Name Description
MessageLengthOrbital

Blocks a user message or a nucleus reply that is too long.
TurnLimitOrbital

Closes the conversation after a number of turns.

MessageLengthOrbital 

MessageLengthOrbital(
    *,
    max_words: int | None = None,
    max_chars: int | None = None,
    reply: str = "Your message is too long. Please shorten it and try again.",
    guidance: str | None = None,
    name: str | None = None,
    timeout: float = 30.0,
    on_error: ErrorPolicy = "block"
)

Bases: Orbital

Blocks a user message or a nucleus reply that is too long.

Words are counted by splitting the text on whitespace, characters with len(); the text is too long when it exceeds any limit that is set. On the input side the user receives reply, the nucleus is not called and the message is replaced by the placeholder in the history. On the output side the nucleus rewrites its reply following the guidance, up to the atom's max_retries; when the retries run out the user receives the atom's refusal.

Every verdict carries the annotations {"words": ..., "chars": ...}; the audit reasons are message_too_long (input) and reply_too_long (output).

Parameters:

Name Type Description Default
max_words int | None

Maximum number of words.

 None
max_chars int | None

Maximum number of characters.

 None
reply str

Sent to the user when their message is too long.

 'Your message is too long. Please shorten it and try again.'
guidance str | None

Instruction for rewriting a reply that is too long; by default it states the limits that are set.

 None
name str | None

Defaults to message_length.

 None
timeout float

Seconds the engine waits for the check.

 30.0
on_error ErrorPolicy

"block" (default) or "allow".

 'block'

Raises:

Type Description
ChatomonConfigError

If no limit is set, a limit is not a positive integer, or a text is empty.

Examples:

import asyncio

from chatomon.orbitals import MessageLengthOrbital
from chatomon.schemas import OrbitalContext

limit = MessageLengthOrbital(max_words=5)
message = OrbitalContext(
    side="input", message="one two three four five six", turn_id="t1"
)
verdict = asyncio.run(limit.check(message))
assert verdict.action == "block"
assert verdict.reason == "message_too_long"
assert verdict.annotations == {"words": 6, "chars": 27}

Methods:

Name Description
check

Measure the message and block it when it exceeds a limit.
config

Return the configuration, including the limits and the texts.

Attributes:

Name Type Description
calls_model

It counts words and characters.
output_guidance str

The guidance for a reply that is too long: guidance, or one stating the limits.
Source code in src/chatomon/orbitals/limits.py 
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
def __init__(
    self,
    *,
    max_words: int | None = None,
    max_chars: int | None = None,
    reply: str = "Your message is too long. Please shorten it and try again.",
    guidance: str | None = None,
    name: str | None = None,
    timeout: float = 30.0,
    on_error: ErrorPolicy = "block",
) -> None:
    super().__init__(name=name, timeout=timeout, on_error=on_error)
    if max_words is None and max_chars is None:
        raise ChatomonConfigError("set max_words, max_chars or both")
    self.max_words = None if max_words is None else _positive_int("max_words", max_words)
    self.max_chars = None if max_chars is None else _positive_int("max_chars", max_chars)
    self.reply = _not_blank("reply", reply)
    self.guidance = None if guidance is None else _not_blank("guidance", guidance)

calls_model class-attribute instance-attribute 

calls_model = False

It counts words and characters.

output_guidance property 

output_guidance: str

The guidance for a reply that is too long: guidance, or one stating the limits.

check async 

check(ctx: OrbitalContext) -> Verdict

Measure the message and block it when it exceeds a limit.

Parameters:

Name Type Description Default
ctx OrbitalContext

The text under check; the history is not used.

 required

Returns:

Type Description
Verdict

allow within the limits; otherwise block with reply on the input
Verdict

side, or with the guidance on the output side.
Source code in src/chatomon/orbitals/limits.py 
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
async def check(self, ctx: OrbitalContext) -> Verdict:
    """Measure the message and block it when it exceeds a limit.

    Args:
        ctx: The text under check; the history is not used.

    Returns:
        `allow` within the limits; otherwise `block` with `reply` on the input
        side, or with the guidance on the output side.
    """
    words = len(ctx.message.split())
    chars = len(ctx.message)
    measures = {"words": words, "chars": chars}
    too_long = (self.max_words is not None and words > self.max_words) or (
        self.max_chars is not None and chars > self.max_chars
    )
    if not too_long:
        return Verdict.allow(annotations=measures)
    if ctx.side == "input":
        return Verdict.block(self.reply, reason="message_too_long", annotations=measures)
    return Verdict.block(
        guidance=self.output_guidance, reason="reply_too_long", annotations=measures
    )

config 

config() -> dict[str, Any]

Return the configuration, including the limits and the texts.

Returns:

Type Description
dict[str, Any]

A JSON-serializable dictionary.
Source code in src/chatomon/orbitals/limits.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
def config(self) -> dict[str, Any]:
    """Return the configuration, including the limits and the texts.

    Returns:
        A JSON-serializable dictionary.
    """
    return {
        **super().config(),
        "max_words": self.max_words,
        "max_chars": self.max_chars,
        "reply": self.reply,
        "guidance": self.guidance,
    }

TurnLimitOrbital 

TurnLimitOrbital(
    *,
    max_turns: int,
    reply: str = "This conversation has reached its limit. Please start a new one.",
    warn_before: int | None = None,
    warning: str = "The conversation will end soon: help the user wrap up.",
    name: str | None = None,
    timeout: float = 30.0,
    on_error: ErrorPolicy = "block"
)

Bases: Orbital

Closes the conversation after a number of turns.

The current turn is the number of user messages in the history plus one; blocked messages count, since they stay in the history as the placeholder. Past max_turns every message is answered with reply without calling the nucleus. With warn_before, the nucleus receives warning as guidance in the last warn_before turns before the limit. chatomon does not end the session: the application reads the turn_limit reason in the turn report and closes it.

The count is right only when the application passes the full history; with a trimmed history it stops growing at the size of the window. Every verdict carries the annotations {"turn": ..., "max_turns": ...}; the audit reasons are turn_limit and turn_limit_warning. Input side only.

Parameters:

Name Type Description Default
max_turns int

Number of turns answered by the nucleus.

 required
reply str

Sent to the user on every turn after the limit.

 'This conversation has reached its limit. Please start a new one.'
warn_before int | None

Number of turns before the limit that carry the warning.

 None
warning str

Guidance for the nucleus in the warning turns.

 'The conversation will end soon: help the user wrap up.'
name str | None

Defaults to turn_limit.

 None
timeout float

Seconds the engine waits for the check.

 30.0
on_error ErrorPolicy

"block" (default) or "allow".

 'block'

Raises:

Type Description
ChatomonConfigError

If max_turns is not a positive integer, warn_before is not a positive integer smaller than max_turns, or a text is empty.

Examples:

import asyncio

from chatomon.orbitals import TurnLimitOrbital
from chatomon.schemas import Message, OrbitalContext

limit = TurnLimitOrbital(max_turns=1, reply="Please start a new chat.")
history = (Message(role="user", content="hi"), Message(role="assistant", content="hello"))
message = OrbitalContext(side="input", message="again", history=history, turn_id="t2")
verdict = asyncio.run(limit.check(message))
assert (verdict.action, verdict.reply) == ("respond", "Please start a new chat.")
assert verdict.annotations == {"turn": 2, "max_turns": 1}

Methods:

Name Description
check

Count the turn and close the conversation past the limit.
config

Return the configuration, including the limit and the texts.

Attributes:

Name Type Description
sides frozenset[Side]

Input side only: the count is taken on the user message.
calls_model

It counts the user messages of the history.
Source code in src/chatomon/orbitals/limits.py 
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
def __init__(
    self,
    *,
    max_turns: int,
    reply: str = "This conversation has reached its limit. Please start a new one.",
    warn_before: int | None = None,
    warning: str = "The conversation will end soon: help the user wrap up.",
    name: str | None = None,
    timeout: float = 30.0,
    on_error: ErrorPolicy = "block",
) -> None:
    super().__init__(name=name, timeout=timeout, on_error=on_error)
    self.max_turns = _positive_int("max_turns", max_turns)
    self.warn_before = (
        None if warn_before is None else _positive_int("warn_before", warn_before)
    )
    if self.warn_before is not None and self.warn_before >= self.max_turns:
        raise ChatomonConfigError(
            f"warn_before must be smaller than max_turns ({self.max_turns}), "
            f"got {self.warn_before!r}"
        )
    self.reply = _not_blank("reply", reply)
    self.warning = _not_blank("warning", warning)

sides class-attribute 

sides: frozenset[Side] = frozenset({'input'})

Input side only: the count is taken on the user message.

calls_model class-attribute instance-attribute 

calls_model = False

It counts the user messages of the history.

check async 

check(ctx: OrbitalContext) -> Verdict

Count the turn and close the conversation past the limit.

Parameters:

Name Type Description Default
ctx OrbitalContext

The user message and the history.

 required

Returns:

Type Description
Verdict

respond with reply past the limit, allow with the warning in the
Verdict

warning turns, allow otherwise.
Source code in src/chatomon/orbitals/limits.py 
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
async def check(self, ctx: OrbitalContext) -> Verdict:
    """Count the turn and close the conversation past the limit.

    Args:
        ctx: The user message and the history.

    Returns:
        `respond` with `reply` past the limit, `allow` with the warning in the
        warning turns, `allow` otherwise.
    """
    turn = sum(1 for message in ctx.history if message.role == "user") + 1
    measures = {"turn": turn, "max_turns": self.max_turns}
    if turn > self.max_turns:
        return Verdict.respond(self.reply, reason="turn_limit", annotations=measures)
    if self.warn_before is not None and turn > self.max_turns - self.warn_before:
        return Verdict.allow(self.warning, reason="turn_limit_warning", annotations=measures)
    return Verdict.allow(annotations=measures)

config 

config() -> dict[str, Any]

Return the configuration, including the limit and the texts.

Returns:

Type Description
dict[str, Any]

A JSON-serializable dictionary.
Source code in src/chatomon/orbitals/limits.py 
238
239
240
241
242
243
244
245
246
247
248
249
250
def config(self) -> dict[str, Any]:
    """Return the configuration, including the limit and the texts.

    Returns:
        A JSON-serializable dictionary.
    """
    return {
        **super().config(),
        "max_turns": self.max_turns,
        "reply": self.reply,
        "warn_before": self.warn_before,
        "warning": self.warning,
    }