AETIUS

LangGraph adapter

Mapping

LangGraph's native model is already a general graph: add_node, add_edge, add_conditional_edges, START/END. agentstage.langgraph.adapt(system) maps OrchestrationGraph onto it directly rather than special-casing each pattern:

  • One node per reconstructed role, via add_node.
  • A from_role with exactly one distinct observed to_role becomes a plain add_edge, deterministic and safe to wire without asking anything of the caller.
  • A from_role with more than one distinct to_role is a real branching decision the traces show happened, but not how it was decided at runtime. That mechanism has to come from the developer, via a routers[from_role] path function passed to adapt(). Without one, adapt() raises rather than guessing which branch to prefer.
  • Every role in terminal_roles gets an edge to END.
  • pattern == "unclassified" additionally requires routers to be passed at all, even {}, as an explicit opt-in. An unclassified graph is the case where structure is least trustworthy, so it never builds silently.
  • Any node with no path from entry_role through the wired edges is refused rather than silently included as a dead node.

adapt() never supplies the model call inside a node. node_factory(role_spec, tools) is required: for each role (with its reconstructed system prompt and replay-backed tools already resolved), the caller returns the actual node callable that invokes their own real model. What adapt() supplies is the reconstructed tools, bound to the replay matcher, and the graph topology itself.

End to end example

The examples below use the local fixture from examples/fixtures.py (see the core concepts page). The node functions here call the bound tools directly with a fixed decision instead of an LLM call, so the example runs with no API keys and no cost. In a real system, node_factory would call your own model instead.

single_agent

Python
from typing import TypedDict
from examples.fixtures import build_environment
from agentstage.langgraph import adapt


class State(TypedDict):
    messages: list


env = build_environment()
system = env.reconstruct("invoice-agent")


def invoice_node_factory(role_spec, tools):
    tool_by_name = {t.name: t for t in tools}

    def node(state):
        last = state["messages"][-1] if state["messages"] else ""
        if "inv_1002" in last:
            result = tool_by_name["lookup_invoice"].invoke({"invoice_id": "inv_1002"})
        else:
            result = tool_by_name["lookup_invoice"].invoke({"invoice_id": "inv_1003"})
        return {"messages": state["messages"] + [f"agent: lookup_invoice -> {result}"]}

    return node


graph = adapt(system, state_schema=State, node_factory=invoice_node_factory)
compiled = graph.compile()

print(compiled.invoke({"messages": ["What is the status of inv_1003?"]}))
print(compiled.invoke({"messages": ["What is the status of inv_1002?"]}))
Output
{'messages': ['What is the status of inv_1003?', "agent: lookup_invoice -> {'status': 'paid', 'amount': 89.99}"]}
{'messages': ['What is the status of inv_1002?', "agent: lookup_invoice -> {'status': 'overdue', 'amount': 54.5}"]}

Both invoice numbers replay their real historical response.

supervisor_delegates without a router

Python
from agentstage.errors import OrchestrationMappingError

team_system = env.reconstruct("support-team")


def team_node_factory(role_spec, tools):
    def node(state):
        return state

    return node


try:
    adapt(team_system, state_schema=State, node_factory=team_node_factory)
except OrchestrationMappingError as e:
    print(f"{type(e).__name__}: {e}")
Output
OrchestrationMappingError: role 'supervisor' branches to ['researcher', 'writer'] in the observed traces, but how that choice was made at runtime isn't captured — pass a path function for it via routers['supervisor'] rather than have this guess which branch to take.

The supervisor role really did hand off to both researcher and writer across the traces. Which one to pick for a specific run is not something the traces settle, so adapt() refuses instead of arbitrarily choosing one.

supervisor_delegates with a router supplied

Python
class TeamState(TypedDict):
    messages: list
    next: str


def team_node_factory_v2(role_spec, tools):
    tool_by_name = {t.name: t for t in tools}

    def node(state):
        if role_spec.role == "supervisor":
            return {"messages": state["messages"] + ["supervisor: routing to researcher"], "next": "researcher"}
        if role_spec.role == "researcher":
            result = tool_by_name["search_docs"].invoke({"query": "refund policy for annual plans"})
            return {"messages": state["messages"] + [f"researcher: search_docs -> {result}"]}
        if role_spec.role == "writer":
            result = tool_by_name["draft_reply"].invoke({"topic": "refund policy"})
            return {"messages": state["messages"] + [f"writer: draft_reply -> {result}"]}
        return state

    return node


def supervisor_router(state):
    return state["next"]


graph = adapt(
    team_system,
    state_schema=TeamState,
    node_factory=team_node_factory_v2,
    routers={"supervisor": supervisor_router},
)
compiled = graph.compile()
print(compiled.invoke({"messages": ["What is our refund policy for annual plans?"], "next": ""}))
Output
{'messages': ['What is our refund policy for annual plans?', 'supervisor: routing to researcher', "researcher: search_docs -> {'results': ['refund-policy.md#annual-plans']}"], 'next': 'researcher'}

The supervisor node's decision (baked into this example as a fixed choice, a real model's decision in production) drives the conditional edge, and the researcher's tool call replays a real historical response.

A Braintrust-sourced system

Everything above used the Langfuse-shaped fixture. agentstage.langgraph.adapt() does not know or care which source built the ReconstructedSystem it is given; the same real compiled-graph execution works unchanged against a Braintrust-sourced one. No live Braintrust project was available while writing this page either, so this uses examples/braintrust_fixtures.py, the same disclosed-fixture pattern as docs/sources/braintrust.md.

