Compound Tools API¶

compound ¶

Compound-tool primitives — bind sub-agents to executors + evaluators.

SubAgentResult ¶

Outcome of a :meth:SubAgentTool.run invocation.

Captures the final result plus a short history of refinement attempts so consumers (metrics, trace spans, LLM tool returns) can surface "how many passes it took" without re-instrumenting.

SubAgentTool ¶

SubAgentTool(
    agent: BaseAgent,
    executor: Callable[[Any, StateT], Awaitable[ResultT]],
    state_updater: Callable[[ResultT, StateT], None],
    *,
    evaluator: Callable[[ResultT], SubAgentEvalResult]
    | None = None,
    max_refinements: int = 0,
    build_refinement_prompt: Callable[[ResultT, str], str]
    | None = None,
    name: str | None = None,
)

Compose a stateless sub-agent with an executor and optional refinement.

Usage — implement three domain functions and hand them to the tool:

executor(agent_output, state) -> Awaitable[ResultT] — turns the sub-agent's structured output into a real-world outcome (e.g. run the MCP pipeline and return a payload).
state_updater(result, state) -> None — mutates the long- lived session state with whatever needs to persist.
evaluator(result) -> SubAgentEvalResult (optional) — decides whether the result is good enough or a refinement cycle should fire.

If max_refinements > 0 and the evaluator rejects the first result, the tool calls build_refinement_prompt(result, prompt) to construct a new prompt and reruns the sub-agent + executor cycle. Refinement exceptions are caught and the best prior result is kept — matching numatics-ai v2's "best effort refinement" rule.

The tool does NOT wrap retry-on-exception; for that, wrap the run call in :func:orqest.agents.retry.run_with_retry.

Configure the compound flow.

Parameters:

Name	Type	Description	Default
`agent`	`BaseAgent`	A stateless :class:`~orqest.agents.base_agent.BaseAgent` whose `.run()` takes `(state, **kwargs)` and returns a structured output (typically a Pydantic model).	required
`executor`	`Callable[[Any, StateT], Awaitable[ResultT]]`	Async function that runs the real-world action using the sub-agent's output and returns a serializable result payload.	required
`state_updater`	`Callable[[ResultT, StateT], None]`	Synchronous mutator that writes the result back onto the state object.	required
`evaluator`	`Callable[[ResultT], SubAgentEvalResult] \| None`	Optional quality check. When provided with `max_refinements > 0`, a failing evaluation triggers a refinement cycle.	`None`
`max_refinements`	`int`	Number of refinement cycles allowed after the first pass. `0` disables refinement.	`0`
`build_refinement_prompt`	`Callable[[ResultT, str], str] \| None`	`(result, original_prompt) -> new_prompt`. Required when `max_refinements > 0`.	`None`
`name`	`str \| None`	Optional tool name used in :class:`SubAgentResult` and error messages. Defaults to `agent.agent_name`.	`None`

Raises:

Type	Description
`ValueError`	If `max_refinements > 0` but `evaluator` or `build_refinement_prompt` is `None`.

Source code in orqest/compound/sub_agent_tool.py

def __init__(
    self,
    agent: BaseAgent,
    executor: Callable[[Any, StateT], Awaitable[ResultT]],
    state_updater: Callable[[ResultT, StateT], None],
    *,
    evaluator: Callable[[ResultT], SubAgentEvalResult] | None = None,
    max_refinements: int = 0,
    build_refinement_prompt: Callable[[ResultT, str], str] | None = None,
    name: str | None = None,
) -> None:
    """Configure the compound flow.

    Args:
        agent: A stateless :class:`~orqest.agents.base_agent.BaseAgent`
            whose ``.run()`` takes ``(state, **kwargs)`` and returns
            a structured output (typically a Pydantic model).
        executor: Async function that runs the real-world action
            using the sub-agent's output and returns a
            serializable result payload.
        state_updater: Synchronous mutator that writes the result
            back onto the state object.
        evaluator: Optional quality check. When provided with
            ``max_refinements > 0``, a failing evaluation triggers
            a refinement cycle.
        max_refinements: Number of refinement cycles allowed after
            the first pass. ``0`` disables refinement.
        build_refinement_prompt: ``(result, original_prompt) ->
            new_prompt``. Required when ``max_refinements > 0``.
        name: Optional tool name used in :class:`SubAgentResult`
            and error messages. Defaults to ``agent.agent_name``.

    Raises:
        ValueError: If ``max_refinements > 0`` but ``evaluator`` or
            ``build_refinement_prompt`` is ``None``.
    """
    if max_refinements > 0 and (
        evaluator is None or build_refinement_prompt is None
    ):
        raise ValueError(
            "SubAgentTool with max_refinements > 0 requires both "
            "evaluator and build_refinement_prompt."
        )

    self._agent = agent
    self._executor = executor
    self._state_updater = state_updater
    self._evaluator = evaluator
    self._max_refinements = max_refinements
    self._build_refinement_prompt = build_refinement_prompt
    self.name = name or getattr(agent, "agent_name", "sub_agent_tool")

run `async` ¶

run(
    state: StateT,
    prompt: str,
    *,
    use_enriched: bool = False,
    **agent_kwargs: Any,
) -> SubAgentResult[ResultT]

Execute one (agent + executor + state-update) cycle, then optionally refine up to max_refinements times.

The first result is always committed to state before refinement begins — so a failed refinement leaves the original result in place (best-effort semantics). Extra keyword arguments are forwarded verbatim to agent.run.