Langgraph

LangGraph Workflows

Summary

LangGraph is a framework for building stateful, multi-agent applications with LLMs. It implements state machines and directed graphs for orchestration, enabling complex workflows with persistent state management, human-in-the-loop support, and time-travel debugging.

Key Innovation: Transforms agent coordination from sequential chains into cyclic graphs with persistent state, conditional branching, and production-grade debugging capabilities.

When to Use

✅ Use LangGraph When:

• Multi-agent coordination required
• Complex state management needs
• Human-in-the-loop workflows (approval gates, reviews)
• Need debugging/observability (time-travel, replay)
• Conditional branching based on outputs
• Building production agent systems
• State persistence across sessions

❌ Don't Use LangGraph When:

• Simple single-agent tasks
• No state persistence needed
• Prototyping/experimentation phase (use simple chains)
• Team lacks graph/state machine expertise
• Stateless request-response patterns

Quick Start

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator

# 1. Define state schema
class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    current_step: str

# 2. Create graph
workflow = StateGraph(AgentState)

# 3. Add nodes (agents)
def researcher(state):
    return {"messages": ["Research complete"], "current_step": "research"}

def writer(state):
    return {"messages": ["Article written"], "current_step": "writing"}

workflow.add_node("researcher", researcher)
workflow.add_node("writer", writer)

# 4. Add edges (transitions)
workflow.add_edge("researcher", "writer")
workflow.add_edge("writer", END)

# 5. Set entry point and compile
workflow.set_entry_point("researcher")
app = workflow.compile()

# 6. Execute
result = app.invoke({"messages": [], "current_step": "start"})
print(result)

---

Core Concepts

StateGraph

The fundamental building block representing a directed graph of agents with shared state.

from langgraph.graph import StateGraph, END
from typing import TypedDict

class AgentState(TypedDict):
    """State schema shared across all nodes."""
    messages: list
    user_input: str
    final_output: str
    metadata: dict

# Create graph with state schema
workflow = StateGraph(AgentState)

Key Properties:

• Nodes: Agent functions that transform state
• Edges: Transitions between nodes (static or conditional)
• State: Shared data structure passed between nodes
• Entry Point: Starting node of execution
• END: Terminal node signaling completion

Nodes

Nodes are functions that receive current state and return state updates.

def research_agent(state: AgentState) -> dict:
    """Node function: receives state, returns updates."""
    query = state["user_input"]

    # Perform research (simplified)
    results = search_web(query)

    # Return state updates (partial state)
    return {
        "messages": state["messages"] + [f"Research: {results}"],
        "metadata": {"research_complete": True}
    }

# Add node to graph
workflow.add_node("researcher", research_agent)

Node Behavior:

• Receives full state as input
• Returns partial state (only fields to update)
• Can be sync or async functions
• Can invoke LLMs, call APIs, run computations

Edges

Edges define transitions between nodes.

Static Edges

# Direct transition: researcher → writer
workflow.add_edge("researcher", "writer")

# Transition to END
workflow.add_edge("writer", END)

Conditional Edges

def should_continue(state: AgentState) -> str:
    """Routing function: decides next node based on state."""
    last_message = state["messages"][-1]

    if "APPROVED" in last_message:
        return END
    elif "NEEDS_REVISION" in last_message:
        return "writer"
    else:
        return "reviewer"

workflow.add_conditional_edges(
    "reviewer",  # Source node
    should_continue,  # Routing function
    {
        END: END,
        "writer": "writer",
        "reviewer": "reviewer"
    }
)

---

Graph Construction

Complete Workflow Example

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator
from langchain_anthropic import ChatAnthropic

# State schema with reducer
class ResearchState(TypedDict):
    topic: str
    research_notes: Annotated[list, operator.add]  # Reducer: appends to list
    draft: str
    revision_count: int
    approved: bool

# Initialize LLM
llm = ChatAnthropic(model="claude-sonnet-4")

# Node 1: Research
def research_node(state: ResearchState) -> dict:
    """Research the topic and gather information."""
    topic = state["topic"]

    prompt = f"Research key points about: {topic}"
    response = llm.invoke(prompt)

    return {
        "research_notes": [response.content]
    }

# Node 2: Write
def write_node(state: ResearchState) -> dict:
    """Write draft based on research."""
    notes = "\n".join(state["research_notes"])

    prompt = f"Write article based on:\n{notes}"
    response = llm.invoke(prompt)

    return {
        "draft": response.content,
        "revision_count": state.get("revision_count", 0)
    }

# Node 3: Review
def review_node(state: ResearchState) -> dict:
    """Review the draft and decide if approved."""
    draft = state["draft"]

    prompt = f"Review this draft. Reply APPROVED or NEEDS_REVISION:\n{draft}"
    response = llm.invoke(prompt)

    approved = "APPROVED" in response.content

    return {
        "approved": approved,
        "revision_count": state["revision_count"] + 1,
        "research_notes": [f"Review feedback: {response.content}"]
    }

