`aaa.algos`¶

Tool-using loops¶

ToolLoop ¶

ToolLoop(
    tools: ToolRegistry,
    *,
    system: str = "",
    model: str | None = None,
    max_turns: int = 10,
    params: dict[str, Any] | None = None,
    tool_names: list[str] | None = None,
    name: str | None = None,
    stream: bool = False,
    recall: RecallConfig | bool | None = None,
)

Bases: LLMLoop

Native function-calling loop: model proposes tool calls, results are fed back, until it answers in plain text. Bad/unknown/failed calls come back to the model as errors.

ReAct ¶

ReAct(tools: ToolRegistry, **kw: Any)

Bases: LLMLoop

Classic text-protocol ReAct (Thought/Action/Observation) for models or gateways without native tool calling. Prefer ToolLoop when tool calling is available.

LLMLoop ¶

LLMLoop(
    tools: ToolRegistry,
    *,
    system: str = "",
    model: str | None = None,
    max_turns: int = 10,
    params: dict[str, Any] | None = None,
    tool_names: list[str] | None = None,
    name: str | None = None,
    stream: bool = False,
    recall: RecallConfig | bool | None = None,
)

Shared engine for tool-using LLM loops. Subclasses decide how the model's reply maps to actions (on_reply) and how tool results are shown back (observations).

first_request ¶

first_request(
    st: LoopState, messages: list[Message]
) -> LLMRequest

The run's first model request. If the input says a provider already holds the start of this conversation (input["provider_state"] = {"state_id": ..., "covered": n}, set by sessions), say so.

with_memory ¶

with_memory(
    messages: list[Message], st: LoopState
) -> list[Message]

Insert the recalled memories as a system message after the existing system prompt.

Planning and delegation¶

PlanAndExecute ¶

PlanAndExecute(
    executor: Algorithm,
    *,
    model: str | None = None,
    max_parallel: int = 3,
    max_replans: int = 1,
    synthesize: bool = True,
    planner_system: str = _PLAN_SYSTEM,
)

Plan once (LLM, structured), execute steps as child runs of executor honoring the dependency DAG with bounded parallelism, replan on step failure (bounded), then synthesize. The plan lives on the log as events, so it is inspectable, resumable and replayable.

Plan ¶

Bases: BaseModel

Step ¶

Bases: BaseModel

Router ¶

Router(
    routes: dict[str, Route | Algorithm],
    *,
    classifier: Callable[[dict[str, Any]], str | None]
    | None = None,
    default: str | None = None,
    model: str | None = None,
)

Pick exactly one specialist and delegate (as a child run). Deterministic classifier first (cheap, testable); the LLM only decides what rules don't. Returns the child's outcome.

Route `dataclass` ¶

Route(algo: Algorithm, description: str = '')

Supervisor ¶

Supervisor(
    workers: dict[str, tuple[Algorithm, str]],
    *,
    extra_tools: list[Tool] | None = None,
    **kw: Any,
)

Bases: ToolLoop

Centralized coordinator: an LLM delegates sub-tasks to named worker algorithms (each a child run: own budget accounting, cancelled with the parent). Parallelism is bounded by the 'delegate' pool. Attach ToolHandler(supervisor.tools).

Verification loops¶

GenerateTestRepair ¶

GenerateTestRepair(validator: str = 'default', **kw: Any)

Bases: _Refine

Candidate is checked by a deterministic named validator (tests, linter, schema, rules) served by ValidationHandler. Accepts only what the validator accepts.

GeneratorCritic ¶

GeneratorCritic(
    critic_system: str = _CRITIC_SYSTEM,
    critic_model: str | None = None,
    **kw: Any,
)

Bases: _Refine

Candidate is judged by a separate LLM critic call. Weaker than deterministic checks (the critic is probabilistic); always bounded by max_iterations.

Deterministic control flow¶

WorkflowGraph ¶

WorkflowGraph(
    nodes: list[Node],
    *,
    output: str
    | Callable[[dict[str, Any]], Any]
    | None = None,
)

Deterministic DAG execution: no LLM decides control flow. Ready nodes run in parallel; tools bring approval gates, timeouts, idempotency and budgets. Cyclic/stateful control flow belongs to FSM. Output = output node id, a callable(results), or all results.

Node `dataclass` ¶

Node(
    id: str,
    tool: str,
    after: list[str] = list(),
    args: Callable[
        [dict[str, Any], dict[str, Any]], dict[str, Any]
    ]
    | dict[str, Any]
    | None = None,
    when: Callable[[dict[str, Any], dict[str, Any]], bool]
    | None = None,
    on_error: str = "fail",
)

One workflow step = one tool call. args(results, input) builds its arguments from upstream outputs; when(results, input) can skip it (skipped nodes yield None).

FSM ¶

FSM(
    initial: str,
    transitions: dict[str, dict[str, Transition]],
    terminal: set[str],
    on_enter: dict[
        str, Callable[[Event, dict[str, Any]], list[Event]]
    ]
    | None = None,
)

Explicit finite-state machine driven by external/internal events. transitions[state] [event_type] is the next state or a function(event, input) -> next state | None (guard). on_enter[state](event, input) may return events (request a tool, an LLM call, ...). Reaching a terminal state completes the run with the path taken. The current state is a fold of aaa.fsm.transitioned events, so it survives restarts.

Ensembles and retrieval¶

SelfConsistency ¶

SelfConsistency(
    n: int = 5,
    *,
    temperature: float = 0.8,
    model: str | None = None,
    system: str = "Think step by step, then end with a line 'Answer: <final answer>'.",
    extract: Callable[[str], str] = default_extract,
)

Sample N independent answers in parallel and return the majority (ties -> earliest). Only worthwhile when answers can be compared exactly (math, classification, extraction).

RAGAgent ¶

RAGAgent(index: Index, *, k: int = 5, **kw: Any)

Bases: ToolLoop

Agentic RAG: a ToolLoop whose tool is retrieval over any Index. The model decides when/what to retrieve and is told to ground and cite. Attach ToolHandler(rag.tools).

Authoring helpers¶

base ¶

llm_request ¶

llm_request(
    parent: Event,
    req: LLMRequest,
    tag: str = "main",
    *,
    prev: str | None = None,
) -> Event

Build an aaa.llm.requested event.

With prev (the id of the previous request event of the SAME conversation), req.messages must hold only the messages appended since that request's reply; the event then stores just that delta plus a pointer. The LLM handler reconstructs the full transcript the provider needs, so the log (and algorithm state) grows linearly with turns, not quadratically.

tool_request ¶

tool_request(
    parent: Event,
    call_id: str,
    name: str,
    arguments: dict[str, Any],
) -> Event

finish ¶

finish(
    parent: Event,
    kind: OutcomeKind,
    value: Any = None,
    reason: str | None = None,
) -> Event

parse_json_obj ¶

parse_json_obj(text: str | None) -> dict[str, Any] | None

Best-effort: a JSON object from raw model text (fenced or surrounded by prose).

strip_fence ¶

strip_fence(text: str) -> str

render_result ¶

render_result(payload: Any) -> str

task_messages ¶

task_messages(
    inp: dict[str, Any], system: str
) -> list[Message]

aaa.algos¶