`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).

request_for ¶

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

The request for the next model call given the loop state (a hook for loops whose tools or model depend on state, like Swarm). Ordinary loops ignore the state.

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]

Reasoning algorithms¶

Reflexion ¶

Reflexion(
    *,
    validator: str | None = None,
    system: str = "",
    model: str | None = None,
    critic_model: str | None = None,
    critic_system: str = _CRITIC_SYSTEM,
    max_trials: int = 3,
    keep_reflections: int = 5,
    params: dict[str, Any] | None = None,
    extract_code: bool = False,
)

Trial -> evaluate -> (if rejected) reflect -> new trial, at most max_trials times.

The evaluator is a deterministic named validator (validator="...", served by ValidationHandler) or, if none is given, an LLM critic. keep_reflections bounds how many lessons enter the prompt (the most recent ones). Result: {"candidate", "trials", "reflections"}; a run that never passes ends NO_SOLUTION carrying the best-effort last attempt and the reflections.

Debate ¶

Debate(
    agents: int = 3,
    rounds: int = 1,
    *,
    judge: bool = True,
    personas: list[str] | None = None,
    model: str | None = None,
    judge_model: str | None = None,
    system: str = "Think step by step, then end with a line 'Answer: <final answer>'.",
    extract: Callable[[str], str] = default_extract,
    temperature: float = 0.7,
)

agents debaters answer rounds + 1 times (round 0 independently, later rounds after reading the others' previous answers). judge=True (default) asks one more model call to rule on the final answers; judge=False takes the majority of the extracted final answers. personas gives each debater its own system prompt.

Result: {"answer", "rounds", "agents", "answers": [[round-0 answers], ...], "judged": bool}.

Swarm ¶

Swarm(
    agents: Mapping[str, Agent],
    tools: ToolRegistry,
    *,
    start: str,
    max_handoffs: int = 6,
    **kw: Any,
)

Bases: ToolLoop

A ToolLoop whose active agent can change. Attach ToolHandler(swarm.tools) as usual (the registry holds every tool; each agent sees only its own). Result is the final agent's reply plus the agent that gave it: {"text", "turns", "agent", "handoffs": [[from, to], ...]}.

Agent `dataclass` ¶

Agent(
    instructions: str,
    tools: list[str] = list(),
    handoffs: list[str] = list(),
    description: str = "",
    model: str | None = None,
)

TreeOfThoughts ¶

TreeOfThoughts(
    *,
    depth: int = 3,
    width: int = 2,
    branching: int = 3,
    stop_score: float | None = 9.0,
    model: str | None = None,
    scorer_model: str | None = None,
    system: str = "You solve problems one reasoning step at a time.",
    temperature: float = 0.8,
)

Beam search over reasoning steps.

At each of up to depth levels, every node of the beam gets branching proposed next steps; each proposal is scored 0-10 by a separate call; the best width survive. A proposal that starts with FINAL: is a complete answer. The search stops when the best node of a level is final, when its score reaches stop_score, or at the last level, and returns the best path.

Result: {"answer", "score", "path": [steps...], "depth", "explored"}.

max_calls `property` ¶

max_calls: int

The most model calls one run can make.

aaa.algos¶

Tool-using loops¶

ToolLoop ¶

ReAct ¶

LLMLoop ¶

request_for ¶

first_request ¶

with_memory ¶

Planning and delegation¶

PlanAndExecute ¶

Plan ¶

Step ¶

Router ¶

Route dataclass ¶

Supervisor ¶

Verification loops¶

GenerateTestRepair ¶

GeneratorCritic ¶

Deterministic control flow¶

WorkflowGraph ¶

Node dataclass ¶

FSM ¶

Ensembles and retrieval¶

SelfConsistency ¶

RAGAgent ¶

Authoring helpers¶

base ¶

llm_request ¶

tool_request ¶

finish ¶

parse_json_obj ¶

strip_fence ¶

render_result ¶

task_messages ¶

Reasoning algorithms¶

Reflexion ¶

Debate ¶

Swarm ¶

Agent dataclass ¶

TreeOfThoughts ¶

max_calls property ¶

`aaa.algos`¶

Route `dataclass` ¶

Node `dataclass` ¶

Agent `dataclass` ¶

max_calls `property` ¶