方寸 Portal

AI Engineer Portal

Your personal operating system for career transition.

Private mode
Back to Learning Paths
Lesson

LangGraph for agent orchestration

Model agents as graphs with StateGraph, route between nodes conditionally, and add checkpointing for human-in-the-loop workflows.

50 min
ai-agents-and-toolsphase-1portfolio

Why this matters

Simple agent loops — call LLM, check if done, call tool, repeat — break down as soon as you add real requirements: parallel branches, human approval steps, recovery from tool failures, or long-running workflows that need to survive process restarts. LangGraph turns the implicit loop into an explicit graph, making complex agent behavior debuggable, testable, and extensible.

For engineers building production agents, graph-based orchestration is the pattern that scales past the demo stage. LangGraph is the most mature implementation of this idea in the Python ecosystem as of 2025.

Core concepts

Graph-based agent architecture vs simple loops

A simple agent loop:

def run_agent(query: str) -> str:
    messages = [{"role": "user", "content": query}]
    while True:
        response = llm.invoke(messages)
        if response.stop_reason == "end_turn":
            return response.content
        # execute tool, append result, continue
        tool_result = execute_tool(response.tool_use)
        messages.append(tool_result)

This works until you need: routing to different behaviors based on state, parallel tool execution, human approval before a destructive action, or checkpointing so the agent can resume after a crash. At that point the loop becomes a tangle of if/elif chains. LangGraph makes the control flow explicit.

StateGraph: nodes, edges, conditional routing

A LangGraph graph has three components:

  1. State — a typed dict passed between nodes
  2. Nodes — functions that take state and return updated state
  3. Edges — fixed or conditional transitions between nodes
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import BaseMessage, HumanMessage

class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    next_step: str

llm = ChatAnthropic(model="claude-3-5-sonnet-20241022")

def research_node(state: AgentState) -> AgentState:
    """Search for information and append findings."""
    last_message = state["messages"][-1]
    response = llm.invoke([
        HumanMessage(content=f"Research this topic and summarize key facts: {last_message.content}")
    ])
    return {"messages": [response], "next_step": "write"}

def write_node(state: AgentState) -> AgentState:
    """Take research findings and produce a final draft."""
    research = "\n".join(m.content for m in state["messages"] if hasattr(m, "content"))
    response = llm.invoke([
        HumanMessage(content=f"Write a clear summary based on this research:\n{research}")
    ])
    return {"messages": [response], "next_step": "done"}

def route(state: AgentState) -> str:
    """Conditional routing function — determines the next node."""
    return state.get("next_step", "research")

# Build the graph
builder = StateGraph(AgentState)
builder.add_node("research", research_node)
builder.add_node("write", write_node)

builder.set_entry_point("research")
builder.add_conditional_edges("research", route, {"write": "write", "done": END})
builder.add_conditional_edges("write", route, {"done": END})

graph = builder.compile()

result = graph.invoke({
    "messages": [HumanMessage(content="Explain vector databases")],
    "next_step": "research",
})
print(result["messages"][-1].content)

The routing function makes the control flow readable. You can trace any execution by following the edges.

Checkpointing and human-in-the-loop patterns

Checkpointing saves graph state at each node, enabling two things: resumption after failure, and human approval gates.

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, END, interrupt

class ReviewState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    draft: str
    approved: bool

def draft_node(state: ReviewState) -> ReviewState:
    response = llm.invoke(state["messages"])
    return {"draft": response.content, "approved": False}

def human_review_node(state: ReviewState) -> ReviewState:
    # interrupt() pauses execution and surfaces state to the caller
    # The human can inspect state["draft"] and decide
    human_decision = interrupt({"draft": state["draft"], "action": "approve_or_revise"})
    return {"approved": human_decision.get("approved", False)}

def publish_node(state: ReviewState) -> ReviewState:
    print(f"Publishing: {state['draft']}")
    return state

def route_after_review(state: ReviewState) -> str:
    return "publish" if state["approved"] else "draft"

builder = StateGraph(ReviewState)
builder.add_node("draft", draft_node)
builder.add_node("human_review", human_review_node)
builder.add_node("publish", publish_node)

builder.set_entry_point("draft")
builder.add_edge("draft", "human_review")
builder.add_conditional_edges("human_review", route_after_review, {
    "publish": "publish",
    "draft": "draft",
})
builder.add_edge("publish", END)

# MemorySaver persists state in memory; use SqliteSaver or PostgresSaver in production
checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer, interrupt_before=["human_review"])

thread_config = {"configurable": {"thread_id": "session-001"}}

# Start execution — pauses at human_review
result = graph.invoke(
    {"messages": [HumanMessage(content="Write a blog post about LangGraph")]},
    config=thread_config,
)
print("Draft ready for review:", result["draft"][:100])

# Resume with human decision
final = graph.invoke(
    {"approved": True},
    config=thread_config,
)

This pattern is the foundation for any agent workflow that requires auditability or human oversight before irreversible actions.

Common mistakes

Putting business logic in edge conditions. Routing functions should be thin — read a field from state and return a string. Complex logic in routing is hard to test. Put the logic in the node and have the node set a next_step field.

Mutable state without the add_messages annotation. LangGraph uses reducers to merge state updates. Without Annotated[list, add_messages], appending to a messages list will overwrite it instead of extending it.

Not using checkpointing in multi-step workflows. A 10-step agent that fails on step 8 without a checkpointer loses all progress. Add MemorySaver during development and swap to a persistent backend before production.

Deeply nested subgraphs for simple branching. LangGraph supports subgraphs (a graph as a node), but two-branch routing does not need one. Use conditional edges for simple cases; reserve subgraphs for genuinely modular, reusable agent components.

Try it yourself

Build a LangGraph agent with three nodes: planner (decomposes a user request into steps), executor (runs one step using a tool of your choice), and reviewer (checks if all steps are done or flags which steps failed). Add a conditional edge from reviewer back to executor if there are incomplete steps, and to END when all steps succeed. Test it with a multi-step research request.

Lesson Deep Dive

Ask a follow-up question about this lesson and get an AI-powered explanation.

Loading previous questions...