Langgraph Implementation

LangGraph Implementation

Core Concepts

LangGraph builds stateful, multi-actor agent applications using a graph-based architecture:

• StateGraph: Builder class for defining graphs with shared state
• Nodes: Functions that read state and return partial updates
• Edges: Define execution flow (static or conditional)
• Channels: Internal state management (LastValue, BinaryOperatorAggregate)
• Checkpointer: Persistence for pause/resume capabilities

Implementation gates

Use these sequenced checks for persistence and human-in-the-loop flows (avoid “it should work” without evidence):

1. Checkpointed runs

• Build config with {"configurable": {"thread_id": "<stable-id>"}} before invoke / ainvoke.
• Pass: The same thread_id is reused for every turn of one conversation; a new conversation uses a new id.

2. State after a step

• Pass: graph.get_state(config).values (or equivalent) contains the keys and reducer outputs your next node or client expects; if not, fix routing, reducers, or node order before continuing.

3. Interrupt and resume (HITL)

• Pass: After a pause, you have inspected pending work (get_state, and your LangGraph version’s interrupt listing if you rely on it) so you know which node is waiting and what resume payload shape to send.
• Pass: Command(resume=...) (or equivalent) includes every field the code path after interrupt() reads.

4. Checkpointer vs environment

• Pass: Tests or local dev use InMemorySaver or disposable SQLite; production uses a durable checkpointer configured for that deployment (not in-memory).

Essential Imports

from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import MessagesState, add_messages
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command, Send, interrupt, RetryPolicy
from typing import Annotated
from typing_extensions import TypedDict

State Schema Patterns

Basic State with TypedDict

import operator

class State(TypedDict):
    counter: int                                    # LastValue - stores last value
    messages: Annotated[list, operator.add]         # Reducer - appends lists
    items: Annotated[list, lambda a, b: a + [b] if b else a]  # Custom reducer

MessagesState for Chat Applications

from langgraph.graph.message import MessagesState

class State(MessagesState):
    # Inherits: messages: Annotated[list[AnyMessage], add_messages]
    user_id: str
    context: dict

Pydantic State (for validation)

from pydantic import BaseModel

class State(BaseModel):
    messages: Annotated[list, add_messages]
    validated_field: str  # Pydantic validates on assignment

Building Graphs

Basic Pattern

builder = StateGraph(State)

# Add nodes - functions that take state, return partial updates
builder.add_node("process", process_fn)
builder.add_node("decide", decide_fn)

# Add edges
builder.add_edge(START, "process")
builder.add_edge("process", "decide")
builder.add_edge("decide", END)

# Compile
graph = builder.compile()

Node Function Signature

def my_node(state: State) -> dict:
    """Node receives full state, returns partial update."""
    return {"counter": state["counter"] + 1}

# With config access
def my_node(state: State, config: RunnableConfig) -> dict:
    thread_id = config["configurable"]["thread_id"]
    return {"result": process(state, thread_id)}

# With Runtime context (v0.6+)
def my_node(state: State, runtime: Runtime[Context]) -> dict:
    user_id = runtime.context.get("user_id")
    return {"result": user_id}

Conditional Edges

from typing import Literal

def router(state: State) -> Literal["agent", "tools", "__end__"]:
    last_msg = state["messages"][-1]
    if hasattr(last_msg, "tool_calls") and last_msg.tool_calls:
        return "tools"
    return END  # or "__end__"

builder.add_conditional_edges("agent", router)

# With path_map for visualization
builder.add_conditional_edges(
    "agent",
    router,
    path_map={"agent": "agent", "tools": "tools", "__end__": END}
)

Command Pattern (Dynamic Routing + State Update)

from langgraph.types import Command

def dynamic_node(state: State) -> Command[Literal["next", "__end__"]]:
    if state["should_continue"]:
        return Command(goto="next", update={"step": state["step"] + 1})
    return Command(goto=END)

# Must declare destinations for visualization
builder.add_node("dynamic", dynamic_node, destinations=["next", END])

Send Pattern (Fan-out/Map-Reduce)

from langgraph.types import Send

def fan_out(state: State) -> list[Send]:
    """Route to multiple node instances with different inputs."""
    return [Send("worker", {"item": item}) for item in state["items"]]

builder.add_conditional_edges(START, fan_out)
builder.add_edge("worker", "aggregate")  # Workers converge

Checkpointing

