Agents API

agents

BaseAgent

BaseAgent(
    agent_name: str,
    system_prompt: str,
    output_type: type[OutputT],
    *,
    model: Model | str,
    api_key: str | None = None,
    retries: int = 3,
    tools: list[Tool] | None = None,
    toolsets: list[Any] | None = None,
    truncated_history: int = 100,
    history_processors: list | None = None,
    result_budget: int | None = 20000,
    context_manager: ContextManager | None = None,
    model_settings: ModelSettings | None = None,
    reasoning: ReasoningEffort | None = None,
    confidence_protocol: Any = None,
)

Abstract base class for orqest agents.

Generic over StateT (input state, a Pydantic model) and OutputT (output, a Pydantic model). Subclasses must implement _run_implementation().

The model parameter is required so that each agent explicitly declares its provider — two agents in the same process can use different models without conflict.

Initialize the agent.

Parameters:

Name	Type	Description	Default
agent_name	str	Name for logging and identification.	required
system_prompt	str	System prompt guiding agent behavior.	required
output_type	type[OutputT]	Pydantic model class for structured output.	required
model	Model | str	A pydantic-ai Model instance, or a 'provider:model_id' string.	required
api_key	str | None	Required when model is a string; passed to the provider.	None
retries	int	Retry attempts for failed LLM calls.	3
tools	list[Tool] | None	Individual Tool instances to register.	None
toolsets	list[Any] | None	Toolset objects providing collections of tools.	None
truncated_history	int	Max recent messages kept by the default history processor.	100
history_processors	list | None	Custom processors; defaults to keep_recent_messages.	None
result_budget	int | None	Max chars for tool result content before truncation. None disables budgeting. Default 20,000.	20000
context_manager	ContextManager | None	Optional ContextManager for token-aware compaction. When present, its compact method is prepended as the first history processor.	None
model_settings	ModelSettings | None	Optional pydantic-ai ModelSettings applied to every model call (e.g. ModelSettings(temperature=0.0, seed=42)). Default None preserves provider defaults.	None
reasoning	ReasoningEffort | None	Optional provider-agnostic reasoning/thinking effort — one of 'minimal' | 'low' | 'medium' | 'high'. Translated to the right provider-specific ModelSettings key (anthropic_thinking / openai_reasoning_effort / google_thinking_config / openrouter_reasoning) and merged into model_settings — explicit model_settings keys win on conflict. The model's provider must be one orqest can resolve, or construction raises ValueError.	None
confidence_protocol	Any	Optional agent-level default ConfidenceProtocol used by :meth:run_enriched. None means run_enriched returns a bare EnrichedOutput unless a protocol is passed per call.	None

Source code in orqest/agents/base_agent.py

