LangGraph实战:构建可处理复杂任务的Agentic RAG智能体架构
30款热门AI模型一站整合DeepSeek/GLM/Qwen 随心用限时 5 折。 点击领海量免费额度最近在帮一个团队做技术选型他们想用大模型做内部知识库问答。聊到具体实现时几乎所有人都提到了“RAG”。但当问及“你们打算怎么处理 Agent 和工具调用以及如何管理复杂的对话流程”时讨论就变得有些模糊了。大家似乎都默认RAG 就是检索增强生成Agent 就是能调用工具的智能体两者结合把知识库塞给 Agent 不就完了这个场景让我想起很多刚开始接触 LangChain 的开发者手里握着锤子RAG看什么都像钉子问答。但当你真正开始构建一个需要理解上下文、调用外部 API、处理多轮对话、并能根据用户意图自主规划步骤的“智能体”时你会发现仅仅把文档切片、向量化、检索出来只是解决了“知识在哪”的问题。而“如何思考”、“如何行动”、“如何从失败中恢复”才是智能体项目的真正难点。这也是为什么尽管 LangChain 的教程铺天盖地但很多照着“快速入门”跑通的 Demo一遇到真实业务就卡壳。问题往往不出在检索的准确率上而是出在流程的僵化、状态管理的缺失以及对“智能体”本质的误解上。今天我们不谈那些基础的pip install和from langchain import ...我们直接切入核心如何用 2026 年的视角意味着更成熟、更工程化的思路理解 LangChain 生态并构建一个真正可用的 Agentic RAG 项目。这里的“可用”指的是能处理边界情况、有清晰的错误处理逻辑、并且架构上易于维护和扩展。1. 重新理解 LangChain 与 LangGraph从链式思维到图式思维如果你搜索“LangChain 快速入门”得到的大概率是一个线性流程加载文档 - 文本分割 - 向量化存储 - 检索 - 组合成 Prompt - 调用 LLM 生成答案。这在 LangChain 中被称为“Chain”链。链是顺序执行的每一步的输出是下一步的输入简单直观适合定义明确的流水线任务。然而智能体Agent的工作模式很少是线性的。想象一个客服智能体用户问“帮我查一下上个月的订单然后取消最新的一单再告诉我退款政策。” 这个任务至少包含三个子步骤1) 查询订单可能需要调用数据库工具2) 执行取消操作调用订单 API3) 检索退款政策从知识库中查找。这些步骤之间存在条件判断如果没找到订单怎么办、循环可能需要用户确认、以及状态传递取消的订单号需要传递给后续步骤。用单一的链来描述这种工作流代码会变得极其复杂和难以维护。这就是LangGraph被引入的意义。它不是取代 LangChain而是对其的增强和补充。你可以把它理解为 LangChain 的“工作流引擎”或“状态机管理器”。LangChain (LCEL) 核心是定义可组合的“单元”Runnable比如一个检索器、一个提示模板、一个 LLM 调用。它擅长构建和连接这些原子操作。LangGraph 核心是定义“图”Graph其中节点是 LCEL 定义的 Runnable 或任意函数边则决定了基于当前“状态”下一步该执行哪个节点。它擅长管理带有分支、循环和状态保持的复杂、长时间运行的流程。两者的区别可以用一个简单的类比LangChain 链像一份菜谱告诉你先放油再放菜最后放盐步骤固定。LangGraph 图像一位厨师他有一个状态锅里有什么根据状态菜熟了没咸不咸决定下一步是翻炒、加水还是出锅并且可以处理意外火太大了怎么办。在 Agentic RAG 项目中我们通常用 LangChain 来构建核心能力组件如检索、工具调用而用 LangGraph 来编排智能体的决策循环。这是从“链式思维”升级到“图式思维”的关键一步。2. 拆解一个 Agentic RAG 智能体的核心架构一个健壮的 Agentic RAG 智能体远不止“检索 生成”。我们需要把它拆解成多个相互协作的模块。下面这个架构图描绘了核心组件及其关系graph TD A[用户输入] -- B(智能体主控/调度器) B -- C{意图分析与规划} C --|需知识检索| D[RAG 检索模块] C --|需执行操作| E[工具调用模块] subgraph RAG模块 D -- D1[文档加载与分割] D1 -- D2[向量化与索引] D2 -- D3[查询检索] D3 -- D4[上下文组装与重排序] end subgraph 工具模块 E -- E1[工具1: 查询API] E -- E2[工具2: 计算器] E -- E3[...] end D4 -- F[信息综合与生成] E1 E2 E3 -- F F -- G[输出答案] G -- H{是否需要进一步行动} H --|是 多轮对话| B H --|否 结束| I[结束] B -- J[记忆模块] J -- B让我们逐一解析图中的关键模块2.1 RAG 模块不只是向量检索RAG 模块负责从知识库中获取相关信息。但“检索”本身就有很多学问文档加载与分割 这是源头。分割策略按字符、按句子、按语义、重叠窗口直接影响检索粒度。对于长文档简单的“一刀切”会导致上下文碎片化。更优的做法是采用层次化分割先按章节分割章节内再按段落或语义分割检索时可以融合不同层次的信息。向量化与索引 选择嵌入模型至关重要。通用模型如text-embedding-ada-002和领域微调模型效果差异可能很大。索引方面除了经典的向量数据库Chroma, Pinecone, Weaviate也可以结合传统全文检索如 Elasticsearch进行混合检索兼顾语义匹配和关键词匹配。查询检索 直接拿用户问题去搜往往不够。更好的做法是进行查询重写或查询扩展。例如利用 LLM 将用户问题“它怎么用”根据对话历史重写为“LangGraph 库的具体使用方法是什么”。这能显著提升召回率。上下文组装与重排序 检索到多个片段后不是简单拼接。需要重排序根据与问题的相关性对片段进行排序并可能过滤掉低分片段。然后以清晰的结构如“根据文档A...根据文档B...”组装进 Prompt帮助 LLM 更好地理解和引用。一个常见误区认为 RAG 的瓶颈只在嵌入模型和向量数据库。实际上分割策略、查询优化和重排序策略对最终效果的影响常常不亚于甚至超过嵌入模型本身。2.2 工具调用模块智能体的手和脚工具是智能体与外部世界交互的接口。在 LangChain 中工具通常是一个 Python 函数用tool装饰器描述。from langchain.tools import tool import requests tool def get_weather(city: str) - str: 根据城市名获取当前天气信息。 # 这里应该是调用真实天气API此处为示例 # response requests.get(fhttps://api.weather.com/v1/city/{city}) # return response.json() return f{city}的天气是晴朗25摄氏度。 tool def search_internal_knowledgebase(query: str) - str: 在公司内部知识库中搜索相关信息。 # 这里可以集成上述的RAG检索模块 # retrieved_docs rag_retriever.invoke(query) # return format_docs(retrieved_docs) return f关于{query}内部知识库中找到3条相关记录...定义工具的关键在于清晰的描述。LLM 根据工具的描述来决定是否以及如何调用它。描述应准确说明功能、输入参数和输出格式。2.3 智能体主控与记忆模块大脑与记忆这是 LangGraph 发挥核心作用的地方。智能体主控是一个“图”其核心是一个循环观察 接收用户输入和当前状态来自记忆。思考 调用 LLM根据当前状态、可用工具和历史决定下一步行动调用哪个工具或直接回复。行动 如果决定调用工具则执行工具并获取结果。反思 将工具结果或直接回复更新到状态中并决定循环是否继续。记忆模块负责持久化这个“状态”。它不仅仅是存储对话历史还包括对话历史 完整的用户-助手交互记录。实体记忆 提取并记住对话中提到的关键实体如人名、订单号、项目名。摘要记忆 对于长对话定期将历史摘要避免 Prompt 过长。 LangChain 提供了多种记忆后端从简单的ConversationBufferMemory到更复杂的ConversationSummaryMemory或ConversationEntityMemory。在 LangGraph 中你可以通过定义State对象来精确控制需要记忆和传递哪些信息。3. 实战用 LangGraph 构建一个客服订单查询智能体让我们构想一个相对复杂的场景一个电商客服智能体它能处理复合查询比如“我上周买的那个蓝色衬衫物流到哪了如果没发货我想换成L码”。这个任务需要1) 识别用户身份可能需要登录态2) 查询订单工具13) 查询物流工具24) 判断条件并执行换货操作工具3。我们使用 LangGraph 来构建。首先定义智能体的状态。这是图的“中央内存”。from typing import TypedDict, Annotated, List from langgraph.graph.message import add_messages import operator class AgentState(TypedDict): # 消息历史 LangGraph 内置支持 messages: Annotated[List, add_messages] # 从对话中提取的用户ID简化示例 user_id: str # 查询到的订单ID order_id: str # 查询到的物流信息 shipping_info: str # 智能体下一步该做什么 need_order_id, need_ship_info, handle_exchange, final_answer next_step: str然后定义工具这里用模拟函数代替真实 API。from langchain.tools import tool tool def find_order_by_description(user_id: str, description: str) - str: 根据用户ID和商品描述如‘蓝色衬衫’查找最近的订单ID。 # 模拟查找 print(f[工具调用] 为用户 {user_id} 查找描述包含 {description} 的订单) return ORDER_12345 # 模拟返回订单号 tool def get_shipping_status(order_id: str) - str: 根据订单ID查询物流状态。 print(f[工具调用] 查询订单 {order_id} 的物流状态) return 已发货正在运输中预计明天送达。 tool def initiate_exchange(order_id: str, new_size: str) - str: 为指定订单发起换货申请请求更换为新尺码。 print(f[工具调用] 为订单 {order_id} 发起换货新尺码: {new_size}) return f换货申请已提交订单 {order_id} - {new_size}。客服将在24小时内确认。接下来是核心构建 LangGraph。图由节点函数和边条件或固定流转组成。from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI # 初始化LLM llm ChatOpenAI(modelgpt-4-turbo-preview) # 创建图构建器 workflow StateGraph(AgentState) # 1. 节点路由器决定下一步 def router(state: AgentState): 根据当前状态和最新消息决定下一步行动。 messages state[messages] last_message messages[-1] # 如果是用户的第一条消息尝试提取用户ID和商品描述 if len(messages) 1: # 这里应该有一个更复杂的LLM调用或规则来提取信息 # 为简化我们假设能提取到 state[user_id] USER_001 # 模拟提取 # 判断用户意图是否需要查订单 if any(word in last_message.content.lower() for word in [订单, 买的, 衬衫, 物流]): state[next_step] need_order_id else: state[next_step] final_answer # 如果已经获取了订单ID且用户问题涉及物流 elif state.get(order_id) and 物流 in last_message.content: state[next_step] need_ship_info # 如果已有物流信息且用户想换货 elif state.get(shipping_info) and 换 in last_message.content: state[next_step] handle_exchange else: state[next_step] final_answer return state # 2. 节点调用工具查找订单 def find_order_node(state: AgentState): 调用工具查找订单ID。 # 从消息中提取商品描述实际应用应用LLM提取 last_msg state[messages][-1].content # 简单模拟提取关键词 description 蓝色衬衫 if 蓝色衬衫 in last_msg else 商品 order_id find_order_by_description.invoke({user_id: state[user_id], description: description}) state[order_id] order_id state[next_step] need_ship_info # 假设找到订单后下一步是查物流 return state # 3. 节点调用工具查询物流 def get_ship_info_node(state: AgentState): 调用工具查询物流信息。 shipping_info get_shipping_status.invoke({order_id: state[order_id]}) state[shipping_info] shipping_info state[next_step] final_answer # 获取物流信息后准备最终回复 return state # 4. 节点调用工具处理换货 def handle_exchange_node(state: AgentState): 调用工具处理换货请求。 # 从消息中提取新尺码实际应用应用LLM提取 last_msg state[messages][-1].content new_size L if L码 in last_msg or l码 in last_msg else M result initiate_exchange.invoke({order_id: state[order_id], new_size: new_size}) # 将结果添加到消息中让LLM生成友好回复 state[messages].append((assistant, f[系统操作结果] {result})) state[next_step] final_answer return state # 5. 节点调用LLM生成最终回复 def final_answer_node(state: AgentState): 汇总所有信息调用LLM生成给用户的自然语言回复。 # 构建给LLM的Prompt context f 对话历史 {state[messages]} 已获取的信息 - 订单ID: {state.get(order_id, 未获取)} - 物流信息: {state.get(shipping_info, 未获取)} # 这里简化了Prompt构建 response llm.invoke(f你是一个客服助手。根据以下上下文生成一段友好、专业的回复直接给用户。\n{context}) state[messages].append((assistant, response.content)) state[next_step] END # 标记流程结束 return state # 将节点添加到图中 workflow.add_node(router, router) workflow.add_node(find_order, find_order_node) workflow.add_node(get_ship_info, get_ship_info_node) workflow.add_node(handle_exchange, handle_exchange_node) workflow.add_node(generate_answer, final_answer_node) # 设置图的入口 workflow.set_entry_point(router) # 定义条件边根据 state[next_step] 决定下一个节点 def decide_next_step(state: AgentState): return state[next_step] workflow.add_conditional_edges( router, decide_next_step, { need_order_id: find_order, need_ship_info: get_ship_info, handle_exchange: handle_exchange, final_answer: generate_answer, } ) # 设置固定边执行完某个节点后固定去往下一个节点 workflow.add_edge(find_order, router) # 找到订单后返回路由器决定下一步 workflow.add_edge(get_ship_info, router) # 查到物流后返回路由器 workflow.add_edge(handle_exchange, router) # 处理换货后返回路由器 workflow.add_edge(generate_answer, END) # 生成答案后结束 # 编译图 app workflow.compile()现在我们可以运行这个智能体# 初始化状态 initial_state AgentState(messages[(user, “我上周买的那个蓝色衬衫物流到哪了如果没发货我想换成L码。)], user_id, order_id, shipping_info, next_step) # 运行图 final_state app.invoke(initial_state) # 查看最终回复 for msg in final_state[messages]: if msg[0] assistant: print(f助手: {msg[1]})这个例子虽然简化但清晰地展示了 LangGraph 如何管理一个多步骤、有条件分支的智能体工作流。router节点充当了“大脑”根据当前状态决定行动路线。每个工具调用节点是“手和脚”而generate_answer节点是“嘴巴”负责组织信息并生成友好回复。4. 超越 Demo项目实战中的关键考量与避坑指南把上述 Demo 跑通只是万里长征第一步。要将其投入真实项目必须考虑以下工程化问题4.1 可靠性错误处理与重试智能体在调用外部工具、API 或 LLM 时失败是常态。你的图必须有健壮的错误处理机制。工具调用超时/失败 在工具节点内使用try...except捕获异常并将错误信息写入状态让路由节点能决定是重试、降级处理还是向用户求助。LLM 输出格式错误 使用 LangChain 的OutputParser或 Pydantic 来结构化输出并设置重试逻辑。对于关键决策如选择哪个工具可以使用with_retry装饰器。状态一致性 确保任何节点的失败都不会导致状态数据损坏。考虑使用更持久化的状态存储并在关键操作前做检查点。4.2 性能与成本优化策略减少不必要的 LLM 调用 LLM 调用是延迟和成本的主要来源。在路由决策时可以先用规则或更小、更快的模型进行过滤。缓存频繁使用的检索结果和 LLM 响应。流式输出 对于生成耗时较长的回复使用流式输出Streaming提升用户体验。LangChain 和 LangGraph 都支持流式响应。异步执行 如果多个工具调用可以并行例如同时查询订单状态和物流信息使用asyncio并行执行可以大幅降低延迟。LangGraph 支持异步节点。4.3 可观测性与调试智能体是个黑盒不你必须把它变成白盒。全面日志记录 记录每个节点的输入、输出、耗时特别是 LLM 的请求和响应注意脱敏。这有助于事后分析和调试。可视化执行轨迹 LangGraph 的一个巨大优势是整个执行过程是一个确定的图。你可以轻松地将一次对话的执行轨迹经过了哪些节点状态如何变化可视化出来这对于调试复杂逻辑至关重要。链路追踪 集成像 OpenTelemetry 这样的标准将智能体的调用纳入整个微服务的可观测性体系。4.4 安全与权限工具权限控制 不是所有用户都能调用所有工具。需要在路由或工具调用层加入权限校验根据用户身份动态过滤可用的工具列表。输入输出过滤与审查 对用户输入和 LLM 输出进行内容安全过滤防止注入攻击或生成不当内容。数据脱敏 在日志和传递给 LLM 的上下文中对敏感信息如手机号、身份证号进行脱敏处理。5. 总结从项目到产品智能体开发的思维转变回顾开头的场景构建一个 Agentic RAG 项目技术选型LangChain vs LangGraph和代码实现只是表层。更深层的转变在于思维模式从“问答系统”到“任务执行系统” 你的目标不再是回答一个问题而是完成一个用户意图。这要求你设计的状态和流程能支撑多轮、多步骤的交互。从“静态管道”到“动态工作流” 接受不确定性。智能体的路径不是预设的而是根据环境用户输入、工具结果动态演进的。LangGraph 的图模型正是为此而生。从“追求准确率”到“管理不确定性” 在复杂任务中100%的准确率不现实。重点转向如何优雅地处理失败重试、降级、澄清询问如何设计确认机制以及如何提供清晰的进度反馈。从“模型中心”到“系统工程” 最终用户体验的瓶颈往往不在模型能力而在系统可靠性、响应速度、错误处理和交互设计。你需要像对待一个分布式系统一样为你的智能体设计监控、告警和回滚机制。因此当你下次启动一个“智能体”项目时不妨先问自己几个问题我的智能体需要管理哪些状态它可能遇到哪些分支情况每个工具调用失败后该怎么办我如何追溯和调试一次完整的对话想清楚这些问题并用 LangGraph 这样的工具将你的设计具象化你才算是真正踏入了智能体开发的大门。这远比纠结于用哪个向量数据库的 recall 高 0.5% 更重要因为前者决定了你的系统能否“活”下去并解决真实问题而后者只是让它解决得更好一点。在智能体时代先让流程“跑通”并“可控”是比“跑得精”更优先的工程原则。 30款热门AI模型一站整合DeepSeek/GLM/Qwen 随心用限时 5 折。 点击领海量免费额度