LangGraph实战:构建可处理复杂任务的Agentic RAG智能体架构
2026/7/9 6:49:41 网站建设 项目流程

🚀 30+款热门AI模型一站整合,DeepSeek/GLM/Qwen 随心用,限时 5 折。 👉 点击领海量免费额度

最近在帮一个团队做技术选型,他们想用大模型做内部知识库问答。聊到具体实现时,几乎所有人都提到了“RAG”。但当问及“你们打算怎么处理 Agent 和工具调用,以及如何管理复杂的对话流程”时,讨论就变得有些模糊了。大家似乎都默认,RAG 就是检索增强生成,Agent 就是能调用工具的智能体,两者结合,把知识库塞给 Agent 不就完了?

这个场景让我想起很多刚开始接触 LangChain 的开发者:手里握着锤子(RAG),看什么都像钉子(问答)。但当你真正开始构建一个需要理解上下文、调用外部 API、处理多轮对话、并能根据用户意图自主规划步骤的“智能体”时,你会发现,仅仅把文档切片、向量化、检索出来,只是解决了“知识在哪”的问题。而“如何思考”、“如何行动”、“如何从失败中恢复”,才是智能体项目的真正难点。

这也是为什么,尽管 LangChain 的教程铺天盖地,但很多照着“快速入门”跑通的 Demo,一遇到真实业务就卡壳。问题往往不出在检索的准确率上,而是出在流程的僵化、状态管理的缺失以及对“智能体”本质的误解上。今天,我们不谈那些基础的pip installfrom langchain import ...,我们直接切入核心:如何用 2026 年的视角(意味着更成熟、更工程化的思路),理解 LangChain 生态,并构建一个真正可用的 Agentic RAG 项目。这里的“可用”,指的是能处理边界情况、有清晰的错误处理逻辑、并且架构上易于维护和扩展。

1. 重新理解 LangChain 与 LangGraph:从链式思维到图式思维

如果你搜索“LangChain 快速入门”,得到的大概率是一个线性流程:加载文档 -> 文本分割 -> 向量化存储 -> 检索 -> 组合成 Prompt -> 调用 LLM 生成答案。这在 LangChain 中被称为“Chain”(链)。链是顺序执行的,每一步的输出是下一步的输入,简单直观,适合定义明确的流水线任务。

然而,智能体(Agent)的工作模式很少是线性的。想象一个客服智能体,用户问:“帮我查一下上个月的订单,然后取消最新的一单,再告诉我退款政策。” 这个任务至少包含三个子步骤:1) 查询订单(可能需要调用数据库工具);2) 执行取消操作(调用订单 API);3) 检索退款政策(从知识库中查找)。这些步骤之间存在条件判断(如果没找到订单怎么办?)、循环(可能需要用户确认)、以及状态传递(取消的订单号需要传递给后续步骤)。用单一的链来描述这种工作流,代码会变得极其复杂和难以维护。

这就是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 模块负责从知识库中获取相关信息。但“检索”本身就有很多学问:

  1. 文档加载与分割: 这是源头。分割策略(按字符、按句子、按语义、重叠窗口)直接影响检索粒度。对于长文档,简单的“一刀切”会导致上下文碎片化。更优的做法是采用层次化分割,先按章节分割,章节内再按段落或语义分割,检索时可以融合不同层次的信息。
  2. 向量化与索引: 选择嵌入模型至关重要。通用模型(如text-embedding-ada-002)和领域微调模型效果差异可能很大。索引方面,除了经典的向量数据库(Chroma, Pinecone, Weaviate),也可以结合传统全文检索(如 Elasticsearch)进行混合检索,兼顾语义匹配和关键词匹配。
  3. 查询检索: 直接拿用户问题去搜,往往不够。更好的做法是进行查询重写查询扩展。例如,利用 LLM 将用户问题“它怎么用?”根据对话历史重写为“LangGraph 库的具体使用方法是什么?”。这能显著提升召回率。
  4. 上下文组装与重排序: 检索到多个片段后,不是简单拼接。需要重排序,根据与问题的相关性对片段进行排序,并可能过滤掉低分片段。然后,以清晰的结构(如“根据文档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(f"https://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 发挥核心作用的地方。智能体主控是一个“图”,其核心是一个循环

  1. 观察: 接收用户输入和当前状态(来自记忆)。
  2. 思考: 调用 LLM,根据当前状态、可用工具和历史,决定下一步行动(调用哪个工具,或直接回复)。
  3. 行动: 如果决定调用工具,则执行工具并获取结果。
  4. 反思: 将工具结果(或直接回复)更新到状态中,并决定循环是否继续。

记忆模块负责持久化这个“状态”。它不仅仅是存储对话历史,还包括:

  • 对话历史: 完整的用户-助手交互记录。
  • 实体记忆: 提取并记住对话中提到的关键实体(如人名、订单号、项目名)。
  • 摘要记忆: 对于长对话,定期将历史摘要,避免 Prompt 过长。 LangChain 提供了多种记忆后端,从简单的ConversationBufferMemory到更复杂的ConversationSummaryMemoryConversationEntityMemory

在 LangGraph 中,你可以通过定义State对象来精确控制需要记忆和传递哪些信息。

3. 实战:用 LangGraph 构建一个客服订单查询智能体

让我们构想一个相对复杂的场景:一个电商客服智能体,它能处理复合查询,比如“我上周买的那个蓝色衬衫,物流到哪了?如果没发货,我想换成L码”。

这个任务需要:1) 识别用户身份(可能需要登录态);2) 查询订单(工具1);3) 查询物流(工具2);4) 判断条件并执行换货操作(工具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(model="gpt-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)和代码实现只是表层。更深层的转变在于思维模式

  1. 从“问答系统”到“任务执行系统”: 你的目标不再是回答一个问题,而是完成一个用户意图。这要求你设计的状态和流程能支撑多轮、多步骤的交互。
  2. 从“静态管道”到“动态工作流”: 接受不确定性。智能体的路径不是预设的,而是根据环境(用户输入、工具结果)动态演进的。LangGraph 的图模型正是为此而生。
  3. 从“追求准确率”到“管理不确定性”: 在复杂任务中,100%的准确率不现实。重点转向如何优雅地处理失败(重试、降级、澄清询问),如何设计确认机制,以及如何提供清晰的进度反馈。
  4. 从“模型中心”到“系统工程”: 最终用户体验的瓶颈,往往不在模型能力,而在系统可靠性、响应速度、错误处理和交互设计。你需要像对待一个分布式系统一样,为你的智能体设计监控、告警和回滚机制。

因此,当你下次启动一个“智能体”项目时,不妨先问自己几个问题:我的智能体需要管理哪些状态?它可能遇到哪些分支情况?每个工具调用失败后该怎么办?我如何追溯和调试一次完整的对话?想清楚这些问题,并用 LangGraph 这样的工具将你的设计具象化,你才算是真正踏入了智能体开发的大门。这远比纠结于用哪个向量数据库的 recall 高 0.5% 更重要,因为前者决定了你的系统能否“活”下去并解决真实问题,而后者只是让它解决得更好一点。在智能体时代,先让流程“跑通”并“可控”,是比“跑得精”更优先的工程原则。

🚀 30+款热门AI模型一站整合,DeepSeek/GLM/Qwen 随心用,限时 5 折。 👉 点击领海量免费额度

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

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

立即咨询