Patterns we used

Reusable shapes that came out of building this agent on AAA.

Verification gate: let code veto the model

GenerateTestRepair takes a named validator, any function (candidate, context) -> Verdict. We use it as an anti-hallucination gate:

def make_validator(get_pages):
    """get_pages(context) -> {url: text}. Registered on ValidationHandler as 'facts'."""

    def validate(candidate: str, context: dict) -> Verdict:
        try:
            ex = parse_extraction(candidate)
        except (ValueError, ValidationError) as e:
            return Verdict(ok=False, feedback=f"Reply with ONLY valid JSON matching the schema. Error: {str(e)[:300]}")
        _, problems = ground(ex, get_pages(context))
        if not problems:
            return Verdict(ok=True)
        shown = "; ".join(problems[:8]) + (f" (+{len(problems) - 8} more)" if len(problems) > 8 else "")
        return Verdict(
            ok=False,
            feedback="Remove or fix every unsupported claim. Never guess; use null/[] when the pages do not say it. Each quote must be ONE short contiguous snippet copied exactly (never stitch lines together or skip text between them). "
            f"Problems: {shown}",
        )

    return validate

The validator parses the model's JSON, checks every claim against pages stored in SQLite, and returns specific feedback (quote not found verbatim on <url>). The model sees that feedback and retries, up to max_iterations. If it never converges, AAA returns no_solution with the best candidate, and we salvage only the claims that pass the same check. The result: output is never worse than "fewer facts", never "invented facts".

Single-tool graphs for fan-out

WorkflowGraph nodes are static, but "fetch these N URLs" is dynamic. Instead of building a graph per call, define a one-node graph whose arguments are the run input, and run it many times concurrently:

def _one(tool, name):
    return _named(WorkflowGraph([Node("r", tool, args=lambda r, inp: dict(inp))], output="r"), name)

results = await asyncio.gather(*(engine.run(fetch_page_graph, {"candidate_id": 7, "url": u}) for u in urls))

You keep tool timeouts, pools and logging, with no per-call algorithm objects.

One long-lived instance per role, unique names

AAA routes and recovers runs by algorithm name, and registering a new instance each call leaks a bus subscription. So we create each algorithm once, give it a unique name, and put per-run variation in the run input, not the object.

algo.name = "bh_extract"

Trusted run metadata for tool scoping

A bounded ToolLoop may only read pages on a product's own site. The allowed host must not come from the model, so it travels in the run's metadata (trusted application data) and the tool reads it from ToolContext:

await engine.run(gapfill_loop, {"task": "..."}, meta={"candidate_id": 42})

@tool
async def read_page(url: str, ctx: ToolContext) -> dict:
    cand = repo.candidate(int(ctx.meta["candidate_id"]))
    if not allowed(url, cand):
        raise ToolError("only pages on the product's own website or GitHub repo may be read")

ctx is injected and hidden from the model's schema, so a prompt injection cannot change which candidate it is acting for.

Route models, not agents

AAA's LLMHandler talks to one LLMClient. To run a small critic and an embedding model locally while extraction stays on a hosted gateway, we wrap two clients in a router that dispatches on the model id:

class RoutingClient:
    """Local models for ids matching `prefixes`; the gateway for the rest."""

    def __init__(self, remote: LLMClient, local: LLMClient, prefixes: tuple[str, ...]) -> None:
        self.remote, self.local, self.prefixes = remote, local, prefixes

    def _pick(self, model: str | None) -> LLMClient:
        return self.local if is_local(model, self.prefixes) else self.remote

    async def complete(self, request: LLMRequest, /) -> LLMResponse:
        return await self._pick(request.model).complete(request)

    async def embed(self, texts: list[str], model: str | None = None) -> EmbedResponse:
        return await self._pick(model).embed(texts, model)

    async def aclose(self) -> None:
        await self.remote.aclose()
        await self.local.aclose()

Anything starting with mlx-community/ runs in-process on Apple silicon (setup); everything else goes to the OpenAI-compatible gateway. Algorithms never know the difference.

Crash safety on SQLite

SQLiteEventStore is WAL-mode and crash-safe, with one engine process per file. We keep AAA's event log (aaa_events.db) separate from our business tables (botshub.db), run a single worker process with asyncio concurrency, and call engine.recover(all_algorithms) at startup.

"Never downgrade" for re-runs

Re-enriching a published listing can fail (LLM outage, thin page). Rather than overwrite good data, the pipeline records kept_previous and leaves the live listing alone. Idempotent upserts keyed by slug mean replays never duplicate.