Subgraphs and Multi-Agent Systems: Supervisor, Swarm, and Handoffs (LangGraph in Production, Module 12)

This is Module 12 of LangGraph in Production, a free 16-module course that takes you from your first StateGraph to a deployed, durable autonomous coding agent on LangGraph v1.x. Start at Module 1 or browse the full syllabus.

---

Picture a workshop where a single worker does everything: reads the job spec, draws the blueprint, cuts the lumber, assembles the pieces, runs the quality check, and fills out the paperwork — alone, on one table, with one set of tools. Nothing wrong with that for a small job. But the moment you want to add a second review station (so bad work never ships) or bring in a documentation specialist, you have to stop the whole line and rearrange the shop floor.

That is Forge after Module 11. forge/graph.py wires every node into one flat graph. forge/nodes.py houses every function from triage through the test-fix loop. There is no Reviewer. There is no Supervisor routing between specialists. And there is no way to test the planning logic without spinning up the entire pipeline.

After this module, Forge runs like a small factory. The edit-test-fix work lives inside an isolated coder subgraph you can test offline in three lines. A hand-built Supervisor reads the state and dispatches to specialists — Coder, Tester, Reviewer — via Command handoffs. The Reviewer examines the edits and either approves them or sends the work back. A frozen review: Review | None field enters ForgeState. And you understand exactly when to reach for a supervisor, a swarm, a hierarchy, or a free-form network — and why the langgraph-supervisor and langgraph-swarm packages are not the answer in a v1.x project.

---

In this module

You’ll learn how to:

• Compose a shared-key subgraph: add a CompiledStateGraph directly as a node when parent and subgraph share state keys (CA12.1).
• Wrap a transform-wrapper subgraph: write a node-function that maps ForgeState → SubState, invokes the subgraph, and maps back (CA12.2).
• Read subgraph checkpoints and stream subgraph events: understand checkpoint_ns, and use subgraphs=True to get (namespace_tuple, chunk) tuples (CA12.3).
• Choose the right multi-agent topology — supervisor, swarm, hierarchical, network — from a comparison table you can actually use (CA12.4).
• Build a Supervisor by hand that routes via Command(goto=...), and write subgraph-to-parent handoffs via Command(graph=Command.PARENT, goto=...) (CA12.5).
• Explain why langgraph-supervisor / langgraph-swarm are broken-forward with create_agent v1 and what the correct v1.x path looks like (CA12.6).
• Situate RemoteGraph as a node: same interface as a compiled subgraph, points at a deployed graph — theory here, BUILD in Module 14 (CA12.7).

You’ll build: The Reviewer specialist agent and an isolated coder subgraph that packages the edit-test-fix loop. A hand-built Supervisor demonstrated with a two-specialist team shows the exact routing pattern. The frozen Review model enters forge/models.py and review: Review | None enters ForgeState.

Concepts covered: CA12 — Composition & multi-agent (core). This module BUILDs CA12.1–CA12.7 of the LangGraph theory inventory.

Prerequisites: Modules 1–11 complete. The forge/ package from Module 11 (state through pull_request, nodes with get_stream_writer, graph with checkpointer + store + durability= + RetryPolicy + CachePolicy). .env with ANTHROPIC_API_KEY (or FORGE_MODEL_PROVIDER=ollama for $0). pytest~=8, uv.

---

Progress checkpoint

• M1–M11 ✅ — Forge is a working monolith: stateful, durable, checkpointed, streamed, with long-term memory.
• M12 👉 — Forge gains a Reviewer specialist, a coder subgraph, and a hand-built Supervisor. The review field enters ForgeState.
• M13–M16 ⬜ — Functional API, deployment, LangSmith evaluation, production review.

State of Forge entering M12: One flat graph, all logic in nodes.py and graph.py, no Reviewer, no Supervisor, review absent from ForgeState.

State of Forge leaving M12: A Reviewer specialist validates edits before the PR interrupt. An isolated coder subgraph is testable offline. A hand-built supervisor demonstrates the routing pattern. The Review model and ForgeState.review field are frozen for all future modules.

Teaser M13: The Functional API (@entrypoint / @task) offers an imperative alternative to the node-and-edge style — Module 13 compares them side by side without rewriting Forge.

Teaser M14: RemoteGraph (theory in this module) gets its BUILD in Module 14 when we deploy Forge on LangGraph Platform.

---

Part 1: Subgraphs — Decomposing a Monolith into Composable Units

The problem with flat graphs

After Module 11, forge/nodes.py is several hundred lines long. It contains intake, triage, recall, planner, plan approval, dispatch, editor, apply edits, tester, fixer, and now the Reviewer. forge/graph.py wires all of it into a single StateGraph. That is not a quality problem yet, but it becomes one the moment you want to:

• Test the edit-test-fix loop in isolation — you cannot; you must invoke the entire graph.
• Add the Reviewer without touching every existing connection.
• Replace the Coder with a different implementation without breaking the graph topology.
• Reuse the planning logic in another graph.

A subgraph solves this. A subgraph is a CompiledStateGraph that you add as a single node inside a parent graph. From the parent’s perspective, it is opaque: it receives state, does its work across multiple internal steps, and returns state. You can test it offline, replace it, and evolve it independently.

LangGraph supports two patterns depending on whether the subgraph and the parent share state keys.

Pattern 1: Shared-key subgraph

When the subgraph uses the same state type as the parent (or a strict subset of the same keys), you add it directly:

# forge/subgraphs.py — the coder loop as a standalone compiled graph
from langgraph.graph import END, START, StateGraph
from langgraph.types import RetryPolicy

from forge import nodes
from forge.state import ForgeState


