Multi-agent graphs

Primitives for dynamic, LLM-routed multi-agent systems where the topology is decided at runtime rather than declared up front.

• AgentPool — a name-keyed registry exposed to agents as a single route(agent_name, task) tool. Lets agents delegate to each other by name (resolving the circular-reference problem that a frozen tool map otherwise imposes) and bounds recursion via max_depth.
• conclude — a non-local exit: any agent, however deeply nested, can call conclude("answer") to end the whole task and return its answer straight to the originating top-level Agent.run. Raised internally as ConcludeSignal (a BaseException) and caught only at the top level.

Pair them with LLMEngine(max_tool_calls_per_turn=1) to keep the graph on a single non-branching path. For narrative usage see Guides → Mid → Dynamic graph.

lazybridge.AgentPool

AgentPool(*, max_depth: int = 25)

Registry of named agents, exposed to the LLM as a single route tool.

Source code in lazybridge/pool.py

def __init__(self, *, max_depth: int = 25) -> None:
    if max_depth < 1:
        raise ValueError(f"max_depth must be >= 1, got {max_depth!r}")
    self._agents: dict[str, Agent] = {}
    self._max_depth = max_depth
    self._depth: contextvars.ContextVar[int] = contextvars.ContextVar("lb_pool_depth", default=0)

register

register(*agents: Agent) -> None

Add agents to the pool, keyed by their name.

Source code in lazybridge/pool.py

def register(self, *agents: Agent) -> None:
    """Add agents to the pool, keyed by their ``name``."""
    for agent in agents:
        if not getattr(agent, "name", None):
            raise ValueError("AgentPool.register() requires agents with an explicit name=.")
        self._agents[agent.name] = agent

roster

roster() -> str

One line per registered agent — drop into the members' system prompt.

Source code in lazybridge/pool.py

def roster(self) -> str:
    """One line per registered agent — drop into the members' system prompt."""
    return "\n".join(f"- {name}: {a.description or ''}" for name, a in self._agents.items())

route async

route(agent_name: str, task: str) -> str

Delegate task to the named specialist agent and return its answer.

Source code in lazybridge/pool.py

async def route(self, agent_name: str, task: str) -> str:
    """Delegate ``task`` to the named specialist agent and return its answer."""
    depth = self._depth.get()
    if depth >= self._max_depth:
        return f"Max routing depth {self._max_depth} reached — call conclude now with your best answer."
    agent = self._agents.get(agent_name)
    if agent is None:
        return f"Unknown agent {agent_name!r}. Available: {list(self._agents)}"
    token = self._depth.set(depth + 1)
    try:
        # ``_run_as_tool`` so a ``conclude`` raised downstream propagates
        # past this level to the original top-level caller.
        return (await agent._run_as_tool(task)).text()
    finally:
        self._depth.reset(token)

as_tool

as_tool(name: str = 'route') -> Tool

Wrap routing as a route(agent_name, task) tool for tools=[...].

Pass a distinct name to give one agent access to several pools without a tool-name collision, e.g. tools=[my_team.as_tool("ask_team"), peers.as_tool("ask_peer")].

Source code in lazybridge/pool.py

def as_tool(self, name: str = "route") -> Tool:
    """Wrap routing as a ``route(agent_name, task)`` tool for ``tools=[...]``.

    Pass a distinct ``name`` to give one agent access to several pools
    without a tool-name collision, e.g.
    ``tools=[my_team.as_tool("ask_team"), peers.as_tool("ask_peer")]``.
    """
    return Tool.wrap(self.route, name=name)

lazybridge.conclude

conclude(message: str) -> str

End the whole task now and return message as the final answer.

Use as a tool in multi-agent graphs: any agent — however deeply nested — can call conclude("…") to short-circuit the entire call chain and return its answer directly to the original caller.

Immediacy note: the exit fires as soon as the turn's tool calls settle. If the model emits conclude alongside other tool calls in the same turn, those siblings still run to completion first (they execute concurrently via asyncio.gather), so a slow sibling can delay the exit. Set LLMEngine(max_tool_calls_per_turn=1) — the recommended multi-agent configuration — to keep one call per turn and avoid this.

Source code in lazybridge/signals.py

def conclude(message: str) -> str:
    """End the whole task now and return ``message`` as the final answer.

    Use as a tool in multi-agent graphs: any agent — however deeply nested —
    can call ``conclude("…")`` to short-circuit the entire call chain and
    return its answer directly to the original caller.

    Immediacy note: the exit fires as soon as the turn's tool calls settle.
    If the model emits ``conclude`` *alongside* other tool calls in the same
    turn, those siblings still run to completion first (they execute
    concurrently via ``asyncio.gather``), so a slow sibling can delay the
    exit. Set ``LLMEngine(max_tool_calls_per_turn=1)`` — the recommended
    multi-agent configuration — to keep one call per turn and avoid this.
    """
    raise ConcludeSignal(message)

lazybridge.ConcludeSignal

ConcludeSignal(message: str)

Bases: BaseException

Raised by :func:conclude to end the whole task with message.

Source code in lazybridge/signals.py

def __init__(self, message: str) -> None:
    self.message = message
    super().__init__(message)