Choosing algorithms

AAA's rule of thumb (from its algorithm guide): start with the simplest algorithm that fits, prefer determinism you can have, and verify with code before verifying with a model. This is how that played out for a real pipeline.

Stage	AAA algorithm	Why this one
Collect signals (GitHub, HN, SearXNG)	WorkflowGraph	The steps are known and independent. A DAG runs them in parallel with zero LLM calls, and on_error="continue" means one dead source never stops the run.
"Is this an AI agent?"	GenerateTestRepair + SelfConsistency	Structured JSON checked by a schema validator. Borderline confidence (0.4 to 0.6) gets a 3-way vote, because the label is an exact answer.
Fetch a page	single-tool WorkflowGraph	Reuses tool timeouts, pools and the execution guard; one run per URL, many in parallel.
Extract facts	GenerateTestRepair with a custom validator	The validator is our code: every claim's quote must appear verbatim on a fetched page. Failure feedback goes back to the model.
Find a missing pricing/features page	ToolLoop (6 turns, 2 tools)	The model must decide where to look, so it earns a loop, but it is bounded and its tools are restricted to the product's own site.
Confirm prices	SelfConsistency	Prices are exact strings, which is where voting works.
Write the overview	GeneratorCritic	The one place we generate prose: only from validated facts, judged by a second model, then re-checked by code.

What we deliberately did not use

Algorithm	Why not
ReAct	Native tool calling is available.
PlanAndExecute	Our steps are known, so a DAG is cheaper and more predictable. Revisit for open-ended research.
Supervisor / Swarm / Router	One pipeline, no specialists to hand off between. (We route models, not tasks; see patterns.)
Reflexion / Debate / TreeOfThoughts	AAA marks their quality gains as unverified with real models, and they cost more calls.
FSM	Listing lifecycle is a status column. An FSM earns its place when something must wait days for a human or a webhook.
Memory libraries (Mem0, Graphiti, ...)	SQLite tables are our memory.

Source

The role-to-instance table is a single small file:

"""One long-lived instance per role, each with a UNIQUE name.

AAA routes and recovers runs by algorithm name, and registering a new instance per call would
leak a bus subscription each time. Per-run variation travels in the run input, not in the object."""

from __future__ import annotations

from dataclasses import dataclass

from aaa.algos import GenerateTestRepair, GeneratorCritic, Node, SelfConsistency, WorkflowGraph

from botshub_agent.config import Settings
from botshub_agent.discovery import classify
from botshub_agent.discovery.workflow import build_graph
from botshub_agent.enrich import extract, gapfill, overview, pricevote


def _named(algo, name: str):
    algo.name = name
    return algo


def _one(tool: str, name: str) -> WorkflowGraph:
    """A single-tool graph whose arguments are the run input."""
    return _named(WorkflowGraph([Node("r", tool, args=lambda r, inp: dict(inp))], output="r"), name)


@dataclass
class Algos:
    discovery: WorkflowGraph
    fetch_page: WorkflowGraph
    github_readme: WorkflowGraph
    count_mentions: WorkflowGraph
    capture: WorkflowGraph
    videos: WorkflowGraph
    classify: GenerateTestRepair
    vote: SelfConsistency
    extract: GenerateTestRepair
    overview: GeneratorCritic
    gapfill: object
    price_vote: SelfConsistency

    def all(self) -> list:
        return [self.discovery, self.fetch_page, self.github_readme, self.count_mentions, self.capture, self.videos,
                self.classify, self.vote, self.extract, self.overview, self.gapfill, self.price_vote]


def make_algos(s: Settings, registry) -> Algos:
    return Algos(
        discovery=_named(build_graph(s), "bh_discovery"),
        fetch_page=_one("fetch_page", "bh_fetch_page"),
        github_readme=_one("github_readme", "bh_github_readme"),
        count_mentions=_one("count_mentions", "bh_count_mentions"),
        capture=_one("capture_screenshot", "bh_capture"),
        videos=_one("find_product_videos", "bh_find_videos"),
        classify=_named(classify.classify_algo(s.llm_model), "bh_classify"),
        vote=_named(classify.vote_algo(s.llm_model), "bh_classify_vote"),
        extract=_named(extract.extraction_algo(s.llm_model), "bh_extract"),
        overview=_named(overview.overview_algo(s.llm_model, s.critic_model), "bh_overview"),
        gapfill=_named(gapfill.gapfill_algo(registry, s.llm_model), "bh_gapfill"),
        price_vote=_named(pricevote.price_vote_algo(s.llm_model, s.price_votes), "bh_price_vote"),
    )