Multi-Agent Workflows & State Management

A single agent looping through a task works fine until the task genuinely needs more than one kind of expertise, research and writing, or diagnosis and fixing. Wiring multiple agents together as separate functions that just call each other gets messy fast: no shared memory, no way to resume if one step fails, no clean place to pause for a human to approve something risky. This lesson covers the real fix, modeling multi-agent coordination as a graph with shared state, using LangGraph, the framework most production multi-agent systems are actually built on.

What you'll learn

LangGraph's state/nodes/edges/checkpointer model
Supervisor-worker patterns for coordinating multiple agents
Human-in-the-loop approval gates inside a cyclical graph

How it works

A simple pipeline where output A feeds into input B breaks down the moment real agent reasoning needs to loop back, backtrack on a bad result, or run two things in parallel. LangGraph models this as a graph instead of a straight line, nodes do work, edges decide what happens next, and a shared state object carries context between every step.

A. Core Graph Components

Nodes: Python functions executing specific computational steps (LLM calls, tool executions, or state mutations).
Edges: Direct connection pathways. Conditional Edges execute routing decisions using an LLM or python conditional check to determine the next node.
State Schema: A persistent, transactionally managed memory store passed to every node. It uses Reducer Functions (such as add_messages) to control how updates are appended to or merged into the state database.
Checkpointers: Snapshots that serialize the active state of the graph and write it to database storage (PostgreSQL/Redis) at every node transition.

B. Micro-Step Persistence & Handshakes

Checkpointers provide thread isolation. If an agent loops over many steps, the state is persisted after every node execution. This allows:

Reentrancy: Resuming execution from the exact node failure point.
Time Travel: Reloading previous thread states to audit agent decision trees.
Thread Isolation: Spawning separate execution threads for different users sharing the same graph structure.

Options & when to use each

Topology	Routing Latency	State Conflict Risk	Token Efficiency	Complexity	Primary Production Bottleneck
Linear Chain	Low (< 200ms)	Low	High	Low	Rigid error propagation (cannot backtrack)
Supervisor-Worker	Moderate	Low	Moderate	Moderate	Router model intelligence constraints
Decentralized Net	High	High (Race conditions)	Low (Info inflation)	High	Cyclic infinite loops and state overrides

Build it: a supervisor routing to two workers

A central supervisor node decides, based on the conversation so far, whether the next step needs a researcher or a writer, the same pattern you'd use for the Foreman capstone's Owl/Gnome split later in this course.

Initialize Folder:

bash

mkdir -p ~/AI_BOOTCAMP/labs/langgraph
cd ~/AI_BOOTCAMP/labs/langgraph

Define Graph Structure: Create agent_graph.py:

python

from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import MemorySaver

### 1. Define the Shared State Schema
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    next_step: str
    decision_payload: str

### 2. Define Node Functions
def supervisor_node(state: AgentState):
    print("🤖 Supervisor evaluating routing...")
    messages = state["messages"]
    last_message = messages[-1].content if messages else ""
    
    # Simple routing logic (in production, use a structured LLM call)
    if "research" in last_message.lower() and state["next_step"] != "researcher":
        return {"next_step": "researcher"}
    elif "draft" in last_message.lower() and state["next_step"] != "writer":
        return {"next_step": "writer"}
    return {"next_step": "end"}

def researcher_node(state: AgentState):
    print("🔍 Researcher executing data query...")
    return {
        "messages": [{"role": "assistant", "content": "RESEARCH_RESULT: Found 12 PostgreSQL instances."}],
        "next_step": "supervisor"
    }

def writer_node(state: AgentState):
    print("✍️ Writer drafting report...")
    return {
        "messages": [{"role": "assistant", "content": "DRAFT_RESULT: Report finalized successfully."}],
        "next_step": "supervisor"
    }

### 3. Compile the State Graph
workflow = StateGraph(AgentState)

### 4. Add Nodes
workflow.add_node("supervisor", supervisor_node)
workflow.add_node("researcher", researcher_node)
workflow.add_node("writer", writer_node)

### 5. Define Edges
workflow.add_edge(START, "supervisor")

### Conditional routing edge
workflow.add_conditional_edges(
    "supervisor",
    lambda state: state["next_step"],
    {
        "researcher": "researcher",
        "writer": "writer",
        "end": END
    }
)

workflow.add_edge("researcher", "supervisor")
workflow.add_edge("writer", "supervisor")

### 6. Add In-Memory State Checkpointer
memory = MemorySaver()
app = workflow.compile(checkpointer=memory)

What goes wrong

Confirm it worked

To verify your graph's routing and state serialization:

Execute the Test Run: Add verification code to agent_graph.py and run the script:

python

if __name__ == "__main__":
    config = {"configurable": {"thread_id": "session_abc123"}}
    print("--- Beginning Graph Execution ---")
    
    # Trigger with a prompt that requests research first
    for event in app.stream(
        {"messages": [{"role": "user", "content": "Run research and then draft it."}], "next_step": "supervisor"},
        config
    ):
        print(event)

Observe Node Transitions: Audit the stdout logging pattern. Confirm that the sequence strictly matches: Supervisor → Researcher → Supervisor → Writer → Supervisor → END.
Inspect State Merges: Observe how add_messages appends data. Print the final state messages list:
python
```
final_state = app.get_state(config)
print(f"Total Messages Count: {len(final_state.values['messages'])}")
```
Confirm that the list contains 3 messages (User prompt + Researcher output + Writer output) rather than just the last worker output.

Next: Day 4 , Security & Evaluation , this multi-agent system is only trustworthy once its output is validated and its execution is guarded.

Multi-Agent Workflows & State Management ​

How it works ​

A. Core Graph Components ​

B. Micro-Step Persistence & Handshakes ​

Options & when to use each ​

Build it: a supervisor routing to two workers ​

What goes wrong ​

Confirm it worked ​