Python
from examples.braintrust_fixtures import build_environment as build_braintrust_environment

bt_env = build_braintrust_environment()
bt_invoice_system = bt_env.reconstruct("invoice-agent")
print("roles:", [r.role for r in bt_invoice_system.roles])


def bt_invoice_node_factory(role_spec, tools):
    tool_by_name = {t.name: t for t in tools}

    def node(state):
        last = state["messages"][-1] if state["messages"] else ""
        if "inv_3002" in last:
            result = tool_by_name["lookup_invoice"].invoke({"invoice_id": "inv_3002"})
        else:
            result = tool_by_name["lookup_invoice"].invoke({"invoice_id": "inv_3001"})
        return {"messages": state["messages"] + [f"agent: lookup_invoice -> {result}"]}

    return node


bt_graph = adapt(bt_invoice_system, state_schema=State, node_factory=bt_invoice_node_factory)
bt_compiled = bt_graph.compile()
print(bt_compiled.invoke({"messages": ["What is the status of inv_3001?"]}))
print(bt_compiled.invoke({"messages": ["What is the status of inv_3002?"]}))
Output
roles: ['invoice-agent']
{'messages': ['What is the status of inv_3001?', "agent: lookup_invoice -> {'status': 'paid', 'amount': 145.0}"]}
{'messages': ['What is the status of inv_3002?', "agent: lookup_invoice -> {'status': 'overdue', 'amount': 62.25}"]}

One real difference from the Langfuse-sourced example above: the role is named invoice-agent, not the generic agent fallback the Langfuse-sourced single_agent system used. Braintrust has no separate trace-level object the way Langfuse does; the grouping key and the root span's own resolved role identity both come from the same metadata.agent_name, so a root span that tags its own identity ends up naming the role after it. See docs/sources/braintrust.md for the full explanation. The graph wiring and execution themselves are identical either way.

The refusal cases

The build prompt for this page asked specifically to show OrchestrationUnavailableError for the case of calling adapt() without routers on an unclassified graph. That is not quite what happens: OrchestrationUnavailableError and OrchestrationMappingError are two different, real errors for two different situations, and it is worth showing both precisely rather than blurring them.

OrchestrationUnavailableError: no orchestration evidence at all

This fires when system.orchestration.source == "unavailable", which only happens when there is no trace data to reconstruct from in the first place:

Python
from agentstage.reconstruction.build import build_reconstructed_system
from agentstage.errors import OrchestrationUnavailableError

empty_system = build_reconstructed_system([], label="empty-system", project_id="demo_project", source_agent_name=None)
print("orchestration.source =", empty_system.orchestration.source)


def empty_node_factory(role_spec, tools):
    return lambda state: state


try:
    adapt(empty_system, state_schema=State, node_factory=empty_node_factory)
except OrchestrationUnavailableError as e:
    print(f"{type(e).__name__}: {e}")
Output
orchestration.source = unavailable
OrchestrationUnavailableError: cannot build a langgraph structure: this system's orchestration is unavailable (no reliable role signal was found in the source traces) — nothing to wire up

OrchestrationMappingError: real structure, but it does not classify

The fixture's multi-hub system has two independent branching hubs, hub_a -> {x, y} and hub_b -> {z, w}, with no edge between them. This is real, observed structure, just not one of the three clean shapes the classifier recognizes:

Python
multi_hub_system = env.reconstruct("multi-hub")
print("pattern:", multi_hub_system.orchestration.pattern)
print("nodes:", [n.role for n in multi_hub_system.orchestration.nodes])
print("edges:", [(e.from_role, e.to_role) for e in multi_hub_system.orchestration.edges])


def passthrough_node_factory(role_spec, tools):
    return lambda state: state


try:
    adapt(multi_hub_system, state_schema=State, node_factory=passthrough_node_factory)
except OrchestrationMappingError as e:
    print(f"{type(e).__name__}: {e}")
Output
pattern: unclassified
nodes: ['hub_a', 'hub_b', 'w', 'x', 'y', 'z']
edges: [('hub_a', 'x'), ('hub_a', 'y'), ('hub_b', 'z'), ('hub_b', 'w')]
OrchestrationMappingError: orchestration graph did not classify into a known shape (pattern='unclassified') — nodes=['hub_a', 'hub_b', 'w', 'x', 'y', 'z'], edges=[('hub_a', 'x'), ('hub_a', 'y'), ('hub_b', 'z'), ('hub_b', 'w')]. Refusing to best-effort-wire this: pass `routers` explicitly (even `{}` if no role branches) to confirm you've reviewed the graph and supply a path function for any role with more than one observed destination.

Passing routers at all is the opt-in. But routers alone are not enough if the resulting graph still leaves nodes unreachable, which is exactly what happens here: hub_a and hub_b are two separate roots with no edge connecting them, so even with both hubs' branching resolved, everything under hub_b is unreachable from entry_role:

Python
try:
    adapt(
        multi_hub_system,
        state_schema=State,
        node_factory=passthrough_node_factory,
        routers={"hub_a": lambda s: "x", "hub_b": lambda s: "z"},
    )
except OrchestrationMappingError as e:
    print(f"{type(e).__name__}: {e}")
Output
OrchestrationMappingError: role(s) ['hub_b', 'w', 'z'] have no path from entry_role 'hub_a' through the observed edges — would be dead nodes in the compiled graph. This usually means the trace batch caught a role's tool calls but never the transition into it; reconstruct with more traces, or pass `routers` covering the missing transition if you know it.

Neither call built a graph. Both told the developer exactly why, and exactly what to supply to move forward.