def build_coder_subgraph():
    """The edit -> test -> fix loop as a compiled subgraph (shared ForgeState)."""
    builder = StateGraph(ForgeState)
    builder.add_node("start_coder", lambda s: {})  # entry passthrough
    builder.add_node("editor", nodes.editor)
    builder.add_node("apply", nodes.apply_edits, defer=True)
    builder.add_node(
        "tester", nodes.tester,
        retry_policy=RetryPolicy(max_attempts=3, retry_on=Exception),
    )
    builder.add_node("fixer", nodes.fixer)

    builder.add_edge(START, "start_coder")
    builder.add_conditional_edges("start_coder", nodes.dispatch, ["editor"])
    builder.add_edge("editor", "apply")
    builder.add_edge("apply", "tester")
    builder.add_conditional_edges(
        "tester", nodes.route_after_tests,
        {"done": END, "give_up": END, "fixer": "fixer"},
    )
    builder.add_edge("fixer", "tester")
    return builder.compile()

Because build_coder_subgraph() returns a CompiledStateGraph that reads and writes ForgeState — the same type as the parent — you can add it as a node directly:

# In the parent graph:
coder_sub = build_coder_subgraph()
builder.add_node("coder_sub", coder_sub)   # shared-key: drop it in directly

LangGraph propagates the shared keys automatically. The subgraph reads repo_path, issue, targets, and writes edits, test_result, attempts into the same channels the parent uses. No mapping code required.

The payoff for testability is immediate. This three-liner runs the entire edit-test-fix loop offline against a real Tally sandbox — no parent graph, no checkpointer:

# tests/test_module_12.py — the coder subgraph runs standalone
sub = build_coder_subgraph()
out = sub.invoke({
    "issue": "report crashes on empty category",
    "repo_path": repo,
    "targets": [FileTarget(path="report.py", instruction="handle empty category")],
})
assert out["test_result"].passed is True

Pattern 2: Transform-wrapper subgraph

When the subgraph needs a smaller or different state type — say, a Reviewer that only cares about edits and test_result, not the full ForgeState — you write a transform-wrapper node. The wrapper maps from the parent’s state before invoking the subgraph and maps back after:

# Sketch — the actual Reviewer in forge/nodes.py is a plain node, not a subgraph wrapper.
# This illustrates the pattern for when you extract a Reviewer into its own subgraph later.
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from forge.models import FileEdit, TestResult, Review


class ReviewerState(TypedDict):
    edits: list[FileEdit]
    test_result: TestResult | None
    review: Review | None


def reviewer_wrapper(state: ForgeState) -> dict:
    """Transform-wrapper: maps ForgeState → ReviewerState, invokes, re-maps."""
    sub_input = {
        "edits": state.get("edits", []),
        "test_result": state.get("test_result"),
        "review": None,
    }
    sub_output = reviewer_subgraph.invoke(sub_input)   # compiled ReviewerState graph
    return {"review": sub_output["review"]}            # writes back to ForgeState


# parent_builder.add_node("reviewer", reviewer_wrapper)

The parent graph has no idea that reviewer_wrapper contains an entire subgraph. That is the point: the boundary is the function signature, not the internal complexity.

The diagram below shows both patterns side by side:

graph LR
  subgraph "Shared-key pattern"
    FS["ForgeState"] -- "shared channels" --> CSG["coder_subgraph\n(CompiledStateGraph)"]
    CSG -- "writes edits, test_result, attempts" --> FS
  end

  subgraph "Transform-wrapper pattern"
    FS2["ForgeState"] -- "edits, test_result" --> RW["reviewer_wrapper\n(plain function)"]
    RW -- "ReviewerState.invoke()" --> RSG["reviewer_subgraph\n(CompiledStateGraph)"]
    RSG -- "review" --> RW
    RW -- "{'review': ...}" --> FS2
  end

When to use which: Shared-key works when the subgraph is a natural phase of the same workflow and shares most state keys. Transform-wrapper works when you want to isolate the subgraph behind a clean interface with a smaller state contract — better for reuse, easier to test in total isolation.

---

Part 2: Subgraph Checkpoints and Streaming

Checkpoint namespaces

When LangGraph executes a subgraph inside a checkpointed parent, it writes subgraph checkpoints under a distinct checkpoint_ns. If the parent node is named coder_sub, the Coder subgraph’s checkpoints land under checkpoint_ns="coder_sub:<uuid>". The Reviewer’s checkpoints land under checkpoint_ns="reviewer:<uuid>". The parent’s own checkpoints use an empty namespace "".

This matters for resumability. If the Coder subgraph crashes mid-execution, a resume picks up from inside the Coder, not from the beginning of Forge. That is checkpoint_ns doing its job — each subgraph execution is its own isolated checkpoint scope.

You can observe this after a multi-subgraph run:

# After graph.invoke(..., {"configurable": {"thread_id": "t1"}}):
history = list(graph.get_state_history({"configurable": {"thread_id": "t1"}}))
for snap in history:
    ns = snap.config.get("configurable", {}).get("checkpoint_ns", "")
    print(f"ns={ns!r:40s}  node={snap.next}")
# Output:
#   ns=''                  node=('intake',)   ← parent checkpoint
#   ns='coder_sub:abc123'  node=('tester',)   ← subgraph checkpoint
#   ns=''                  node=('reviewer',) ← back in parent

The acquis from Module 6 (checkpoint_ns / checkpoint_id) applies directly here. Each namespace is independently addressable: you can call graph.get_state(config) with a checkpoint_ns to inspect a subgraph’s internal state at a past point.

Streaming subgraph events

The subgraphs=True flag you passed to graph.stream() in Module 11 now has real subgraphs to surface. Without subgraphs, every event comes as (mode, chunk). With subgraphs=True, each event becomes (namespace_tuple, mode, chunk) — where namespace_tuple is () for the parent and ("coder_sub:<uuid>",) for a subgraph:

