Build stateful AI agents and agentic workflows with LangGraph in Python...
Build stateful AI agents and workflows by defining graphs of nodes (steps) connected by edges (transitions).
Minimal chatbot with memory:
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AnyMessage
from typing_extensions import TypedDict, Annotated
import operator
# 1. Define state
class State(TypedDict):
messages: Annotated[list[AnyMessage], operator.add] # Append mode
# 2. Define node
llm = ChatOpenAI(model="gpt-4")
def chat(state: State) -> dict:
response = llm.invoke(state["messages"])
return {"messages": [response]}
# 3. Build graph
graph = StateGraph(State)
graph.add_node("chat", chat)
graph.add_edge(START, "chat")
graph.add_edge("chat", END)
# 4. Compile with memory
chain = graph.compile(checkpointer=InMemorySaver())
# 5. Invoke with thread_id for persistence
result = chain.invoke(
{"messages": [HumanMessage(content="Hello!")]},
config={"configurable": {"thread_id": "user-123"}}
)
print(result["messages"][-1].content)
Key patterns:
Annotated[list, operator.add] ā append to list instead of replaceInMemorySaver() ā enables memory across invocationsthread_id ā identifies conversation for persistenceThe Quick Start above covers this. Add more nodes for preprocessing or postprocessing as needed.
Agent that calls external tools (APIs, calculators, search) in a loop until task complete. ā See references/tool-agent-pattern.md
Multi-step pipeline with conditional branches, parallel execution, or prompt chaining. ā See references/workflow-patterns.md
Persist conversation across sessions, enable time-travel debugging, survive crashes. ā See references/persistence-memory.md
Pause for human approval, correction, or additional input mid-workflow. ā See references/hitl-patterns.md
Unit test nodes, visualize graphs, trace with LangSmith. ā See references/debugging-monitoring.md
Build supervisor or swarm-based multi-agent workflows with handoff tools. ā See references/multi-agent-patterns.md
Deploy to LangGraph Platform (cloud/self-hosted) or custom infrastructure. ā See references/production-deployment.md
Learn core concepts: State, Nodes, Edges, Graph APIs. ā See references/core-api.md
Store facts, not formatted prompts. Each node can format data as needed.
# ā Good: raw data
class State(TypedDict):
user_question: str
retrieved_docs: list[str]
intent: str
# ā Bad: pre-formatted
class State(TypedDict):
full_prompt: str # Mixes data with formatting
Each node does one thing. Name it descriptively.
# ā Good: clear responsibilities
graph.add_node("classify_intent", classify_intent)
graph.add_node("search_knowledge", search_knowledge)
graph.add_node("generate_response", generate_response)
Use conditional edges for decisions. Don't hide routing logic inside nodes.
def route_by_intent(state) -> str:
if state["intent"] == "billing":
return "billing_handler"
return "general_handler"
graph.add_conditional_edges("classify", route_by_intent,
["billing_handler", "general_handler"])
Any list field that accumulates values needs operator.add:
class State(TypedDict):
messages: Annotated[list, operator.add] # ā Appends
current_step: str # Replaces (no annotation)
| Error Type | Strategy |
|---|---|
| Transient (network) | Use RetryPolicy on node |
| LLM-recoverable (parse fail) | Feed error to LLM via state, loop back |
| User-fixable (missing info) | Use interrupt() to pause and ask |
| Unexpected (bugs) | Let bubble up for debugging |
def node(state) -> dict for each stepadd_node(), add_edge(), add_conditional_edges()graph.compile(), test with sample inputsoperator.add on ListsSymptom: Messages disappear, only last message retained.
# ā Wrong: messages: list[AnyMessage]
# ā Fix: messages: Annotated[list[AnyMessage], operator.add]
thread_id for MemorySymptom: Agent forgets previous turns.
# ā Fix: Always pass config with thread_id
chain.invoke(input, config={"configurable": {"thread_id": "unique-id"}})
Symptom: AttributeError on graph object.
# ā Wrong: graph.invoke(input)
# ā Fix: chain = graph.compile(); chain.invoke(input)
Symptom: Different results on resume from checkpoint.
from langgraph.func import task
@task # Wrap for durable execution
def fetch_data(state):
return {"data": requests.get(url).json()}
Symptom: ImportError when defining state classes.
# ā Fix: Use string annotations
from __future__ import annotations
# Core
pip install -U langgraph
# LLM providers (pick one or more)
pip install langchain-openai
pip install langchain-anthropic
# Production persistence
pip install langgraph-checkpoint-postgres
# Observability
pip install langsmith
Environment variables:
export OPENAI_API_KEY="sk-..."
export ANTHROPIC_API_KEY="sk-ant-..."
export LANGSMITH_API_KEY="ls-..."
export LANGSMITH_TRACING=true
python -c "import langgraph; print(langgraph.__version__)" worksOPENAI_API_KEY or ANTHROPIC_API_KEY)LANGSMITH_API_KEY for tracingchain = graph.compile()print(chain.get_graph().draw_mermaid())chain.invoke({...})operator.add annotations)thread_id twice)# Imports
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from typing_extensions import TypedDict, Annotated
import operator
# State with append-mode list
class State(TypedDict):
messages: Annotated[list, operator.add]
# Node signature
def node(state: State) -> dict:
return {"messages": [new_message]}
# Graph construction
graph = StateGraph(State)
graph.add_node("name", node_fn)
graph.add_edge(START, "name")
graph.add_edge("name", END)
# Conditional routing
graph.add_conditional_edges("from", router_fn, ["option1", "option2", END])
# Compile and run
chain = graph.compile(checkpointer=InMemorySaver())
result = chain.invoke(input, config={"configurable": {"thread_id": "id"}})
# Visualization
print(chain.get_graph().draw_mermaid())
For detailed API reference ā See references/core-api.md