Custom architectures

If none of the built-in architectures fits your loop, write your own: implement the Architecture protocol (three members) and pass an instance to Agent(architecture=...).

The protocol

from collections.abc import AsyncIterator
from typing import Protocol, runtime_checkable
from loomflow import Agent
from loomflow.architecture.base import AgentSession, Dependencies
from loomflow.core.types import Event
 
@runtime_checkable
class Architecture(Protocol):
    name: str
 
    def declared_workers(self) -> dict[str, Agent]: ...
 
    def run(
        self,
        session: AgentSession,
        deps: Dependencies,
        prompt: str,
    ) -> AsyncIterator[Event]:
        ...

• name. Short identifier for telemetry / logging (e.g. "react", "my-strategy").
• declared_workers(). The sub-agents the architecture delegates to, keyed by role name. Used for graph visualization. Return {} for single-agent loops.
• run(session, deps, prompt). Async generator (async def with yield) yielding Events as iteration progresses. prompt is the user’s message for this pass. The Agent may call run again on the same session when a stop hook asks it to continue; then session.messages is not empty and prompt is a new user turn to append. See Architecture protocols.

What AgentSession and Dependencies give you

AgentSession is the mutable per-run state:

Dependencies is the read-only protocol bundle:

Setup events (Event.started) and teardown events (Event.completed) are emitted by Agent, NOT the architecture. Architectures yield the events that happen during iteration. Per-turn, per-tool, per-step, budget warnings, errors.

Minimal example: a “single-shot” architecture

from collections.abc import AsyncIterator
from loomflow import Agent
from loomflow.architecture.base import AgentSession, Dependencies
from loomflow.core.types import Event, Message, Role
 
class SingleShot:
    """One model call, no tools, return the answer. For testing."""
 
    name: str = "single-shot"
 
    def declared_workers(self) -> dict[str, Agent]:
        return {}
 
    async def run(
        self,
        session: AgentSession,
        deps: Dependencies,
        prompt: str,
    ) -> AsyncIterator[Event]:
        if not session.messages:
            session.messages.append(
                Message(role=Role.SYSTEM, content=session.instructions)
            )
        session.messages.append(Message(role=Role.USER, content=prompt))
 
        text = ""
        async for chunk in deps.model.stream(session.messages):
            if chunk.kind == "text":
                text += chunk.text or ""
            yield Event.model_chunk(session.id, chunk)
 
        session.messages.append(Message(role=Role.ASSISTANT, content=text))
        session.output = text
        session.turns += 1
        # No need to yield Event.completed — Agent emits that.
 
agent = Agent("...", model="claude-opus-4-7", architecture=SingleShot())

This sketch skips the budget checks and usage accounting the built-in architectures do (through budget_gate and consume_usage in loomflow.architecture.helpers), and it doesn’t recall memory.

Replay-correctness contract

If you make external calls inside run, wrap them with runtime.step(...) so they’re cached on the journal:

result = await deps.runtime.step(
    f"my_call_{session.turns}",
    my_async_callable,
    arg1, arg2,
)

Without this, a crashed run can’t resume cleanly. The call would re-execute and produce a different result. See Runtime.

Telemetry-correctness contract

Open a span per logical “turn” so the OTel trace matches the iteration:

async with deps.telemetry.trace(
    "myarch.turn",
    turn=session.turns,
    session_id=session.id,
):
    ...

ReAct opens loom.turn per iteration, loom.model.complete (or loom.model.stream when streaming) per model call and loom.tool per tool call. Match that pattern for compatible dashboards. Give model-call spans the gen_ai.* attributes from loomflow.observability.semconv.chat_attrs(deps.model), and tool spans tool_attrs(name, call_id=...): OTelTelemetry then exports them under the GenAI spec names (chat {model}, execute_tool {tool}) like the built-in spans, keeping your name in the loom.span.name attribute. See GenAI semantic conventions.