async for namespace, chunk in graph.astream(
    input_state, config, subgraphs=True
):
    if namespace == ():
        print(f"[parent]    {chunk}")
    elif namespace[0].startswith("coder_sub:"):
        print(f"[coder]     {chunk}")
    elif namespace[0].startswith("reviewer:"):
        print(f"[reviewer]  {chunk}")

The synchronous .stream() form is identical; swap async for for for. In Module 11 you used this flag with a flat graph — the namespace tuple was always (). Now it carries real routing information.

---

Part 3: Four Multi-Agent Topologies — Choose Your Architecture

Common misconception — langgraph-supervisor and langgraph-swarm

You might reach for from langgraph_supervisor import create_supervisor or from langgraph_swarm import create_swarm. Do not. Both packages are broken-forward with create_agent v1 from langchain.agents. The breakage is tracked in GitHub issue langgraphjs#1739 and manifests as interface mismatches between the new create_agent signature and the prebuilt supervision abstractions. There is no patched release of those packages that works cleanly with a v1.x project as of June 2026.

The correct path in v1.x is to build the supervisor by hand: a single node that reads state and returns Command(goto=...). That is fewer than 20 lines of code and gives you full control. The packages langgraph-supervisor and langgraph-swarm do not appear anywhere in the Forge codebase — not even in pyproject.toml.

(If you see a tutorial that imports from langgraph_supervisor import ... alongside from langchain.agents import create_agent, it was written for a pre-v1.x stack and is broken by design.)

With that settled, here are the four topologies and when each one earns its keep:

Topology	Structure	Handoff mechanism	Use when	Forge decision
Supervisor	One central node decides who goes next — deterministic logic or an LLM call	Command(goto="worker") from the supervisor node	Routing decisions are centralized; transitions are predictable or rule-based	Forge M12: Supervisor routes Coder → Tester → Reviewer via Command
Swarm	Each agent decides for itself who gets the next turn	Command(goto="other_agent") from the current agent	Agents have overlapping expertise and can negotiate locally; routing is emergent	Forge doesn’t need this — the transitions follow a clear sequence
Hierarchical	Supervisors of supervisors — layered routing	Command(graph=Command.PARENT, goto=...) to climb one level; nested StateGraph compilation	Large systems with multiple specialist groups, each with their own coordination	Possible Forge extension: a top-level Supervisor over a Testing Supervisor and a Review Supervisor
Network	Free-form, any agent can message any other	Command(goto=any_node) with global state access	Truly exploratory workflows where all agents need to see and respond to everything	Rarely recommended — hard to debug, hard to reason about convergence

Forge uses the supervisor topology. The Supervisor is a deterministic node — no LLM call needed — that reads test_result, review, and attempts from ForgeState and routes to the next specialist. Simple, auditable, testable.

The swarm topology is tempting when you want agents to feel autonomous, but autonomy without structure is a debugging nightmare. If you cannot answer “why did agent X hand off to agent Y on this run?”, you will lose days chasing non-deterministic failures. Keep the routing logic explicit until the problem genuinely demands emergent coordination.

---

Part 4: Building the Supervisor and Handoffs by Hand

A concrete supervisor: two-specialist team

forge/supervisor.py demonstrates the supervisor pattern with a two-specialist team (researcher and writer). This is the exact mechanism used for any hand-built supervisor:

# forge/supervisor.py — hand-built supervisor, NO langgraph-supervisor/-swarm (broken-forward)
from __future__ import annotations

import operator
from typing import Annotated, Literal, TypedDict

from langgraph.graph import END, START, StateGraph
from langgraph.types import Command

_ORDER = ["researcher", "writer"]


class TeamState(TypedDict, total=False):
    task: str
    log: Annotated[list, operator.add]   # fan-in accumulator
    done: list


def _supervisor(state: TeamState) -> Command[Literal["researcher", "writer", "__end__"]]:
    remaining = [s for s in _ORDER if s not in state.get("done", [])]
    return Command(goto=(remaining[0] if remaining else END))


def _specialist(name: str):
    def node(state: TeamState) -> Command[Literal["supervisor"]]:
        return Command(
            goto="supervisor",
            update={"log": [f"{name} ran"], "done": [*state.get("done", []), name]},
        )
    return node

The supervisor node returns Command(goto=...) — not a string, not a dict, but a Command object. That is the v1.x handoff primitive. The annotation -> Command[Literal["researcher", "writer", "__end__"]] documents the reachable destinations so the graph renderer can draw the edges.

Each specialist returns Command(goto="supervisor", update={...}) — it hands control back to the supervisor and updates state atomically in one step. The supervisor then re-reads the updated state and decides the next destination.

The graph assembly is the same as any StateGraph:

def build_supervisor():
    builder = StateGraph(TeamState)
    builder.add_node("supervisor", _supervisor)
    builder.add_node("researcher", _specialist("researcher"))
    builder.add_node("writer", _specialist("writer"))
    builder.add_edge(START, "supervisor")
    return builder.compile()

The test verifies routing:

# tests/test_module_12.py
def test_hand_built_supervisor_routes_specialists():
    graph = build_supervisor()
    out = graph.invoke({"task": "demo", "log": [], "done": []})
    assert out["log"] == ["researcher ran", "writer ran"]
    assert set(out["done"]) == {"researcher", "writer"}

The distinction between in-graph and parent handoffs

Two forms of Command handle routing, and confusing them is the most common bug in multi-agent code:

• Command(goto="supervisor") — routes within the current graph. Use this when a specialist node hands control back to the supervisor in the same graph (the pattern above).
• Command(graph=Command.PARENT, goto="next_node") — routes from inside a subgraph back to the parent graph. Use this in the final node of a subgraph when the subgraph is embedded in a larger graph and needs to signal completion upstream.

