6.1 Stateful Orchestration: Directed Cyclic Graphs & LangGraph 1.0

Canonical Curriculum Reference: docs/plan/06_phase6_stateful_graphs_hitl.md
Textbook Reference: AI Agents and Applications (Manning Publications), Chapter 11, Sections 11.3–11.4 ("Assembling the agent graph & structure", pp. 305–307) & Chapter 12 ("Multi-agent systems", pp. 321–335).

---

🔄 1. Beyond Linear Pipelines: The Shift to Directed Cyclic Graphs (DCGs)

In Phase 3, we built declarative pipelines using LangChain Expression Language (LCEL). LCEL compiles pipelines into Directed Acyclic Graphs (DAGs):

$$A⟶B⟶C$$

DAGs are mathematically incapable of representing cycles. Information moves strictly downstream in a single forward pass.

However, real-world agentic reasoning is fundamentally non-linear and iterative:

• An agent formulates a code fix, runs tests, observes a failure, and must loop back to modify its code.
• A research assistant retrieves documents, discovers the findings are insufficient, and must cycle back to issue new queries.
• A human reviewer inspects a drafted contract, requests revisions, and redirects the workflow back to the drafting node.

To model iterative, self-correcting workflows, industry architecture transitioned to Directed Cyclic Graphs (DCGs) powered by LangGraph 1.0:

Linear DAG (LCEL): Strictly Forward
Input ---> [ Node A ] ---> [ Node B ] ---> [ Node C ] ---> Output

Directed Cyclic Graph (LangGraph 1.0): Iterative Self-Correction
                   +---------------------------+
                   |                           |
                   v                           | (Iterate if tests fail)
Input ---> [ Draft Code ] ---> [ Run Tests ] --+
                                     |
                                     +---> (Pass) ---> [ Deploy ]

---

🧩 2. Core Architectural Primitives of LangGraph

A LangGraph workflow is modeled as a mathematical graph $G=(V,E,S)$ where:

1. Nodes ($V$): Discrete Python functions or Runnables that receive the current state, perform computation (LLM call, database query, tool execution), and return a state update.
2. Edges ($E$): Control-flow connections directing the transition from one node to the next.
   • Normal Edges: Deterministic transitions ($A\to B$).
   • Conditional Edges: Dynamic routing functions that inspect the state and return the name of the next destination node ($A\to \text{route}(S)\in \{B,C,\text{END}\}$).
3. State ($S$): The shared, typed context object passed between nodes (acting as a durable "backpack" of data).

---

🏛️ 3. Core Multi-Agent Graph Topologies

AI Agents and Applications (Chapter 12, pp. 321–335) details two canonical architectural topologies for orchestrating multi-agent systems:

1. The Router Pattern (Chapter 12.2, p. 326)

A single classifier/router node analyzes the incoming user request and delegates execution to exactly one specialist agent. Once that specialist finishes, the workflow terminates:

2. The Supervisor Pattern: "Agent of Agents" (Chapter 12.3, pp. 330–334)

In complex multi-step missions, a single router is insufficient. The Supervisor Pattern establishes a hierarchical orchestrator that coordinates multiple worker agents using "return-ticket" interactions:

• The Supervisor dispatches work to Worker A (e.g. Flight Booker).
• Worker A executes its tools and returns control back to the Supervisor.
• The Supervisor inspects the updated global state, determines that hotel accommodation is now required, and dispatches to Worker B.
• When all sub-goals are satisfied, the Supervisor synthesizes the final client response.

---

💻 4. Production Implementation: StateGraph with Cyclic Feedback

Below is a production-grade Python 3.12 implementation of a self-correcting StateGraph with conditional loop-back:

from typing import Annotated, Literal, TypedDict
from langgraph.graph import StateGraph, START, END
from pydantic import BaseModel, ConfigDict, Field


class AgentState(TypedDict):
    """Strongly typed shared state object."""
    draft_code: str
    test_results: str
    iteration_count: int
    is_verified: bool


def drafting_node(state: AgentState) -> dict:
    """Generates or refines code based on current state and test feedback."""
    iteration = state.get("iteration_count", 0) + 1
    feedback = state.get("test_results", "")
    
    # In production, call LLM with feedback
    new_code = f"def solution(): return True  # Iteration {iteration}"
    return {
        "draft_code": new_code,
        "iteration_count": iteration
    }


def testing_node(state: AgentState) -> dict:
    """Executes deterministic test runner against draft code."""
    iteration = state["iteration_count"]
    # Simulate: Fails on iteration 1, passes on iteration 2
    passed = (iteration >= 2)
    results = "All tests passed." if passed else "AssertionError: Expected True, got False."
    return {
        "test_results": results,
        "is_verified": passed
    }


def route_after_testing(state: AgentState) -> Literal["drafting_node", "__end__"]:
    """Conditional edge: Determines whether to loop back or terminate."""
    if state["is_verified"]:
        return END
    if state["iteration_count"] >= 3:
        # Bounded iteration guardrail
        return END
    return "drafting_node"


# 1. Initialize StateGraph with typed schema
builder = StateGraph(AgentState)

# 2. Add Nodes
builder.add_node("drafting_node", drafting_node)
builder.add_node("testing_node", testing_node)

# 3. Add Edges
builder.add_edge(START, "drafting_node")
builder.add_edge("drafting_node", "testing_node")
builder.add_conditional_edges(
    "testing_node",
    route_after_testing,
    {
        "drafting_node": "drafting_node",
        END: END
    }
)

# 4. Compile Graph
app = builder.compile()

---

🧪 5. Automated Verification with Pytest

import pytest

def test_state_graph_cyclic_execution():
    initial_state = {
        "draft_code": "",
        "test_results": "",
        "iteration_count": 0,
        "is_verified": False
    }
    
    final_state = app.invoke(initial_state)
    
    # Proves the graph successfully looped back and passed on iteration 2
    assert final_state["is_verified"] is True
    assert final_state["iteration_count"] == 2
    assert "All tests passed." in final_state["test_results"]