def __init__(
    self,
    agent_name: str,
    system_prompt: str,
    output_type: type[OutputT],
    *,
    model: Model | str,
    api_key: str | None = None,
    retries: int = 3,
    tools: list[Tool] | None = None,
    toolsets: list[Any] | None = None,
    truncated_history: int = 100,
    history_processors: list | None = None,
    result_budget: int | None = 20_000,
    context_manager: ContextManager | None = None,
    model_settings: ModelSettings | None = None,
    reasoning: ReasoningEffort | None = None,
    confidence_protocol: Any = None,
):
    """Initialize the agent.

    Args:
        agent_name: Name for logging and identification.
        system_prompt: System prompt guiding agent behavior.
        output_type: Pydantic model class for structured output.
        model: A pydantic-ai Model instance, or a 'provider:model_id' string.
        api_key: Required when model is a string; passed to the provider.
        retries: Retry attempts for failed LLM calls.
        tools: Individual Tool instances to register.
        toolsets: Toolset objects providing collections of tools.
        truncated_history: Max recent messages kept by the default history processor.
        history_processors: Custom processors; defaults to keep_recent_messages.
        result_budget: Max chars for tool result content before truncation.
            None disables budgeting. Default 20,000.
        context_manager: Optional ContextManager for token-aware compaction.
            When present, its compact method is prepended as the first
            history processor.
        model_settings: Optional pydantic-ai ``ModelSettings`` applied to
            every model call (e.g. ``ModelSettings(temperature=0.0,
            seed=42)``). Default ``None`` preserves provider defaults.
        reasoning: Optional provider-agnostic reasoning/thinking effort —
            one of ``'minimal'`` | ``'low'`` | ``'medium'`` | ``'high'``.
            Translated to the right provider-specific ``ModelSettings``
            key (``anthropic_thinking`` / ``openai_reasoning_effort`` /
            ``google_thinking_config`` / ``openrouter_reasoning``) and
            merged into ``model_settings`` — explicit ``model_settings``
            keys win on conflict. The model's provider must be one
            orqest can resolve, or construction raises ``ValueError``.
        confidence_protocol: Optional agent-level default
            ``ConfidenceProtocol`` used by :meth:`run_enriched`.
            ``None`` means ``run_enriched`` returns a bare
            ``EnrichedOutput`` unless a protocol is passed per call.

    """
    if isinstance(model, str):
        if api_key is None:
            raise ValueError(
                "api_key is required when model is a string name. "
                "Pass api_key explicitly or provide a Model instance instead."
            )
        self._model = resolve_model(model, api_key=api_key)
    elif isinstance(model, Model):
        self._model = model
    else:
        raise TypeError(
            f"model must be a pydantic-ai Model instance or a 'provider:model_id' string, "
            f"got {type(model).__name__}"
        )

    self.reasoning = reasoning
    if reasoning is not None:
        provider = model if isinstance(model, str) else self._model.system
        reasoning_settings = resolve_reasoning_settings(
            provider, reasoning, base=model_settings
        )
        # Reasoning fills in; explicit model_settings keys win on conflict.
        model_settings = {**reasoning_settings, **(model_settings or {})}

    self.agent_name = agent_name
    self.system_prompt = system_prompt
    _validate_output_type_for_llm_structured_output(agent_name, output_type)
    self.output_type = output_type
    self.retries = retries
    self.truncated_history = truncated_history
    self.model_settings = model_settings

    self.tools: list[Tool] = [
        t if isinstance(t, Tool) else Tool(t) for t in (tools or [])
    ]
    self.toolsets = list(toolsets) if toolsets else []

    if history_processors is not None:
        self._history_processors = list(history_processors)
    else:
        self._history_processors = [
            partial(keep_recent_messages, max_messages=truncated_history)
        ]

    if result_budget is not None:
        self._history_processors.insert(
            0, partial(budget_tool_results, max_result_chars=result_budget)
        )

    if context_manager is not None:
        self._history_processors.insert(0, context_manager.compact)

    # ALWAYS run an orphan-tool-return sweep last. Any earlier processor
    # (custom or built-in) can split a tool_call/tool_return pair across
    # the trim boundary; both the OpenAI chat/completions transport AND
    # the Responses transport reject orphan tool returns with a 400.
    # See :func:`orqest.agents.context_manager._repair_orphan_tool_returns`.
    from orqest.agents.context_manager import _repair_orphan_tool_returns
    self._history_processors.append(_repair_orphan_tool_returns)

    self._confidence_protocol = confidence_protocol
    self._agent: Agent | None = None

model property

model: Model

The resolved pydantic-ai Model instance.

confidence_protocol property

confidence_protocol: Any

The agent-level default ConfidenceProtocol, or None.

Set via the constructor's confidence_protocol keyword; used by :meth:run_enriched and consulted by callers (e.g. :class:~orqest.orchestration.loop.RefinementLoop) that need the agent to be confidence-aware.

agent property

agent: Agent

Lazily constructed pydantic-ai Agent.

add_tool

add_tool(tool: Tool) -> None

Add a :class:pydantic_ai.Tool at runtime; invalidate the cache.

Appends tool to self.tools and resets the cached internal pydantic_ai.Agent so the next call to :attr:agent rebuilds with the new tool list.

Idempotent for tools sharing a name: an existing entry is replaced (last write wins). Useful when an LLM-spawned tool needs to update its implementation.

In-flight runs do NOT see the new tool — the rebuild happens on the next :attr:agent access; any Agent.run() call already in flight uses the tool list it captured at start time.

Source code in orqest/agents/base_agent.py

def add_tool(self, tool: Tool) -> None:
    """Add a :class:`pydantic_ai.Tool` at runtime; invalidate the cache.

    Appends *tool* to ``self.tools`` and resets the cached internal
    ``pydantic_ai.Agent`` so the next call to :attr:`agent` rebuilds
    with the new tool list.

    Idempotent for tools sharing a ``name``: an existing entry is
    replaced (last write wins). Useful when an LLM-spawned tool needs
    to update its implementation.

    **In-flight runs do NOT see the new tool** — the rebuild happens
    on the next :attr:`agent` access; any ``Agent.run()`` call already
    in flight uses the tool list it captured at start time.
    """
    if not isinstance(tool, Tool):
        raise TypeError(
            f"add_tool expects a pydantic_ai.Tool instance, got {type(tool).__name__}"
        )
    # Replace existing entry with same name (idempotent / last-write-wins)
    self.tools = [t for t in self.tools if t.name != tool.name]
    self.tools.append(tool)
    # Critical: invalidate the cached pydantic_ai.Agent so the next
    # access rebuilds with the updated tool list.
    self._agent = None

call_model async

call_model(prompt: Prompt, state: StateT) -> AgentRunResult

Run the pydantic-ai agent with conversation history from state.

Passes state.message_history into Agent.run() and stores the updated history back on state after the run. This is the recommended way to call the LLM from _run_implementation() when you want multi-turn conversation support.

For stateless one-shot calls, use self.agent.run() directly instead.

