chatomon.engine ¶

def start_turn(self, *, conversation_id: str | None = None) -> TurnState:
    """Start a new turn.

    Args:
        conversation_id: Identifier of the conversation, when the framework has one.

    Returns:
        The state the connector passes to the other methods.
    """
    return TurnState(
        turn_id=uuid.uuid4().hex,
        conversation_id=conversation_id,
        started_at=datetime.now(UTC),
        started_perf=time.perf_counter(),
    )

async def run_input(
    self, turn: TurnState, *, message: str, history: Sequence[Message] = ()
) -> InputDecision:
    """Run the input chain on the user message.

    Args:
        turn: The turn started with `start_turn`.
        message: The user message.
        history: Earlier messages, with blocked user messages already replaced
            by the placeholder.

    Returns:
        What the connector must do next.

    Raises:
        RuntimeError: If the turn is finished or the input chain already ran.
    """
    if turn.finished:
        raise RuntimeError("this turn is already finished")
    if turn.input_decision is not None:
        raise RuntimeError("run_input was already called for this turn")
    turn.user_message = message
    guidance: list[str] = []
    decision: InputDecision | None = None
    with self._tracer.start_as_current_span(
        "chatomon.input", attributes=self._turn_attributes(turn)
    ) as span:
        for orb in self._input:
            verdict = await self._run_orbital(
                orb,
                turn,
                side="input",
                message=message,
                history=history,
                attempt=1,
                results=turn.input_results,
            )
            if verdict.action == "allow":
                if verdict.guidance:
                    guidance.append(verdict.guidance)
                continue
            decision = self._stop_input(turn, verdict)
            break
        if decision is None:
            decision = InputDecision(kind="proceed", guidance=tuple(guidance))
        span.set_attribute("chatomon.decision.kind", decision.kind)
    turn.input_decision = decision
    turn.input_annotations = dict(turn.annotations)
    self._log_decision(turn, "input", decision.kind)
    return decision

async def run_output(
    self, turn: TurnState, *, reply: str, history: Sequence[Message] = ()
) -> OutputDecision:
    """Run the output chain on a final reply of the nucleus.

    Call it on every final reply until the decision is not `regenerate`. Every
    call is one attempt of the turn's report, with the reply recorded according
    to `capture_content`.

    Args:
        turn: The turn whose input chain already ran.
        reply: The nucleus reply.
        history: Earlier messages including the current user message
            (replaced by the placeholder if the input decision asked for it).

    Returns:
        What the connector must do with the reply.

    Raises:
        RuntimeError: If the input chain did not run, already produced the
            reply, or the turn already has its final reply.
    """
    if turn.finished:
        raise RuntimeError("this turn is already finished")
    if turn.input_decision is None:
        raise RuntimeError("call run_input before run_output")
    if turn.input_decision.kind == "reply":
        raise RuntimeError(
            "the input chain already produced the reply; run_output must not be called"
        )
    if turn.final_action is not None:
        raise RuntimeError("this turn already has its final reply")
    turn.attempt += 1
    attempt = turn.attempt
    turn.annotations = dict(turn.input_annotations)
    # In the state before the chain runs, with the content policy applied: a turn
    # closed while an orbital runs reports the attempt with the results it has.
    record = _Attempt(nucleus_reply=self._content.text(reply))
    turn.output_attempts.append(record)
    decision: OutputDecision | None = None
    with self._tracer.start_as_current_span(
        "chatomon.output",
        attributes={**self._turn_attributes(turn), "chatomon.attempt": attempt},
    ) as span:
        for orb in self._output:
            verdict = await self._run_orbital(
                orb,
                turn,
                side="output",
                message=reply,
                history=history,
                attempt=attempt,
                results=record.results,
            )
            if verdict.action == "allow":
                continue
            decision = self._stop_output(turn, verdict, attempt)
            break
        if decision is None:
            decision = OutputDecision(kind="accept")
            turn.final_reply = reply
            turn.final_action = (
                "guided_refusal" if turn.input_decision.kind == "guided_refusal" else "answered"
            )
        span.set_attribute("chatomon.decision.kind", decision.kind)
    record.decision = decision.kind
    self._log_decision(turn, "output", decision.kind)
    return decision

def reopen_output(self, turn: TurnState) -> None:
    """Allow a new reply to be checked after the turn already had one.

    Connectors call it when the framework asks the model again after chatomon
    accepted or replaced a reply (for example a PydanticAI output validator
    raising `ModelRetry`). The next `run_output` counts as a new attempt.

    Args:
        turn: The turn to reopen.

    Raises:
        RuntimeError: If the turn is finished or its reply came from the input chain.
    """
    if turn.finished:
        raise RuntimeError("this turn is already finished")
    if turn.input_decision is None or turn.input_decision.kind == "reply":
        raise RuntimeError("the input chain produced the reply; there is no output to reopen")
    turn.final_action = None
    turn.final_reply = None

async def finish_turn(self, turn: TurnState) -> TurnReport:
    """Close the turn, build its report and deliver it to `on_turn`.

    An exception raised by `on_turn` is logged and never propagates.

    Args:
        turn: The turn, which must have its final reply.

    Returns:
        The audit report of the turn.

    Raises:
        RuntimeError: If the turn has no final reply yet or was already finished.
    """
    if turn.finished:
        raise RuntimeError("finish_turn was already called for this turn")
    if turn.final_action is None:
        raise RuntimeError("the turn has no final reply yet")
    return await self._close(turn, turn.final_action)

async def abort_turn(self, turn: TurnState, error: BaseException) -> TurnReport:
    """Close a turn that the nucleus or the framework could not complete.

    Connectors call it when the model call or the framework raises, or the run
    is cancelled, so every turn leaves an audit record. The exception type is
    always recorded; its message only with `capture_content="full"`.

    Args:
        turn: The turn to close, at any stage.
        error: The exception that ended the turn.

    Returns:
        The audit report, with `final_action="aborted"`.

    Raises:
        RuntimeError: If the turn was already finished.
    """
    if turn.finished:
        raise RuntimeError("this turn was already finished")
    turn.final_reply = None
    return await self._close(turn, "aborted", error)