If a subgraph’s final node returns Command(goto="supervisor") when the supervisor lives in the parent, the routing stays trapped inside the subgraph and the parent never regains control. The graph appears to hang. The fix is Command(graph=Command.PARENT, goto="supervisor").

Forge’s Reviewer: the real specialist added to the pipeline

In the actual Forge pipeline for Module 12, the Reviewer is a specialist node inserted after the test-fix loop. It uses create_agent with response_format=Review — the same pattern the Planner uses:

# forge/agents.py — the Reviewer specialist (added at Module 12)
from langchain.agents import create_agent
from forge import config, tools
from forge.models import Review

REVIEWER_SYSTEM = (
    "You are Forge's code reviewer (a specialist agent). Given the issue and the edits "
    "made, read the changed files with your tools and decide whether the change is correct "
    "and minimal. Return approved + a short list of comments."
)


def build_reviewer(repo_path: str, model=None):
    model = model or config.get_model("smart")
    return create_agent(
        model,
        tools=tools.make_read_tools(repo_path),
        system_prompt=REVIEWER_SYSTEM,
        response_format=Review,
    )


def review_change(repo_path: str, edits, model=None) -> Review:
    agent = build_reviewer(repo_path, model)
    summary = "\n".join(f"- {e.path}: {e.summary}" for e in edits) or "(no edits)"
    result = agent.invoke({"messages": [{"role": "user", "content": f"Edits:\n{summary}\nReview them."}]})
    review = result.get("structured_response")
    if review is None:
        raise RuntimeError("reviewer returned no structured_response")
    return review

The reviewer node in forge/nodes.py calls this seam:

# forge/nodes.py — Module 12 addition
def reviewer(state: ForgeState) -> dict:
    """A second specialist agent: review the edits, then ship or request rework."""
    review = agents.review_change(state["repo_path"], state.get("edits", []))
    verdict = "approved" if review.approved else "changes requested"
    return {"review": review, "messages": [AIMessage(content=f"Reviewer: {verdict}")]}


def route_after_review(state: ForgeState) -> str:
    """Handoff: ship if approved (or budget spent), else hand back to the fixer."""
    review = state.get("review")
    if (review and review.approved) or state.get("attempts", 0) >= MAX_ATTEMPTS:
        return "ship"
    return "rework"

The routing function route_after_review has a practical guard: if the attempt budget is spent, the code ships regardless of the Reviewer’s verdict. Infinite rework loops are worse than imperfect code in a production-bound agent.

The Review model and ForgeState.review — frozen contracts

The Review Pydantic model is introduced at Module 12 and frozen from this point on. Exactly two fields:

# forge/models.py — added at Module 12
class Review(BaseModel):
    """The Reviewer specialist's verdict — frozen at Module 12."""

    approved: bool
    comments: list[str]

No score, no confidence, no severity. If you add fields to Review after Module 12, you break the downstream modules that read it.

ForgeState gains one field, appended at the end:

# forge/state.py — Module 12 addition, appended to the 11 existing fields
review: Review | None    # M12 — verdict of the Reviewer specialist (absent before M12)

The eleven fields introduced in Modules 2 through 7 (issue, messages, repo_path, route, attempts, test_result, plan, targets, edits, plan_approved, pull_request) are untouched. edits: Annotated[list[FileEdit], operator.add] remains the only field with a custom reducer.

Forge’s complete graph topology after Module 12

Here is the full pipeline after Module 12, including the Reviewer and the conditional handoff back to the Fixer on rejection:

flowchart TD
    START --> intake
    intake --> triage
    triage -->|"route"| recall
    recall --> planner
    planner --> plan_approval["plan_approval\n⏸ interrupt()"]
    plan_approval -->|"approved"| editor["editor × N\n(Send fan-out)"]
    plan_approval -->|"rejected"| abandon
    editor --> apply["apply\n(defer=True)"]
    apply --> tester
    tester -->|"passed=True"| reviewer
    tester -->|"passed=False & budget"| fixer
    fixer --> tester
    reviewer -->|"approved=True"| pr_approval["pr_approval\n⏸ interrupt()"]
    reviewer -->|"approved=False"| fixer
    pr_approval --> END
    abandon --> END

Notice that the Reviewer has its own exit back to the Fixer. The Fixer re-applies edits, the Tester runs again, and if tests are still green the Reviewer gets another look. This is a bounded handoff loop: route_after_review ships unconditionally when attempts >= MAX_ATTEMPTS, so the loop cannot run forever.

forge/graph.py assembles this by adding reviewer as a node and wiring two conditional edges:

# forge/graph.py (excerpt) — Module 12 wires the Reviewer into the pipeline
builder.add_node("reviewer", nodes.reviewer)
builder.add_conditional_edges(
    "tester", nodes.route_after_tests,
    {"done": "reviewer", "give_up": END, "fixer": "fixer"},
)
builder.add_conditional_edges(
    "reviewer", nodes.route_after_review,
    {"ship": "pr_approval", "rework": "fixer"},
)

---

Part 5: The Frozen Models, the Store, and RemoteGraph

Why review_change is a LLM seam

Every LLM call in Forge is isolated behind a function that tests can monkeypatch. The review_change function in forge/agents.py is the Reviewer’s seam. Offline tests swap it for a lambda that returns a canned Review:

# tests/test_pipeline.py — offline replacement of the Reviewer seam
monkeypatch.setattr(
    agents, "review_change",
    lambda repo_path, edits, model=None: Review(approved=True, comments=[]),
)

The test test_reviewer_rework_then_ship pushes the rework loop: the Reviewer rejects once, then approves on the second call. This exercises the full reviewer → fixer → tester → reviewer → ship path without any API key:

