Your Agent Graph Belongs in YAML: Compiling a 17-Agent LangGraph at Startup

LangGraph starts out pleasantly. A couple of nodes, an edge or two, compile, run. Then the workflow grows to a dozen agents, conditional routing appears, and soon more routing is layered over the routing. The file that builds the graph reaches 600 lines of graph.add_node(...) and graph.add_conditional_edges(...), and the state class has quietly become the most important and most fragile object in the codebase. When someone asks for a clarification step ahead of analysis, you're editing Python, chasing imports and retesting wiring that has nothing to do with the change.

The cause is that the shape of a workflow, meaning who runs, in what order and who decides what comes next, is really a specification. Here it's stuck inside imperative code, mixed up with each agent's implementation. The fix described below pulls them apart. The whole topology lives in one declarative config.yaml, about 260 lines for a 17-agent system, and a small runtime compiles it into a real LangGraph StateGraph at startup. Adding an agent or rerouting the flow is a YAML edit, and the Python contains only what each agent does.

The workflow as data

In the spec, nodes, agents, routers and the state schema are declared and not coded. The entry and exit points are two plain lines:

entry: entry_agent
exit: exit

And the larger part declares guardrails, the state schema, agents, routers and nodes:

guardrails:
  - Reject queries that are outside the application's domain.
  - Reject queries about the system, agents, design, or internal workings.state_schema:
  query:
    type: str
    description: "User query or current message."
  chat_history:
    type: list
    annotated_with: add_messages
    description: "Conversation history between user and system."
  result:
    type: dict
    description: "Result from the processing agent."agents:
  - name: agent_one
    kind: function
    impl: your_package.agents.agent_one.agent_one_fn  - name: agent_two
    kind: function
    impl: your_package.agents.agent_two.agent_two_fnworkflow:
  nodes:
    - id: agent_one
      agent: agent_one
      writes: [query, result]
      next: decision_router    - id: decision_router
      kind: router
      router:
        impl: your_package.agents.routers.route_after_agent_one
        reads: [result]
        edges:
          agent_two: agent_two
          human_agent: human_agent

Take a router block. A conditional branch, normally a pile of if state["route"] == ... checks, collapses into a function that returns a label plus an edges map from each label to its next node, three lines in total. Policy for routing stays as code, and the topology of routing becomes data.

Generating the state class at runtime

LangGraph wants a typed state, normally a hand-written TypedDict that must stay in step with reality. Here it's generated from the state_schema block:

# your_package/orchestrator/schema.py
annotations = {}
for key, value in state_schema.items():
    type_str = value.get("type", "str")
    # Convert YAML string to Python type
    py_type = eval(type_str)
    if value.get("annotated_with") == "add_messages":
        py_type = Annotated[list, {}]
    annotations[key] = py_type

The result is handed to the spec:

spec = Spec(
    ...
    state=TypedDict("State", annotations),   # <- dynamic class, born at boot
    ...
)

Two things are going on. The functional form TypedDict("State", annotations) manufactures a new type from a dict of field names and types, so the state is built by the loader on every boot rather than maintained by hand. Meanwhile annotated_with: add_messages plugs into LangGraph's reducers, where a field accumulates across nodes (chat history, say) and isn't overwritten. Declaring this in YAML spares whoever designs the conversation from knowing what Annotated[list, add_messages] means.

A caution about eval: the loader uses py_type = eval(type_str), which evaluates a string from a config file. That's acceptable only because config.yaml is trusted and owned by developers. If untrusted parties could supply the schema, swap it for an explicit allow-list such as {"str": str, "list": list, "dict": dict, "int": int}, a one-line change.

The benefit is that state definition and agent topology share a file, so they're reviewed and changed together. The bug where a field was added to the graph but not the TypedDict goes away.

Agents as dotted strings

In the YAML, each agent is only a string:

impl: your_package.agents.agent_one.agent_one_fn

