LangGraph: State Graph-Based Agent Development

Overview

LangGraph is LangChain's framework for building stateful, agentic applications using graph-based workflow design. It reached production stability (v1.0) in 2024 and is the most widely adopted agent framework in 2025-2026.

Key insight: LangGraph makes agent state and control flow explicit via directed graphs, making systems more debuggable and testable.

---

Why LangGraph?

Core Strengths

Aspect	Benefit
Explicit State	State is first-class, not implicit
Deterministic Flow	Graph defines exact execution path
Debuggable	Trace exactly what happened
Testable	Each node can be unit tested
Persistent	Can checkpoint and resume
Multi-agent	Natural support for multiple agents

Market Position (2025)

• Market share: ~35% (largest)
• Growth: Growing fastest
• Production readiness: Stable 1.0+ API
• Enterprise adoption: 60%+ of Fortune 500 companies using

---

Core Concepts

1. State Graph

Workflow represented as a directed graph:

from langgraph.graph import StateGraph, START, END

# Define state schema
class AgentState(TypedDict):
    messages: list[str]
    goal: str
    tools_used: list[str]

# Create graph
graph = StateGraph(AgentState)

# Add nodes (actions)
graph.add_node("analyze", analyze_node)
graph.add_node("plan", plan_node)
graph.add_node("execute", execute_node)

# Add edges (transitions)
graph.add_edge(START, "analyze")
graph.add_edge("analyze", "plan")
graph.add_edge("plan", "execute")
graph.add_edge("execute", END)

2. Nodes

Functions that process state and return updated state:

def analyze_node(state: AgentState) -> AgentState:
    """Analyze goal and context"""
    analysis = llm.analyze(state["goal"])
    return {
        **state,
        "messages": state["messages"] + [analysis]
    }

def plan_node(state: AgentState) -> AgentState:
    """Create execution plan"""
    plan = llm.create_plan(
        goal=state["goal"],
        context=state["messages"]
    )
    return {
        **state,
        "messages": state["messages"] + [plan]
    }

3. Edges

Transitions between nodes (can be conditional):

# Simple edge
graph.add_edge("analyze", "plan")

# Conditional edge
def should_execute(state: AgentState):
    if state["confidence"] > 0.8:
        return "execute"
    else:
        return "analyze"  # Re-analyze

graph.add_conditional_edges("plan", should_execute)

4. Compilation and Execution

# Compile graph
agent = graph.compile()

# Run agent
initial_state = {
    "messages": [],
    "goal": "Find recent papers on quantum computing",
    "tools_used": []
}

result = agent.invoke(initial_state)
print(result["messages"][-1])  # Final result

---

Complete Example: Research Agent

from langgraph.graph import StateGraph, START, END
from langgraph.types import Command
from typing import TypedDict, Literal

class ResearchState(TypedDict):
    goal: str
    papers: list[dict]
    analysis: str
    messages: list[str]

# Define nodes
def search_node(state: ResearchState) -> ResearchState:
    """Search for papers"""
    papers = search_academic_db(state["goal"])
    return {
        **state,
        "papers": papers,
        "messages": state["messages"] + [f"Found {len(papers)} papers"]
    }

def filter_node(state: ResearchState) -> ResearchState:
    """Filter papers by relevance"""
    filtered = [p for p in state["papers"] if p["relevance"] > 0.7]
    return {
        **state,
        "papers": filtered,
        "messages": state["messages"] + [f"Filtered to {len(filtered)} papers"]
    }

def analyze_node(state: ResearchState) -> ResearchState:
    """Analyze key findings"""
    summary = llm.analyze_papers(state["papers"])
    return {
        **state,
        "analysis": summary,
        "messages": state["messages"] + [summary]
    }

# Build graph
graph = StateGraph(ResearchState)

graph.add_node("search", search_node)
graph.add_node("filter", filter_node)
graph.add_node("analyze", analyze_node)

graph.add_edge(START, "search")
graph.add_edge("search", "filter")
graph.add_edge("filter", "analyze")
graph.add_edge("analyze", END)

# Run
agent = graph.compile()
result = agent.invoke({"goal": "Quantum error correction 2024"})

---

Advanced Features

1. Persistence (Checkpointing)

Save and resume agent execution:

from langgraph.checkpoint.memory import MemorySaver

# Add memory backend
memory = MemorySaver()
agent = graph.compile(checkpointer=memory)

# Run and create checkpoint
config = {"configurable": {"thread_id": "user-123"}}
result = agent.invoke(initial_state, config)