Source code in orqest/agents/base_agent.py

async def call_model(self, prompt: Prompt, state: StateT) -> AgentRunResult:
    """Run the pydantic-ai agent with conversation history from state.

    Passes state.message_history into Agent.run() and stores the updated
    history back on state after the run. This is the recommended way to
    call the LLM from _run_implementation() when you want multi-turn
    conversation support.

    For stateless one-shot calls, use self.agent.run() directly instead.
    """
    history = getattr(state, "message_history", None) or None
    result = await self.agent.run(prompt, message_history=history)
    if hasattr(state, "message_history"):
        state.message_history = result.all_messages()
    return result

call_model_stream async

call_model_stream(
    prompt: Prompt, state: StateT
) -> AsyncIterator[StreamedRunResult]

Stream the pydantic-ai agent with conversation history from state.

Async context manager that wraps Agent.run_stream(). Yields a StreamedRunResult for full control over the stream. Updates state.message_history after the context exits (once the stream is consumed).

This is the low-level streaming primitive — stream_text() and stream_output() are built on top of it.

Source code in orqest/agents/base_agent.py

@asynccontextmanager
async def call_model_stream(
    self, prompt: Prompt, state: StateT
) -> AsyncIterator[StreamedRunResult]:
    """Stream the pydantic-ai agent with conversation history from state.

    Async context manager that wraps Agent.run_stream(). Yields a
    StreamedRunResult for full control over the stream. Updates
    state.message_history after the context exits (once the stream
    is consumed).

    This is the low-level streaming primitive — stream_text() and
    stream_output() are built on top of it.
    """
    history = getattr(state, "message_history", None) or None
    async with self.agent.run_stream(prompt, message_history=history) as streamed:
        yield streamed
    if hasattr(state, "message_history"):
        state.message_history = streamed.all_messages()

stream_output async

stream_output(
    prompt: Prompt,
    state: StateT,
    *,
    debounce_by: float | None = None,
) -> AsyncIterator[OutputT]

Stream partial structured output from the LLM.

Async generator yielding partial OutputT Pydantic model instances as the model produces them. Each yield is the latest validated partial of the structured output.

Parameters:

Name	Type	Description	Default
prompt	Prompt	The user prompt to send.	required
state	StateT	Conversation state — history is read and updated.	required
debounce_by	float | None	Optional minimum interval (seconds) between yields.	None

Source code in orqest/agents/base_agent.py

async def stream_output(
    self, prompt: Prompt, state: StateT, *, debounce_by: float | None = None
) -> AsyncIterator[OutputT]:
    """Stream partial structured output from the LLM.

    Async generator yielding partial OutputT Pydantic model instances
    as the model produces them. Each yield is the latest validated
    partial of the structured output.

    Args:
        prompt: The user prompt to send.
        state: Conversation state — history is read and updated.
        debounce_by: Optional minimum interval (seconds) between yields.

    """
    kwargs: dict[str, Any] = {}
    if debounce_by is not None:
        kwargs["debounce_by"] = debounce_by
    async with self.call_model_stream(prompt, state) as streamed:
        async for partial in streamed.stream_output(**kwargs):
            yield partial

stream_events async

stream_events(
    prompt: Prompt, state: StateT
) -> AsyncIterator[AgentStreamEvent]

Stream all agent events including model responses and tool calls.

Async generator yielding AgentStreamEvent instances as the agent runs. This includes model response tokens (PartStartEvent, PartDeltaEvent, PartEndEvent, FinalResultEvent) and tool execution events (FunctionToolCallEvent, FunctionToolResultEvent).

Uses Agent.iter() under the hood for full node-by-node control, making tool calls visible during streaming.

Parameters:

Name	Type	Description	Default
prompt	Prompt	The user prompt to send.	required
state	StateT	Conversation state — history is read and updated.	required

Source code in orqest/agents/base_agent.py

async def stream_events(
    self, prompt: Prompt, state: StateT
) -> AsyncIterator[AgentStreamEvent]:
    """Stream all agent events including model responses and tool calls.

    Async generator yielding AgentStreamEvent instances as the agent runs.
    This includes model response tokens (PartStartEvent, PartDeltaEvent,
    PartEndEvent, FinalResultEvent) and tool execution events
    (FunctionToolCallEvent, FunctionToolResultEvent).

    Uses Agent.iter() under the hood for full node-by-node control,
    making tool calls visible during streaming.

    Args:
        prompt: The user prompt to send.
        state: Conversation state — history is read and updated.

    """
    history = getattr(state, "message_history", None) or None
    async with self.agent.iter(prompt, message_history=history) as agent_run:
        async for node in agent_run:
            if Agent.is_model_request_node(node):
                async with node.stream(agent_run.ctx) as stream:
                    async for event in stream:
                        yield event
            elif Agent.is_call_tools_node(node):
                async with node.stream(agent_run.ctx) as stream:
                    async for event in stream:
                        yield event
    if hasattr(state, "message_history"):
        state.message_history = agent_run.result.all_messages()