Enable Persistence

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.checkpoint.sqlite import SqliteSaver  # Development
from langgraph.checkpoint.postgres import PostgresSaver  # Production

# In-memory (testing only)
graph = builder.compile(checkpointer=InMemorySaver())

# SQLite (development)
with SqliteSaver.from_conn_string("checkpoints.db") as checkpointer:
    graph = builder.compile(checkpointer=checkpointer)

# Thread-based invocation
config = {"configurable": {"thread_id": "user-123"}}
result = graph.invoke({"messages": [...]}, config)

State Management

# Get current state
state = graph.get_state(config)

# Get state history
for state in graph.get_state_history(config):
    print(state.values, state.next)

# Update state manually
graph.update_state(config, {"key": "new_value"}, as_node="node_name")

Human-in-the-Loop

Using interrupt()

from langgraph.types import interrupt, Command

def review_node(state: State) -> dict:
    # Pause and surface value to client
    human_input = interrupt({"question": "Please review", "data": state["draft"]})
    return {"approved": human_input["approved"]}

# Resume with Command
graph.invoke(Command(resume={"approved": True}), config)

Interrupt Before/After Nodes

graph = builder.compile(
    checkpointer=checkpointer,
    interrupt_before=["human_review"],  # Pause before node
    interrupt_after=["agent"],          # Pause after node
)

# Check pending interrupts
state = graph.get_state(config)
if state.next:  # Has pending nodes
    # Resume
    graph.invoke(None, config)

Streaming

# Stream modes: "values", "updates", "custom", "messages", "debug"

# Updates only (node outputs)
for chunk in graph.stream(input, stream_mode="updates"):
    print(chunk)  # {"node_name": {"key": "value"}}

# Full state after each step
for chunk in graph.stream(input, stream_mode="values"):
    print(chunk)

# Multiple modes
for mode, chunk in graph.stream(input, stream_mode=["updates", "messages"]):
    if mode == "messages":
        print("Token:", chunk)

# Custom streaming from within nodes
from langgraph.config import get_stream_writer

def my_node(state):
    writer = get_stream_writer()
    writer({"progress": 0.5})  # Custom event
    return {"result": "done"}

Subgraphs

# Define subgraph
sub_builder = StateGraph(SubState)
sub_builder.add_node("step", step_fn)
sub_builder.add_edge(START, "step")
subgraph = sub_builder.compile()

# Use as node in parent
parent_builder = StateGraph(ParentState)
parent_builder.add_node("subprocess", subgraph)
parent_builder.add_edge(START, "subprocess")

# Subgraph checkpointing
subgraph = sub_builder.compile(
    checkpointer=None,   # Inherit from parent (default)
    # checkpointer=True,   # Use persistent checkpointing
    # checkpointer=False,  # Disable checkpointing
)

Retry and Caching

from langgraph.types import RetryPolicy, CachePolicy

retry = RetryPolicy(
    initial_interval=0.5,
    backoff_factor=2.0,
    max_attempts=3,
    retry_on=ValueError,  # Or callable: lambda e: isinstance(e, ValueError)
)

cache = CachePolicy(ttl=3600)  # Cache for 1 hour

builder.add_node("risky", risky_fn, retry_policy=retry, cache_policy=cache)

Prebuilt Components

create_react_agent (moved to langchain.agents in v1.0)

from langgraph.prebuilt import create_react_agent, ToolNode

# Simple agent
graph = create_react_agent(
    model="anthropic:claude-3-5-sonnet",
    tools=[my_tool],
    prompt="You are a helpful assistant",
    checkpointer=InMemorySaver(),
)

# Custom tool node
tool_node = ToolNode([tool1, tool2])
builder.add_node("tools", tool_node)

Common Patterns

Agent Loop

def should_continue(state) -> Literal["tools", "__end__"]:
    if state["messages"][-1].tool_calls:
        return "tools"
    return END

builder.add_node("agent", call_model)
builder.add_node("tools", ToolNode(tools))
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", should_continue)
builder.add_edge("tools", "agent")

Parallel Execution

# Multiple nodes execute in parallel when they share the same trigger
builder.add_edge(START, "node_a")
builder.add_edge(START, "node_b")  # Runs parallel with node_a
builder.add_edge(["node_a", "node_b"], "join")  # Wait for both

See PATTERNS.md for advanced patterns including multi-agent systems, hierarchical graphs, and complex workflows.

Advanced LangGraph Patterns

