langgraph

LangGraph Patterns

LangGraph builds stateful multi-step LLM workflows as directed graphs. Each node is a Python function; edges define routing between them.

When to Activate

• Building a multi-step LLM pipeline (research → draft → review → publish)
• Implementing human-in-the-loop interrupts or approval steps
• Designing conditional routing based on LLM output
• Adding persistence/memory to an agent across sessions
• Streaming intermediate results to the client
• Coordinating multiple agents as subgraphs
• Debugging InvalidUpdateError, cycle errors, or state shape issues

---

Core Concepts

StateGraph
├── State        — TypedDict that flows through every node
├── Nodes        — functions: State → State update (partial dict)
├── Edges        — unconditional routing A → B
├── Conditional  — function decides which node to go to next
└── Checkpointer — persists state between invocations (memory)

---

Minimal Example

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI

# 1. Define state — Annotated[list, add_messages] appends instead of replacing
class State(TypedDict):
    messages: Annotated[list, add_messages]

llm = ChatOpenAI(model="gpt-4o-mini")

# 2. Define a node — receives full state, returns partial update
def chatbot(state: State) -> dict:
    return {"messages": [llm.invoke(state["messages"])]}

# 3. Build the graph
graph = (
    StateGraph(State)
    .add_node("chatbot", chatbot)
    .add_edge(START, "chatbot")
    .add_edge("chatbot", END)
    .compile()
)

# 4. Invoke
result = graph.invoke({"messages": [{"role": "user", "content": "Hello!"}]})
print(result["messages"][-1].content)

---

State Design

from typing import TypedDict, Annotated
from operator import add

# Annotated reducers control how values merge on update
class ResearchState(TypedDict):
    # add_messages: appends new messages, deduplicates by ID
    messages: Annotated[list, add_messages]

    # add (operator.add): appends items from each node update
    sources: Annotated[list[str], add]

    # Last-write-wins (default — no annotation needed)
    query: str
    status: str
    final_report: str | None

    # Optional fields
    error: str | None