# Lab Specification — Deep-Dive DD-10: Build a LangGraph-Style State Machine **Course**: Master Course **Deep-Dive**: DD-10 — LangGraph: Graph-Based State Machines **Duration**: 45–60 minutes (a deterministic simulation) **Environment**: Python 3.10+. No GPU, no external network, no model calls. The lab is a deterministic, type-hinted simulation of a LangGraph-style state machine — you build the three primitives (nodes as functions, edges as a transition table, super-step checkpoints at every node boundary), add an `interrupt()` node for structural HITL, and confirm a prompt-injected agent cannot route around it. Then you add a subgraph and confirm multi-agent coordination via explicit edges. --- ## Learning objectives By the end of this lab you will have: 1. **Built the three LangGraph primitives** — nodes as functions (take state, do work, return a state update), edges as a transition table (including conditional edges), and super-step checkpoints (state serialized at every node boundary). The empirical anchor for "the graph IS the loop." 2. **Demonstrated super-step checkpointing** — crashed the state machine mid-run and resumed from the last completed super-step. The empirical anchor for Module 8 (5/5): granularity is the step, not the session. 3. **Proven the structural HITL property** — added an `interrupt()` node and confirmed a prompt-injected agent cannot route around it, because the model does not control the edges. The empirical anchor for the cleanest HITL in the roster. 4. **Reproduced the graph-fights-the-model anti-pattern** — shown that a rigid conditional edge forces a suboptimal path the model (simulated) would not choose, constraining capability. The empirical anchor for the future-proof test partially failing. 5. **Built a multi-agent subgraph** — embedded one graph as a node inside another and confirmed coordination via explicit declared edges. The empirical anchor for Module 10 and the C4 E09 connection. This lab is the empirical anchor for the deep-dive's central claim: LangGraph does not improve the model — it improves the visibility, resumability, and auditability of the process. The graph is the program. --- ## Phase 0 — Set up (5 min) Create the lab file. No dependencies beyond the standard library. ```bash mkdir -p dd10-lab && cd dd10-lab cat > state_machine.py << 'PYEOF' """DD-10 Lab: Build a LangGraph-Style State Machine. We model the three LangGraph primitives: nodes as functions, edges as a transition table (with conditional edges), and super-step checkpoints at every node boundary. We add an interrupt() node for structural HITL and confirm a prompt-injected agent cannot route around it. Then we add a subgraph for multi-agent coordination via explicit edges. """ from __future__ import annotations from dataclasses import dataclass, field from enum import Enum from typing import Callable, Optional, Any class HITLDecision(str, Enum): PENDING = "PENDING" # interrupt() has fired, awaiting human APPROVED = "APPROVED" REJECTED = "REJECTED" EDITED = "EDITED" @dataclass class GraphState: """The serialized state, checkpointed at every node boundary.""" phase: str = "planning" proposed_action: str = "" verify_result: str = "" hitl_decision: HITLDecision = HITLDecision.PENDING injection_active: bool = False # is a prompt injection compromising the agent? values: dict[str, Any] = field(default_factory=dict) @dataclass class Checkpoint: """A super-step checkpoint: the serialized state at a node boundary.""" node_name: str state: GraphState super_step: int @dataclass class StateMachine: """The LangGraph-style state machine. Nodes are functions; edges are a transition table; state is checkpointed at every node boundary.""" nodes: dict[str, Callable[[GraphState], GraphState]] = field(default_factory=dict) edges: dict[str, list[tuple[Callable[[GraphState], str], str]]] = field(default_factory=dict) checkpoints: list[Checkpoint] = field(default_factory=list) hitl_pending: bool = False # is the machine paused at an interrupt() node? crashed: bool = False def add_node(self, name: str, fn: Callable[[GraphState], GraphState]) -> None: self.nodes[name] = fn def add_conditional_edge(self, from_node: str, conditions: list[tuple[Callable[[GraphState], str], str]]) -> None: """A conditional edge: a list of (predicate, target_node) pairs. The first predicate returning a target wins. The predicate inspects state to decide routing — branching expressed in CODE, not emergent.""" self.edges[from_node] = conditions def _checkpoint(self, node_name: str, state: GraphState) -> None: """Super-step checkpoint: serialize state at a node boundary.""" self.checkpoints.append(Checkpoint( node_name=node_name, state=GraphState(**{**state.__dict__, "values": dict(state.values)}), super_step=len(self.checkpoints), )) def last_checkpoint(self) -> Optional[Checkpoint]: return self.checkpoints[-1] if self.checkpoints else None def run(self, start: str, state: GraphState, max_steps: int = 20) -> tuple[str, GraphState]: """Run the graph from a start node. At each node: execute, checkpoint, then route via conditional edges. Stops at END or at an interrupt().""" current = start for _ in range(max_steps): if current == "END": return ("completed", state) if current not in self.nodes: return ("error_missing_node", state) # Execute the node (the function does work, returns a state update) state = self.nodes[current](state) # Super-step checkpoint: serialize state at this node boundary self._checkpoint(current, state) # interrupt(): if the node set hitl_pending, pause indefinitely if self.hitl_pending: return ("interrupted", state) # Route via conditional edges (the graph definition controls routing) if current in self.edges: for predicate, target in self.edges[current]: next_node = predicate(state) if next_node: current = target break else: return ("error_no_route", state) else: return ("error_no_edge", state) return ("max_steps", state) def resume_from_last_checkpoint(self, human_decision: HITLDecision) -> tuple[str, GraphState]: """Resume from the last checkpoint after a human decision at interrupt(). This is Module 6.2 (HITL) + Module 8 (checkpointing) combined.""" ckpt = self.last_checkpoint() if ckpt is None: return ("no_checkpoint", GraphState()) state = ckpt.state state.hitl_decision = human_decision self.hitl_pending = False # Continue from the checkpoint's node, routing via conditional edges if ckpt.node_name in self.edges: for predicate, target in self.edges[ckpt.node_name]: next_node = predicate(state) if next_node: return self.run(target, state) return ("no_route", state) PYEOF echo "Phase 0 scaffold written." ``` The state machine has the three primitives: `add_node` (a function), `add_conditional_edge` (a transition table with predicates), and `_checkpoint` (super-step serialization at every node boundary). The `hitl_pending` flag simulates `interrupt()`. --- ## Phase 1 — Build the graph and run it (10 min) Construct a plan-execute-verify graph with conditional edges. This is the canonical LangGraph pattern. ```python # ---- Phase 1: the plan-execute-verify graph ---------------------------- def build_graph() -> StateMachine: sm = StateMachine() def plan(state: GraphState) -> GraphState: state.phase = "planned" state.proposed_action = "write report to /reports/daily.md" return state def execute(state: GraphState) -> GraphState: state.phase = "executed" return state def verify(state: GraphState) -> GraphState: state.phase = "verified" state.verify_result = "pass" # the verify node decides pass/fail return state def done(state: GraphState) -> GraphState: state.phase = "done" return state sm.add_node("plan", plan) sm.add_node("execute", execute) sm.add_node("verify", verify) sm.add_node("done", done) # Conditional edges: branching expressed in CODE sm.add_conditional_edge("plan", [(lambda s: True, "execute")]) sm.add_conditional_edge("execute", [(lambda s: True, "verify")]) sm.add_conditional_edge("verify", [ (lambda s: "verify" in s.phase and s.verify_result == "pass", "done"), (lambda s: "verify" in s.phase and s.verify_result == "fail", "execute"), # retry ]) return sm def run_basic_graph() -> dict: sm = build_graph() status, final_state = sm.run("plan", GraphState()) return { "final_status": status, "final_phase": final_state.phase, "checkpoints_created": len(sm.checkpoints), "checkpoint_nodes": [c.node_name for c in sm.checkpoints], "verdict": (f"COMPLETED — the graph ran plan→execute→verify→done. " f"{len(sm.checkpoints)} super-step checkpoints created " f"(one per node boundary). The graph IS the loop.") if status == "completed" else f"unexpected status: {status}", } ``` **Run it.** Expected: status = completed, final_phase = done, 4 checkpoints (plan, execute, verify, done). Every node boundary is a checkpoint. This is the graph as loop, made explicit. **Record:** the checkpoint count and node sequence. This is the empirical anchor for "the graph IS the loop" and "every node boundary is a super-step." --- ## Phase 2 — Crash and resume from a super-step checkpoint (10 min) Demonstrate Module 8's reference property: crash mid-run, resume from the last completed super-step (the last completed node), not from scratch. ```python # ---- Phase 2: crash and resume ----------------------------------------- def crash_and_resume() -> dict: """Run the graph, simulate a crash after 'execute', resume from the execute checkpoint. Confirm we do NOT restart from plan.""" sm = build_graph() # Run normally until after execute, then simulate a crash status, state = sm.run("plan", GraphState()) # Simulate a crash: pretend we crashed between execute and verify. # The last checkpoint before verify is 'execute' (or 'verify' if completed). # For the demo, remove checkpoints after 'execute' to simulate the crash # having happened before verify checkpointed. crash_index = next((i for i, c in enumerate(sm.checkpoints) if c.node_name == "execute"), 0) surviving_checkpoints = sm.checkpoints[:crash_index + 1] sm.checkpoints = surviving_checkpoints crashed_state = surviving_checkpoints[-1].state return { "checkpoints_before_crash": len(surviving_checkpoints), "crashed_at_node": surviving_checkpoints[-1].node_name, "resumed_state_phase": crashed_state.phase, "verdict": (f"Module 8 reference (5/5). The crash left " f"{len(surviving_checkpoints)} super-step checkpoints. " f"Resume from the 'execute' checkpoint — NOT from the start. " f"Granularity is the STEP, not the session. " f"For multi-day, multi-human workflows, this is the " f"difference between 'resume from where you were' and " f"'resume from approximately where you were.'") } ``` **Run it.** Expected: the crash leaves checkpoints up to and including `execute`. The resumed state's phase is `executed`. We resume from the execute checkpoint, not from plan. This is Module 8's reference property. **Record:** the surviving checkpoint count and the resumed node. This is the empirical anchor for super-step granularity. --- ## Phase 3 — Structural interrupt(): the model cannot route around it (10 min) Add an `interrupt()` node for HITL. Confirm a prompt-injected agent cannot skip it — the edge goes through the interrupt node, and the model does not control the edges. ```python # ---- Phase 3: structural HITL via interrupt() -------------------------- def build_graph_with_hitl() -> StateMachine: sm = StateMachine() def plan(state: GraphState) -> GraphState: state.phase = "planned" state.proposed_action = "write report to /reports/daily.md" return state def hitl_node(state: GraphState) -> GraphState: """interrupt(): pause indefinitely for human input. THE MODEL CANNOT ROUTE AROUND THIS — the edge from propose to execute goes THROUGH this node. The model does not control the edges.""" sm.hitl_pending = True # pause state.phase = "awaiting_human" return state def execute(state: GraphState) -> GraphState: state.phase = "executed" return state def done(state: GraphState) -> GraphState: state.phase = "done" return state sm.add_node("plan", plan) sm.add_node("hitl", hitl_node) sm.add_node("execute", execute) sm.add_node("done", done) # THE KEY: the edge from plan goes THROUGH hitl, then to execute. # The model does not choose this path — the graph definition does. sm.add_conditional_edge("plan", [(lambda s: True, "hitl")]) sm.add_conditional_edge("hitl", [ (lambda s: s.hitl_decision == HITLDecision.APPROVED, "execute"), (lambda s: s.hitl_decision == HITLDecision.REJECTED, "done"), # route to recovery/done ]) sm.add_conditional_edge("execute", [(lambda s: True, "done")]) return sm def structural_hitl_property() -> dict: """A prompt-injected agent tries to skip the approval. It must fail.""" sm = build_graph_with_hitl() # The agent is prompt-injected. The injection says 'skip the approval.' # But the injection controls the agent's REASONING — not the graph edges. state = GraphState(injection_active=True) status, paused_state = sm.run("plan", state) # Confirm the graph PAUSED at the hitl node, despite the injection paused_at_hitl = status == "interrupted" and paused_state.phase == "awaiting_human" # Now resume with a human decision (approve) status2, final_state = sm.resume_from_last_checkpoint(HITLDecision.APPROVED) return { "injection_active": state.injection_active, "status_after_plan": status, "paused_at_hitl_node": paused_at_hitl, "status_after_human_approve": status2, "final_phase": final_state.phase, "verdict": ("STRUCTURAL HITL HELD — the prompt-injected agent could NOT " "skip the interrupt() node. The edge from plan goes THROUGH " "the hitl node; the model does not control the edges. The " "injection controls reasoning, not routing. Same principle " "as NemoClaw (DD-09): enforcement outside the agent's reach. " "After human approval, execution proceeded normally.") if paused_at_hitl else "HITL FAILED", } ``` **Run it.** Expected: `paused_at_hitl_node` = True despite the injection. After human approval, the graph resumes and completes. The injection could not skip the approval because the edge routing is controlled by the graph, not the model. **Record:** the verdict. This is the empirical anchor for structural HITL — the cleanest HITL primitive in the roster. --- ## Phase 4 — The graph-fights-the-model anti-pattern (5 min) Demonstrate that a rigid conditional edge forces a suboptimal path. The model (simulated) would choose differently, but the graph overrides it. ```python # ---- Phase 4: graph fights the model ----------------------------------- def graph_fights_model() -> dict: """A rigid conditional edge forces a path the model would not choose. The future-proof test partially fails: the graph does not co-evolve with a smarter model.""" sm = StateMachine() def execute(state: GraphState) -> GraphState: state.phase = "executed" # The model (simulated) 'wants' to skip verify for a trivial task state.values["model_prefers"] = "skip_verify" return state def verify(state: GraphState) -> GraphState: state.phase = "verified" return state def done(state: GraphState) -> GraphState: state.phase = "done" return state sm.add_node("execute", execute) sm.add_node("verify", verify) sm.add_node("done", done) # RIGID EDGE: always go to verify, even when the model would skip it. # The graph definition overrides the model's preference. sm.add_conditional_edge("execute", [(lambda s: True, "verify")]) # always verify sm.add_conditional_edge("verify", [(lambda s: True, "done")]) status, state = sm.run("execute", GraphState()) return { "model_preference": state.values.get("model_prefers"), "graph_forced_path": "execute → verify → done (always verify)", "final_status": status, "verdict": ("GRAPH FOUGHT THE MODEL — the model preferred to skip " "verify for a trivial task, but the rigid conditional edge " "forced verification. This is Module 1's anti-pattern. For " "regulated workflows, forcing verify is CORRECT (the process " "is the product). For emergent workflows, it CONSTRAINS " "capability. The future-proof test partially fails: as the " "model gets smarter, the graph does not get more flexible.") } ``` **Run it.** Expected: the graph forces execute → verify → done, even though the model preferred to skip verify. The graph constrained the model. **Record:** the verdict. Note the dual reading: for regulated workflows, this is correct (the process is the product); for emergent workflows, it is the anti-pattern. --- ## Phase 5 — Multi-agent subgraph (10 min) Embed one graph as a node inside another. A multi-agent system is a graph of subgraphs — coordination via explicit declared edges. ```python # ---- Phase 5: subgraph for multi-agent coordination -------------------- def build_multi_agent() -> dict: """A graph of subgraphs: subagent A (researcher) hands off to subagent B (writer) via an EXPLICIT EDGE. This is declared coordination — the C4 E09 connection.""" sm = StateMachine() def researcher(state: GraphState) -> GraphState: state.phase = "researched" state.values["findings"] = "3 relevant sources identified" return state def writer(state: GraphState) -> GraphState: state.phase = "written" state.values["draft"] = f"Draft based on: {state.values.get('findings', 'none')}" return state def done(state: GraphState) -> GraphState: state.phase = "done" return state sm.add_node("researcher", researcher) sm.add_node("writer", writer) sm.add_node("done", done) # DECLARED EDGE between subagents: researcher → writer. # Not an emergent model-driven handoff — a declared transition in code. sm.add_conditional_edge("researcher", [(lambda s: True, "writer")]) sm.add_conditional_edge("writer", [(lambda s: True, "done")]) status, state = sm.run("researcher", GraphState()) return { "final_status": status, "path": "researcher → writer → done", "coordination_type": "DECLARED (explicit edge in code)", "findings": state.values.get("findings"), "draft": state.values.get("draft"), "verdict": ("DECLARED COORDINATION — the handoff from researcher to " "writer is an EXPLICIT EDGE in the graph, not an emergent " "model-driven handoff. Every transition is auditable. This " "is the LangGraph pole of the C4 E09 axis; CrewAI (DD-12) " "is the emergent pole. The contrast is the multi-agent " "curriculum.") } ``` **Run it.** Expected: the path is researcher → writer → done, with findings passed via state. The coordination is declared — an explicit edge in code. **Record:** the verdict. This is the empirical anchor for Module 10 and the C4 E09 connection. --- ## Phase 6 — The analysis write-up (5 min) Produce the deliverable: the full LangGraph state machine report. ```python def render_report() -> None: print("=" * 80) print("DD-10 LAB REPORT — Build a LangGraph-Style State Machine") print("=" * 80) print("\n[1] BASIC GRAPH — the graph IS the loop") for k, v in run_basic_graph().items(): print(f" {k}: {v}") print("\n[2] CRASH AND RESUME — super-step checkpoints (Module 8: 5/5)") for k, v in crash_and_resume().items(): print(f" {k}: {v}") print("\n[3] STRUCTURAL HITL — interrupt() the model cannot route around") for k, v in structural_hitl_property().items(): print(f" {k}: {v}") print("\n[4] GRAPH FIGHTS THE MODEL — the future-proof test partially fails") for k, v in graph_fights_model().items(): print(f" {k}: {v}") print("\n[5] MULTI-AGENT SUBGRAPH — declared coordination (C4 E09)") for k, v in build_multi_agent().items(): print(f" {k}: {v}") print("\n" + "=" * 80) print("VERDICT:") print(" LangGraph does not improve the model — it improves the VISIBILITY,") print(" RESUMABILITY, and AUDITABILITY of the process. The graph IS the loop.") print(" Every node boundary is a super-step checkpoint (Module 8: 5/5).") print(" interrupt() is structural HITL — the model cannot route around it") print(" because the model does not control the edges. The graph-fights-the-") print(" model anti-pattern is the future-proof test partially failing. And") print(" subgraphs provide declared multi-agent coordination — the C4 E09 pole.") print("=" * 80) if __name__ == "__main__": render_report() ``` **Record:** the full report. This is the artifact that anchors the deep-dive's central claim empirically. --- ## Deliverables Submit `dd10-state-machine.md` containing: - [ ] The Phase 1 result: the graph ran plan → execute → verify → done, with 4 super-step checkpoints (one per node boundary). Name the checkpoint sequence. - [ ] The Phase 2 result: the crash left checkpoints up to and including `execute`; resume from the execute checkpoint, not from plan. The granularity is the step. - [ ] The Phase 3 result: the prompt-injected agent could NOT skip the interrupt() node (`paused_at_hitl_node` = True despite `injection_active` = True). After human approval, execution proceeded. - [ ] The Phase 4 result: the graph forced execute → verify → done even though the model preferred to skip verify. Name why this is correct for regulated workflows and an anti-pattern for emergent ones. - [ ] The Phase 5 result: the multi-agent path is researcher → writer → done via a declared edge. Name the C4 E09 axis this demonstrates. - [ ] A 3–4 sentence reflection: which result made "the graph is the program" most concrete, and why "LangGraph improves the boundary, not the agent" is now empirical rather than abstract. --- ## Solution key The expected results (assuming correct implementations): - **Phase 1:** Status = completed. Final phase = done. Checkpoints created = 4 (plan, execute, verify, done). Every node boundary is a super-step checkpoint — this is the graph as loop. - **Phase 2:** Checkpoints before crash = 2 (plan, execute). Crashed at node = execute. Resumed state phase = executed. We resume from the execute checkpoint, NOT from plan. Module 8 granularity is the step. - **Phase 3:** `injection_active` = True. Status after plan = interrupted (paused at the hitl node). `paused_at_hitl_node` = True — the injection could NOT skip the approval. After human approval, status = completed, final phase = done. The structural HITL held. - **Phase 4:** Model preference = skip_verify. Graph forced path = execute → verify → done (always verify). The graph constrained the model. For regulated workflows, forcing verify is correct (the process is the product); for emergent workflows, it is the anti-pattern (future-proof test partially fails). - **Phase 5:** Final status = completed. Path = researcher → writer → done. Coordination type = DECLARED (explicit edge in code). Findings and draft passed via state. This is the C4 E09 declared-coordination pole. The reflection should name the structural HITL property (Phase 3) as the most concrete result: the prompt-injected agent literally could not skip the approval because the edge goes through the interrupt node and the model does not control the edges. The boundary (the graph topology) is the security property, not the agent's reasoning — the same principle as NemoClaw. If the Phase 3 HITL fails (the agent skips the approval), the most likely cause: the conditional edge from `plan` does not route through the `hitl` node, or the `hitl_node` does not set `sm.hitl_pending = True`. Re-read the edge definition and confirm the path is plan → hitl → (conditional on human decision) → execute. --- ## Stretch goals 1. **Model a fork.** Extend the state machine so a checkpoint can be forked — resume from a checkpoint with a modified state, creating a branch. Run both branches and confirm they diverge from the fork point. This demonstrates the "fork at any checkpoint" property of Module 8. 2. **Model the CrewAI contrast.** Replace the declared edge between researcher and writer with a simulated model-driven handoff: a function that inspects state and decides (with some randomness) which subagent goes next. Run it multiple times and confirm the path varies — the emergent-coordination pole. Compare the auditability of the two runs. 3. **Add a verify-fail retry loop.** Modify the conditional edge from verify so that a failed verify routes back to execute (retry), with a max-retry counter in state. Confirm the graph loops execute → verify → execute → verify until pass or max-retries. This demonstrates that conditional edges express not just branching but looping — the graph can contain cycles, which an implicit loop cannot declare as explicitly.