def test_reviewer_rework_then_ship(monkeypatch, make_routed_model):
    calls = {"n": 0}

    def review(repo_path, edits, model=None):
        calls["n"] += 1
        return Review(
            approved=calls["n"] >= 2,
            comments=[] if calls["n"] >= 2 else ["tighten it"],
        )

    monkeypatch.setattr(agents, "review_change", review)
    # ... run the graph ...
    assert calls["n"] >= 2   # Reviewer ran twice: reject, then approve

That is the concrete proof that the handoff loop works: the Reviewer got a second chance because the Supervisor (the conditional edge in route_after_review) sent the work back to the Fixer.

RemoteGraph: theory now, BUILD in Module 14

Once a subgraph is deployed to LangGraph Platform, you can use it from another graph through RemoteGraph. The interface is identical to a CompiledStateGraph:

# Theory only — not in the lab code, BUILD in Module 14
from langgraph.pregel.remote import RemoteGraph

remote_planner = RemoteGraph(
    "planner-graph",
    url="https://your-langgraph-server.com",
    api_key="...",
)
parent_builder.add_node("planner", remote_planner)   # exact same syntax as a compiled subgraph

RemoteGraph supports .invoke(), .stream(), and .get_state(). From the parent graph’s perspective, it is indistinguishable from a local subgraph. This is the architectural foundation for decomposing Forge into independently deployed services — covered in Module 14.

---

Hands-On Lab: Module 12

Lab path: langgraph-prod-labs/module-12/

The lab has two deliverables: the coder subgraph (extracting the edit-test-fix loop into a standalone CompiledStateGraph) and the Reviewer specialist (adding the Review model, ForgeState.review, the reviewer node, and the route_after_review conditional edge to the main pipeline). The hand-built supervisor in forge/supervisor.py is a standalone demonstration of the routing pattern.

Step 1: Add the Review model to forge/models.py

Append exactly two fields — nothing more. The prior models (TestResult, Plan, PlanStep, FileTarget, FileEdit, PullRequest) are untouched:

# forge/models.py — append after PullRequest
class Review(BaseModel):
    """The Reviewer specialist's verdict — frozen at Module 12."""

    approved: bool
    comments: list[str]

Verify the contract:

# tests/test_module_12.py
def test_review_contract():
    r = Review(approved=True, comments=["looks good"])
    assert r.approved and r.comments == ["looks good"]

Step 2: Add review to ForgeState

Append the field at the end of the TypedDict. The existing fields — including edits: Annotated[list[FileEdit], operator.add] — are untouched:

# forge/state.py — append as the 12th field
from forge.models import FileEdit, FileTarget, Plan, PullRequest, Review, TestResult

class ForgeState(TypedDict, total=False):
    # ... fields M2 through M7, unchanged ...
    review: Review | None    # M12 — verdict of the Reviewer (absent before M12)

Step 3: Add the Reviewer seam to forge/agents.py

Add build_reviewer and review_change alongside the existing build_planner / plan_change. Import Review from forge.models. The create_agent call with response_format=Review is the same pattern as the Planner — use tier "smart":

# forge/agents.py — add after plan_change
from forge.models import Plan, Review

REVIEWER_SYSTEM = (
    "You are Forge's code reviewer (a specialist agent). Given the issue and the edits "
    "made, read the changed files with your tools and decide whether the change is correct "
    "and minimal. Return approved + a short list of comments."
)


def build_reviewer(repo_path: str, model=None):
    model = model or config.get_model("smart")
    return create_agent(
        model,
        tools=tools.make_read_tools(repo_path),
        system_prompt=REVIEWER_SYSTEM,
        response_format=Review,
    )


def review_change(repo_path: str, edits, model=None) -> Review:
    agent = build_reviewer(repo_path, model)
    summary = "\n".join(f"- {e.path}: {e.summary}" for e in edits) or "(no edits)"
    result = agent.invoke({"messages": [{"role": "user", "content": f"Edits:\n{summary}\nReview them."}]})
    review = result.get("structured_response")
    if review is None:
        raise RuntimeError("reviewer returned no structured_response")
    return review

Step 4: Add the Reviewer node and routing to forge/nodes.py

# forge/nodes.py — Module 12 additions
from forge.models import FileEdit, FileTarget, Plan, PullRequest   # Review imported via agents

def reviewer(state: ForgeState) -> dict:
    """Specialist agent: review the edits; write Review to ForgeState."""
    review = agents.review_change(state["repo_path"], state.get("edits", []))
    verdict = "approved" if review.approved else "changes requested"
    return {"review": review, "messages": [AIMessage(content=f"Reviewer: {verdict}")]}


def route_after_review(state: ForgeState) -> str:
    """Conditional edge: ship if approved or budget spent, else rework."""
    review = state.get("review")
    if (review and review.approved) or state.get("attempts", 0) >= MAX_ATTEMPTS:
        return "ship"
    return "rework"

Verify the seam and routing:

# tests/test_module_12.py
def test_reviewer_node_uses_seam(monkeypatch):
    monkeypatch.setattr(agents, "review_change",
                        lambda repo, edits, model=None: Review(approved=True, comments=[]))
    out = nodes.reviewer({"repo_path": "/r", "edits": []})
    assert out["review"].approved is True

def test_route_after_review():
    assert nodes.route_after_review(
        {"review": Review(approved=True, comments=[])}
    ) == "ship"
    assert nodes.route_after_review(
        {"review": Review(approved=False, comments=["x"]), "attempts": 1}
    ) == "rework"
    # Budget spent → ship regardless
    assert nodes.route_after_review(
        {"review": Review(approved=False, comments=["x"]), "attempts": MAX_ATTEMPTS}
    ) == "ship"

Step 5: Wire the Reviewer into forge/graph.py

Two additions: a builder.add_node and two builder.add_conditional_edges (the Tester’s exit now routes to reviewer on success, and the Reviewer routes to pr_approval or fixer):