# Routing logic
def should_continue_writing(state: ResearchState) -> str:
    """Decide next step after review."""
    if state["approved"]:
        return END
    elif state["revision_count"] >= 3:
        return END  # Max revisions reached
    else:
        return "writer"

# Build graph
workflow = StateGraph(ResearchState)

workflow.add_node("researcher", research_node)
workflow.add_node("writer", write_node)
workflow.add_node("reviewer", review_node)

# Static edges
workflow.add_edge("researcher", "writer")
workflow.add_edge("writer", "reviewer")

# Conditional edge from reviewer
workflow.add_conditional_edges(
    "reviewer",
    should_continue_writing,
    {
        END: END,
        "writer": "writer"
    }
)

workflow.set_entry_point("researcher")

# Compile
app = workflow.compile()

# Execute
result = app.invoke({
    "topic": "AI Safety",
    "research_notes": [],
    "draft": "",
    "revision_count": 0,
    "approved": False
})

print(f"Final draft: {result['draft']}")
print(f"Revisions: {result['revision_count']}")

---

State Management

State Schema with TypedDict

from typing import TypedDict, Annotated, Literal
import operator

class WorkflowState(TypedDict):
    # Simple fields (replaced on update)
    user_id: str
    request_id: str
    status: Literal["pending", "processing", "complete", "failed"]

    # List with reducer (appends instead of replacing)
    messages: Annotated[list, operator.add]

    # Dict with custom reducer
    metadata: Annotated[dict, lambda x, y: {**x, **y}]

    # Optional fields
    error: str | None
    result: dict | None

State Reducers

Reducers control how state updates are merged.

import operator
from typing import Annotated

# Built-in reducers
Annotated[list, operator.add]  # Append to list
Annotated[set, operator.or_]   # Union of sets
Annotated[int, operator.add]   # Sum integers

# Custom reducer
def merge_dicts(existing: dict, update: dict) -> dict:
    """Deep merge dictionaries."""
    result = existing.copy()
    for key, value in update.items():
        if key in result and isinstance(result[key], dict) and isinstance(value, dict):
            result[key] = merge_dicts(result[key], value)
        else:
            result[key] = value
    return result

class State(TypedDict):
    config: Annotated[dict, merge_dicts]

State Updates

Nodes return partial state - only fields to update:

def node_function(state: State) -> dict:
    """Return only fields that should be updated."""
    return {
        "messages": ["New message"],  # Will be appended
        "status": "processing"  # Will be replaced
    }
    # Other fields remain unchanged

---

Conditional Routing

Basic Routing Function

def route_based_on_score(state: State) -> str:
    """Route to different nodes based on state."""
    score = state["quality_score"]

    if score >= 0.9:
        return "publish"
    elif score >= 0.6:
        return "review"
    else:
        return "revise"

workflow.add_conditional_edges(
    "evaluator",
    route_based_on_score,
    {
        "publish": "publisher",
        "review": "reviewer",
        "revise": "reviser"
    }
)

Dynamic Routing with LLM

from langchain_anthropic import ChatAnthropic

def llm_router(state: State) -> str:
    """Use LLM to decide next step."""
    llm = ChatAnthropic(model="claude-sonnet-4")

    prompt = f"""
    Current state: {state['current_step']}
    User request: {state['user_input']}

    Which agent should handle this next?
    Options: researcher, coder, writer, FINISH

    Return only one word.
    """

    response = llm.invoke(prompt)
    next_agent = response.content.strip().lower()

    if next_agent == "finish":
        return END
    else:
        return next_agent

workflow.add_conditional_edges(
    "supervisor",
    llm_router,
    {
        "researcher": "researcher",
        "coder": "coder",
        "writer": "writer",
        END: END
    }
)

Multi-Condition Routing

def complex_router(state: State) -> str:
    """Route based on multiple conditions."""
    # Check multiple conditions
    has_errors = bool(state.get("errors"))
    is_approved = state.get("approved", False)
    iteration_count = state.get("iterations", 0)

    # Priority-based routing
    if has_errors:
        return "error_handler"
    elif is_approved:
        return END
    elif iteration_count >= 5:
        return "escalation"
    else:
        return "processor"

workflow.add_conditional_edges(
    "validator",
    complex_router,
    {
        "error_handler": "error_handler",
        "processor": "processor",
        "escalation": "escalation",
        END: END
    }
)

---

Multi-Agent Patterns

Supervisor Pattern

One supervisor coordinates multiple specialized agents.

from langgraph.graph import StateGraph, END
from langchain_anthropic import ChatAnthropic
from typing import TypedDict, Literal

class SupervisorState(TypedDict):
    task: str
    agent_history: list
    result: dict
    next_agent: str

