如果你正在做 AI 应用开发,尤其是 Agent 方向的开发,最近一定绕不开一个名字:LangGraph。从 LangChain 到 LangGraph,再到各类 Agent 框架的涌现,这一波技术演进的速度非常快,很多开发者已经明显感觉到,过去“调模型、拼 Prompt、串 API”就能交付 AI 应用的阶段正在过去,真正决定应用上限的,已经从模型能力转移到了流程控制、状态管理和工程化落地上。
这篇文章不是简单的 “LangGraph 入门文档翻译”,而是一份面向 CSDN 技术读者的实战型解析。我会从实际开发中最关心的问题出发:LangGraph 到底解决了 LangChain 的什么痛点?Agent、工作流、图执行这些概念的本质是什么?如何从零构建一个带条件分支、循环、持久化和人工审批的真实 Agent?在进入企业级生产环境时,有哪些容易踩的坑和必须做的设计?
如果你正在准备 AI 大模型应用开发、Agent 开发相关的技术面试,或者想在自己的项目中引入更可控的 Agent 编排能力,这篇文章建议收藏后用。
1. 先搞清楚:LangGraph 到底解决了什么问题
先说一个核心判断:LangGraph 不是 LangChain 的简单升级版,而是一次设计范式上的补充。LangChain 核心提供的是一套标准化的组件抽象:模型封装、Prompt 模板、检索器、工具调用。你在 LangChain 里写 AI 应用,本质上是把一段段逻辑串联成链(Chain)。链的问题在于它更像一条直线,虽然可以做分支判断,但一旦涉及循环、回退、分支合并、人工介入、状态回滚这类真实的 Agent 场景,代码会迅速变得混乱且不可维护。
很多人第一次接触 LangGraph 时会问:我不是用 LangChain 也能写 Agent 吗?确实是。你可以用代码手写一个while循环,反复调用模型,判断要不要继续调工具。这在 Demo 阶段没有问题,但到了企业级场景,你需要考虑几个实际问题:
- 会话状态如何持久化?进程重启后,Agent 的上下文还在吗?
- 某一步执行失败,如何精确回滚或重试?
- 需要人工审核的环节,如何暂停流程等待用户输入,再继续执行?
- 多个分支并行执行,如何合并结果并维护清晰的数据流?
- 如何记录每一步的耗时、 Token 消耗和中间输出,做到可观测和审计?
LangGraph 给出的答案是:把 Agent 的执行过程建模为一张图。每个节点是一个计算步骤,每条边定义了步骤之间的流转关系,再加上一个全局的 State 对象承载数据、一个 Checkpointer 负责持久化和断点恢复。这套抽象借鉴了数据流编程和图计算的思想,放到 Agent 编排场景里,天然地解决了循环、分支、并行和人工介入这些核心问题。
如果说 LangChain 解决的是“模型和工具怎么封装”,LangGraph 解决的是“整个 Agent 过程的执行逻辑怎么编排才可控”。
从当下 AI 应用开发的大趋势看,2026 年最值得投入的不再是简单的 Prompt 拼接,而是 Agent 的工程化控制能力。LangGraph 在这一点上提供了近乎工业级的参考实现。
2. 基础概念:State、Node、Edge 与 Checkpointer
LangGraph 的核心概念并不复杂,但初次接触时容易被术语绕晕,尤其是 State 和 LangChain 中的 Message 搞混。下面逐个拆解。
2.1 State:Agent 的全局中央状态
State 是 LangGraph 中的灵魂对象。你可以把它理解成一个随着图执行不断更新的字典,里面保存了所有节点之间需要共享的数据。不同于普通函数参数,State 的设计有几个关键点:
- 所有节点共享同一个 State 对象。
- 每个节点执行完后,可以返回一份更新,LangGraph 会把更新合并回 State。
- 更新方式可以自定义:是覆盖字段,还是往列表字段里追加元素。
举个例子,简单定义一个 State:
from typing_extensions import TypedDict class AgentState(TypedDict): messages: list current_step: str retry_count: int这里messages保存对话历史,current_step记录当前流程阶段,retry_count用于控制循环次数。在实际执行中,不同节点会修改不同字段,共同维护 Agent 的完整上下文。
2.2 Node:每个节点就是一个执行单元
Node 就是图中的节点,可以是任意一个 Python 函数。函数签名是固定的:接收一个 State,返回 State 的部分更新或新值。例如一个简单节点:
def call_model(state: AgentState): messages = state["messages"] response = llm.invoke(messages) return {"messages": messages + [response]}注意,你没有必要把整个 State 都返回。只返回需要更新的字段即可,LangGraph 会自动合并。
2.3 Edge:定义流转路径
Edge 分为普通边和条件边。普通边表示“执行完 A 后固定执行 B”,条件边则根据当前 State 的值动态决定下一步走向哪个节点。条件边是 Agent 具备决策能力的关键。
graph.add_conditional_edges( "analyze", route_by_intent, { "answer": "generate_answer", "use_tool": "call_tool", "human": "ask_human" } )上面这段代码的意思是:执行完analyze节点后,调用route_by_intent函数,根据返回的字符串选择下一步进入哪个节点。
2.4 Checkpointer:让 Agent 拥有记忆和执行可恢复能力
Checkpointer 是 LangGraph 比较独特的机制。它会定期把 State 的完整快照保存下来。基于这个机制,你可以实现:
- 时间旅行:回到历史某个步骤重新执行。
- 断点恢复:Agent 执行到一半,进程崩溃,可以从最近检查点恢复。
- 人工介入:Agent 遇到需要人工审批的节点时暂停,等待人工输入后再继续。
from langgraph.checkpoint.memory import MemorySaver checkpointer = MemorySaver() graph = workflow.compile(checkpointer=checkpointer)开发阶段用MemorySaver就足够,生产环境建议使用 Postgres 或 Redis 等外部存储实现持久化检查点。
2.5 和 LangChain 的关系
可以这样理解:LangChain 是面向模型、Prompt、工具的组件库,LangGraph 是面向流程控制的编排引擎。LangGraph 本身不依赖 LangChain,但二者配合效果最好:用 LangChain 封装模型和工具,用 LangGraph 控制流程。两者定位并不冲突,有大量项目是 LangChain 负责 RAG 和工具接入,LangGraph 负责整体 Agent 状态机。
3. 环境准备与依赖安装
本文的实操部分使用 Python,建议版本为 Python 3.9 及以上。LangGraph 从 0.4 版本开始 API 已经相对稳定,安装时建议直接安装较新版本,确保和当前生态兼容。
创建虚拟环境并安装依赖:
mkdir langgraph-demo cd langgraph-demo python -m venv .venv # Linux / macOS source .venv/bin/activate # Windows # .venv\Scripts\activate安装核心依赖:
pip install langgraph langchain langchain-openai这里我使用 OpenAI 接口作为示例,你可以根据实际情况替换为 DeepSeek、通义千问、智谱等国内模型的 OpenAI 兼容接口。配置方式是通过环境变量设置 API Key 和 Base URL:
export OPENAI_API_KEY="your-api-key" export OPENAI_BASE_URL="https://api.openai.com/v1"使用国内兼容接口时,将OPENAI_BASE_URL换成服务商对应的地址即可。所有对话模型都走 OpenAI 兼容协议,LangChain 接入后不需要改业务代码。
验证安装是否成功:
python -c "from langgraph.graph import StateGraph; print('LangGraph OK')"如果这一行没有报错,说明环境已经就绪。
4. 从零实现一个企业级客服工单 Agent
理论知识容易理解,真正有价值的是把各种机制组合到一个完整的场景中。下面我们用 LangGraph 实现一个接近真实业务场景的客服工单处理 Agent。
这个 Agent 需要做的事:
- 接收用户提交的工单内容。
- 分析工单并判断意图。
- 能回答的智能问答直接生成答复。
- 需要查询知识库或外部系统的走工具查询。
- 识别到用户情绪强烈或问题复杂时,自动转人工并等待审批。
- 对 AI 生成的答复进行质量评估,不满意时自动重写,最多重写两轮。
- 整个流程状态可持久化,业务方可以随时查看进度,甚至中断回滚。
4.1 定义 State
from typing_extensions import TypedDict class TicketState(TypedDict): ticket_id: str user_message: str intent: str ai_response: str response_rating: int rewrite_count: int need_human: bool history: list4.2 定义节点函数
先定义 5 个节点:意图识别、知识库检索、AI 生成、质量评估、转人工。每个节点本质上是一个接收 State、返回部分更新的 Python 函数。
# 文件路径:demo/nodes.py def analyze_intent(state: TicketState): """识别工单意图""" prompt = f""" 用户提交了新的客服工单,请你分析其意图。 工单内容:{state["user_message"]} 只返回以下类型之一:一般咨询、退款申请、产品故障、投诉。 """ resp = llm.invoke(prompt) intent = resp.content.strip() return { "intent": intent, "history": state["history"] + [{"step": "analyze_intent", "result": intent}] } def retrieve_knowledge(state: TicketState): """根据意图检索知识库,这里用关键字模拟,实际项目可替换为向量检索""" knowledge_db = { "退款申请": "退款流程:用户可在订单页面发起退款申请,3 个工作日内审核完成。", "产品故障": "请用户先尝试重启应用,若问题仍存在,提供日志文件以便进一步排查。", "投诉": "需要优先安抚用户情绪,并记录问题细节,转交高级客服处理。" } answer = knowledge_db.get(state["intent"], "暂时没有检索到匹配的解决方案。") return {"ai_response": answer, "history": state["history"] + [{"step": "retrieve", "result": answer}]} def generate_answer(state: TicketState): """生成最终回复文案""" prompt = f""" 你是客服助手,请根据已知信息和工单内容生成一段友好、专业、简洁的回复。 工单内容:{state["user_message"]} 已知解决方案:{state["ai_response"]} 注意控制回复字数在 100 字以内。 """ resp = llm.invoke(prompt) return {"ai_response": resp.content.strip(), "history": state["history"] + [{"step": "generate", "result": resp.content.strip()}]} def evaluate_response(state: TicketState): """给回复质量打分,分数低于 6 分视为不合格""" prompt = f""" 请评价下面这条客服回复的完整度、语气和专业性,只输出 0 到 10 的数字。 用户问题:{state["user_message"]} 客服回复:{state["ai_response"]} """ resp = llm.invoke(prompt) try: score = int(resp.content.strip()) except ValueError: score = 5 return {"response_rating": score, "history": state["history"] + [{"step": "evaluate", "result": score}]} def ask_human(state: TicketState): """转人工审批,流程在这里会暂停等待人工处理""" return {"need_human": True, "history": state["history"] + [{"step": "human_handoff", "result": "waiting"}]}4.3 构建图和条件边
上面完成了节点定义,接下来把这些节点串联成一张有分支、有循环的图。
# 文件路径:demo/graph.py from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import MemorySaver # 1. 创建图实例 builder = StateGraph(TicketState) # 2. 添加节点 builder.add_node("analyze", analyze_intent) builder.add_node("retrieve", retrieve_knowledge) builder.add_node("generate", generate_answer) builder.add_node("evaluate", evaluate_response) builder.add_node("human", ask_human) # 3. 设置入口 builder.add_edge(START, "analyze") # 4. 根据意图决定下一步 def route_by_intent(state: TicketState): if state["intent"] == "投诉": return "human" return "retrieve" builder.add_conditional_edges( "analyze", route_by_intent, {"human": "human", "retrieve": "retrieve"} ) # 5. 工具知识库检索后生成回答 builder.add_edge("retrieve", "generate") # 6. 生成回答后进入质量评估 builder.add_edge("generate", "evaluate") # 7. 如果质量不合格且未超过重写次数限制,重新生成 def route_after_evaluate(state: TicketState): if state["response_rating"] < 6 and state["rewrite_count"] < 2: return "rewrite" return "end" builder.add_conditional_edges( "evaluate", route_after_evaluate, { "rewrite": "generate", "end": END } ) # 8. 人工审批结束回到生成节点,继续完成回复 builder.add_edge("human", "generate") # 9. 编译图,传入 checkpointer 以支持断点与持久化 checkpointer = MemorySaver() graph = builder.compile(checkpointer=checkpointer)注意这里第 7 步的条件路由有个细节:rewrite_count需要在某个节点递增,否则会因为重写次数一直为 0 而无限循环。可以在generate_answer中增加计数:
def generate_answer(state: TicketState): # 原有逻辑... current_count = state.get("rewrite_count", 0) return { "ai_response": resp.content.strip(), "rewrite_count": current_count + 1, "history": state["history"] + [{"step": "generate", "result": resp.content.strip()}] }每次重新生成都会让rewrite_count加 1,当超过 2 时,route_after_evaluate会返回end,流程自然终止。
这里真正容易踩坑的地方是:如果不修改rewrite_count,条件边会一直认为“次数没有达到上限”,从而反复调用生成节点,造成 Token 费用失控。在真实项目中,凡是有条件边的地方,都要确保对应的 State 字段会在某个节点被正确更新。
4.4 条件边与循环机制的原理
上面的代码展示了两种典型的条件路由:
- 意图分流:
analyze之后根据意图去不同节点。 - 质量驱动的循环:
evaluate之后决定是重新生成还是结束。
LangGraph 的循环实现并没有特殊的语法,它就是靠条件边指向前置节点实现的。图软件层面允许有环,状态机天然支持这种回退逻辑。这也是 LangGraph 比普通 Chain 灵活很多的核心原因。
你可以把 LangGraph 想象成一个带反馈回路的流程图,而 LangChain 的 Chain 是一根从输入到输出的直线。前者可以处理“一次结果不满意再重来一次”这类真实业务需求,后者只能在链路上做有限的前置处理。
4.5 人工审批:interrupt 让流程暂停等待用户输入
企业级场景中,Agent 全自动处理并不是常态,更多是“AI 完成大部分工作,关键节点交给人确认”。LangGraph 通过interrupt机制实现这种人工介入。
在ask_human节点中,如果希望流程真正暂停、等待人工输入而不是立即返回,可以使用interrupt:
from langgraph.types import interrupt def ask_human(state: TicketState): """暂停流程,等待人工审核结果""" human_review = interrupt({ "ticket_id": state["ticket_id"], "user_message": state["user_message"], "reason": "工单被判定为投诉,需要人工介入处理" }) return { "need_human": False, "human_feedback": human_review, "history": state["history"] + [{"step": "human_approved", "result": human_review}] }使用interrupt后,图执行到这个节点会暂停,并返回一个__interrupt__对象。外部系统收到这个对象后,可以在人工审核通过后通过Command恢复执行。这种模式非常适合工单审批、交易确认、知识库更新审核等场景。
完整的恢复调用方式需要在图中引入一个显式状态字段:
from langgraph.types import Command from typing_extensions import Annotated class TicketState(TypedDict): # ...原有字段... human_feedback: str后续通过保存的 checkpoint 标识(即线程 ID)和 Command 恢复执行:
graph.invoke( Command(resume="人工已确认,继续生成回复"), config={"configurable": {"thread_id": "ticket-10001"}} )这个步骤在企业 Agent 开发中是关键能力。没有它,人工介入只能是“流程外介入”,一旦人审通过,Agent 无法从断点恢复,整个状态机就要重建。
5. 运行测试与效果验证
构建完整的测试脚本,验证你刚才设计的图是否能跑通。
# 文件路径:demo/main.py config = {"configurable": {"thread_id": "ticket-10086"}} # 第一轮:一般咨询工单 initial_state = { "ticket_id": "ticket-10086", "user_message": "我想咨询一下订单的发货时间为什么延迟了?", "intent": "", "ai_response": "", "response_rating": 0, "rewrite_count": 0, "need_human": False, "history": [] } result = graph.invoke(initial_state, config) print("最终处理结果:") print("意图识别:", result["intent"]) print("AI 回复:", result["ai_response"]) print("回复质量分:", result["response_rating"]) print("是否转人工:", result["need_human"]) print("执行历史节点:", [h["step"] for h in result["history"]])运行脚本:
python demo/main.py预期输出效果类似:
意图识别: 一般咨询 AI 回复: 您好,非常抱歉给您带来不便。您的订单发货延迟可能与物流高峰期相关,查询到最新进展后我们会第一时间联系您。 回复质量分: 9 是否转人工: False 执行历史节点: ['analyze_intent', 'retrieve', 'generate', 'evaluate']如果流程运行成功,可以看到执行历史节点顺序:先分析意图,再检索知识库,生成回复,最后评估质量。评估分数合格,流程正常结束。
想测试循环重写功能,可以在测试数据中把模型刻意调低输出质量,或者在 Prompt 中要求“故意生成不完整的回复”,观察rewrite_count是否递增,执行历史中是否出现两次generate。
想测试人工转交,把工单内容改成“我要投诉你们的服务质量”,预期执行历史中会包含human_handoff。
6. 企业级 Agent 工程化:持久化、监控与安全
Demo 跑通只是第一步。从企业级角度看,需要关注几个关键问题:
6.1 持久化检查点的实现选型
前面用的是MemorySaver,它把检查点保存在内存中,进程重启就丢失。生产环境需要把检查点持久化到外部存储。
from langgraph.checkpoint.postgres import PostgresSaver DB_URI = "postgresql://user:password@localhost:5432/langgraph" with PostgresSaver.from_conn_string(DB_URI) as checkpointer: graph = builder.compile(checkpointer=checkpointer)使用 PostgresSaver 后,每个线程(即一次会话流程)的执行状态都会保存在数据库中。即使应用重启,也可以根据thread_id恢复到原来的执行位置。
6.2 使用 LangSmith 做全链路可观测
Agent 环境和传统 Web 后端不同,同一个流程中可能发生多次 LLM 调用,且每次调用的输入输出都受前一步影响。排错时必须能看清每一步的输入、输出、耗时、Token 消耗。
LangSmith 是目前 LangChain 生态中主流的可观测平台。在代码中简单配置:
export LANGCHAIN_TRACING_V2=true export LANGCHAIN_API_KEY="your-langsmith-api-key" export LANGCHAIN_PROJECT="ticket-agent"配置完成后,LangGraph 的每一步执行都会自动上报到 LangSmith,后台可以看到完整的 Trace 链路。在一个生产级 Agent 项目中,可观测能力是上线前必须解决的问题,否则根本无法分析“为什么 Agent 在某个特殊输入上表现异常”。
6.3 版本管理与回滚
LangGraph 的图不是一个线上不可变组件。随着业务发展,你会不断调整节点逻辑、修改 Prompt、替换模型。每一次调整都可能影响整个执行链路。
推荐的做法是引入版本化部署机制:
- 图逻辑与配置分离:Prompt 模板放在配置中心,不写死在代码里。
- 保留旧版本图实例:发布新版本时,旧版本保留一段时间,用于快速回滚。
- 使用线程 ID 隔离:不同版本的 Agent 服务使用不同的线程 ID 前缀,避免状态串扰。
6.4 安全边界与敏感信息处理
Agent 的核心机制是“把工具交还给模型决定使用”,这本身就是安全风险。你必须明确画出一条安全边界:
- 所有涉及资金、数据删除、权限变更的操作,必须经过人工审批节点,Agent 没有最终执行权。
- 对 Tool 的输入参数做白名单校验,禁止传入任意代码或任意路径。
- State 中不要存放敏感明文信息。
- 对外暴露 API 时,必须做用户身份校验和鉴权,不能让人随意传入
thread_id读取他人的执行状态。
在客服工单这个场景中,转人工审批必须走interrupt机制,确保最敏感的一步始终有人参与决策。
7. 常见报错与排查方法
LangGraph 使用过程中,报错最多的地方集中在几个方向:状态更新格式错误、条件边返回值不匹配、检查点持久化配置出错、循环次数控制失效。下面用一个常见问题排查表整理。
| 问题现象 | 可能原因 | 排查方式 | 解决方案 |
|---|---|---|---|
| 执行时报错 “Invalid update for field” | State 节点返回值与字段类型不一致,或返回了未定义的字段 | 查看错误堆栈中字段名 | 检查 State 定义,确保节点返回的键存在于 State 中,且类型一致 |
| 条件边报错 “No node found for value” | 条件函数返回了图上不存在的节点名 | 打印条件函数的返回值 | 将条件函数返回值与 add_conditional_edges 的映射字典对齐 |
| 使用 checkpointer 后流程从旧状态恢复 | 多个请求复用了同一个 thread_id | 查看每次请求的 thread_id | 每次新的业务请求使用新的 thread_id |
| 手动恢复中断时一直报错 | 调用的 thread_id 不存在,或该流程没有执行到 interrupt | 查询检查点状态 | 确认上一次 invoke 返回了interrupt对象后再调用 Command(resume=...) |
| 循环节点无限执行 | State 中没有维护循环计数器,或计数器没有递增 | 在历史记录中观察 generate 节点出现次数 | 在循环路径上的节点内对计数器字段递增,并设置上限 |
| Token 消耗远超预期 | 每次模型调用都输出了超长内容,循环重写逻辑触发多次 | 查看每次调用的 Token 用量的 Trace | 合理设置 max_tokens,优化 Prompt 输出约束,在条件边上限制重写次数 |
| PostgresSaver 连接失败 | 数据库连接串配置错误,或依赖库版本不匹配 | 打印数据库连接异常 | 检查连接串、驱动安装和网络策略 |
8. LangGraph 学习路径与进阶建议
很多开发者问,学 LangGraph 应该从哪里开始,如何避免走弯路。如果从头规划一条学习路径,我会建议分四个阶段推进。
第一阶段:理解核心抽象。用最小示例跑通 State、Node、Edge、Checkpointer 这四件事。不要急着写复杂业务逻辑,先写一个只有两个节点的图,手工触发几次,观察 State 的变化和 Checkpoint 的持久化效果。
第二阶段:掌握条件边与循环。找三个典型的流程模式练手:意图分流、工具调用循环、质量评估重写。这三个模式基本覆盖了日常 Agent 开发中的大部分控制流需求。
第三阶段:熟悉中断与人工审批。实现一个带人工确认的订单审批流程。理解interrupt的行为:它如何暂停执行、如何携带数据给外部系统、如何通过Command(resume=...)恢复执行。
第四阶段:企业级工程化。把 PostgresSaver 集成进来,接入 LangSmith 追踪,设计安全的 Tool 边界,做版本管理和灰度发布。在这个阶段你可以去研究langgraph-platform等更上层的部署方案。
这里也想给出一个判断:LangGraph 的学习重点从来不是记住 Api 的调用格式,而是建模能力。拿到一个业务场景,能不能拆出节点、状态和条件边,能不能合理设计循环和中断,才是拉开初级开发者与资深 Agent 工程师差距的关键。
9. 总结与最后几点提醒
LangGraph 之所以在 Agent 开发领域快速流行,是因为它真正解决了流程控制不确定性的问题。它给 Agent 开发带来了状态图这种工程化建模方式,把持久化、断点恢复、人工介入这些企业级需求落到了标准机制中,而不是靠开发者自己手写一整套状态机。
如果你当前正在做一个 Agent 项目,我的建议是:先用 LangGraph 把完整链路跑通,重点验证状态流转和分支逻辑是否正确,再逐步补上持久化和可观测能力,最后才考虑复杂工具链和多智能体协作。直接上多 Agent 架构很容易失败,因为问题往往出在基础控制流不扎实。
最后提醒一点:数据安全和权限管控在任何 Agent 项目中都是最高优先级。一条不变的原则是,涉及资金变动、数据删除、权限修改的操作,永远把决策权保留给人类。LangGraph 的interrupt机制已经提供了足够的工具,剩下的,取决于你如何设计整个流程的安全边界。