# forge/graph.py — Module 12 changes (excerpt)
builder.add_node("reviewer", nodes.reviewer)

builder.add_conditional_edges(
    "tester", nodes.route_after_tests,
    {"done": "reviewer", "give_up": END, "fixer": "fixer"},
)
builder.add_conditional_edges(
    "reviewer", nodes.route_after_review,
    {"ship": "pr_approval", "rework": "fixer"},
)

The build_graph() signature and compile() call are unchanged — checkpointer, store, and cache pass through identically.

Step 6: Build the coder subgraph

Create forge/subgraphs.py with build_coder_subgraph(). Reuse the existing node functions from forge/nodes.py — they live in the subgraph without modification:

# forge/subgraphs.py — the edit-test-fix loop as a standalone compiled graph
from langgraph.graph import END, START, StateGraph
from langgraph.types import RetryPolicy
from forge import nodes
from forge.state import ForgeState


def build_coder_subgraph():
    builder = StateGraph(ForgeState)
    builder.add_node("start_coder", lambda s: {})
    builder.add_node("editor", nodes.editor)
    builder.add_node("apply", nodes.apply_edits, defer=True)
    builder.add_node(
        "tester", nodes.tester,
        retry_policy=RetryPolicy(max_attempts=3, retry_on=Exception),
    )
    builder.add_node("fixer", nodes.fixer)
    builder.add_edge(START, "start_coder")
    builder.add_conditional_edges("start_coder", nodes.dispatch, ["editor"])
    builder.add_edge("editor", "apply")
    builder.add_edge("apply", "tester")
    builder.add_conditional_edges(
        "tester", nodes.route_after_tests,
        {"done": END, "give_up": END, "fixer": "fixer"},
    )
    builder.add_edge("fixer", "tester")
    return builder.compile()

Verify it runs standalone against a real Tally sandbox:

# tests/test_module_12.py — the signal: pytest of Tally passes, no API needed
def test_coder_subgraph_runs_standalone(monkeypatch, make_routed_model):
    tally_src = Path(__file__).resolve().parents[1] / "tally"
    repo = sandbox.make_sandbox(tally_src)
    sandbox.write_file(repo, "tests/test_issue_empty_category.py", REPRO_TEST)
    fixed = (Path(repo) / "report.py").read_text().replace(
        "totals[cat]", "totals.get(cat, 0.0)"
    )
    routed = make_routed_model(
        [("Edit file report.py", fixed)], default="# noop\n"
    )
    monkeypatch.setattr(nodes.config, "get_model", lambda tier="smart", **k: routed)

    sub = build_coder_subgraph()
    out = sub.invoke({
        "issue": "report crashes on empty category",
        "repo_path": repo,
        "targets": [FileTarget(path="report.py", instruction="handle empty category")],
    })
    assert out["test_result"].passed is True

The make_routed_model fixture is a fake chat model that returns canned responses without any API call. The Tally pytest suite runs for real in the sandbox — that is the deterministic signal.

Step 7: Build the hand-built supervisor

forge/supervisor.py demonstrates the routing pattern with a clean two-specialist system. Wire the graph and test it:

# tests/test_module_12.py
def test_hand_built_supervisor_routes_specialists():
    graph = build_supervisor()
    out = graph.invoke({"task": "demo", "log": [], "done": []})
    assert out["log"] == ["researcher ran", "writer ran"]
    assert set(out["done"]) == {"researcher", "writer"}

Running the full test suite

$ uv run pytest -q
...
tests/test_module_01.py   1 passed
tests/test_module_02.py   2 passed
tests/test_module_03.py   4 passed
...
tests/test_module_12.py   5 passed
tests/test_pipeline.py    6 passed
# All modules M1–M12 offline, zero API calls

The full pipeline test test_reviewer_rework_then_ship exercises the rework handoff loop with the Reviewer rejecting once then approving. You can watch the Reviewer’s update field and the Fixer’s second pass in the history:

$ uv run pytest tests/test_pipeline.py::test_reviewer_rework_then_ship -v
PASSED   [100%]

Try it yourself

1. Propagate Reviewer comments to the Fixer. When route_after_review returns "rework", pass review.comments into the messages so the Fixer sees what the Reviewer objected to. Look at how the plan_approval node passes an edited plan back through state — the same pattern applies here.

2. Add a security_check node to the coder subgraph. The parent graph has no knowledge of it. Add it between apply and tester in build_coder_subgraph() without touching forge/graph.py. This is the isolation guarantee subgraphs provide.

3. Observe checkpoint_ns in a live run. After test_hitl_approve_plan_and_pr completes, iterate graph.get_state_history(cfg) and print the checkpoint_ns field of each snapshot’s config. You will see the flat graph’s namespaces (empty string) appear with every other checkpoint — because this graph’s subgraph is a standalone graph invoked inside a node, not an embedded graph node yet. In a fully embedded subgraph, you would see "coder_sub:<uuid>" namespaces interleaved.

---

In Production

Subgraph granularity. The right boundary for a subgraph is where the work is independently testable and deployable. Too fine (one subgraph per node) and you pay checkpoint overhead for nothing. Too coarse (the entire pipeline as one “subgraph”) and you have gained nothing. A practical rule: if you would write a separate pytest test file for a phase of work, it earns a subgraph.

The Supervisor is a policy. In Forge, the Supervisor is three conditional edges — deterministic, no LLM call. In a more exploratory system, you might want an LLM-powered Supervisor that can express uncertainty (“I am not sure whether this needs a Reviewer or more editing”). That is a valid choice but adds latency and cost on every routing decision. Start deterministic; reach for an LLM Supervisor only when the routing logic genuinely requires language understanding.

