☰
LangGraph实战:构建会拆任务、查数据、自我纠错的深度智能体
2026/10/22 9:50:41 网站建设 项目流程

有没有想过一个问题:同样是接大模型API,为什么有些程序只会一问一答,而有些程序却能在收到指令后自动拆任务、查数据、核对结果、再修正方向,最后像模像样地给你一份完整结论?差别通常不在模型本身,而在代码架构。我这两年做AI应用,踩过最多的坑就是“模型能力很强,但程序不知道什么时候该调工具、什么时候该停”。围绕LangChain加LangGraph打造深度智能体的思路,恰好能解决这个问题。这篇文章我直接用Python实战代码拆解整个搭建过程,从状态管理到工具调用,从反射纠错到人机协同,把复杂任务编排的底层逻辑一次讲清楚。

这篇文章适合谁?已经跑通过大模型接口、想从“单轮对话Demo”升级到“能自主干活Agent”的开发者,以及被复杂业务逻辑折磨、想用结构化方案管理AI流程的工程师。看完你至少能获得三样东西:一套可以直接改来用的代码骨架、几个容易踩到死的坑、以及“为什么图结构比链式调用更适合复杂任务”的判断力。

1. 从普通脚本到深度智能体:为什么要用LangGraph

1.1 普通对话机器人与智能体之间差了一个“状态机”

先说个扎心的事实:很多所谓的AI应用,本质就是一个for循环里反复调用model.invoke(),把上次的回复拼进对话历史再丢回去。这种方式对简单问答够用,但一旦任务变成“先查A系统,再根据结果判断要不要查B系统,最后输出报告”,这种线性写法直接原地爆炸。因为逻辑分支多了以后,代码会退化成一团if-else裹着prompt的意大利面,改一处崩三处。

智能体的核心差异在于:它不再是被动地“处理一条消息”,而是主动地“执行一个任务”。这个任务有目标、有中间产物、有完成条件,甚至允许在执行过程中修改自己的计划。这种情况下,你必须把“执行过程”本身当成一等公民来管理,也就是引入状态机。程序跑到哪个环节、当前收集了哪些证据、还差哪些信息、是否已经超过重试次数,这些信息都需要被显式记录和流转。

1.2 LangChain与LangGraph的定位:胶水层与编排层

很多人分不清LangChain和LangGraph的关系,其实特别简单。LangChain负责“连接”,提供模型封装、提示词模板、工具调用协议、输出解析这些基础能力,相当于给你一堆质量不错的乐高零件。LangGraph负责“编排”,它把零件按图的方式拼起来,让你能定义节点、边、条件跳转和状态流转,相当于设计师手里的拼装图纸。

我实际用下来的体会是:如果你只是做原型Demo,光用LangChain就够;一旦你认真规划一个要上线的复杂任务,LangGraph的价值立刻显现。最典型的一个例子是“多步决策”:你希望智能体在执行完一个工具调用后,根据结果决定下一步是继续调用其他工具、还是进入反思环节、还是直接生成最终答案。用纯LangChain实现这种动态路由,你得手写大量while循环和break条件,而用LangGraph只是一条条件边的事。

1.3 复杂任务的“图思维”:节点、边、状态

图思维的精髓可以概括成一句话:把整个执行流程画成一张有向图,每个圆角矩形是一个节点,箭头是流转条件。代码里对应就是StateGraph、add_node、add_edge和add_conditional_edges四个核心API,几句话就能描述一个完整流程,而且流程图和代码是一一对应的,写错一眼就能看出来。

这里要特别强调状态(State)的设计。LangGraph的状态本质上是一个字典,最常见的写法是TypedDict或dataclass。复杂任务对状态设计的要求非常高,因为每个节点都在往状态里写数据、读数据,状态定义得好,后续加节点、改流程就非常顺;定义得烂,后面改需求时会发现所有节点函数都要跟着改签名。我见过不少新手一上来就把所有临时变量都塞进State里,结果状态越滚越大,每次进节点都要做一堆字段判空,这就是典型的图思维没建立起来。

2. 开工前要准备的三件事:依赖、模型与工具注册