Multi-Agent Supervisor Pattern

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

class SupervisorState(TypedDict):
    messages: Annotated[list, add_messages]
    next_agent: str

def supervisor(state: SupervisorState) -> Command[Literal["researcher", "coder", "__end__"]]:
    """Route to appropriate agent based on task."""
    # LLM decides which agent to use
    decision = llm.invoke(state["messages"])
    if "research" in decision.content.lower():
        return Command(goto="researcher")
    elif "code" in decision.content.lower():
        return Command(goto="coder")
    return Command(goto=END)

def researcher(state: SupervisorState) -> dict:
    result = research_agent.invoke(state["messages"])
    return {"messages": [result]}

def coder(state: SupervisorState) -> dict:
    result = coding_agent.invoke(state["messages"])
    return {"messages": [result]}

builder = StateGraph(SupervisorState)
builder.add_node("supervisor", supervisor, destinations=["researcher", "coder", END])
builder.add_node("researcher", researcher)
builder.add_node("coder", coder)
builder.add_edge(START, "supervisor")
builder.add_edge("researcher", "supervisor")
builder.add_edge("coder", "supervisor")

Map-Reduce Pattern

from langgraph.types import Send

class MapReduceState(TypedDict):
    topics: list[str]
    results: Annotated[list[str], operator.add]  # Reducer aggregates

def distribute(state: MapReduceState) -> list[Send]:
    """Fan out to process each topic."""
    return [Send("process_topic", {"topic": t}) for t in state["topics"]]

def process_topic(state: dict) -> dict:
    """Process individual topic (receives Send payload)."""
    result = analyze(state["topic"])
    return {"results": [result]}

def aggregate(state: MapReduceState) -> dict:
    """Combine all results."""
    summary = summarize(state["results"])
    return {"summary": summary}

builder = StateGraph(MapReduceState)
builder.add_conditional_edges(START, distribute)
builder.add_edge("process_topic", "aggregate")
builder.add_edge("aggregate", END)

Hierarchical Graph Pattern

# Inner graph - specialized task
class InnerState(TypedDict):
    query: str
    result: str

inner_builder = StateGraph(InnerState)
inner_builder.add_node("search", search_fn)
inner_builder.add_node("analyze", analyze_fn)
inner_builder.add_edge(START, "search")
inner_builder.add_edge("search", "analyze")
inner_builder.add_edge("analyze", END)
inner_graph = inner_builder.compile()

# Outer graph - orchestration
class OuterState(TypedDict):
    messages: Annotated[list, add_messages]
    research_result: str

def prepare_research(state: OuterState) -> dict:
    """Transform outer state for inner graph."""
    return {"query": state["messages"][-1].content}

def process_result(state: OuterState) -> dict:
    """Handle result from inner graph."""
    return {"messages": [AIMessage(content=state["research_result"])]}

outer_builder = StateGraph(OuterState)
outer_builder.add_node("prepare", prepare_research)
outer_builder.add_node("research", inner_graph)  # Subgraph as node
outer_builder.add_node("process", process_result)
outer_builder.add_edge(START, "prepare")
outer_builder.add_edge("prepare", "research")
outer_builder.add_edge("research", "process")
outer_builder.add_edge("process", END)

Reflection/Self-Correction Pattern

class ReflectionState(TypedDict):
    draft: str
    feedback: str
    revision_count: int

def generate(state: ReflectionState) -> dict:
    if state.get("feedback"):
        prompt = f"Revise based on: {state['feedback']}\n\nDraft: {state['draft']}"
    else:
        prompt = "Generate initial draft"
    return {"draft": llm.invoke(prompt).content}

def reflect(state: ReflectionState) -> dict:
    feedback = critic_llm.invoke(f"Critique this: {state['draft']}").content
    return {"feedback": feedback, "revision_count": state.get("revision_count", 0) + 1}

def should_continue(state: ReflectionState) -> Literal["generate", "__end__"]:
    if state["revision_count"] >= 3:
        return END
    if "looks good" in state["feedback"].lower():
        return END
    return "generate"

builder = StateGraph(ReflectionState)
builder.add_node("generate", generate)
builder.add_node("reflect", reflect)
builder.add_edge(START, "generate")
builder.add_edge("generate", "reflect")
builder.add_conditional_edges("reflect", should_continue)

Plan-and-Execute Pattern

class PlanExecuteState(TypedDict):
    objective: str
    plan: list[str]
    completed_steps: Annotated[list[str], operator.add]
    current_step: int