The rework loop budget. The attempts >= MAX_ATTEMPTS guard in route_after_review is not a compromise; it is a production requirement. An unbounded rework loop can exhaust your token budget on a single issue. In practice, cap the Reviewer’s rework passes at 1 or 2 beyond the initial test-fix budget. Log every rejection with its comments to LangSmith (covered in Module 15) so you can tune the Reviewer’s threshold.

RemoteGraph and service boundaries. When Forge is deployed (Module 14), the Planner subgraph and the Coder subgraph can be deployed as independent LangGraph Platform services. The parent graph switches from coder_sub = build_coder_subgraph() to coder_sub = RemoteGraph("coder-graph", url=...) — one line change. Independent deployment means independent scaling, independent rollback, and independent versioning. That is the production argument for subgraphs beyond testability.

Checkpoint storage at scale. Each subgraph execution writes its own checkpoint rows under a distinct checkpoint_ns. With many subgraphs and high concurrency, this multiplies your checkpoint storage. PostgreSQL checkpointers (available via langgraph-checkpoint-postgres) handle this better than SQLite; plan for it before you hit the SQLite connection-pool limit.

---

Mastery Corner

What to really understand here

Module 12 is about composability: the idea that a large graph is best understood and maintained as a composition of smaller, independently testable graphs. The two key mechanisms are:

• Shared-key subgraph: add a compiled subgraph as a node when state schemas are compatible. Zero mapping code, zero overhead.
• Transform-wrapper: write a node function that maps state in, calls subgraph.invoke(), maps state out. Full schema isolation.

The second theme is handoffs: how agents transfer control in a multi-agent system. In v1.x, every handoff is a Command — either Command(goto="node") within the current graph, or Command(graph=Command.PARENT, goto="node") to escape a subgraph and return control to the parent.

The third theme is the topology decision: supervisor gives you central, auditable routing; swarm gives you emergent, local routing; hierarchical gives you both at scale; network gives you chaos. Choose intentionally.

Finally: langgraph-supervisor and langgraph-swarm are broken-forward with create_agent v1. Build the supervisor by hand. It is 15 lines.

---

Quiz

Question 1. You want to extract the Reviewer into its own subgraph. The Reviewer only needs edits, test_result, and review from ForgeState. Your colleague says: “Just use the shared-key pattern — add the compiled subgraph directly.” What is the trade-off?

A. The shared-key pattern is always wrong when the subgraph only uses a subset of keys.
B. The shared-key pattern works but gives the subgraph read access to the entire ForgeState; a transform-wrapper creates a narrower contract, improves isolation, and makes the subgraph independently testable with a minimal state fixture.
C. The transform-wrapper is required by LangGraph when the subgraph has fewer keys than the parent.
D. The shared-key pattern copies the entire state into the subgraph on every invocation, causing performance issues.

---

Question 2. After adding the Coder subgraph, you notice that when the Coder subgraph finishes its work the Supervisor never regains control — the run appears to hang. You inspect the final node of the subgraph and see:

def coder_done(state):
    return Command(goto="supervisor", update={"edits": state["edits"]})

What is wrong and how do you fix it?

A. The update keyword is not valid in a Command inside a subgraph. Remove it.
B. Command(goto="supervisor") routes within the subgraph; if supervisor is not a node in the subgraph, LangGraph silently drops the command. Change to Command(graph=Command.PARENT, goto="supervisor", update={"edits": state["edits"]}).
C. The goto value must be a list, not a string, when targeting the parent.
D. Subgraphs cannot return Command objects — the node must return a plain dict.

---

Question 3. You want the Reviewer to sometimes hand off directly to the Docs-writer without going back through the Supervisor. Which topology does this reflect, and is it a good idea for Forge?

A. This is the hierarchical topology. It is the recommended pattern whenever two agents need to communicate directly.
B. This is the swarm topology — agents make local handoff decisions. It can work but makes routing harder to audit and debug; the emergent path between Reviewer and Docs-writer is implicit in the Reviewer’s LLM output, not in your graph edges.
C. This is the supervisor topology; all handoffs go through a supervisor by definition.
D. This is the network topology and is only supported via Command(graph=Command.PARENT).

---

Question 4. A developer on your team imports from langgraph_supervisor import create_supervisor in a project that uses from langchain.agents import create_agent (v1). The import succeeds but the graph fails at runtime. What is the root cause?