2.1 安装依赖与模型接入

新的LangGraph版本已经把很多操作简化了,建议直接用新版API。命令行安装非常简单:

pip install -U langchain langchain-openai langgraph

模型方面,我建议选支持工具调用(function calling)的对话模型,不管你是用国内还是国外的模型服务,只要协议兼容就行。代码里我用ChatOpenAI类来做统一封装,通过base_url可以轻松指向任意兼容接口:

from langchain_openai import ChatOpenAI llm = ChatOpenAI( model="your-model-name", # 替换成你实际用的模型名 base_url="https://your-api-endpoint", api_key="your-api-key", temperature=0.3, # 复杂任务建议调低 )

这里有个小经验:复杂任务的temperature不建议调高。编排型任务需要确定性,哪怕牺牲一点“创造性”都值得。我自己做Agent时基本固定在0.2到0.4之间,太高的温度会导致模型在规划阶段输出不稳定,今天生成的计划能用,明天换个问法就崩了。

2.2 工具注册:让模型拥有“手”

深度智能体区别于纯聊天的关键,就是它能调用工具。工具可以是查询API、操作数据库、跑一段计算、搜索文档,本质上都是“把真实世界的能力封装成函数交给模型”。

在LangChain里,注册工具最常见的方式是@tool装饰器。每个工具都需要写清楚name和description,这两个字段直接决定模型能不能正确选择工具。description尤其重要,它相当于工具的自述,写得越具体、越贴近业务场景,模型选错工具的概率越低。举个实际例子:

from langchain_core.tools import tool @tool def query_order_status(order_id: str) -> str: """查询订单状态,适用于用户询问订单是否发货、是否签收的场景。传入订单号,返回最新物流节点。""" # 这里对接你实际的订单系统API return f"订单{order_id}已发货,预计3天内到达"

注册完成后,把工具列表绑定到模型上,这样模型在对话时如果发现需要查订单,就会在回复里输出一个结构化的工具调用请求,而不是直接编一个答案。这一步是后续所有编排的基础,工具没绑好,后面图写得再漂亮也是空中楼阁。

2.3 状态定义与图结构初探:四阶段工作流的设计构想

在写节点之前,先把状态设计出来。我给这个实战项目定的目标是:智能体收到一个复杂任务后,自动经历“规划、执行、反思、汇总”四个阶段。对应到状态上就需要用到TypedDict:

from typing import TypedDict, Annotated class AgentState(TypedDict): query: str # 用户的原始请求 plan: str # 模型生成的执行计划 tool_results: list # 已采集到的工具结果 intermediate_answers: list # 反思过程的中间结论 retries: int # 当前已重试次数 final_answer: str # 最终输出

四个阶段各司其职:规划阶段让模型拆解任务,定出步骤;执行阶段逐个调用工具;反思阶段检查结果够不够支撑最终答案;汇总阶段把收集到的信息整理成结构化的回复。如果反思阶段发现信息不足,就带着“为什么不够”的结论重新回到执行阶段,形成一个闭环。这种“规划-执行-反思-汇总”的循环结构,是应对复杂任务最稳的一套模式,比单次调用可靠得多。

3. 完整实战:构建一个会思考、会纠错的多工具智能体

3.1 场景设计:为什么选“跨系统信息整合”这个任务

为了演示复杂任务编排,我设计了一个模拟场景:智能体需要根据用户的一句话,同时查询订单信息、库存信息和物流时效,最后给出一个综合性的处理建议。这个任务的特点是:单一工具搞不定、信息需要交叉验证、结果必须经过判断而不是简单拼接。用它来展示LangGraph的编排能力,再合适不过。

你可以把这三个查询工具理解成三个独立部门:订单部门、库存部门、物流部门。没有Agent的时候,用户得像踢皮球一样分别去三个系统里查;有了Agent,它自动按顺序把所有信息捞回来,还替你做了交叉判断。这就是深度智能体带来的体验升级。

3.2 核心代码实现:工具集定义

先把三个模拟工具写好。为了让Demo能直接跑,我用模拟数据代替真实接口,但结构上跟真实项目完全一致:

@tool def query_order(order_id: str) -> str: """根据订单号查询订单的基础信息,返回下单时间、商品名称和当前处理状态。""" mock_data = { "A1001": "商品:机械键盘,状态:已签收", "A1002": "商品:显示器,状态:运输中", } return mock_data.get(order_id, "未找到该订单") @tool def check_stock(product_id: str) -> str: """根据商品编码查询库存余量,返回当前可售数量。""" mock_data = { "KB01": "库存充足,剩余120件", "MN02": "库存不足,仅剩3件", } return mock_data.get(product_id, "未找到商品编码") @tool def estimate_delivery(city: str) -> str: """根据收货城市估算配送时效,返回预计到达天数。""" mock_data = { "北京": "预计2天到达", "深圳": "预计4天到达", } return mock_data.get(city, "该城市暂无配送信息")

注意,三个工具的description我都尽量写清楚了“什么场景下用”,这是实战中最容易忽略的细节。模型不像人,它不会“顺着语境猜”工具用途,描述越精确,路由越可靠。

3.3 核心代码实现:四个节点函数

接下来是重头戏,用LangGraph把四个节点串起来。先定义规划节点,它的任务是让模型生成一份执行计划:

def plan_node(state: AgentState) -> dict: prompt = f"""你是任务规划助手。用户的问题是: {state["query"]} 请分析这个问题需要调用哪些工具,按顺序列出执行步骤。只输出计划,不要执行。""" response = llm.invoke(prompt) return {"plan": response.content}

执行节点稍微复杂一些。它需要做两件事:识别计划中的工具名称,然后调用对应工具并捕获结果。因为是演示代码,我用关键词匹配的方式从计划文本里提取工具需求:

def execute_node(state: AgentState) -> dict: tools_map = { "订单": query_order, "库存": check_stock, "物流": estimate_delivery, } results = state.get("tool_results", []).copy() plan_text = state.get("plan", "") for keyword, tool_func in tools_map.items(): if keyword in plan_text: # 简化处理:从查询中提取参数 if keyword == "订单": result = tool_func.invoke("A1002") elif keyword == "库存": result = tool_func.invoke("MN02") else: result = tool_func.invoke("深圳") results.append(f"{keyword}查询结果:{result}") return {"tool_results": results}

这里刻意用了简化逻辑,真实场景中应该让模型输出结构化参数。但为了让你看清流程骨架,参数硬编码并不是问题,替换成真实参数解析逻辑即可。

反思节点是深度智能体的灵魂。它负责审视已收集的信息,判断是否足以回答用户的问题,如果不足则明确指出缺口:

def reflect_node(state: AgentState) -> dict: prompt = f""" 原始问题:{state["query"]} 执行计划:{state["plan"]} 已收集信息:{state["tool_results"]} 请判断现有信息是否足够回答原始问题。如果足够,回复“足够”;如果不够,说明还缺少什么信息。 """ response = llm.invoke(prompt) return {"intermediate_answers": [response.content]}

最后是汇总节点,它把所有工具结果整理成一份面向用户的完整答案。这个环节同样交给模型,因为复杂的业务概括能力恰恰是模型的强项:

def finalize_node(state: AgentState) -> dict: prompt = f""" 请根据以下信息生成一份简洁完整的答复,要求条理清晰、结论明确。 原始问题:{state["query"]} 已收集信息:{state["tool_results"]} """ response = llm.invoke(prompt) return {"final_answer": response.content}

3.4 组图、条件跳转与运行效果拆解

节点函数写好之后,就轮到LangGraph登场了。先构建状态图,把所有节点注册进去:

from langgraph.graph import StateGraph, END graph = StateGraph(AgentState) graph.add_node("plan", plan_node) graph.add_node("execute", execute_node) graph.add_node("reflect", reflect_node) graph.add_node("finalize", finalize_node) graph.set_entry_point("plan") graph.add_edge("plan", "execute") graph.add_edge("execute", "reflect")

关键在条件边:反思节点之后,系统判断信息是否足够,决定跳到汇总还是回到执行:

def should_continue(state: AgentState) -> str: last_reflection = state["intermediate_answers"][-1] if "足够" in last_reflection or state.get("retries", 0) >= 2: return "finalize" else: return "execute" graph.add_conditional_edges( "reflect", should_continue, { "finalize": "finalize", "execute": "execute", }, ) graph.add_edge("finalize", END) app = graph.compile()

运行的时候,只要一行代码就能把整个智能体跑起来:

state = app.invoke({"query": "帮我查一下A1002订单的物流情况,顺便看下相关商品库存还够不够,最后告诉我什么时候能收到货。"}) print(state["final_answer"])

运行效果是这样的:模型先输出一份计划;执行节点依次查了订单、库存和物流;反思节点审视后觉得信息齐全;汇总节点整理成一段自然语言答案。整个流程看起来就像真人助手在处理工单,而不是简单地把几个工具输出拼在一起。如果你把intermediate_answers和tool_results打印出来,能清楚看到每一步的思考痕迹,排查问题非常直观。

4. 应对复杂任务的四板斧:并行、重试、人机协同与记忆增强

4.1 并行分支:用Send把多个子任务同时丢出去

真实的复杂任务往往包含多个相互独立的子任务,比如既要查订单又要查库存,两者没有依赖关系,串行执行既慢又浪费。LangGraph的SendAPI就是专门解决这个问题的。它允许从一个节点动态发起多个并行分支,每个分支带着自己的参数独立跑。

用法很简单:在条件边函数里返回一个Send列表,每个Send指明要唤醒的节点和数据:

from langgraph.types import Send def continue_to_parallel(state: AgentState): steps = state["plan_steps"] return [Send("execute_one", {"step": s}) for s in steps]

并行分支适合“数据互不依赖”的任务。我在实际项目里经常同时发起三到五个工具调用,整体耗时可以压缩到串行方案的几分之一。但要注意:并行并非万能,如果后面的步骤依赖前面步骤的输出,那乖乖串行才是正道。

4.2 错误恢复与限流重试:Agent不能一挂到底

工具调用是跟外部系统打交道,任何意外都可能发生:接口超时、返回异常、参数格式不合法。深度智能体跟普通脚本的差别在于,它必须学会“带伤作战”——某个工具挂了,它还能绕路完成目标。

我最常用的策略是在执行节点里做try-except,并把错误信息作为工具结果的一部分交给反思节点。这样模型看到错误后,会判断是重试、换工具,还是直接把异常情况反馈给用户。顺便说一句,调用外部API时一定要做指数退避重试,不要一失败就拼命重试,那只会把限流阈值顶得更狠。简单的退避逻辑可以写在工具装饰器外面:

import time def safe_call(func, *args, max_retry=3, **kwargs): for i in range(max_retry): try: return func(*args, **kwargs) except Exception as e: time.sleep(2 ** i) if i == max_retry - 1: return f"调用失败:{e}"

4.3 Human-in-the-Loop:关键决策让真人拍板

有些任务的敏感度很高,比如“给客户发送一封包含折扣的邮件”,这种动作如果完全交给模型自动完成,风险不小。LangGraph提供了interrupt_before这样的机制,可以指定在某个节点执行前暂停流程,等人工确认后再恢复。

实操中我会把“人工审批”设计成一个特殊节点,或者直接在编译图的时候指定中断点:

app = graph.compile(interrupt_before=["finalize"])

这样流程跑到汇总节点之前会停下来,你可以把当前的状态展示给审核人员,确认没问题后再调用app.invoke(None)继续往下走。这个机制让Agent既有自动化效率,又有可审计的把关环节,特别适合“半自动”的业务场景。

4.4 短期记忆与长期记忆:Agent需要“上下文”

对话型智能体靠消息列表维持短期记忆,这个LangChain已经给了很好的支持。真正容易忽略的是长期记忆:同一个用户,上周问过的偏好、之前处理过的工单,能不能在下一次对话中被自动带出来?只靠把历史消息全部塞进上下文,既费token又不精准。

成熟的方案是给Agent加一个“记忆读写工具”:需要记信息时,通过工具写入外部存储(比如向量库或普通数据库);需要回忆用户历史时,通过另一个工具检索相关片段再注入上下文。这样模型完全不需要在上下文里背负全部历史,只在决策时按需调取。LangGraph的状态因为可以自由增加字段,配合这类记忆工具非常顺手。我强烈建议在做生产级Agent时,从第一天就设计记忆读写接口,而不是等项目膨胀后再补。

5. 实战中的暗坑:常见问题与排查实录

5.1 模型死活不调用工具?三种姿势排查

这是新手遇到最多的问题,现象是运行半天模型只是输出一段话,工具一次都没触发。我排查顺序是:先看工具的描述是否有歧义,再看模型的工具协议是否启用,最后检查工具名称和参数是不是被模型“理解歪了”。其中工具描述是重灾区,很多团队几十个工具堆在一起,描述写得含糊其辞,模型根本分不清该用哪个。这里有一个非常实用的判断标准:如果你把这个描述发给一个不了解背景的同事,他能根据描述判断出什么时候该用这个工具吗?如果不能,那模型也同样不能。

5.2 图状态被覆盖、循环卡死等编排问题

状态覆盖是LangGraph新手最常见的bug。比如你在execute_node里写了return {"tool_results": results},但忘了把之前的plan字段也带回来,那么状态中除tool_results之外的其他键会被整体覆盖清空。LangGraph的节点返回值默认是完整替换对应字段的,除非你用Annotated配合operator.add等操作符做合并。所以最好在设计状态的时候就明确哪些字段是可叠加列表、哪些是单一快照,用Annotated标注清楚。

循环卡死的问题同样隐蔽。反思节点如果一直不满意,就会无限循环。我的兜底方案很简单:在状态里放一个retries计数,条件边判断时一旦超过上限就强制走汇总节点。这个经验价值非常高,没有兜底的循环结构上线后迟早出事。

5.3 上下文爆炸与Token管理策略

深度智能体执行几次工具调用后,状态里的信息量会快速膨胀,尤其当工具返回的是完整JSON或长文本时。如果每次反思都把全部历史塞给模型,很快你的token账单就要出事。我的习惯是:每轮工具返回后立刻做信息压缩,只保留结论性内容。比如工具返回原始订单数据,我在入库前就抽取出关键字段拼成一句话。这样既保证模型能看到核心信息,又极大节省上下文空间。

另外,规划节点里的长指令没必要每次都重复传给模型。可以把不变量提成系统提示词,动态部分只留实际任务内容,省下的大多数是重复token。真到上下文快满的时候,考虑用滑动窗口策略,丢弃早期轮次的细粒度数据,只保留摘要。

5.4 常见问题速查表

为了方便你排查,我把实战中高频出现的几个问题整理成一张速查表:

现象常见原因推荐处理方式
模型不调用任何工具工具description模糊或未绑定重写描述,用业务场景描述触发条件
工具被重复调用形成死循环缺少终止条件增加retries上限,或对相似结果去重
节点返回值把其他字段清空状态字段被整体覆盖用Annotated标注合并策略,或显式return所有字段
上下文长度迅速超限工具原始结果未压缩入库前抽取关键字段,保留摘要
外部API偶发超时导致流程中断缺少异常捕获与重试工具调用外层包try-except,加退避重试
反思节点总能得到“不够”的结论反思没有可执行标准在prompt中给出“足够”的明确判定条件

写在最后的实操体会

这套“规划-执行-反思-汇总”的架构,我个人试过不下十个项目变体,从行业分析报告生成到售后工单自动处理,稳定性和可解释性都要远好于单轮大模型调用。最大的心得是:别迷信模型的“自主性”,真正稳定运行的Agent靠的还是工程结构上的约束——明确的节点边界、可追踪的状态、强制的终止条件,这些缺一不可。

最后再分享一个小技巧:调试复杂智能体时不要直接看最终输出,要一层层看中间状态。LangGraph里可以设置回调函数,把每个节点的输入输出都打印出来。这个习惯能帮你省下至少一半的排查时间,强烈建议从第一次运行就养成。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询