1.1 什么是 LangGraph
LangGraph 是 LangChain 团队推出的有状态、多步骤工作流框架。如果说 LangChain 是「一条流水线」,那 LangGraph 就是「一张导航图」。
图 1-1:LangChain vs LangGraph 架构对比
面试要点
LangGraph 本质上是一个有向图,每个节点可以是 LLM、工具或任意 Python 函数,边定义了节点间的流向。
1.2 为什么需要 LangGraph
当你的 AI 应用需要多步骤推理、循环迭代或人工介入时,LangChain 的线性链式调用就不够用了。
| 场景 | LangChain | LangGraph |
|---|---|---|
| 简单问答 | ✅ 适合 | ✅ 可以 |
| ReAct 推理 | ⚠️ 需特殊处理 | ✅ 原生支持 |
| 多 Agent 协作 | ❌ 不支持 | ✅ 完整支持 |
| 人机交互 | ❌ 不支持 | ✅ 支持中断恢复 |
| 状态持久化 | ❌ 不支持 | ✅ Checkpointer |
图 1-2:State 在节点间的流转过程
1.3 核心概念
1.3.1 State(状态)
State 是贯穿整个图的共享数据容器。你可以定义任意结构的状态:
pythonfrom typing import TypedDict, Annotated
from operator import add
class AgentState(TypedDict):
# 自动累加消息(不覆盖,追加)
messages: Annotated[list, add]
# 当前城市
city: str
# 天气信息(可为 None)
weather: str | None
Reducer 是什么?
Annotated[list, add] 告诉 LangGraph:每次更新 messages 时,用 add 函数合并。这样新消息会追加到历史,而不是覆盖。
1.3.2 Nodes(节点)
节点是处理函数,接收当前状态,返回要更新的字段:
python# 节点函数签名:输入 State,返回 dict
def extract_city(state: AgentState) -> dict:
user_msg = state["messages"][-1].content
city = extract_from_text(user_msg) # 简单解析
return {"city": city}
def fetch_weather(state: AgentState) -> dict:
weather = call_weather_api(state["city"])
return {"weather": weather}
图 1-3:Node 执行模型 — 输入 State,输出差量更新
关键理解
Node 只返回它想更新的字段,不需要返回整个 State。LangGraph 会用 Reducer 自动合并更新。比如只改了 city,messages 保持不变。
1.3.3 Edges(边)
边定义节点之间的流向:
pythonfrom langgraph.graph import StateGraph, START, END
# 构建图
graph = StateGraph(AgentState)
# 添加节点
graph.add_node("extract_city", extract_city)
graph.add_node("fetch_weather", fetch_weather)
# 添加边
graph.add_edge(START, "extract_city") # 起点
graph.add_edge("extract_city", "fetch_weather") # 普通边
graph.add_edge("fetch_weather", END) # 终点
图 1-2:LangGraph 三要素
1.4 环境安装
bash# 基础安装
pip install langgraph langchain-openai
# 可选:状态持久化后端
pip install langgraph-checkpoint
# 验证版本
python -c "import langgraph; print(langgraph.__version__)"
版本要求
LangGraph 需要 Python 3.10+ 和 LangChain 0.1.20+。建议使用虚拟环境隔离依赖。
1.5 第一个 LangGraph 应用
实战:构建一个天气问答助手,从用户问题中提取城市并查询天气。
步骤 1:定义状态
pythonfrom typing import TypedDict, Annotated
from operator import add
class WeatherState(TypedDict):
messages: Annotated[list, add]
city: str
weather: str | None
步骤 2:定义节点
pythondef extract_city(state: WeatherState):
last_msg = state["messages"][-1].content
city = extract_city_from_text(last_msg)
return {"city": city}
def fetch_weather(state: WeatherState):
weather = weather_api.get(state["city"])
return {"weather": weather}
步骤 3:构建图
pythonfrom langgraph.graph import StateGraph, START, END
graph = StateGraph(WeatherState)
graph.add_node("extract_city", extract_city)
graph.add_node("fetch_weather", fetch_weather)
graph.add_edge(START, "extract_city")
graph.add_edge("extract_city", "fetch_weather")
graph.add_edge("fetch_weather", END)
app = graph.compile() # 编译后才能执行
步骤 4:执行
pythonresult = app.invoke({
"messages": [{"role": "user", "content": "北京今天天气怎么样?"}],
"city": "",
"weather": None
})
print(result["weather"])
图 1-4:LangGraph 执行流程 — compile() + invoke() 全链路
1.6 本章小结
- LangGraph 是有状态的多步骤工作流框架
- 核心三要素:State(状态)、Nodes(节点)、Edges(边)
- State 使用
Annotated[type, reducer]定义,reducer 决定如何合并更新 - 图必须编译(compile)后才能执行
- 与 LangChain 的关系:LangGraph 构建在 LangChain 之上,专注复杂工作流
下章预告
第 2 章我们将深入学习 State 的高级用法:条件更新、多个 Reducer、以及如何设计复杂的状态结构。