No import statement ties the graph to that function. The loader resolves it with importlib:

# your_package/orchestrator/schema.py
def _import_from_path(dotted: str) -> Callable[..., Any]:
    """Import a callable from a dotted path like 'package.module.function'."""
    if not dotted or "." not in dotted:
        raise ImportError(f"Invalid impl path: {dotted!r}")
    mod_path, attr = dotted.rsplit(".", 1)
    mod = importlib.import_module(mod_path)
    fn = getattr(mod, attr)
    if not callable(fn):
        raise TypeError(f"Imported object is not callable: {dotted}")
    return fn

def agent_impl_map(spec: Spec) -> Dict[str, Optional[Callable]]:
    """Map agent name -> callable (or None if impl missing)."""
    return {a.name: _import_from_path(a.impl) if a.impl else None
            for a in spec.agents}

This is the plugin pattern. To add an agent you write the function and add three YAML lines pointing at it, never touching the graph builder, and the framework finds your code by name at runtime, as a plugin host does. Routers resolve in exactly the same way through router_impl_map.

The compiler

The core is GraphRunner.build(), which walks the spec and builds the actual StateGraph:

# your_package/orchestrator/runner.py
def build(self):
    graph = StateGraph(state_schema=self.spec.state)   # our generated TypedDict

    def _add_task_node(node):
        async def _node(state: Dict[str, Any]) -> Dict[str, Any]:
            res = await self._call_agent(node.agent, state, node.id)
            if getattr(node, "writes", None):
                if isinstance(res, dict):
                    # Only let the node write the keys it declared in YAML
                    filtered = {k: v for k, v in res.items() if k in node.writes}
                    return filtered or res
                key = node.writes[0]
                return {key: res}
            return res
        graph.add_node(node.id, _node)    # Build every node
    for node in self.spec.workflow.nodes:
        if getattr(node, "router", None):
            _add_router_node(node)
        else:
            _add_task_node(node)    graph.set_entry_point(entry)    # Inline "next:" edges from YAML become static edges
    for node in self.spec.workflow.nodes:
        if getattr(node, "next", None):
            graph.add_edge(node.id, node.next)    # Terminal nodes wire to END
    for node in self.spec.workflow.nodes:
        if getattr(node, "terminal", False):
            graph.add_edge(node.id, END)    self._runnable = graph.compile(checkpointer=self.checkpoint)
    return self

A few points deserve attention. The writes key works as an access-control list, and isn't merely documentation. When a node declares writes: [query, result], the compiler cuts the agent's returned dict down to exactly those keys, so an agent can't corrupt state fields it didn't declare. The spec becomes a contract that is actually enforced. There are three declarative edge types: explicit edges, the inline next: shorthand, and terminal: true, which connects to LangGraph's END automatically. After compiling, the runner also renders the graph to a PNG with draw_mermaid_png(), so the diagram can't go stale: it comes from the same source that runs in production.

Routers show the imperative-to-declarative gain most clearly:

# your_package/orchestrator/runner.py
def _add_router_node(node):
    router = self.router_fns[node.id]

    def _router_fn():
        def _f(state):
            out = router(state)
            # Routers may return either a label, or (state_updates, label)
            if isinstance(out, tuple):
                updates, label = out
                if isinstance(updates, dict):
                    for k, v in updates.items():
                        state[k] = v
            else:
                label = out
            return label
        return _f    graph.add_node(node.id, lambda s: {})
    graph.add_conditional_edges(node.id, _router_fn(), node.router.edges)

A router function returns a label, a plain string such as "next_agent", and LangGraph's add_conditional_edges looks it up in the YAML map under node.router.edges to find the next node. The router knows outcomes, not node IDs, so changing where an outcome leads means editing the edges map and leaving the router code alone. A router may also return (updates, label), adjusting state and choosing a branch in one return without a separate node to record the decision.

Adaptive calls

