LangGraph学习
从零开始掌握 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 核心优势
langgraph为任何长期运行的有状态工作流或者代理提供底层基础设施,有以下核心优势:
- 持久执行构建能够在失败中持续运行、并能从中断处恢复运行代理。
- 人机参与:通过检查和修改代理状态,纳入人工监督。
- 综合记忆:创建具有短期工作记忆(用于持续推理)和跨会话长期记忆的有状态智能体。
- 使用langsmith进行调试:通过可视化工具,追踪执行路径、捕捉状态转换并提供详细的运行时指标,深入了解复杂代理行为。
- 生产准备部署:自信的部署复杂的代理系统,配备可扩展的基础设施,以应对有的状态长期运行的独立工作流挑战。
4. 快速上手:Hello LangGraph
让我们通过一个简单的代码示例来感受 LangGraph 的工作流。我们将构建一个简单的“加法器 Agent",它可以连续接收数字并累加。
4.1 环境准备
pipinstalllanggraph langchain-core4.2 定义状态与节点
fromtypingimportTypedDict,Annotated,Listfromlanggraph.graphimportStateGraph,ENDimportoperator# 1. 定义状态classAgentState(TypedDict):numbers:Annotated[List[int],operator.add]# 使用 operator.add 实现列表自动合并total:int# 2. 定义节点函数definput_number(state:AgentState):# 模拟用户输入一个数字new_num=5print(f"输入数字:{new_num}")return{"numbers":[new_num]}defcalculate_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()4.3 运行
initial_state={"numbers":[],"total":0}result=app.invoke(initial_state)print(result)# 输出:{'numbers': [5], 'total': 5}虽然这个例子很简单,但它展示了State如何在节点间传递,以及Graph如何编排逻辑。
5. 进阶特性:LangGraph 的杀手锏
真正让 LangGraph 强大的是以下三个高级特性,它们解决了生产级 AI 应用的核心痛点。
5.1 循环与 Agent 逻辑
在 LangGraph 中,你可以轻松创建循环。这是构建ReAct Agent(推理 - 行动)的基础。
- 逻辑:
LLM 思考->决定调用工具->执行工具->返回结果给 LLM->LLM 再次思考。 - 实现:通过条件边(Conditional Edges),如果 LLM 认为任务未完成,将流程指回 LLM 节点。
5.2 持久化与时间旅行(Persistence & Time Travel)
LangGraph 内置了Checkpointer。
- 断点续传:如果程序崩溃,可以从上一步的状态继续执行,无需重新计算。
- 人机协作:在敏感操作前暂停流程,等待人类批准。
- 时间旅行:你可以查看历史状态快照,甚至回滚到之前的某个步骤重新修改参数运行。这对于调试 Agent 极其有用。
fromlanggraph.checkpoint.memoryimportMemorySaver memory=MemorySaver()app=workflow.compile(checkpointer=memory)# 运行时传入 thread_id 即可保存状态config={"configurable":{"thread_id":"user_123"}}app.invoke(input_state,config)5.3 多 Agent 协作(Multi-Agent)
你可以将不同的 LangGraph 图嵌套在一起。例如:
- 主管节点:负责分发任务。
- 子图 A:负责写代码。
- 子图 B:负责测试代码。
- 子图 C:负责写文档。
这种模块化设计让复杂的系统变得可维护。
6. LangChain vs LangGraph:我该选哪个?
| 特性 | LangChain (Chains) | LangGraph |
|---|---|---|
| 控制流 | 线性为主 (DAG) | 支持循环 (Cycles) |
| 状态管理 | 隐式,传递较难 | 显式,全局共享 State |
| 持久化 | 需自行实现 | 内置 Checkpointer |
| 人机交互 | 较复杂 | 原生支持中断/恢复 |
| 适用场景 | 简单问答、RAG 流水线 | 复杂 Agent、工作流自动化 |
建议:如果你的应用只是简单的“检索 - 生成”,LangChain 足够且更轻量。如果你需要构建能自主规划、多步执行、需要人工干预的 Agent,LangGraph 是必选项。
7. 学习建议与避坑指南
在深入学习 LangGraph 的过程中,我总结了几条经验:
- 状态设计是关键:不要把所有数据都塞进 State。只存放需要在节点间共享的、持久化的数据。临时变量请在节点函数内部处理。
- 善用 LangSmith:LangGraph 的执行轨迹非常复杂,配合 LangSmith 的可视化追踪(Tracing),你能清晰地看到每一步的状态变化,调试效率提升 10 倍。
- 注意类型安全:尽量使用 Pydantic 模型来定义 State,这样 IDE 能提供自动补全,减少字段拼写错误。
- 避免无限循环:在编写条件边逻辑时,务必设置最大迭代次数(recursion_limit),防止 Agent 陷入死循环消耗 Token。