run async

run(state: StateT, **kwargs: Any) -> OutputT

Execute the agent. Exceptions propagate to the caller.

Source code in orqest/agents/base_agent.py

async def run(self, state: StateT, **kwargs: Any) -> OutputT:
    """Execute the agent. Exceptions propagate to the caller."""
    return await self._run_implementation(state, **kwargs)

run_enriched async

run_enriched(
    state: StateT,
    *,
    confidence_protocol: Any = None,
    **kwargs: Any,
) -> Any

Execute the agent and pair its output with self-assessment.

Returns :class:~orqest.metacognition.EnrichedOutput[OutputT]. Best-effort: a protocol that fails surfaces confidence=None and a protocol_error in the metadata; the underlying output is always returned. Exceptions raised by _run_implementation propagate, exactly like :meth:run.

Parameters:

Name	Type	Description	Default
state	StateT	The agent's state, same as run.	required
confidence_protocol	Any	Per-call override of the agent-level default (set via constructor). When neither is provided, returns EnrichedOutput(output=...) with all metacognitive fields at their defaults.	None
**kwargs	Any	Forwarded to _run_implementation and to the protocol's enrich call.	{}

Source code in orqest/agents/base_agent.py

async def run_enriched(
    self,
    state: StateT,
    *,
    confidence_protocol: Any = None,
    **kwargs: Any,
) -> Any:
    """Execute the agent and pair its output with self-assessment.

    Returns :class:`~orqest.metacognition.EnrichedOutput[OutputT]`.
    Best-effort: a protocol that fails surfaces ``confidence=None``
    and a ``protocol_error`` in the metadata; the underlying output
    is always returned. Exceptions raised by ``_run_implementation``
    propagate, exactly like :meth:`run`.

    Args:
        state: The agent's state, same as ``run``.
        confidence_protocol: Per-call override of the agent-level
            default (set via constructor). When neither is provided,
            returns ``EnrichedOutput(output=...)`` with all
            metacognitive fields at their defaults.
        **kwargs: Forwarded to ``_run_implementation`` and to the
            protocol's ``enrich`` call.

    """
    from orqest.metacognition.enriched import EnrichedOutput

    output = await self._run_implementation(state, **kwargs)
    protocol = confidence_protocol or self._confidence_protocol
    if protocol is None:
        return EnrichedOutput(output=output)
    try:
        return await protocol.enrich(self, state, output, **kwargs)
    except Exception as exc:
        from loguru import logger as _logger

        _logger.debug(
            "ConfidenceProtocol {p} failed: {e}",
            p=getattr(protocol, "name", type(protocol).__name__),
            e=str(exc),
        )
        return EnrichedOutput(
            output=output,
            metadata={"protocol_error": type(exc).__name__},
        )

CompoundTool

CompoundTool(
    agent: BaseAgent,
    executor: Callable[[OutputT, StateT], Awaitable[Any]],
    *,
    state_updater: Callable[[StateT, Any], StateT]
    | None = None,
    hooks: HookRunner | None = None,
    name: str | None = None,
)

Combines an agent call with a tool execution and state update.

Pattern: agent produces structured output, executor runs a tool/action using that output, state is updated with the result. Hooks fire before/after the execution step and can influence flow via :class:HookDecision.

Initialize the compound tool.

Parameters:

Name	Type	Description	Default
agent	BaseAgent	The agent that produces structured output.	required
executor	Callable[[OutputT, StateT], Awaitable[Any]]	Async callable that acts on (agent_output, state).	required
state_updater	Callable[[StateT, Any], StateT] | None	Optional callable to update state with the result.	None
hooks	HookRunner | None	HookRunner for before/after/error callbacks.	None
name	str | None	Tool name for hook dispatch. Defaults to agent.agent_name.	None

Source code in orqest/agents/compound_tool.py

def __init__(
    self,
    agent: BaseAgent,
    executor: Callable[[OutputT, StateT], Awaitable[Any]],
    *,
    state_updater: Callable[[StateT, Any], StateT] | None = None,
    hooks: HookRunner | None = None,
    name: str | None = None,
) -> None:
    """Initialize the compound tool.

    Args:
        agent: The agent that produces structured output.
        executor: Async callable that acts on (agent_output, state).
        state_updater: Optional callable to update state with the result.
        hooks: HookRunner for before/after/error callbacks.
        name: Tool name for hook dispatch. Defaults to agent.agent_name.

    """
    self.name = name or agent.agent_name
    self._agent = agent
    self._executor = executor
    self._state_updater = state_updater
    self._hooks = hooks or HookRunner()

run async

run(
    state: StateT, prompt: str, **kwargs: Any
) -> tuple[OutputT, Any]

Execute the compound tool: agent, decide, execute, update state.