# Specialized agents
class ResearchAgent:
    def __init__(self):
        self.llm = ChatAnthropic(model="claude-sonnet-4")

    def run(self, state: SupervisorState) -> dict:
        response = self.llm.invoke(f"Research: {state['task']}")
        return {
            "agent_history": [f"Research: {response.content}"],
            "result": {"research": response.content}
        }

class CodingAgent:
    def __init__(self):
        self.llm = ChatAnthropic(model="claude-sonnet-4")

    def run(self, state: SupervisorState) -> dict:
        response = self.llm.invoke(f"Code: {state['task']}")
        return {
            "agent_history": [f"Code: {response.content}"],
            "result": {"code": response.content}
        }

class SupervisorAgent:
    def __init__(self):
        self.llm = ChatAnthropic(model="claude-sonnet-4")

    def route(self, state: SupervisorState) -> dict:
        """Decide which agent to use next."""
        history = "\n".join(state.get("agent_history", []))

        prompt = f"""
        Task: {state['task']}
        Progress: {history}

        Which agent should handle the next step?
        Options: researcher, coder, FINISH

        Return only one word.
        """

        response = self.llm.invoke(prompt)
        next_agent = response.content.strip().lower()

        return {"next_agent": next_agent}

# Build supervisor workflow
def create_supervisor_workflow():
    workflow = StateGraph(SupervisorState)

    # Initialize agents
    research_agent = ResearchAgent()
    coding_agent = CodingAgent()
    supervisor = SupervisorAgent()

    # Add nodes
    workflow.add_node("supervisor", supervisor.route)
    workflow.add_node("researcher", research_agent.run)
    workflow.add_node("coder", coding_agent.run)

    # Conditional routing from supervisor
    def route_from_supervisor(state: SupervisorState) -> str:
        next_agent = state.get("next_agent", "FINISH")
        if next_agent == "finish":
            return END
        return next_agent

    workflow.add_conditional_edges(
        "supervisor",
        route_from_supervisor,
        {
            "researcher": "researcher",
            "coder": "coder",
            END: END
        }
    )

    # Loop back to supervisor after each agent
    workflow.add_edge("researcher", "supervisor")
    workflow.add_edge("coder", "supervisor")

    workflow.set_entry_point("supervisor")

    return workflow.compile()

# Execute
app = create_supervisor_workflow()
result = app.invoke({
    "task": "Build a REST API for user management",
    "agent_history": [],
    "result": {},
    "next_agent": ""
})

Hierarchical Multi-Agent

Nested supervisor pattern with sub-teams.

class TeamState(TypedDict):
    task: str
    team_results: dict

def create_backend_team():
    """Sub-graph for backend development."""
    workflow = StateGraph(TeamState)

    workflow.add_node("api_designer", design_api)
    workflow.add_node("database_designer", design_db)
    workflow.add_node("implementer", implement_backend)

    workflow.add_edge("api_designer", "database_designer")
    workflow.add_edge("database_designer", "implementer")
    workflow.add_edge("implementer", END)

    workflow.set_entry_point("api_designer")

    return workflow.compile()

def create_frontend_team():
    """Sub-graph for frontend development."""
    workflow = StateGraph(TeamState)

    workflow.add_node("ui_designer", design_ui)
    workflow.add_node("component_builder", build_components)
    workflow.add_node("integrator", integrate_frontend)

    workflow.add_edge("ui_designer", "component_builder")
    workflow.add_edge("component_builder", "integrator")
    workflow.add_edge("integrator", END)

    workflow.set_entry_point("ui_designer")

    return workflow.compile()

# Top-level coordinator
def create_project_workflow():
    workflow = StateGraph(TeamState)

    # Add team sub-graphs as nodes
    backend_team = create_backend_team()
    frontend_team = create_frontend_team()

    workflow.add_node("backend_team", backend_team)
    workflow.add_node("frontend_team", frontend_team)
    workflow.add_node("integrator", integrate_teams)

    # Parallel execution of teams
    workflow.add_edge("backend_team", "integrator")
    workflow.add_edge("frontend_team", "integrator")
    workflow.add_edge("integrator", END)

    workflow.set_entry_point("backend_team")

    return workflow.compile()

Swarm Pattern (2025)

Dynamic agent hand-offs with collaborative decision-making.

from typing import Literal

class SwarmState(TypedDict):
    task: str
    messages: list
    current_agent: str
    handoff_reason: str

def create_swarm():
    """Agents can dynamically hand off to each other."""

    def research_agent(state: SwarmState) -> dict:
        llm = ChatAnthropic(model="claude-sonnet-4")

        # Perform research
        response = llm.invoke(f"Research: {state['task']}")

        # Decide if handoff needed
        needs_code = "implementation" in response.content.lower()

        if needs_code:
            return {

…

## Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

- **Author:** [bobmatnyc](https://github.com/bobmatnyc)
- **Source:** [bobmatnyc/claude-mpm-skills](https://github.com/bobmatnyc/claude-mpm-skills)
- **License:** MIT

Install and usage instructions live in the source repository linked above.