# Later: resume from checkpoint
result = agent.invoke(next_input, config)
# Agent continues from where it left off

2. Streaming

Get results as they're generated:

# Stream node outputs
for event in agent.stream(initial_state):
    print(event)
    # Output: {"search": {...}}
    # Output: {"filter": {...}}
    # Output: {"analyze": {...}}

3. Parallel Branches

Execute multiple paths simultaneously:

# Add multiple branches
graph.add_node("web_search", web_search_node)
graph.add_node("db_search", db_search_node)

graph.add_edge("analyze", "web_search")
graph.add_edge("analyze", "db_search")

# Merge results
def merge_node(state):
    combined = state["web_results"] + state["db_results"]
    return {**state, "all_results": combined}

graph.add_node("merge", merge_node)
graph.add_edge("web_search", "merge")
graph.add_edge("db_search", "merge")

4. Human-in-the-Loop

Pause for human approval:

def should_execute(state):
    if state["cost"] > 100:
        return Command(
            goto="approve",
            update={"needs_approval": True}
        )
    else:
        return "execute"

def approve_node(state):
    # Wait for human approval
    approved = wait_for_human_approval(state)
    if approved:
        return Command(goto="execute")
    else:
        return Command(goto="END")

---

Comparison: LangGraph vs Alternatives

Feature	LangGraph	CrewAI	AutoGen
State	Explicit	Implicit	Implicit
Control	Graph	Config	Conversation
Multi-agent	Natural	Specialized	Good
Debugging	Excellent	Good	Fair
Learning curve	Medium	Low	Medium
Production use	60%+	40%+	30%+

---

Production Patterns with LangGraph

Pattern 1: Tool-Using Agent

tools = [search_tool, fetch_tool, summarize_tool]

def tool_node(state):
    # Decide which tool to use
    tool_choice = llm.choose_tool(state["goal"])
    result = tools[tool_choice](...)
    return {"messages": state["messages"] + [result]}

graph.add_node("tools", tool_node)

Pattern 2: Multi-Step Workflow

# Research workflow
graph.add_edge(START, "search")
graph.add_edge("search", "fetch")
graph.add_edge("fetch", "analyze")
graph.add_edge("analyze", "write")
graph.add_edge("write", END)

Pattern 3: Conditional Routing

def route_by_type(state):
    category = classify_request(state["goal"])
    if category == "research":
        return "research_flow"
    elif category == "analysis":
        return "analysis_flow"
    else:
        return "general_flow"

graph.add_conditional_edges("route", route_by_type)

---

Deployment (2025 Best Practices)

Local Development

pip install langgraph

Production Deployment

# Use LangGraph Cloud (hosted)
from langgraph_sdk import get_client

client = get_client()
graph_id = client.publish_graph(agent)
# Graph deployed, accessible via API

Scaling

• Per-request inference: ~100ms-2sec
• Concurrent requests: Handle 1000+/sec
• Memory: ~100MB per agent instance
• Cost: ~$0.001-0.01 per agent invocation

---

Observability with LangGraph

Built-in Tracing

from langsmith import Client

client = Client()

# Automatic logging
result = agent.invoke(state)
# Logs to LangSmith dashboard
# See: request, steps, tokens, cost, latency

Custom Logging

def my_node(state):
    result = process(state)
    logger.info(f"Node decision: {result}")
    return result

---

Common Patterns & Best Practices

Pattern	Use Case	Benefit
Reflection loop	Improve quality	Self-critique
Tool-using agent	Take action	Affect world
Router	Specialize	Better accuracy
Multi-step workflow	Complex tasks	Clear flow
Memory	Context management	Persist info

---

Debugging LangGraph Agents

# 1. Inspect state at each node
graph.add_node("debug", lambda s: logger.info(s) or s)

# 2. Use LangSmith for tracing
with client.tracing_context():
    result = agent.invoke(state)

# 3. Step through execution
for step in agent.stream(state):
    print(f"Step: {step}")
    # See intermediate states

# 4. Unit test nodes
def test_analyze_node():
    state = {"goal": "test"}
    result = analyze_node(state)
    assert "analysis" in result

---

Key Takeaways

1. State is first-class - Makes debugging easier
2. Graphs are explicit - Control flow is clear
3. Production-ready - v1.0 stable, 60%+ adoption
4. Flexible - Handles simple to complex workflows
5. Observable - Built-in tracing and logging
6. Scalable - Designed for production deployment

---

Next Steps

• Read About Crewai - Alternative multi-agent framework
• See Langgraph In Action - Real-world examples

---

Last Updated: August 9, 2026