LangGraph是构建大模型Agent的革命性框架,相比传统LangChain的链式结构,它引入了循环能力和状态管理机制。通过State(状态)、Nodes(节点)和Edges(边)三个核心概念,开发者可以构建能自我纠错、具有长期记忆甚至人机协作的复杂Agent。文章通过实战演示了如何构建会反思的Agent,并介绍了其持久化记忆和人机交互特性,为开发者提供了从入门到实践的完整路径。
在过去的一年多里,LLM 应用开发经历了一个显著的范式转移:我们正在从构建 Chains 转向构建 Agents。如果你是 LangChain 的老用户,你可能习惯了 LangChain Expression Language那种行云流水的链式调用。然而,当你试图构建一个能自我纠错、长期记忆、甚至能模拟多人协作的复杂 Agent 时,你会发现“链”这种结构变得捉襟见肘。
实不相瞒,之前写Agents,我基本不依赖外部框架,一般直接用Python代码手写。最近接手一个其他同事的项目,是用LangGraph写的,所以就俗速成了一下,把整理的文档分享出来吧。
一、 为什么我们需要 LangGraph?
要理解 LangGraph,首先要理解 DAG(有向无环图) 的局限性。
1.1 链的困境:开弓没有回头箭
传统的 LangChain 就像一条工厂流水线。输入(原料)进去,经过步骤 A、步骤 B、步骤 C,最后输出(产品)。
- 特点:确定性强,路径单一。
- 缺点:一旦开始,无法回头。如果步骤 C 发现步骤 A 的结果有误,整条链无法自动跳转回 A 重做。
1.2 图的革命:引入“循环”与“状态”
真实的智能往往包含循环(Loops)。
比如写代码:写代码 -> 运行 -> 报错 -> 看错误日志 -> 修改代码 -> 再运行。这是一个循环迭代的过程,直到成功为止。
LangGraph 的核心哲学在于:
- 循环能力(Cyclic):允许控制流回到之前的节点。
- 状态主导(Stateful):不再只是传递字符串,而是维护一个全局的“状态对象”(State),所有节点都在读取和修改这个共享状态。
打个比方:
- LangChain 是接力赛跑,棒子(数据)传给下一个人,跑完就结束。
- LangGraph 是一个项目会议室。白板上写着项目状态(State),不同的专家(Nodes)轮流走上白板更新内容,项目经理(Edges)根据白板上的最新进展,决定下一个叫谁上来,或者是不是该打回去重做。
二、 LangGraph 的重点概念
在使用 LangGraph 之前,我们需要了解几个基本概念:
1. State(状态):应用的记忆中枢
这是图的“白板”。通常是一个 PythonTypedDict或 Pydantic 模型。它定义了在这个应用生命周期内,我们需要追踪哪些数据(聊天记录、中间结果、工具输出等)。
2. Nodes(节点):执行者
节点本质上就是 Python 函数。
ounter(line func(state) -> new_state_update它们接收当前状态,执行逻辑(调用 LLM、查数据库、运行代码),然后返回需要更新的数据。
3. Edges(边):控制流
边决定了“下一步去哪”。
- Normal Edge:由于 A,所以 B。
- Conditional Edge:这是 Agent 的“大脑”。根据当前状态(例如 LLM 是否决定调用工具),动态决定跳到哪个节点。
三、 实战:构建一个会反思的 Agent
在这个 Demo 中,我们不仅让 AI 调用工具,还赋予它循环执行的能力。如果 LLM 觉得需要工具,它会进入工具节点,执行完后强制循环回到 LLM 节点,让 LLM 根据工具结果生成最终答案。
3.1 环境准备
ounter(line pip install langgraph langchain langchain-openai3.2 代码实现
ounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(line import operator from typing import Annotated, TypedDict, Union, Literal from langchain_openai import ChatOpenAI from langchain_core.tools import tool from langchain_core.messages import BaseMessage, HumanMessage, AIMessage from langgraph.graph import StateGraph, START, END from langgraph.prebuilt import ToolNode # --- 1. 定义状态 (The Memory) --- # 使用 add_messages 策略:新的消息会 append 到列表末尾,而不是覆盖 from langgraph.graph.message import add_messages class AgentState(TypedDict): # Annotated 允许我们定义状态更新的策略 messages: Annotated[list, add_messages] # --- 2. 定义工具 (The Tools) --- @tool def get_weather(city: str): """获取指定城市的实时天气""" # 模拟 API 返回 if "上海" in city: return "上海今天多云,22摄氏度。" return "未知地区天气数据。" tools = [get_weather] # --- 3. 初始化模型 (The Brain) --- # 绑定工具,让 LLM 知道它有哪些能力 llm = ChatOpenAI(model="gpt-4o", temperature=0) llm_with_tools = llm.bind_tools(tools) # --- 4. 定义节点逻辑 (The Workers) --- def agent_node(state: AgentState): """ Agent 节点:负责思考和生成回复,或者发起工具调用请求 """ messages = state["messages"] response = llm_with_tools.invoke(messages) # 返回的内容会自动 merge 到 state["messages"] 中 return {"messages": [response]} # 利用 LangGraph 预构建的 ToolNode,它会自动执行 tool_calls tool_node = ToolNode(tools) # --- 5. 定义条件边逻辑 (The Router) --- def should_continue(state: AgentState) -> Literal["tools", END]: """ 决策逻辑:检查上一条消息是否包含工具调用 """ last_message = state["messages"][-1] # 如果 LLM 决定调用工具 if last_message.tool_calls: return "tools" # 否则,任务结束 return END # --- 6. 构建图 (The Graph Assembly) --- workflow = StateGraph(AgentState) # 添加节点 workflow.add_node("agent", agent_node) workflow.add_node("tools", tool_node) # 设置入口 workflow.add_edge(START, "agent") # 添加条件边:从 agent 出来后,走 should_continue 逻辑 workflow.add_conditional_edges( "agent", should_continue, ) # 关键循环:工具执行完,必须回到 agent,让 LLM 看到结果并生成最终回复 workflow.add_edge("tools", "agent") # 编译应用 app = workflow.compile() # --- 7. 运行 (Execution) --- print("=== 开始运行 Agent ===") inputs = {"messages": [HumanMessage(content="上海今天天气怎么样?")]} # stream 模式让我们能看到每一步的思考过程 for chunk in app.stream(inputs): for node_name, node_state in chunk.items(): print(f"--- 节点: {node_name} 完成执行 ---") print(f"最新消息: {node_state['messages'][-1].content}") print("-------------------------------")四、 LangGraph 的重要特性
如果仅仅是上面的循环,普通的 Pythonwhile循环也能写。LangGraph 真正的威力在于以下两点:
1. Persistence(持久化与记忆)
LangGraph 内置了 Checkpointer 机制。它可以将图的每一步状态自动保存到数据库(Redis, Postgres 等)。
这意味着什么?
- 长程对话:即使用户隔了三天再来,你只需提供
thread_id,Agent 就能加载之前的状态(State),无缝继续。 - 时光回溯:你可以获取 Agent 以前的某个状态,修改它,然后从那里重新分叉运行。这对于调试和“反事实推理”极其有用。
2. Human-in-the-loop
在企业级应用中,全自动的 Agent 是危险的(比如自动退款、群发邮件)。LangGraph 允许你在特定的节点前设置interrupt_before。
工作流如下:
- Agent 思考并决定:“我要给用户退款 1000 元”。
- LangGraph 检测到下一步是敏感操作,暂停运行,保存当前状态。
- 系统通知管理员。
- 管理员在后台查看:“Agent 要退款 1000 元,是否批准?”
- 管理员点击“批准”或“修改为 500 元”。
- LangGraph 恢复运行,Agent 继续执行。
ounter(lineounter(lineounter(lineounter(lineounter(lineounter(line # 简单的 Human-in-the-loop 示例 memory = MemorySaver() app = workflow.compile( checkpointer=memory, interrupt_before=["tools"] # 在执行工具前暂停 )最后,如果你想真正掌握 LangGraph,建议从以下路径入手:
- 跑通上面的 Demo。
- 尝试添加一个
checkpointer实现对话记忆。 - 去官网查看 Multi-Agent 的案例,学习如何让一个“写手 Agent”和一个“审核 Agent”互相配合工作。
如何学习大模型 AI ?
由于新岗位的生产效率,要优于被取代岗位的生产效率,所以实际上整个社会的生产效率是提升的。
但是具体到个人,只能说是:
“最先掌握AI的人,将会比较晚掌握AI的人有竞争优势”。
这句话,放在计算机、互联网、移动互联网的开局时期,都是一样的道理。
我在一线互联网企业工作十余年里,指导过不少同行后辈。帮助很多人得到了学习和成长。
我意识到有很多经验和知识值得分享给大家,也可以通过我们的能力和经验解答大家在人工智能学习中的很多困惑,所以在工作繁忙的情况下还是坚持各种整理和分享。但苦于知识传播途径有限,很多互联网行业朋友无法获得正确的资料得到学习提升,故此将并将重要的AI大模型资料包括AI大模型入门学习思维导图、精品AI大模型学习书籍手册、视频教程、实战学习等录播视频免费分享出来。
第一阶段(10天):初阶应用
该阶段让大家对大模型 AI有一个最前沿的认识,对大模型 AI 的理解超过 95% 的人,可以在相关讨论时发表高级、不跟风、又接地气的见解,别人只会和 AI 聊天,而你能调教 AI,并能用代码将大模型和业务衔接。
- 大模型 AI 能干什么?
- 大模型是怎样获得「智能」的?
- 用好 AI 的核心心法
- 大模型应用业务架构
- 大模型应用技术架构
- 代码示例:向 GPT-3.5 灌入新知识
- 提示工程的意义和核心思想
- Prompt 典型构成
- 指令调优方法论
- 思维链和思维树
- Prompt 攻击和防范
- …
第二阶段(30天):高阶应用
该阶段我们正式进入大模型 AI 进阶实战学习,学会构造私有知识库,扩展 AI 的能力。快速开发一个完整的基于 agent 对话机器人。掌握功能最强的大模型开发框架,抓住最新的技术进展,适合 Python 和 JavaScript 程序员。
- 为什么要做 RAG
- 搭建一个简单的 ChatPDF
- 检索的基础概念
- 什么是向量表示(Embeddings)
- 向量数据库与向量检索
- 基于向量检索的 RAG
- 搭建 RAG 系统的扩展知识
- 混合检索与 RAG-Fusion 简介
- 向量模型本地部署
- …
第三阶段(30天):模型训练
恭喜你,如果学到这里,你基本可以找到一份大模型 AI相关的工作,自己也能训练 GPT 了!通过微调,训练自己的垂直大模型,能独立训练开源多模态大模型,掌握更多技术方案。
到此为止,大概2个月的时间。你已经成为了一名“AI小子”。那么你还想往下探索吗?
- 为什么要做 RAG
- 什么是模型
- 什么是模型训练
- 求解器 & 损失函数简介
- 小实验2:手写一个简单的神经网络并训练它
- 什么是训练/预训练/微调/轻量化微调
- Transformer结构简介
- 轻量化微调
- 实验数据集的构建
- …
第四阶段(20天):商业闭环
对全球大模型从性能、吞吐量、成本等方面有一定的认知,可以在云端和本地等多种环境下部署大模型,找到适合自己的项目/创业方向,做一名被 AI 武装的产品经理。
- 硬件选型
- 带你了解全球大模型
- 使用国产大模型服务
- 搭建 OpenAI 代理
- 热身:基于阿里云 PAI 部署 Stable Diffusion
- 在本地计算机运行大模型
- 大模型的私有化部署
- 基于 vLLM 部署大模型
- 案例:如何优雅地在阿里云私有部署开源大模型
- 部署一套开源 LLM 项目
- 内容安全
- 互联网信息服务算法备案
- …
学习是一个过程,只要学习就会有挑战。天道酬勤,你越努力,就会成为越优秀的自己。
如果你能在15天内完成所有的任务,那你堪称天才。然而,如果你能完成 60-70% 的内容,你就已经开始具备成为一名大模型 AI 的正确特征了。