prompt is injected into state as a user message before the agent runs (when state supports add_message), so the wrapped agent receives it through the same channel as as_tool / AgentStep rather than being silently dropped.

Returns (agent_output, execution_result). Honors :class:Skip (returns stub_result in place of executor), :class:Redirect (mutates args/name; bounded one re-execution on after_tool redirect), and :class:Abort (raises :class:HookAbortError to the caller).

Source code in orqest/agents/compound_tool.py

async def run(
    self, state: StateT, prompt: str, **kwargs: Any
) -> tuple[OutputT, Any]:
    """Execute the compound tool: agent, decide, execute, update state.

    ``prompt`` is injected into ``state`` as a user message before the
    agent runs (when ``state`` supports ``add_message``), so the wrapped
    agent receives it through the same channel as ``as_tool`` /
    ``AgentStep`` rather than being silently dropped.

    Returns ``(agent_output, execution_result)``. Honors
    :class:`Skip` (returns ``stub_result`` in place of executor),
    :class:`Redirect` (mutates args/name; bounded one re-execution
    on ``after_tool`` redirect), and :class:`Abort` (raises
    :class:`HookAbortError` to the caller).
    """
    if hasattr(state, "add_message"):
        state.add_message("user", prompt)
    agent_output = await self._agent.run(state, **kwargs)

    args = {"agent_output": agent_output, "prompt": prompt}
    effective_name = self.name
    effective_args: dict[str, Any] = args

    before_decision = await self._hooks.run_before(
        effective_name, effective_args, state
    )

    if isinstance(before_decision, Skip):
        result = before_decision.stub_result
        await self._hooks.run_after(
            effective_name, effective_args, result, state, 0.0
        )
        if self._state_updater is not None:
            state = self._state_updater(state, result)
        return agent_output, result

    if isinstance(before_decision, Redirect):
        if before_decision.new_args is not None:
            effective_args = {**effective_args, **before_decision.new_args}
        if before_decision.new_tool is not None:
            effective_name = before_decision.new_tool

    start = time.monotonic()
    try:
        result = await self._executor(agent_output, state)
        duration_ms = (time.monotonic() - start) * 1000
    except HookAbortError:
        raise
    except Exception as exc:
        error_decision = await self._hooks.run_error(
            effective_name, effective_args, exc, state
        )
        if not isinstance(error_decision, Redirect):
            raise
        # on_error Redirect → bounded single retry of the executor
        # (e.g. DiscoveryHook registered the missing tool). A second
        # failure is not retried — it propagates.
        if error_decision.new_args is not None:
            effective_args = {**effective_args, **error_decision.new_args}
        if error_decision.new_tool is not None:
            effective_name = error_decision.new_tool
        start = time.monotonic()
        result = await self._executor(agent_output, state)
        duration_ms = (time.monotonic() - start) * 1000

    after_decision = await self._hooks.run_after(
        effective_name, effective_args, result, state, duration_ms
    )

    # Bounded one re-execution on after_tool redirect.
    if isinstance(after_decision, Redirect):
        if after_decision.new_args is not None:
            effective_args = {**effective_args, **after_decision.new_args}
        start = time.monotonic()
        try:
            result = await self._executor(agent_output, state)
            duration_ms = (time.monotonic() - start) * 1000
            # Final after_tool — any further Redirect from this call is
            # ignored (logged inside HookRunner) to bound the loop.
            await self._hooks.run_after(
                effective_name, effective_args, result, state, duration_ms
            )
        except HookAbortError:
            raise
        except Exception as exc:
            await self._hooks.run_error(
                effective_name, effective_args, exc, state
            )
            raise

    if self._state_updater is not None:
        state = self._state_updater(state, result)

    return agent_output, result

ContextManager

ContextManager(
    token_budget: int = 128000,
    reserve: int = 20000,
    summarize_threshold: float = 0.6,
    truncate_threshold: float = 0.85,
    min_recent_turns: int = 5,
    min_recent_tokens: int = 10000,
    *,
    salience_fn: Callable[[Any], float] | None = None,
)

Progressive context compaction based on token budget.

Three layers of compaction, activated by token usage thresholds: 1. Tool result snipping (handled by budget_tool_results, separate processor) 2. Turn summarization — at summarize_threshold, compress old tool-call turns 3. Emergency truncation — at truncate_threshold, aggressive message dropping

Source code in orqest/agents/context_manager.py

def __init__(
    self,
    token_budget: int = 128_000,
    reserve: int = 20_000,
    summarize_threshold: float = 0.60,
    truncate_threshold: float = 0.85,
    min_recent_turns: int = 5,
    min_recent_tokens: int = 10_000,
    *,
    salience_fn: Callable[[Any], float] | None = None,
):
    self.effective_budget = token_budget - reserve
    self.summarize_threshold = summarize_threshold
    self.truncate_threshold = truncate_threshold
    self.min_recent_turns = min_recent_turns
    self.min_recent_tokens = min_recent_tokens
    self._salience_fn = salience_fn

