30款热门AI模型一站整合DeepSeek/GLM/Qwen 随心用限时 5 折。 点击领海量免费额度1. 这篇文章真正要解决的问题如果你正在尝试构建一个能回答复杂问题、引用可靠来源的智能助手那么你很可能已经接触过 RAG检索增强生成。但你是否发现一个简单的 RAG 系统在真实业务场景中往往“不堪一击”它可能无法理解多跳查询无法验证信息来源更无法在信息不足时主动探索。这正是当前 AI 应用从“玩具”走向“生产级”的核心瓶颈。本文要解决的正是如何跨越这道鸿沟。我们不再讨论基础的 RAG 概念而是聚焦于“工程化 Agentic RAG 系统”。这不仅仅是给 RAG 套上一个 Agent 的壳而是构建一个具备自主规划、工具调用、验证与反思能力的可信赖 AI 系统。其核心挑战在于如何将 Google Search 这类强大的外部工具无缝、安全、可控地集成到 AI 的工作流中并最终形成一个能在生产环境稳定运行、输出可信结果的智能体。读完本文你将清晰地理解 Agentic RAG 的工程架构核心掌握从零搭建一个具备联网搜索能力的智能体的关键步骤并规避从原型到生产过程中最常见的“坑”。无论你是希望升级现有问答系统还是从头构建一个行业专家助手这篇文章都将提供一条可落地的技术路径。2. 基础概念与核心原理从 RAG 到 Agentic RAG在深入工程细节前我们必须厘清几个关键概念以及它们是如何演进和结合的。传统 RAG检索增强生成其工作流是线性的“检索-生成”。用户提问 → 系统从向量库检索相关文档片段 → 将片段和问题一起喂给大模型 → 模型生成答案。它的局限性很明显检索一次生成一次缺乏对复杂问题的分解能力也无法在首次检索失败时采取补救措施。AI Agent智能体在这里我们指的是能感知环境、规划目标、执行动作通过调用工具并基于结果进行反思的智能系统。一个典型的 Agent 循环是思考规划步骤→ 行动调用搜索、计算器等工具→ 观察获取工具结果→ 反思评估结果是否足够决定下一步。Agentic RAG这是上述两者的融合。它将 RAG 的检索与生成能力转变为 Agent 可以调用的“工具”之一。同时Agent 的规划能力被用来管理 RAG 的流程。例如对于一个复杂问题Agent 可能会规划需要先理解问题拆解成多个子问题。行动1调用“搜索引擎工具”获取最新的背景信息。行动2调用“向量数据库检索工具”查找内部专业知识。观察整合来自外部搜索和内部检索的信息。反思信息是否足够、是否冲突是否需要进一步搜索或验证最终行动调用“文本生成工具”基于所有核实过的信息生成最终答案。生产级可信这是工程化的目标。它意味着系统需要具备可靠性高可用低延迟错误有兜底。可观测性每一步的思考、行动、观察都可记录、可追溯、可调试。安全性控制工具调用的权限和频率防止无限循环或有害请求。结果可信答案附带来源引用对可能的不确定性进行标注。理解了这些我们就能看到工程化的核心在于设计一个稳健的Agent 运行循环并将 RAG 及其它工具如 Google Search作为可插拔的组件接入这个循环。3. 环境准备与前置条件在开始构建之前请确保你的开发环境满足以下要求。我们将以一个基于 Python 的流行技术栈为例。操作系统Linux (Ubuntu 20.04)、macOS 或 WSL2 (Windows)。生产环境推荐 Linux。Python版本 3.9 或 3.10。避免使用 3.11 可能存在的某些包兼容性问题。关键依赖包LangChain / LangGraph用于编排 Agent 工作流的核心框架。LangGraph 特别适合构建有状态的、循环的 Agent 系统。OpenAI API / 或其他大模型 API作为 Agent 的“大脑”LLM。我们将使用 GPT-4 或 GPT-3.5-Turbo。Tavily API / SerpAPI提供联网搜索能力的工具。鉴于直接使用 Google Search API 的复杂性我们使用这些聚合服务。请注意你必须合法合规地获取和使用这些服务的 API Key并严格遵守其服务条款。Chroma / Weaviate / Pinecone向量数据库用于存储和管理内部知识库。Sentence-Transformers用于生成文本嵌入向量。FastAPI / Flask用于构建提供服务的 Web API。Pydantic用于数据验证和设置管理。日志与监控structlog或loguru用于结构化日志Prometheus和Grafana用于生产环境监控后续阶段。环境变量管理使用.env文件管理敏感信息如 API Keys。# .env 文件示例 OPENAI_API_KEYsk-your-openai-key-here TAVILY_API_KEYyour-tavily-key-here MODEL_NAMEgpt-4-turbo-preview使用python-dotenv在应用中加载# config.py from pydantic_settings import BaseSettings from dotenv import load_dotenv load_dotenv() class Settings(BaseSettings): openai_api_key: str tavily_api_key: str model_name: str gpt-4-turbo-preview settings Settings()4. 核心流程拆解Agentic RAG 系统架构一个工程化的 Agentic RAG 系统通常包含以下核心模块和流程。我们将以“回答一个需要最新市场数据和内部报告的综合问题”为例。步骤 1问题接收与解析做什么接收用户原始查询。为什么需要对查询进行预处理例如检查是否包含敏感词、进行意图分类是简单问答、复杂分析还是需要搜索。关键点这里可以集成一个轻量级分类模型或规则引擎决定是否触发 Agent 流程。对于简单问题可能直接走传统 RAG 更高效。步骤 2Agent 规划与任务分解做什么LLM作为规划器分析问题并生成一个执行计划。为什么复杂问题无法一步解决。规划步骤将问题拆解为顺序或并行的子任务例如“1. 搜索2024年Q1智能手机全球出货量。2. 从内部数据库检索我们的产品市场份额报告。3. 对比数据分析差距。”关键点使用 LangChain 的PlanAndExecute或 LangGraph 的StateGraph来建模此流程。规划提示词Prompt的设计至关重要。步骤 3工具调用与执行做什么Agent 根据规划依次调用工具。工具类型网络搜索工具调用 Tavily API 获取最新、最相关的网页摘要和链接。向量检索工具从 Chroma 向量库中检索相关的内部文档片段。计算器/代码解释器工具如需进行数值计算或数据分析。为什么每个工具负责获取特定类型的信息或完成特定操作。工具的设计应单一职责并返回结构化的结果。步骤 4信息验证与合成做什么Agent 对收集到的信息进行交叉验证和去重。为什么网络信息可能矛盾内部数据可能过时。此步骤需要 LLM 判断信息的一致性、时效性和可靠性。关键点这是实现“可信”的关键。可以设计一个“验证”节点让 LLM 评估信息的质量并决定是否需要重新搜索或检索。步骤 5反思与迭代做什么评估当前收集的信息是否足以回答原始问题。为什么如果信息不足或质量不高Agent 需要回到规划步骤提出新的、更具体的问题继续搜索或检索形成循环。关键点使用 LangGraph 的循环机制可以优雅地实现这一点。必须设置最大迭代次数以防止无限循环。步骤 6最终答案生成与溯源做什么使用所有经过验证的信息生成最终答案。为什么答案需要自然、准确并且必须明确标注信息来源。关键点强制 LLM 在答案中引用来源如[1],[2]并在最后列出详细的参考链接和文档标题。这直接提升了答案的可信度和可审计性。5. 完整示例与代码实现下面我们使用LangGraph来构建一个具备搜索和检索能力的 Agentic RAG 系统。LangGraph 通过“图”的概念来定义状态和节点间的流转非常适合此类工作流。首先定义系统的核心状态State# agent_state.py from typing import TypedDict, List, Annotated import operator class AgentState(TypedDict): Agent 工作流的状态定义 question: str # 原始问题 plan: List[str] # 执行计划 search_results: List[dict] # 网络搜索结果 retrieved_docs: List[str] # 向量检索结果 validated_info: List[dict] # 经过验证的信息片段 answer: str # 最终答案 iteration_count: int # 迭代次数用于控制循环接着定义各个工具节点。首先是网络搜索工具# tools/search_tool.py import os from tavily import TavilyClient from langchain.tools import tool tool def tavily_search(query: str) - str: 使用 Tavily API 进行网络搜索返回相关的摘要和链接。 tavily_client TavilyClient(api_keyos.getenv(TAVILY_API_KEY)) try: # 限制搜索深度和数量控制成本与延迟 response tavily_client.search(query, max_results5, search_depthbasic) # 将结果格式化为结构化字符串 formatted_results [] for result in response.get(results, []): formatted_results.append( f标题: {result.get(title, N/A)}\n f链接: {result.get(url, N/A)}\n f内容: {result.get(content, N/A)[:500]}...\n f--- ) return \n.join(formatted_results) if formatted_results else 未找到相关网络信息。 except Exception as e: return f搜索工具出错: {str(e)}然后是向量检索工具# tools/retrieval_tool.py from langchain.vectorstores import Chroma from langchain.embeddings import OpenAIEmbeddings from langchain.tools import tool import os # 初始化向量库假设已存在 embeddings OpenAIEmbeddings(openai_api_keyos.getenv(OPENAI_API_KEY)) vectorstore Chroma( persist_directory./chroma_db, embedding_functionembeddings ) retriever vectorstore.as_retriever(search_kwargs{k: 4}) # 检索4个最相关片段 tool def retrieve_internal_knowledge(query: str) - str: 从内部知识库中检索与问题相关的文档片段。 try: docs retriever.get_relevant_documents(query) formatted_docs [] for i, doc in enumerate(docs): formatted_docs.append( f[内部文档片段 {i1}]:\n f来源: {doc.metadata.get(source, 未知)}\n f内容: {doc.page_content[:600]}...\n f--- ) return \n.join(formatted_docs) if formatted_docs else 未找到相关内部资料。 except Exception as e: return f检索工具出错: {str(e)}现在使用 LangGraph 构建主工作流图# agent_graph.py from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain.prompts import ChatPromptTemplate from agent_state import AgentState from tools.search_tool import tavily_search from tools.retrieval_tool import retrieve_internal_knowledge import os llm ChatOpenAI(modelos.getenv(MODEL_NAME), temperature0, openai_api_keyos.getenv(OPENAI_API_KEY)) # 1. 规划节点 def plan_node(state: AgentState): 分析问题制定执行计划。 planner_prompt ChatPromptTemplate.from_messages([ (system, 你是一个任务规划专家。请将以下复杂问题分解为具体的、可执行的子任务。子任务应明确说明需要调用什么工具如‘网络搜索’或‘内部检索’。), (human, 问题{question}\n当前已验证信息{validated_info}\n请给出计划。) ]) chain planner_prompt | llm plan_text chain.invoke({question: state[question], validated_info: state[validated_info]}).content # 简单按行分割作为计划步骤 plan_steps [step.strip() for step in plan_text.split(\n) if step.strip()] return {plan: plan_steps} # 2. 工具路由与执行节点 def execute_tools(state: AgentState): 根据计划执行工具。这是一个简化示例实际中需要更复杂的路由逻辑。 plan state[plan] search_results [] retrieved_docs [] for step in plan: if 搜索 in step or search in step.lower(): # 提取搜索关键词这里简化处理实际应用需要更精细的解析 search_query state[question] # 简化用原问题搜索 result tavily_search.invoke(search_query) search_results.append({step: step, result: result}) elif 内部 in step or 检索 in step: retrieval_query state[question] # 简化用原问题检索 result retrieve_internal_knowledge.invoke(retrieval_query) retrieved_docs.append({step: step, result: result}) return {search_results: search_results, retrieved_docs: retrieved_docs} # 3. 验证与合成节点 def validate_and_synthesize_node(state: AgentState): 验证收集到的信息并合成有效信息片段。 all_info [] for sr in state[search_results]: all_info.append(f来源[网络]: {sr[result]}) for rd in state[retrieved_docs]: all_info.append(f来源[内部]: {rd[result]}) validator_prompt ChatPromptTemplate.from_messages([ (system, 你是一个信息验证员。请评估以下信息的可靠性、时效性和相关性剔除矛盾或不可信的部分合成一个简洁、准确的信息摘要。对于网络信息请特别留意其来源是否权威。), (human, 原始问题{question}\n收集到的原始信息\n{raw_info}\n请提供验证后的信息摘要。) ]) chain validator_prompt | llm raw_info_text \n---\n.join(all_info) validated_summary chain.invoke({question: state[question], raw_info: raw_info_text}).content # 将本次迭代验证后的信息追加到历史中 new_validated_info state[validated_info] [validated_summary] return {validated_info: new_validated_info} # 4. 判断是否继续循环的节点 def should_continue_node(state: AgentState): 判断信息是否足够或是否达到最大迭代次数。 max_iterations 3 if state[iteration_count] max_iterations: return end # 让LLM判断当前信息是否足以回答问题 judge_prompt ChatPromptTemplate.from_messages([ (system, 请判断基于当前已验证的信息是否足以清晰、准确地回答用户问题如果信息仍不足或有明显缺失请回答‘不足’否则回答‘足够’。), (human, 问题{question}\n当前已验证信息{validated_info}\n判断) ]) chain judge_prompt | llm decision chain.invoke({question: state[question], validated_info: state[validated_info]}).content.strip() if 足够 in decision: return end else: # 增加迭代计数准备进入下一轮 return continue # 5. 最终答案生成节点 def generate_answer_node(state: AgentState): 基于所有验证后的信息生成最终答案。 answer_prompt ChatPromptTemplate.from_messages([ (system, 你是一个专业的分析师。请基于以下所有已验证的信息生成一个全面、准确、结构清晰的答案。答案必须包含引用格式如 [信息1], [信息2]。最后请列出所有参考来源的简要说明。), (human, 问题{question}\n已验证信息\n{validated_info}\n请生成最终答案) ]) chain answer_prompt | llm final_answer chain.invoke({question: state[question], validated_info: state[validated_info]}).content return {answer: final_answer} # 构建图 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(plan, plan_node) workflow.add_node(execute, execute_tools) workflow.add_node(validate, validate_and_synthesize_node) workflow.add_node(generate_answer, generate_answer_node) # 设置边和流程 workflow.set_entry_point(plan) workflow.add_edge(plan, execute) workflow.add_edge(execute, validate) # 条件边验证后判断是继续循环还是结束 workflow.add_conditional_edges( validate, should_continue_node, { continue: plan, # 信息不足返回规划节点开始新一轮 end: generate_answer # 信息足够去生成答案 } ) workflow.add_edge(generate_answer, END) # 编译图 app workflow.compile()6. 运行结果与效果验证现在我们可以运行这个 Agent 来处理一个复杂查询。创建一个主执行文件# main.py from agent_graph import app from agent_state import AgentState import asyncio async def run_agentic_rag(question: str): 运行Agentic RAG工作流 # 初始化状态 initial_state: AgentState { question: question, plan: [], search_results: [], retrieved_docs: [], validated_info: [], answer: , iteration_count: 0 } print(f开始处理问题: {question}\n) # 执行图 final_state None async for event in app.astream(initial_state, stream_modevalues): node_name list(event.keys())[0] state event[node_name] print(f[节点: {node_name}]) if node_name plan: print(f 生成计划: {state[plan]}) elif node_name execute: print(f 执行工具完成。搜索结果数: {len(state[search_results])}, 检索结果数: {len(state[retrieved_docs])}) elif node_name validate: print(f 验证并合成信息。当前已验证信息块数: {len(state[validated_info])}) print(f 最新信息摘要: {state[validated_info][-1][:200]}...) # 打印前200字符 elif node_name generate_answer: final_state state print(f 生成最终答案。) print(- * 50) if final_state: print(\n *60) print(最终答案) print(*60) print(final_state[answer]) print(\n *60) print(f总迭代次数: {final_state.get(iteration_count, N/A)}) return final_state if __name__ __main__: # 示例问题一个需要结合最新动态和内部知识的复杂问题 query 对比分析特斯拉在2024年第一季度的全球电动汽车交付量数据以及我们公司内部对2024年市场增长的预测报告指出我们的预测在哪些区域可能过于保守或激进 asyncio.run(run_agentic_rag(query))预期输出与验证 运行上述脚本你应该能看到控制台输出工作流的执行步骤。一个成功的运行将展示[节点: plan]显示生成的计划例如[1. 使用网络搜索工具查找特斯拉2024年Q1全球交付量数据。, 2. 使用内部检索工具查找公司2024年市场增长预测报告。]。[节点: execute]显示工具调用完成并获取到结果数量。[节点: validate]显示验证后的信息摘要。根据信息充足与否可能会在plan - execute - validate之间循环多次。最终进入[节点: generate_answer]并输出结构化的答案。验证成功的关键标志答案包含引用答案中应有类似[1],[2]的标记。答案综合了多方信息既提到了特斯拉的交付量来自网络搜索也结合了内部预测报告。有明确的结论指出了预测可能保守或激进的区域。流程可追溯通过日志可以回溯 Agent 的每一步思考和行动。如果运行失败第一步应检查API Key确保OPENAI_API_KEY和TAVILY_API_KEY已正确设置。网络连接确保能正常访问外部 API。向量库确保./chroma_db路径下存在已构建好的向量库。依赖版本检查langchain,langgraph,tavily-python等包的版本兼容性。7. 常见问题与排查思路在开发和部署 Agentic RAG 系统时你会遇到一些典型问题。下表列出了常见现象、原因及解决方案。问题现象可能原因排查方式解决方案Agent 陷入无限循环不停止。1.should_continue_node中的 LLM 判断逻辑不清晰始终返回“不足”。2. 最大迭代次数设置过高或未生效。1. 检查该节点的 Prompt确保指令明确。可让 LLM 输出“是/否”而非自由文本。2. 打印state[‘iteration_count’]检查计数逻辑。1. 优化判断 Prompt例如“如果信息已覆盖问题的 A、B、C 三点则回答‘足够’否则回答‘不足’。”2. 在状态中硬编码max_iterations并在判断节点强制拦截。网络搜索返回结果不相关或质量差。1. 搜索查询Query构建不佳直接使用了原始长问题。2. Tavily API 参数如search_depth设置不当。1. 在调用搜索工具前增加一个“查询重写”节点让 LLM 将问题提炼成搜索关键词。2. 检查 Tavily 返回的原始 JSON看score相关性分数是否过低。1. 添加查询优化步骤optimized_query llm.invoke(f“将以下问题转化为简洁的搜索关键词{question}”)。2. 调整max_results和search_depth或考虑使用高级搜索Advanced模式成本更高。向量检索返回的内容不准确。1. 原始文档切分Chunk策略不合理导致语义不完整。2. 嵌入模型Embedding Model与任务不匹配。3. 检索的 top-k 值不合适。1. 检查被检索出的文档片段看其开头结尾是否完整。2. 尝试不同的嵌入模型如text-embedding-3-small。3. 调整retriever.search_kwargs[“k”]的值。1. 优化文本分割器尝试按语义句子或固定长度重叠分割。2. 在向量库中混合使用不同嵌入模型进行测试选择召回率最高的。3. 引入“重排序Re-ranking”模型对初步检索结果进行二次精排。最终答案不引用来源或引用错误。1. 生成答案的 Prompt 指令不够强制。2. 传递给最终生成环节的“已验证信息”结构混乱LLM 无法关联。1. 检查generate_answer_node的 Prompt是否明确要求引用格式。2. 查看state[‘validated_info’]的内容是否每个信息块都有清晰标识。1. 在 Prompt 中使用更严格的指令例如“你必须为答案中的每一个关键数据或论断标注来源格式为 [来源X]其中 X 对应下方信息块的编号。”2. 在验证节点为每个信息块生成一个唯一 ID如info_1,info_2并结构化存储。系统延迟非常高。1. 串行调用工具网络 I/O 等待时间长。2. LLM 调用规划、验证、生成次数过多。3. 向量检索范围过大。1. 使用异步Async方式并发调用可以同时执行的工具。2. 统计每个节点的耗时定位瓶颈。3. 检查向量索引是否已构建优化。1. 使用asyncio.gather()并发执行独立的搜索和检索任务。2. 对于简单判断使用更便宜、更快的模型如 GPT-3.5-Turbo。3. 对向量数据库进行性能调优或使用更高效的向量库如 Weaviate, Qdrant。API 调用成本失控。1. Agent 循环次数过多每次循环都调用 LLM 和工具。2. 搜索工具每次返回大量、冗长的结果全部送入 LLM 上下文。1. 监控日志统计每次查询的平均 Token 消耗和 API 调用次数。2. 检查搜索工具返回的内容长度。1. 严格限制最大迭代次数如 3 次。实现成本预算超出则提前终止。2. 在搜索工具中只返回摘要和关键信息而非完整网页内容。对长文本进行压缩Summarization后再送入后续节点。8. 最佳实践与工程建议要将一个原型系统推进到生产级必须考虑以下工程化实践1. 可观测性与日志结构化日志使用structlog或loguru记录每个节点的输入、输出、耗时和错误。关键信息包括session_id,node_name,input_snapshot,output_snapshot,duration_ms,llm_calls。链路追踪为每个用户查询生成唯一的trace_id贯穿整个工作流方便问题排查。监控指标暴露 Prometheus 指标如agent_invocation_total,agent_iteration_count,tool_call_duration_seconds,llm_token_usage。2. 稳定性与容错超时与重试为所有外部 API 调用LLM、搜索、向量库设置合理的超时和重试机制如使用tenacity库。断路器模式当某个工具如搜索引擎连续失败时暂时将其熔断避免拖垮整个系统并降级到备用方案如仅使用内部知识库。验证与过滤在信息合成节点之前增加一个“事实性检查”或“毒性内容过滤”节点拦截明显错误或有害的信息。3. 性能优化异步并发如前所述将可并行的工具调用如同时搜索A和B改为异步。缓存策略对频繁出现的、结果不变的查询如“公司的核心价值观是什么”进行缓存。可以使用Redis缓存 LLM 响应或工具结果。向量检索优化使用 HNSW 等高效索引算法对向量进行量化PQ以减少内存占用和加速检索。4. 安全与权限工具权限控制不是所有 Agent 都能调用所有工具。根据用户角色或查询类型动态启用或禁用某些工具如内部文档检索可能仅限于内部员工。输入输出审查对用户输入和 Agent 生成的输出进行审查防止 Prompt 注入或输出有害内容。API 密钥管理使用HashiCorp Vault或云服务商的密钥管理服务而非硬编码在环境变量或代码中。5. 配置与版本管理配置外部化将 Prompt 模板、模型参数、工具参数、迭代次数等全部抽取到配置文件如config.yaml中。版本化 Prompt将 Prompt 视为代码进行版本控制。每次修改 Prompt 都可能显著影响 Agent 行为。实验管理使用MLflow或Weights Biases跟踪不同的 Agent 架构、Prompt 和参数下的实验效果。9. 总结与后续学习方向构建一个工程化的 Agentic RAG 系统本质上是将大模型的推理能力、外部工具的获取能力与内部知识的管理能力通过一个可控、可观测、可迭代的自动化流程串联起来。本文带你走完了从核心概念理解到具体代码实现的全过程关键在于掌握“状态管理”和“循环控制”这两个 LangGraph 的核心思想。你下一步可以沿着这几个方向深入更复杂的规划策略探索ReAct、Plan-and-Solve、Chain-of-Thought等更高级的规划范式并将其集成到图中。多智能体协作引入不同的“专家”Agent如研究员、分析师、校对员让他们通过协作来解决问题这可以通过 LangGraph 的多子图Multi-Agent功能实现。工具扩展除了搜索和检索可以集成代码执行器、数据库查询、API 调用等更多工具让 Agent 的能力边界无限扩展。评估与持续改进建立一套离线评估体系用一批测试问题来量化 Agent 回答的准确性、相关性和可信度从而驱动 Prompt 和流程的迭代优化。记住一个强大的生产级 Agent 系统不是一蹴而就的。它始于一个清晰的工作流设计成长于持续的迭代、测试和监控。建议你从本文的示例出发先在一个明确的业务场景中跑通闭环再逐步纳入上述工程化实践最终打造出真正可靠、可信的 AI 智能体。 30款热门AI模型一站整合DeepSeek/GLM/Qwen 随心用限时 5 折。 点击领海量免费额度