A. langgraph-supervisor requires Python 3.11 minimum; the project runs 3.10.
B. create_supervisor from langgraph-supervisor is incompatible with the v1 create_agent interface (GitHub issue langgraphjs#1739) — the prebuilt supervision abstraction was written against a pre-v1 create_agent signature and has not been updated. The correct path is to build the supervisor by hand: a node that reads state and returns Command(goto=...).
C. langgraph-supervisor only works with langgraph>=2.0, which is not yet released.
D. The create_supervisor function requires all worker agents to be compiled graphs, not create_agent instances.

---

Question 5. After a multi-subgraph Forge run, you call graph.get_state_history(config). You see checkpoints with checkpoint_ns="" and others with checkpoint_ns="coder_sub:abc123". You cannot find the overall “current state” of the full Forge run. What is happening?

A. The empty-namespace checkpoints are corrupt and should be discarded.
B. The parent graph’s checkpoints live under checkpoint_ns="" (empty string). That is where the overall Forge state is. The coder_sub:abc123 checkpoints are the Coder subgraph’s internal steps. To read the Forge-level state, filter for snapshots where checkpoint_ns == "".
C. The checkpoint_ns field is only set for subgraphs; parent checkpoints have no checkpoint_ns field at all.
D. You need to call graph.get_subgraph_history() separately to see parent checkpoints.

---

Answers

Q1 → B. The shared-key pattern is perfectly valid when the subgraph uses a subset of keys — LangGraph only reads the keys the subgraph schema declares. But it gives the subgraph implicit access to the full ForgeState namespace. The transform-wrapper creates a narrower interface: the Reviewer subgraph only receives edits, test_result, and review, and can be tested with a ReviewerState fixture instead of a full ForgeState. Neither is “wrong”; the choice is about isolation and testability.

Q2 → B. Command(goto="supervisor") routes within the graph that contains the current node. If supervisor is a node in the parent, not in the subgraph, the routing has nowhere to go — the run stalls. The fix is Command(graph=Command.PARENT, goto="supervisor"), which tells LangGraph to exit the current subgraph and resume the parent at the supervisor node.

Q3 → B. Direct agent-to-agent handoffs without a central router are the swarm topology. The concern with swarm is auditability: the path a specific run takes is encoded in LLM outputs, not in graph edges. For Forge, where routing is entirely deterministic (tests pass or fail; review approves or rejects), the supervisor topology keeps every routing decision visible in route_after_review. Use swarm when routing genuinely requires the agent’s domain expertise — not when a simple conditional covers it.

Q4 → B. langgraph-supervisor and langgraph-swarm were built against the pre-v1 create_agent API. The v1 create_agent from langchain.agents has a different signature and the prebuilt packages have not been reconciled with it. This is documented in langgraphjs#1739. The portable solution — and the only one guaranteed to work in a v1.x project — is a hand-built supervisor node that returns Command(goto=...).

Q5 → B. Parent checkpoints use checkpoint_ns="" (the empty string). Subgraph checkpoints use checkpoint_ns="<node_name>:<uuid>". When you iterate get_state_history(), filter on checkpoint_ns == "" to see only the parent-level snapshots. Calling graph.get_state(config) without a checkpoint_ns in the configurable returns the current parent state, which is almost always what you want when debugging a Forge run.

---

Traps to avoid

Trap 1 (v0 → v1 freshness). Using langgraph-supervisor or langgraph-swarm with a create_agent v1 project. These packages are broken-forward (langgraphjs#1739). They do not appear in pyproject.toml, they do not appear in forge/. Build the supervisor by hand: one node, returns Command(goto=...), 15 lines. You gain the same routing behavior and full control over the logic.

Trap 2. Confusing Command(goto="supervisor") and Command(graph=Command.PARENT, goto="supervisor"). The first routes within the current graph — subgraph or parent. The second escapes a subgraph and routes in the parent. Using the first form in a subgraph’s final node when the target node is in the parent is the single most common handoff bug. The symptom is a run that appears to finish (no exception) but the parent never progresses past the subgraph node.

Trap 3. Assuming the shared-key pattern copies state. It does not. The subgraph reads and writes directly into the parent’s state channels. A write to state["plan"] inside the Planner subgraph is immediately visible to the parent graph at the next superstep — it is not buffered or delayed. The checkpoint write happens at the end of the subgraph’s superstep, under its own checkpoint_ns, and the parent reads the updated values when it regains control.

---

Key Takeaways

• A subgraph is a CompiledStateGraph added as a node inside a parent graph. It isolates a phase of work: independently testable, replaceable, and reusable.
• Shared-key pattern: add the compiled subgraph directly when parent and subgraph share state keys — zero mapping code, LangGraph handles channel propagation.
• Transform-wrapper pattern: write a node function that maps parent state to sub-state, calls subgraph.invoke(), and maps back — use when you want a narrower interface or strict schema isolation.
• Subgraph checkpoints write under checkpoint_ns="<node_name>:<uuid>"; the parent’s checkpoints use checkpoint_ns="". Resume after a crash picks up from inside the subgraph.
• Command(goto="node") routes within the current graph; Command(graph=Command.PARENT, goto="node") exits a subgraph and routes in the parent. Confusing these two is the most common multi-agent bug.
• The four topologies — supervisor, swarm, hierarchical, network — differ in where routing decisions are made. Supervisor (centralized, deterministic) is the right default; reach for swarm only when routing genuinely requires emergent, agent-local expertise.
• langgraph-supervisor and langgraph-swarm are broken-forward with create_agent v1 (langgraphjs#1739). Build supervisors by hand: one node, returns Command(goto=...), testable with a normal graph.invoke() call.
• RemoteGraph (from langgraph.pregel.remote import RemoteGraph) gives a deployed graph the same interface as a local compiled subgraph — the bridge between local composition and production deployment, covered in Module 14.

---

What’s Next

Want the full LangGraph mastery question bank (interview-style) and the next module in your inbox? Subscribe here — it’s free, like everything in this series.

Module 13: The Functional API — @entrypoint, @task, and When to Use It. The Graph API you have used since Module 2 is not LangGraph’s only interface. The Functional API lets you write stateful, durable, streamable agent logic as ordinary Python functions — no StateGraph, no explicit edges. Module 13 compares the two approaches on the same Forge problem so you can make an informed choice in your own projects.

---

Links: Module 11 — Streaming | Module 13 — The Functional API | Module 6 — Persistence & Threads | Module 14 — Deployment | Course index | Lab code

---

References

1. LangGraph — How to add and use subgraphs — official guide to both shared-key and transform-wrapper patterns, with streaming examples.
2. LangGraph — Multi-agent architectures — canonical comparison of supervisor, swarm, hierarchical, and network topologies with v1.x Command examples.
3. LangGraph — How to create and manage multi-agent systems — hands-on walkthrough of handoffs via Command and Command(graph=Command.PARENT, ...).
4. LangGraph — RemoteGraph — how to use a deployed graph as a subgraph node; preview of what Module 14 builds.
5. LangGraph — Streaming subgraph events — subgraphs=True with namespace tuple interpretation and checkpoint_ns behavior.
6. LangGraph — Persistence: checkpoint namespaces — technical reference for checkpoint_ns and how subgraph checkpoints are scoped, building on the Module 6 acquis.