compact

compact(messages: list[ModelMessage]) -> list[ModelMessage]

Apply progressive compaction based on token usage.

Returns a new list — never mutates the input. Whatever path runs, the result is post-processed by :func:_repair_orphan_tool_returns so a ToolReturnPart never appears without its matching ToolCallPart earlier in the same window. The Responses API rejects such orphan returns with "No tool call found for function call output with call_id ..." (observed 2026-05-16 mid-Koopman-run), and chat/completions rejects them with "messages with role 'tool' must be a response to a preceding message with 'tool_calls'".

Source code in orqest/agents/context_manager.py

def compact(self, messages: list[ModelMessage]) -> list[ModelMessage]:
    """Apply progressive compaction based on token usage.

    Returns a new list — never mutates the input. Whatever path runs,
    the result is post-processed by :func:`_repair_orphan_tool_returns`
    so a ``ToolReturnPart`` never appears without its matching
    ``ToolCallPart`` earlier in the same window. The Responses API
    rejects such orphan returns with
    ``"No tool call found for function call output with call_id ..."``
    (observed 2026-05-16 mid-Koopman-run), and chat/completions
    rejects them with ``"messages with role 'tool' must be a response
    to a preceding message with 'tool_calls'"``.
    """
    if not messages:
        return list(messages)

    tokens = estimate_tokens(messages)

    if tokens > self.effective_budget * self.truncate_threshold:
        compacted = self._emergency_truncate(messages)
    elif tokens > self.effective_budget * self.summarize_threshold:
        compacted = self._summarize_old_turns(messages)
    else:
        return list(messages)
    return _repair_orphan_tool_returns(compacted)

BaseSessionState

GlobalState extended with session tracking and native serialization.

Subclass for domain-specific state. model_dump() and model_validate() handle message_history correctly thanks to the :data:SerializableMessageHistory annotation.

serialize

serialize() -> dict[str, Any]

Return a JSON-safe dict representation of this state.

Thin wrapper over model_dump(mode="json"); kept for callers that prefer method-style access.

Source code in orqest/agents/session_state.py

def serialize(self) -> dict[str, Any]:
    """Return a JSON-safe dict representation of this state.

    Thin wrapper over ``model_dump(mode="json")``; kept for callers
    that prefer method-style access.
    """
    return self.model_dump(mode="json")

deserialize classmethod

deserialize(data: dict[str, Any]) -> Self

Reconstruct a state from a serialized dict.

Thin wrapper over model_validate; corrupt message_history is handled inside the :data:SerializableMessageHistory validator.

Source code in orqest/agents/session_state.py

@classmethod
def deserialize(cls, data: dict[str, Any]) -> Self:
    """Reconstruct a state from a serialized dict.

    Thin wrapper over ``model_validate``; corrupt ``message_history``
    is handled inside the :data:`SerializableMessageHistory` validator.
    """
    return cls.model_validate(data)

GlobalState

Shared conversation state for orqest agents.

add_message

add_message(
    role: str, content: str | Sequence[UserContent]
) -> None

Append a role/content pair to the conversation log.

Source code in orqest/agents/state.py

def add_message(self, role: str, content: str | Sequence[UserContent]) -> None:
    """Append a role/content pair to the conversation log."""
    self.messages.append({"role": role, "content": content})

get_latest_message

get_latest_message(
    role: str,
) -> str | Sequence[UserContent] | None

Return the content of the most recent message with the given role.

Source code in orqest/agents/state.py

def get_latest_message(self, role: str) -> str | Sequence[UserContent] | None:
    """Return the content of the most recent message with the given role."""
    for message in reversed(self.messages):
        if message.get("role") == role:
            return message.get("content")
    return None

budget_tool_results

budget_tool_results(
    messages: list[ModelMessage],
    *,
    max_result_chars: int = 20000,
    preview_chars: int = 2000,
) -> list[ModelMessage]

Truncate oversized ToolReturnPart content in message history.

Walks all messages, finds ToolReturnPart instances with content exceeding max_result_chars, and replaces with a preview prefix plus truncation notice. Returns a new list — never mutates input.

Source code in orqest/agents/base_agent.py