When 17 agents are written by different people at different times, signatures will be inconsistent. One developer writes async def agent(state):, another async def agent(query):, a third async def agent(**kwargs):. A normal graph would force conformity or hand-written adapters. Here the runner inspects each function's signature at call time:

# your_package/orchestrator/runner.py
async def _adapt_and_call(self, fn, state, node_id):
    """
    Adaptively call agent functions so implementations receive what they expect:
    - def agent(**kwargs):        → pass **state (+ inject 'query' if missing)
    - def agent(query, **kwargs): → pass query=..., plus any **extra
    - def agent(state):           → pass state
    - def agent(query):           → pass query
    - def agent():                → call without args
    """
    sig = inspect.signature(fn)
    params = sig.parameters
    has_var_kw = any(p.kind == inspect.Parameter.VAR_KEYWORD
                     for p in params.values())

    kwargs = {}
    if has_var_kw:
        kwargs.update(state)
    if "state" in params:
        kwargs["state"] = state
    if "query" in params or has_var_kw:
        kwargs.setdefault("query", self._fallback_query(state))    # A lone positional 'query' → call it positionally
    if (len(params) == 1
            and next(iter(params.keys())) == "query"):
        return await _maybe_await(fn(self._fallback_query(state)))    res = fn(**kwargs)
    return await res if hasattr(res, "__await__") else res

inspect.signature reveals the parameters a function declares, and the runner assembles exactly that call. Sync and async are handled uniformly: if the result is awaitable it's awaited, otherwise used directly. Authors write whichever signature reads naturally. A _fallback_query helper adds a convenience: if an intermediate node failed to carry the original query forward in state, the runner falls back to the seed state captured when the run began, so any agent can still get the user's question.

Swapping a node per session

Treating the graph as data pays off again with human-in-the-loop steps. A single node, human_agent, halts the run, puts a clarifying question to the user through a WebSocket and blocks until an answer arrives. The trouble is that every session has its own socket, which can't be fixed into a graph compiled once at startup. Because agents are just entries in the dict self.agent_fns, the runner swaps that one node's implementation for each live session, installing a handler bound to that user's connection:

# your_package/services/session_service.py (paraphrased)
if websocket is not None:
    session_handler = SessionHandler(websocket, user_id=user_id, session_id=session_id, ...)
    _runner.agent_fns["human_agent"] = _import_from_function(
        make_input_method(session_handler)
    )

The topology stays fixed and precompiled while one node's behavior is replaced per request. A stable structure with replaceable implementations falls out of the design nearly for free.

What changes

Compare the two approaches by concern:

• Adding an agent: edit the graph builder, imports and tests, versus three lines of YAML.
• Reshaping the flow: rewrite add_edge and add_conditional_edges calls, versus editing next or edges.
• State schema: a hand-written TypedDict kept in sync by discipline, versus one generated from state_schema.
• Routing logic: if/elif buried in graph code, versus a label-returning function plus a YAML map.
• State safety: a convention that agents shouldn't write certain keys, versus enforcement through writes filtering.
• Architecture diagram: drawn by hand and soon stale, versus rendered automatically from the live graph.
• Who can change the flow: someone fluent in the graph API, versus anyone who reads YAML.

The last point is the biggest. A workflow specification that product managers can read, that a reviewer unfamiliar with LangGraph can check in a pull request and that diffs as plain text means the diff itself is the change. Nobody has to mentally execute 40 lines of reorganized Python.

Multi-agent frameworks provide nodes and edges, the mechanism, but not a tidy place for policy, meaning the shape of the workflow kept apart from what each step does. If an agent graph has become wiring that everyone fears to touch, try pulling the topology out into data. The loader takes only a few hundred lines: a dynamic TypedDict, importlib for plugins, a signature-adaptive call, and loops translating nodes and edges into graph calls. A month on, when someone wants a step added before analysis, the answer is a three-line YAML diff, with no afternoon lost in the graph builder.