def planner(state: PlanExecuteState) -> dict:
    plan = planning_llm.invoke(f"Create plan for: {state['objective']}")
    steps = parse_steps(plan.content)
    return {"plan": steps, "current_step": 0}

def executor(state: PlanExecuteState) -> dict:
    step = state["plan"][state["current_step"]]
    result = execute_step(step)
    return {
        "completed_steps": [f"{step}: {result}"],
        "current_step": state["current_step"] + 1
    }

def should_continue(state: PlanExecuteState) -> Literal["executor", "__end__"]:
    if state["current_step"] >= len(state["plan"]):
        return END
    return "executor"

builder = StateGraph(PlanExecuteState)
builder.add_node("planner", planner)
builder.add_node("executor", executor)
builder.add_edge(START, "planner")
builder.add_edge("planner", "executor")
builder.add_conditional_edges("executor", should_continue)

Human Approval Gate Pattern

from langgraph.types import interrupt, Command

class ApprovalState(TypedDict):
    action: str
    approved: bool
    result: str

def propose_action(state: ApprovalState) -> dict:
    action = determine_action(state)
    return {"action": action}

def human_review(state: ApprovalState) -> dict:
    decision = interrupt({
        "action": state["action"],
        "message": "Please approve or reject this action"
    })
    return {"approved": decision.get("approved", False)}

def execute_action(state: ApprovalState) -> dict:
    if state["approved"]:
        result = execute(state["action"])
    else:
        result = "Action rejected by human"
    return {"result": result}

def route_after_review(state: ApprovalState) -> Literal["execute", "__end__"]:
    return "execute" if state["approved"] else END

builder = StateGraph(ApprovalState)
builder.add_node("propose", propose_action)
builder.add_node("review", human_review)
builder.add_node("execute", execute_action)
builder.add_edge(START, "propose")
builder.add_edge("propose", "review")
builder.add_conditional_edges("review", route_after_review)
builder.add_edge("execute", END)

graph = builder.compile(checkpointer=checkpointer)

# Usage
config = {"configurable": {"thread_id": "1"}}
result = graph.invoke({"action": ""}, config)
# Graph pauses at review node

# Resume with approval
graph.invoke(Command(resume={"approved": True}), config)

Branching and Joining

class BranchState(TypedDict):
    input: str
    branch_a_result: str
    branch_b_result: str
    final_result: str

builder = StateGraph(BranchState)
builder.add_node("branch_a", branch_a_fn)
builder.add_node("branch_b", branch_b_fn)
builder.add_node("join", join_fn)

# Fan out - both run in parallel
builder.add_edge(START, "branch_a")
builder.add_edge(START, "branch_b")

# Fan in - wait for both
builder.add_edge(["branch_a", "branch_b"], "join")
builder.add_edge("join", END)

Looping with Counter

class LoopState(TypedDict):
    value: int
    iterations: int

def increment(state: LoopState) -> dict:
    return {
        "value": state["value"] * 2,
        "iterations": state["iterations"] + 1
    }

def should_loop(state: LoopState) -> Literal["increment", "__end__"]:
    if state["iterations"] >= 5:
        return END
    if state["value"] >= 1000:
        return END
    return "increment"

builder = StateGraph(LoopState)
builder.add_node("increment", increment)
builder.add_edge(START, "increment")
builder.add_conditional_edges("increment", should_loop)

Error Recovery Pattern

from langgraph.types import RetryPolicy

class ErrorRecoveryState(TypedDict):
    input: str
    result: str
    error: str
    attempts: int

def risky_operation(state: ErrorRecoveryState) -> dict:
    try:
        result = dangerous_api_call(state["input"])
        return {"result": result, "error": ""}
    except Exception as e:
        return {"error": str(e), "attempts": state.get("attempts", 0) + 1}

def fallback(state: ErrorRecoveryState) -> dict:
    return {"result": f"Fallback result for: {state['input']}"}

def route_after_operation(state: ErrorRecoveryState) -> Literal["fallback", "__end__"]:
    if state["error"] and state["attempts"] >= 3:
        return "fallback"
    if state["error"]:
        return "risky_operation"  # Retry
    return END

# With RetryPolicy for automatic retries
retry = RetryPolicy(max_attempts=3, retry_on=ConnectionError)
builder.add_node("risky_operation", risky_operation, retry_policy=retry)