def budget_tool_results(
    messages: list[ModelMessage],
    *,
    max_result_chars: int = 20_000,
    preview_chars: int = 2_000,
) -> list[ModelMessage]:
    """Truncate oversized ToolReturnPart content in message history.

    Walks all messages, finds ToolReturnPart instances with content
    exceeding max_result_chars, and replaces with a preview prefix
    plus truncation notice. Returns a new list — never mutates input.
    """
    result: list[ModelMessage] = []
    for msg in messages:
        if isinstance(msg, ModelRequest):
            new_parts = []
            modified = False
            for part in msg.parts:
                if isinstance(part, ToolReturnPart):
                    content_str = str(part.content)
                    if len(content_str) > max_result_chars:
                        truncated = (
                            content_str[:preview_chars]
                            + f"\n\n[TRUNCATED — {len(content_str)} chars total. "
                            f"Full result was too large for context window.]"
                        )
                        new_parts.append(dc_replace(part, content=truncated))
                        modified = True
                    else:
                        new_parts.append(part)
                else:
                    new_parts.append(part)
            if modified:
                result.append(dc_replace(msg, parts=new_parts))
            else:
                result.append(msg)
        else:
            result.append(msg)
    return result

keep_recent_messages

keep_recent_messages(
    messages: list[ModelMessage], *, max_messages: int = 100
) -> list[ModelMessage]

Truncate message history while preserving the first message and turn integrity.

Returns a new list — never mutates the input.

The first message is always preserved because it typically contains the initial user prompt that establishes context. When truncation would split a request/response pair (tool call followed by tool return), the preceding response is included to maintain a valid message sequence.

Source code in orqest/agents/base_agent.py

def keep_recent_messages(
    messages: list[ModelMessage],
    *,
    max_messages: int = 100,
) -> list[ModelMessage]:
    """Truncate message history while preserving the first message and turn integrity.

    Returns a new list — never mutates the input.

    The first message is always preserved because it typically contains the initial
    user prompt that establishes context. When truncation would split a
    request/response pair (tool call followed by tool return), the preceding
    response is included to maintain a valid message sequence.
    """
    if not messages or max_messages <= 0:
        return list(messages)

    n = max(1, max_messages)
    if len(messages) <= n:
        return list(messages)

    truncated = list(messages[-n:])

    # If truncation split a request/response pair, include the preceding response.
    # A ModelRequest right at the boundary that follows a ModelResponse means we
    # cut in the middle of a tool-call turn.
    boundary_idx = len(messages) - n
    first = truncated[0]
    if isinstance(first, ModelRequest) and boundary_idx > 0:
        preceding = messages[boundary_idx - 1]
        if isinstance(preceding, ModelResponse):
            truncated = [preceding] + truncated

    # Always preserve the first message in the full history.
    if truncated[0] is not messages[0]:
        return [messages[0]] + truncated

    return truncated

run_with_retry async

run_with_retry(
    operation: Callable[[str], Awaitable[str]],
    *,
    tool_name: str,
    args: dict[str, Any],
    state: Any,
    hooks: HookRunner,
    note: str,
    max_attempts: int = 2,
    enrich_note: Callable[[str, str], str] | None = None,
    is_retryable: Callable[[Exception], bool] | None = None,
) -> str

Run a compound tool operation with retry, enrichment, and hook dispatch.

Fires hooks.run_before once at the start and hooks.run_after once with the final result — whether success or exhausted failure. The caller does not need to fire hooks itself.

Honors :class:HookDecision directives: - :class:Skip from before_tool → short-circuit; return stub_result (JSON-serialised if non-string). - :class:Redirect(new_args=...) from before_tool → if new_args["note"] is set, override the note for this run. - :class:Abort → :class:HookAbortError propagates.

Parameters:

Name	Type	Description	Default
operation	Callable[[str], Awaitable[str]]	Async callable that takes a (possibly enriched) note and returns a serialized result string. Raises on failure.	required
tool_name	str	Name passed to hooks.run_before / run_after for dispatch.	required
args	dict[str, Any]	Arguments dict passed to hooks for observability.	required
state	Any	Opaque state object passed to hooks (e.g. SessionState).	required
hooks	HookRunner	HookRunner that dispatches lifecycle events.	required
note	str	Original note/prompt passed to operation on the first attempt.	required
max_attempts	int	Total number of attempts including the first.	2
enrich_note	Callable[[str, str], str] | None	(original_note, last_error_str) -> new_note builder invoked before every retry. Defaults to _default_enrich.	None
is_retryable	Callable[[Exception], bool] | None	(exception) -> bool predicate. If False, abort retrying and emit a failure payload. Default: always retry.	None

Returns:

Type	Description
str	The operation's successful result string on any attempt, or a
str	JSON-serialized failure payload ({success, error, attempts}) when
str	attempts are exhausted or a non-retryable error occurs.

Source code in orqest/agents/retry.py

