LangGraph学习
摘要: LangGraph作为LangChain生态的扩展,专为解决复杂AI应用中的非线性流程而设计。传统线性Chain难以处理循环、条件分支和持久化状态等需求,而LangGraph通过状态管理、节点函数和流程边三大核心概念,将LLM应用升级为可编排的"流程图"。其支持循环逻辑、持久化状态恢复和多Agent协作等高级特性,尤其适合构建自主Agent系统。本文通过加法器Agent
从零开始掌握 LangGraph:构建下一代 AI Agent 的终极指南
摘要:当 LangChain 的线性链(Chain)无法满足复杂的业务逻辑时,我们需要更强大的工具。本文将带你深入理解 LangGraph 的核心概念,通过实战代码掌握状态管理、循环逻辑与人机协作,开启构建复杂 AI Agent 的大门。
1. 为什么我们需要 LangGraph?
在 LLM 应用开发的早期,LangChain 凭借其 Chain(链)的概念迅速流行。它非常适合处理线性的任务流程:输入 -> 提示词 -> 模型 -> 输出。
然而,现实世界的应用场景往往不是线性的。想象一下你要构建一个自主 Agent:
- 它需要思考(Thinking)
- 它需要调用工具(Tool Use)
- 如果工具报错,它需要重试(Retry Loop)
- 在关键操作前,它需要人类确认(Human-in-the-loop)
这种包含循环(Cycles)、条件分支和持久化状态的逻辑,用传统的线性 Chain 很难优雅地实现。
这就是 LangGraph 登场的时刻。它是 LangChain 生态的一部分,专门用于构建有状态的(Stateful)、多角色(Multi-actor) 的应用程序。简单来说,它把 LLM 应用从“流水线”变成了“流程图”。
2. LangGraph 的核心概念
理解 LangGraph,只需要掌握三个核心要素。你可以把它想象成在画一张流程图:
2.1 状态(State)
这是 LangGraph 的灵魂。整个图共享一个状态对象(通常是一个 TypedDict 或 Pydantic 模型)。每个节点都可以读取和更新这个状态。
- 比喻:就像游戏中的“存档”,记录着当前的对话历史、变量值和执行进度。
2.2 节点(Nodes)
节点是执行具体工作的 Python 函数。
- 它可以是调用 LLM。
- 它可以是执行一段代码。
- 它可以是调用外部 API。
- 关键点:节点接收当前状态,返回更新后的状态。
2.3 边(Edges)
边决定了流程的走向。
- 普通边:从节点 A 无条件走到节点 B。
- 条件边:根据节点 A 的输出结果,决定走到节点 B 还是节点 C(例如:如果意图是“查询天气”走 A,否则走 B)。
3. 快速上手:Hello LangGraph
让我们通过一个简单的代码示例来感受 LangGraph 的工作流。我们将构建一个简单的“加法器 Agent",它可以连续接收数字并累加。
3.1 环境准备
pip install langgraph langchain-core
3.2 定义状态与节点
from typing import TypedDict, Annotated, List
from langgraph.graph import StateGraph, END
import operator
# 1. 定义状态
class AgentState(TypedDict):
numbers: Annotated[List[int], operator.add] # 使用 operator.add 实现列表自动合并
total: int
# 2. 定义节点函数
def input_number(state: AgentState):
# 模拟用户输入一个数字
new_num = 5
print(f"输入数字:{new_num}")
return {"numbers": [new_num]}
def calculate_sum(state: AgentState):
current_sum = sum(state['numbers'])
print(f"当前总和:{current_sum}")
return {"total": current_sum}
# 3. 构建图
workflow = StateGraph(AgentState)
# 添加节点
workflow.add_node("input", input_number)
workflow.add_node("calculate", calculate_sum)
# 添加边 (设置入口和流程)
workflow.set_entry_point("input")
workflow.add_edge("input", "calculate")
workflow.add_edge("calculate", END)
# 编译
app = workflow.compile()
3.3 运行
initial_state = {"numbers": [], "total": 0}
result = app.invoke(initial_state)
print(result)
# 输出:{'numbers': [5], 'total': 5}
虽然这个例子很简单,但它展示了 State 如何在节点间传递,以及 Graph 如何编排逻辑。
4. 进阶特性:LangGraph 的杀手锏
真正让 LangGraph 强大的是以下三个高级特性,它们解决了生产级 AI 应用的核心痛点。
4.1 循环与 Agent 逻辑
在 LangGraph 中,你可以轻松创建循环。这是构建 ReAct Agent(推理 - 行动)的基础。
- 逻辑:
LLM 思考->决定调用工具->执行工具->返回结果给 LLM->LLM 再次思考。 - 实现:通过条件边(Conditional Edges),如果 LLM 认为任务未完成,将流程指回 LLM 节点。
4.2 持久化与时间旅行(Persistence & Time Travel)
LangGraph 内置了 Checkpointer。
- 断点续传:如果程序崩溃,可以从上一步的状态继续执行,无需重新计算。
- 人机协作:在敏感操作前暂停流程,等待人类批准。
- 时间旅行:你可以查看历史状态快照,甚至回滚到之前的某个步骤重新修改参数运行。这对于调试 Agent 极其有用。
from langgraph.checkpoint.memory import MemorySaver
memory = MemorySaver()
app = workflow.compile(checkpointer=memory)
# 运行时传入 thread_id 即可保存状态
config = {"configurable": {"thread_id": "user_123"}}
app.invoke(input_state, config)
4.3 多 Agent 协作(Multi-Agent)
你可以将不同的 LangGraph 图嵌套在一起。例如:
- 主管节点:负责分发任务。
- 子图 A:负责写代码。
- 子图 B:负责测试代码。
- 子图 C:负责写文档。
这种模块化设计让复杂的系统变得可维护。
5. LangChain vs LangGraph:我该选哪个?
| 特性 | LangChain (Chains) | LangGraph |
|---|---|---|
| 控制流 | 线性为主 (DAG) | 支持循环 (Cycles) |
| 状态管理 | 隐式,传递较难 | 显式,全局共享 State |
| 持久化 | 需自行实现 | 内置 Checkpointer |
| 人机交互 | 较复杂 | 原生支持中断/恢复 |
| 适用场景 | 简单问答、RAG 流水线 | 复杂 Agent、工作流自动化 |
建议:如果你的应用只是简单的“检索 - 生成”,LangChain 足够且更轻量。如果你需要构建能自主规划、多步执行、需要人工干预的 Agent,LangGraph 是必选项。
6. 学习建议与避坑指南
在深入学习 LangGraph 的过程中,我总结了几条经验:
- 状态设计是关键:不要把所有数据都塞进 State。只存放需要在节点间共享的、持久化的数据。临时变量请在节点函数内部处理。
- 善用 LangSmith:LangGraph 的执行轨迹非常复杂,配合 LangSmith 的可视化追踪(Tracing),你能清晰地看到每一步的状态变化,调试效率提升 10 倍。
- 注意类型安全:尽量使用 Pydantic 模型来定义 State,这样 IDE 能提供自动补全,减少字段拼写错误。
- 避免无限循环:在编写条件边逻辑时,务必设置最大迭代次数(recursion_limit),防止 Agent 陷入死循环消耗 Token。
更多推荐

所有评论(0)