Writing your own algorithm¶

An algorithm is a small object with three members, and no state of its own:

class MyAlgorithm:
    name = "my_algorithm"  # unique; how runs are routed to you

    def subscriptions(self) -> list[str]:  # event-type globs you react to
        return ["aaa.run.started", "aaa.llm.completed"]

    async def on_event(self, e: Event, ctx: Ctx) -> list[Event]:  # (event, run state) -> new events
        ...

Here is a complete one: translate a text into several languages in parallel, then collect the results.

"""Write your own algorithm: a fan-out/fan-in that translates into N languages in parallel.

An algorithm is just: which events it reacts to, and a function from (event, run state) to new
events. It holds no state of its own, so it survives restarts and can run on any worker."""

import asyncio

from pydantic import BaseModel, Field

from aaa.algos.base import finish, llm_request
from aaa.core import OutcomeKind
from aaa.events import Event, Projection
from aaa.llm import FakeLLMClient, LLMHandler, LLMRequest, LLMResponse, Message
from aaa.llm.fake import text
from aaa.runtime import Ctx, Engine


class Translations(BaseModel):  # run state = a fold over the run's events
    wanted: list[str] = Field(default_factory=list)
    done: dict[str, str] = Field(default_factory=dict)


def apply(state: Translations, e: Event) -> Translations:
    if e.type == "aaa.run.started":
        state.wanted = e.data["input"]["languages"]
    elif e.type == "aaa.llm.completed":
        state.done[e.data["tag"]] = (
            LLMResponse.model_validate(e.data["response"]).message.content or ""
        )
    elif e.type == "aaa.llm.failed":
        state.done[e.data["tag"]] = f"<failed: {e.data['error']['type']}>"
    return state


TRANSLATIONS = Projection(
    Translations, apply, name="example.translations", version=1, state_type=Translations
)


class Translate:
    name = "translate"

    def subscriptions(self) -> list[str]:
        return ["aaa.run.started", "aaa.llm.completed", "aaa.llm.failed"]

    async def on_event(self, e: Event, ctx: Ctx) -> list[Event]:
        if e.type == "aaa.run.started":  # fan out: one tagged LLM request per language
            body, langs = e.data["input"]["text"], e.data["input"]["languages"]
            return [
                llm_request(
                    e,
                    LLMRequest(messages=[Message(role="user", content=f"To {lang}: {body}")]),
                    tag=lang,
                )
                for lang in langs
            ]
        state = await ctx.project(TRANSLATIONS)  # fan in: wait until every language reported
        if len(state.done) < len(state.wanted):
            return []
        return [finish(e, OutcomeKind.SUCCESS, state.done)]


def model(request):
    lang = request.messages[-1].content.split(":")[0].replace("To ", "")
    return text({"French": "Bonjour", "German": "Hallo", "Spanish": "Hola"}[lang])


async def main() -> None:
    async with Engine() as engine:
        LLMHandler(FakeLLMClient(model)).attach(engine)
        outcome = await engine.run(
            Translate(), {"text": "Hello", "languages": ["French", "German", "Spanish"]}
        )
    print(dict(sorted(outcome.value.items())))


asyncio.run(main())

{'French': 'Bonjour', 'German': 'Hallo', 'Spanish': 'Hola'}

The recipe¶

Describe the run state as a projection (a Pydantic model plus a fold function). See Projections and snapshots.
React to aaa.run.started by emitting the first request(s).
React to results by projecting state and deciding. Return [] when you are still waiting.
Finish with finish(e, OutcomeKind.SUCCESS, value).

Helpers (`aaa.algos.base`)¶

Helper
`llm_request(parent, LLMRequest, tag="main", prev=None)`	An `aaa.llm.requested` event. `tag` correlates the reply; `prev` continues a conversation as a delta (see below).
`tool_request(parent, call_id, name, arguments)`	An `aaa.tool.requested` event.
`finish(parent, kind, value, reason)`	The terminal `aaa.run.completed` event.
`parse_json_obj(text)`	Best-effort JSON object from model text (strips code fences, tolerates prose).
`strip_fence(text)`, `render_result(x)`, `task_messages(input, system)`	Small utilities.

`Ctx`¶


`await ctx.project(PROJECTION)`	The run's current state: snapshot + tail.
`await ctx.started()`	The `aaa.run.started` event (input, budget).
`await ctx.history()`	Every event (use `project` instead for anything on the hot path).
`ctx.ledger()`	Spend so far.
`ctx.engine`, `ctx.run_id`	For starting child runs or emitting events directly.

Rules that keep it correct¶

Decide from state, not from the triggering event. The same event may be seen twice; a pure function of the projected state gives the same answer both times.
Derive events with e.caused(...) (every helper does). That gives them deterministic ids, so re-deriving after a crash is absorbed by the store instead of duplicated.
No I/O and no clocks in on_event. Anything with a side effect belongs in a tool or a handler. That is what makes a run replayable.
Tag parallel requests and key your projection by tag, as above. Return [] until all have reported.
Child runs: await ctx.engine.start(algo, input, parent=ctx.run_id, tag=..., run_id=stable_id(ctx.run_id, tag)). A deterministic run_id makes the start idempotent after a crash; the result arrives as aaa.child.completed.
Emitting more than one event in a defined order where a later step reads the earlier one: call await ctx.engine.emit(event) for the first, then project again (PlanAndExecute does this when it creates a plan then schedules from it).
Unique names. Two instances with the same name cannot both run in one cluster. Pass distinct names.

Continuing a conversation cheaply¶

For multi-turn loops, send only the new messages and point at the previous request:

first = llm_request(e, LLMRequest(messages=[system, user]))  # full messages
second = llm_request(e, LLMRequest(messages=[tool_result_message]), prev=state.request_id)

Store request_id (the id of the last main request) in your projection instead of the transcript. The LLMHandler reconstructs the full conversation for the provider. This keeps log size and state linear in the number of turns.

Testing it¶

Use FakeLLMClient with a function of the request, and assert on the outcome and the event list. See Testing your agents.