async def run_with_retry(
    operation: Callable[[str], Awaitable[str]],
    *,
    tool_name: str,
    args: dict[str, Any],
    state: Any,
    hooks: HookRunner,
    note: str,
    max_attempts: int = 2,
    enrich_note: Callable[[str, str], str] | None = None,
    is_retryable: Callable[[Exception], bool] | None = None,
) -> str:
    """Run a compound tool operation with retry, enrichment, and hook dispatch.

    Fires ``hooks.run_before`` once at the start and ``hooks.run_after`` once
    with the final result — whether success or exhausted failure. The caller
    does not need to fire hooks itself.

    Honors :class:`HookDecision` directives:
      - :class:`Skip` from ``before_tool`` → short-circuit; return
        ``stub_result`` (JSON-serialised if non-string).
      - :class:`Redirect(new_args=...)` from ``before_tool`` → if
        ``new_args["note"]`` is set, override the note for this run.
      - :class:`Abort` → :class:`HookAbortError` propagates.

    Args:
        operation: Async callable that takes a (possibly enriched) note and
            returns a serialized result string. Raises on failure.
        tool_name: Name passed to hooks.run_before / run_after for dispatch.
        args: Arguments dict passed to hooks for observability.
        state: Opaque state object passed to hooks (e.g. SessionState).
        hooks: HookRunner that dispatches lifecycle events.
        note: Original note/prompt passed to operation on the first attempt.
        max_attempts: Total number of attempts including the first.
        enrich_note: ``(original_note, last_error_str) -> new_note`` builder
            invoked before every retry. Defaults to ``_default_enrich``.
        is_retryable: ``(exception) -> bool`` predicate. If False, abort
            retrying and emit a failure payload. Default: always retry.

    Returns:
        The operation's successful result string on any attempt, or a
        JSON-serialized failure payload (``{success, error, attempts}``) when
        attempts are exhausted or a non-retryable error occurs.
    """
    before_decision = await hooks.run_before(tool_name, args, state)

    # Skip short-circuits the whole operation.
    if isinstance(before_decision, Skip):
        stub = before_decision.stub_result
        result_str = stub if isinstance(stub, str) else json.dumps(stub)
        await hooks.run_after(tool_name, args, result_str, state, 0.0)
        return result_str

    # Redirect can override the note for this run via new_args["note"].
    if isinstance(before_decision, Redirect):
        if before_decision.new_args and "note" in before_decision.new_args:
            note = str(before_decision.new_args["note"])

    start = time.monotonic()
    enrich = enrich_note or _default_enrich
    last_error: str | None = None
    attempt = 0

    for attempt in range(max_attempts):
        current_note = note if last_error is None else enrich(note, last_error)
        try:
            result_str = await operation(current_note)
        except Exception as exc:
            last_error = str(exc)
            if is_retryable is not None and not is_retryable(exc):
                logger.warning(
                    "{tool} attempt {n} failed with non-retryable error: {err}",
                    tool=tool_name,
                    n=attempt + 1,
                    err=exc,
                )
                break
            if attempt < max_attempts - 1:
                logger.warning(
                    "{tool} attempt {n} failed, retrying: {err}",
                    tool=tool_name,
                    n=attempt + 1,
                    err=exc,
                )
            continue

        duration_ms = (time.monotonic() - start) * 1000
        await hooks.run_after(tool_name, args, result_str, state, duration_ms)
        return result_str

    result_str = json.dumps({
        "success": False,
        "error": last_error,
        "attempts": attempt + 1,
    })
    duration_ms = (time.monotonic() - start) * 1000
    await hooks.run_after(tool_name, args, result_str, state, duration_ms)
    return result_str

as_tool

as_tool(
    agent: BaseAgent,
    *,
    name: str | None = None,
    description: str,
) -> Tool

Wrap a BaseAgent as a pydantic-ai Tool.

The orchestrating agent's LLM sees a tool with a single query parameter. When invoked, a fresh GlobalState is created, the query is added as a user message, and the wrapped agent runs statelessly. The output is returned as a JSON string.

Parameters:

Name	Type	Description	Default
agent	BaseAgent	The BaseAgent instance to wrap.	required
name	str | None	Tool name visible to the LLM. Defaults to agent.agent_name.	None
description	str	Tool description visible to the LLM. Should clearly explain what the agent does and when to use it.	required

Source code in orqest/agents/tool_wrapper.py

def as_tool(
    agent: BaseAgent,
    *,
    name: str | None = None,
    description: str,
) -> Tool:
    """Wrap a BaseAgent as a pydantic-ai Tool.

    The orchestrating agent's LLM sees a tool with a single `query` parameter.
    When invoked, a fresh GlobalState is created, the query is added as a user
    message, and the wrapped agent runs statelessly. The output is returned as
    a JSON string.

    Args:
        agent: The BaseAgent instance to wrap.
        name: Tool name visible to the LLM. Defaults to agent.agent_name.
        description: Tool description visible to the LLM. Should clearly explain
            what the agent does and when to use it.
    """
    tool_name = name or agent.agent_name

    async def _run(query: str) -> str:
        """Execute the wrapped agent with a fresh state.

        Args:
            query: The input to send to the agent.
        """
        state = GlobalState()
        state.add_message("user", query)
        output = await agent.run(state)
        return output.model_dump_json()

    _run.__name__ = tool_name
    _run.__qualname__ = tool_name

    return Tool(_run, name=tool_name, description=description)