LangGraph 入门教程
LangGraph 是由 LangChain 团队开发的一个低层级 Agent 编排框架，专为构建有状态（Stateful）、长时运行的 AI 工作流而设计。
与传统的线性 LLM 调用链不同，LangGraph 将工作流建模为有向图（Directed Graph）：
- 节点（Node）：执行具体操作的函数（如调用 LLM、执行工具、处理数据）
- 边（Edge）：定义节点之间的流转路径，支持条件分支
- 状态（State）：在整个工作流中共享并传递的数据
开源地址：https://github.com/langchain-ai/langgraph。
想象你正在指挥一场交响乐演出：传统的 LLM Chain 就像演奏一首从头到尾的曲子，只能顺序播放；而 LangGraph 则像一位指挥家，可以根据现场观众的反应随时调整演奏顺序，让某个乐章重复，或者跳转到特定段落。它让 AI 工作流拥有了"指挥"的智慧——能够循环、分支、回溯，真正实现复杂的自主决策。
为什么选择 LangGraph？
下表对比了传统 LLM Chain 与 LangGraph 的主要差异：
| 特性 | 传统 LLM Chain | LangGraph |
|---|---|---|
| 工作流结构 | 线性，单向执行 | 图结构，支持循环 |
| 状态管理 | 需手动管理 | 内置状态持久化 |
| 条件路由 | 实现复杂 | 原生支持 |
| 人机协作 | 需要额外开发 | 内置支持 interrupt |
| 多 Agent 协调 | 实现困难 | 一流支持 |
| 调试工具 | 有限 | LangGraph Studio |
适用场景
- 对话机器人：需要记忆多轮对话上下文
- 自主 Agent：能够规划、使用工具、迭代思考
- 多 Agent 系统：多个 AI 协同完成复杂任务
- 审批工作流：需要人工审核的自动化流程
- 研究助手：需要多步骤推理和信息检索
核心概念
在开始编写代码之前，先理解 LangGraph 的三大核心概念。
Graph（图）
Graph 是整个工作流的蓝图，定义了 Agent 的完整逻辑结构。它由节点（Nodes）和边（Edges）组成：
StateGraph
|-- Nodes（节点）
| |-- node_a
| |-- node_b
| +-- node_c
+-- Edges（边）
|-- START -> node_a
|-- node_a -> node_b（条件边）
|-- node_a -> node_c（条件边）
+-- node_b -> END
State（状态）
State 是贯穿整个图的共享数据结构。每个节点可以读取和更新 State，更新后的 State 会传递给下一个节点。
from typing import TypedDict, Annotated
from langgraph.graph import add_messages
class MyState(TypedDict):
messages: Annotated[list, add_messages] # 消息列表（自动追加）
user_name: str # 用户名称
step_count: int # 步骤计数
重要：
Annotated[list, add_messages] 表示该字段使用 add_messages 作为 reducer——新消息会追加到列表而不是覆盖。这是 LangGraph 状态管理的核心机制。
Nodes（节点）
节点是普通的 Python 函数，接收当前 State，返回更新后的 State（部分字段）。
def my_node(state: MyState) -> dict:
# 读取状态
messages = state["messages"]
# 执行操作...
result = "处理结果"
# 返回更新的字段（不需要返回所有字段）
return {"messages": [{"role": "ai", "content": result}]}
Edges（边）
边定义节点之间的流转方式：
- 普通边：固定路径，
node_a -> node_b - 条件边：根据 State 动态路由，
node_a -> node_b 或 node_c - 起始边：
START -> 第一个节点 - 结束边：
某节点 -> END
环境搭建
安装依赖
使用国内镜像安装 LangGraph 和相关依赖：
# 创建虚拟环境（推荐） python -m venv venv source venv/bin/activate # macOS/Linux # venv\Scripts\activate # Windows # 安装 LangGraph 和 LangChain pip install langgraph langchain langchain-openai python-dotenv -i https://mirrors.aliyun.com/pypi/simple/ # 可选：安装开发工具 pip install langgraph-cli jupyter -i https://mirrors.aliyun.com/pypi/simple/
配置 API Key
在项目根目录创建 .env 文件，同时配置 OpenAI 和 DeepSeek：
# .env 文件内容 # OpenAI 配置（国外用户） OPENAI_API_KEY=sk-xxx OPENAI_BASE_URL=https://api.openai.com/v1 # DeepSeek 配置（国内用户推荐） DEEPSEEK_API_KEY=sk-xxx DEEPSEEK_BASE_URL=https://api.deepseek.com DEEPSEEK_MODEL=deepseek-v4-pro
提示：本文示例同时支持 OpenAI 和 DeepSeek。推荐国内用户使用 DeepSeek，API 地址配置为
https://api.deepseek.com（不带 /v1），LangChain 会自动拼接路径。DeepSeek API Key 可在 https://platform.deepseek.com/api_keys 创建。
验证安装
import langgraph
print(f"LangGraph 版本: {langgraph.__version__}")
第一个 LangGraph 程序
让我们从最简单的例子开始——一个只有两个节点的线性工作流。
实例
from langgraph.graph import StateGraph, START, END
from typing import TypedDict
# Step 1: 定义 State
class SimpleState(TypedDict):
message: str
processed: bool
# Step 2: 定义节点函数
def greet_node(state: SimpleState) -> dict:
"""欢迎节点：生成问候语"""
print(f"[greet_node] 收到消息: {state['message']}")
return {"message": f"你好！{state['message']}"}
def process_node(state: SimpleState) -> dict:
"""处理节点：标记为已处理"""
print(f"[process_node] 处理消息: {state['message']}")
return {"processed": True}
# Step 3: 构建图
builder = StateGraph(SimpleState)
# 添加节点
builder.add_node("greet", greet_node)
builder.add_node("process", process_node)
# 添加边
builder.add_edge(START, "greet")
builder.add_edge("greet", "process")
builder.add_edge("process", END)
# Step 4: 编译图
graph = builder.compile()
# Step 5: 运行
result = graph.invoke({
"message": "世界",
"processed": False
})
print(f"\n最终结果: {result}")
from typing import TypedDict
# Step 1: 定义 State
class SimpleState(TypedDict):
message: str
processed: bool
# Step 2: 定义节点函数
def greet_node(state: SimpleState) -> dict:
"""欢迎节点：生成问候语"""
print(f"[greet_node] 收到消息: {state['message']}")
return {"message": f"你好！{state['message']}"}
def process_node(state: SimpleState) -> dict:
"""处理节点：标记为已处理"""
print(f"[process_node] 处理消息: {state['message']}")
return {"processed": True}
# Step 3: 构建图
builder = StateGraph(SimpleState)
# 添加节点
builder.add_node("greet", greet_node)
builder.add_node("process", process_node)
# 添加边
builder.add_edge(START, "greet")
builder.add_edge("greet", "process")
builder.add_edge("process", END)
# Step 4: 编译图
graph = builder.compile()
# Step 5: 运行
result = graph.invoke({
"message": "世界",
"processed": False
})
print(f"\n最终结果: {result}")
运行结果：
[greet_node] 收到消息: 世界
[process_node] 处理消息: 你好！世界
最终结果: {'message': '你好！世界', 'processed': True}
可视化图结构
在 Jupyter Notebook 中可以直接可视化图结构：
# 在 Jupyter Notebook 中可视化 from IPython.display import Image Image(graph.get_graph().draw_mermaid_png()) # 或者打印 Mermaid 格式 print(graph.get_graph().draw_mermaid())
State 状态管理
使用 TypedDict 定义状态
from typing import TypedDict, Annotated, Optional
from langgraph.graph.message import add_messages
class AgentState(TypedDict):
# 消息历史（add_messages reducer 自动追加而非覆盖）
messages: Annotated[list, add_messages]
# 普通字段（直接覆盖）
user_id: str
session_id: str
# 可选字段
error: Optional[str]
# 计数器（使用 operator.add 作为 reducer）
retry_count: Annotated[int, lambda x, y: x + y]
使用 Pydantic 定义状态（推荐用于生产）
from pydantic import BaseModel, Field
from typing import Annotated
from langgraph.graph.message import add_messages
class ProductionState(BaseModel):
messages: Annotated[list, add_messages] = Field(default_factory=list)
user_id: str = ""
confidence_score: float = 0.0
class Config:
arbitrary_types_allowed = True
MessagesState（内置快捷状态）
LangGraph 提供了内置的 MessagesState，专为对话场景设计：
from langgraph.graph import MessagesState # MessagesState 等价于: # class MessagesState(TypedDict): # messages: Annotated[list[AnyMessage], add_messages] # 直接使用，无需自定义 builder = StateGraph(MessagesState)
Nodes 节点
普通函数节点
def simple_node(state: AgentState) -> dict:
# 读取状态
last_message = state["messages"][-1]
# 执行操作
response = f"收到: {last_message.content}"
# 返回部分状态更新
return {
"messages": [{"role": "assistant", "content": response}]
}
LLM 调用节点
实例
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage
load_dotenv()
# 使用 DeepSeek 模型
llm = ChatOpenAI(
model=os.getenv('DEEPSEEK_MODEL', 'deepseek-v4-pro'),
openai_api_key=os.getenv('DEEPSEEK_API_KEY'),
openai_api_base=os.getenv('DEEPSEEK_BASE_URL', 'https://api.deepseek.com'),
temperature=0
)
def llm_node(state: dict) -> dict:
"""调用 LLM 的节点"""
system_prompt = SystemMessage(content="你是一个有帮助的助手。")
# 将系统提示与对话历史合并
messages = [system_prompt] + state["messages"]
# 调用 LLM
response = llm.invoke(messages)
return {"messages": [response]}
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage
load_dotenv()
# 使用 DeepSeek 模型
llm = ChatOpenAI(
model=os.getenv('DEEPSEEK_MODEL', 'deepseek-v4-pro'),
openai_api_key=os.getenv('DEEPSEEK_API_KEY'),
openai_api_base=os.getenv('DEEPSEEK_BASE_URL', 'https://api.deepseek.com'),
temperature=0
)
def llm_node(state: dict) -> dict:
"""调用 LLM 的节点"""
system_prompt = SystemMessage(content="你是一个有帮助的助手。")
# 将系统提示与对话历史合并
messages = [system_prompt] + state["messages"]
# 调用 LLM
response = llm.invoke(messages)
return {"messages": [response]}
异步节点
import asyncio
async def async_node(state: AgentState) -> dict:
"""异步节点，适合 I/O 密集型操作"""
# 模拟异步操作（如 API 调用、数据库查询）
await asyncio.sleep(0.1)
result = await some_async_api_call(state["messages"][-1].content)
return {"messages": [{"role": "assistant", "content": result}]}
# 使用异步图
result = await graph.ainvoke({"messages": [...]})
使用类作为节点
class RouterNode:
def __init__(self, llm, system_prompt: str):
self.llm = llm
self.system_prompt = system_prompt
def __call__(self, state: AgentState) -> dict:
"""类实例可以作为节点使用"""
messages = [
SystemMessage(content=self.system_prompt),
*state["messages"]
]
response = self.llm.invoke(messages)
return {"messages": [response]}
# 添加类节点
router = RouterNode(llm, "你是一个专业的路由助手。")
builder.add_node("router", router)
Edges 边与条件路由
普通边
# 固定路径：node_a 完成后始终执行 node_b
builder.add_edge("node_a", "node_b")
# 结束：node_a 完成后图结束
builder.add_edge("node_a", END)
条件边
条件边是 LangGraph 的核心功能，根据当前 State 动态决定下一步。
def route_after_llm(state: AgentState) -> str:
"""
路由函数：根据 LLM 的最新输出决定走哪条路径
返回值必须是已注册节点名称或 END
"""
last_message = state["messages"][-1]
# 如果 LLM 请求使用工具
if hasattr(last_message, "tool_calls") and last_message.tool_calls:
return "tools"
# 否则结束
return END
# 添加条件边
builder.add_conditional_edges(
"llm", # 源节点
route_after_llm, # 路由函数
{
"tools": "tool_executor", # 返回 "tools" 时 -> tool_executor 节点
END: END # 返回 END 时 -> 结束
}
)
并行执行（Fan-out）
# 从一个节点并行分叉到多个节点
builder.add_edge("start_node", "branch_a")
builder.add_edge("start_node", "branch_b")
builder.add_edge("start_node", "branch_c")
# 多个节点汇聚到一个节点（Fan-in）
builder.add_edge("branch_a", "merge_node")
builder.add_edge("branch_b", "merge_node")
builder.add_edge("branch_c", "merge_node")
完整条件路由示例
实例
import os
from dotenv import load_dotenv
from langgraph.graph import StateGraph, START, END, MessagesState
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage
load_dotenv()
# 初始化 LLM
llm = ChatOpenAI(
model=os.getenv('DEEPSEEK_MODEL', 'deepseek-v4-pro'),
openai_api_key=os.getenv('DEEPSEEK_API_KEY'),
openai_api_base=os.getenv('DEEPSEEK_BASE_URL', 'https://api.deepseek.com'),
temperature=0.7
)
# 路由函数
def classify_intent(state: MessagesState) -> str:
"""根据用户意图路由到不同的 Agent"""
last_message = state["messages"][-1]
content = last_message.content.lower()
if "天气" in content or "温度" in content:
return "weather_agent"
elif "代码" in content or "编程" in content:
return "code_agent"
elif "再见" in content or "退出" in content:
return "farewell"
else:
return "general_agent"
# 定义各个 Agent 节点
def router_node(state: MessagesState) -> dict:
"""路由节点：不做处理，只用于触发路由判断"""
return {}
def weather_node(state: MessagesState) -> dict:
"""天气 Agent"""
response = llm.invoke([
SystemMessage(content="你是一个天气助手，友好地回答天气相关问题。如果没有实时数据，可以给出一般性建议。"),
*state["messages"]
])
return {"messages": [response]}
def code_node(state: MessagesState) -> dict:
"""代码 Agent"""
response = llm.invoke([
SystemMessage(content="你是一个编程助手，擅长解答代码问题并给出清晰的代码示例。"),
*state["messages"]
])
return {"messages": [response]}
def general_node(state: MessagesState) -> dict:
"""通用 Agent"""
response = llm.invoke([
SystemMessage(content="你是一个友善的 AI 助手，可以回答各种问题。"),
*state["messages"]
])
return {"messages": [response]}
def farewell_node(state: MessagesState) -> dict:
"""告别节点"""
return {"messages": [{"role": "assistant", "content": "再见！期待下次与你交流。"}]}
# 构建图
builder = StateGraph(MessagesState)
# 添加节点
builder.add_node("router", router_node)
builder.add_node("weather_agent", weather_node)
builder.add_node("code_agent", code_node)
builder.add_node("general_agent", general_node)
builder.add_node("farewell", farewell_node)
# 添加边
builder.add_edge(START, "router")
builder.add_conditional_edges(
"router",
classify_intent,
{
"weather_agent": "weather_agent",
"code_agent": "code_agent"
from dotenv import load_dotenv
from langgraph.graph import StateGraph, START, END, MessagesState
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage
load_dotenv()
# 初始化 LLM
llm = ChatOpenAI(
model=os.getenv('DEEPSEEK_MODEL', 'deepseek-v4-pro'),
openai_api_key=os.getenv('DEEPSEEK_API_KEY'),
openai_api_base=os.getenv('DEEPSEEK_BASE_URL', 'https://api.deepseek.com'),
temperature=0.7
)
# 路由函数
def classify_intent(state: MessagesState) -> str:
"""根据用户意图路由到不同的 Agent"""
last_message = state["messages"][-1]
content = last_message.content.lower()
if "天气" in content or "温度" in content:
return "weather_agent"
elif "代码" in content or "编程" in content:
return "code_agent"
elif "再见" in content or "退出" in content:
return "farewell"
else:
return "general_agent"
# 定义各个 Agent 节点
def router_node(state: MessagesState) -> dict:
"""路由节点：不做处理，只用于触发路由判断"""
return {}
def weather_node(state: MessagesState) -> dict:
"""天气 Agent"""
response = llm.invoke([
SystemMessage(content="你是一个天气助手，友好地回答天气相关问题。如果没有实时数据，可以给出一般性建议。"),
*state["messages"]
])
return {"messages": [response]}
def code_node(state: MessagesState) -> dict:
"""代码 Agent"""
response = llm.invoke([
SystemMessage(content="你是一个编程助手，擅长解答代码问题并给出清晰的代码示例。"),
*state["messages"]
])
return {"messages": [response]}
def general_node(state: MessagesState) -> dict:
"""通用 Agent"""
response = llm.invoke([
SystemMessage(content="你是一个友善的 AI 助手，可以回答各种问题。"),
*state["messages"]
])
return {"messages": [response]}
def farewell_node(state: MessagesState) -> dict:
"""告别节点"""
return {"messages": [{"role": "assistant", "content": "再见！期待下次与你交流。"}]}
# 构建图
builder = StateGraph(MessagesState)
# 添加节点
builder.add_node("router", router_node)
builder.add_node("weather_agent", weather_node)
builder.add_node("code_agent", code_node)
builder.add_node("general_agent", general_node)
builder.add_node("farewell", farewell_node)
# 添加边
builder.add_edge(START, "router")
builder.add_conditional_edges(
"router",
classify_intent,
{
"weather_agent": "weather_agent",
"code_agent": "code_agent"