当RAG学会思考用LangGraph构建Agentic RAG系统完全指南2024年我们还在为能用的RAG系统欢呼2025年RAG已经成为企业AI的标配但到了2026年如果你还在用传统的线性RAG——检索、增强、生成——你可能正在浪费用户的时间。根据2025年的生产环境数据高达30%的RAG响应存在事实性错误而这些错误的根源几乎都指向同一个问题检索质量。这篇文章将带你从传统RAG的困境出发深入理解Agentic RAG的设计哲学并通过LangGraph构建一个真正会思考的检索系统。一、传统RAG的致命缺陷1.1 线性管道的脆弱性传统RAG的工作流程非常直观用户提问 → 向量检索 → 拼接上下文 → LLM生成回答这个流程看起来完美但实际生产中却处处踩坑问题一检索器装聋作哑当用户问Lilian Weng关于reward hacking的类型有什么观点时检索器可能只返回了reward hacking相关的片段却漏掉了types这个关键限定词。结果LLM基于不完整的信息生成了一个看似合理但完全偏离重点的回答。问题二查询歧义无法处理用户问那个产品怎么样——哪个产品传统RAG没有任何机制来澄清或细化查询只能硬着头皮检索然后生成一个模糊的回答。问题三错误无法自纠正当检索到的文档与问题无关时传统RAG没有意识到错误的能力。它只会忠实地把无关内容喂给LLM然后LLM会产生一个自信但错误的回答——这就是所谓的幻觉问题。1.2 为什么简单方案不够用你可能会想“那我们加个re-ranker或者调大top-k”这些都是治标不治本。真正的问题在于传统RAG是被动的而复杂问题需要主动的推理。就像一个只会按图索骥的图书管理员传统RAG从不质疑用户的查询从不反思检索结果的质量也从不主动寻找更多信息。而Agentic RAG就是要让系统具备这种主动性。二、什么是Agentic RAG2.1 从管道到循环范式的根本转变Agentic RAG的核心转变是从线性管道到推理循环。这不是简单的技术升级而是思维方式的根本改变。传统RAG的工作模式就像一个只会执行命令的流水线工人1.收到问题2.去仓库找资料3.把资料交给写手4.写手生成答案问题出在哪里每个环节都是盲执行。工人不会质疑问题是否清晰不会检查找来的资料是否相关更不会在发现问题后主动修正。Agentic RAG的工作模式则像一个有经验的咨询顾问5.先理解你的问题必要时追问澄清6.判断需要查哪些资料去哪里查7.拿到资料后评估是否真的有用8.如果资料不对换个角度再查9.确认信息可靠后才给出专业答案这就是从被动执行到主动推理的质变。2.2 Think-Act-Observe循环Agentic RAG的核心是Think-Act-Observe思考-行动-观察循环这个循环会持续进行直到系统认为已经获得足够信息来回答问题。Think思考LLM分析当前状态决定下一步做什么·这个问题需要查资料吗·用户真正想问的是什么·我现有的信息足够吗Act行动执行具体的操作·调用检索工具·改写查询词·选择合适的数据源Observe观察评估行动结果·检索到的文档相关吗·信息是否完整·需要进一步行动吗2.3 五大核心能力详解一个成熟的Agentic RAG系统具备以下五种核心能力它们共同构成了系统的智能能力1智能查询理解系统能够深入理解用户问题的真实意图而不是简单地匹配关键词。工作原理·意图识别分析问题是问什么、“怎么问”、“为什么问”·实体抽取识别问题中涉及的关键实体人名、产品名、概念等·复杂度评估判断问题是否需要多步推理或简单查询就能解决实际例子用户问“那个产品怎么样”→ 意图识别用户想了解某个产品→ 实体抽取那个没有指明具体产品→ 系统回复“您指的是哪个产品能提供更具体的名称吗”能力2动态检索策略系统不会用固定的检索策略而是根据问题类型动态选择。支持的策略·向量检索基于语义相似度适合概念性问题·关键词检索精确匹配术语适合事实查询·混合检索结合向量和关键词取长补短·多跳检索当结果不足时自动扩展搜索范围能力3自我反思与纠错当检索结果不理想时系统能够意识到问题并主动纠正。纠错机制·相关性评分判断检索到的内容是否真的有用·质量评估检查信息是否过时、是否有偏见·改写决策当文档不相关时决定如何改写查询能力4多工具协作系统不局限于向量数据库可以调用多种外部工具。常用工具·向量数据库存储文档的检索器·Web搜索API搜索实时信息·数据库查询查询结构化数据·计算工具执行数学运算·外部API调用第三方服务能力5记忆与上下文系统记住对话历史在后续交互中保持连贯性。记忆类型·短期记忆当前会话的上下文·长期记忆跨会话的历史信息·工作记忆中间推理步骤的保存2.4 Agentic RAG vs 传统RAG特性传统RAGAgentic RAG执行模式线性管道循环推理检索方式被动检索主动决策错误处理无法纠错自我反思工具支持单一向量库多工具协作幻觉率~30%5%响应延迟低略高什么时候选择Agentic RAG·检索失败时需要自动重试·答案不在知识库中时需要追问·需要多轮交互·对质量要求高2.5 核心架构组件一个完整的Agentic RAG系统通常包含以下组件查询分析器Query Analyzer·判断是否需要检索·识别查询意图和实体·生成检索策略检索器Retriever·向量相似度检索·关键词匹配·混合检索策略查询改写器Query Rewriter·扩展查询词·消除歧义·语义增强文档评分器Document Grader·评估文档相关性·过滤无关内容·生成质量信号回答生成器Answer Generator·基于有效上下文生成回答·标注引用来源·处理不确定情况2.6 为什么选择LangGraph在众多Agent框架中LangGraph有几个独特优势状态管理内置的状态机制让复杂的多步推理变得可控。每一步的状态变化都有迹可循便于调试和优化。可视化可以清晰地看到每一步的决策过程。出了问题一眼就能看出卡在哪一步。持久化支持会话记忆和断点恢复。长对话不会丢失上下文用户体验更连贯。灵活性既可以用预置模板快速开始也可以完全自定义。从简单到复杂平滑过渡。生产就绪支持流式输出、错误重试、并发控制等企业级特性。最重要的是LangGraph的设计哲学与Agentic RAG高度契合——用图来表示状态流转用节点来封装逻辑用边来定义决策。三、用LangGraph构建Agentic RAG3.1 环境准备首先安装必要的依赖[bash]pip install langgraph langchain langchain-openai chromadb你需要准备·OpenAI API Key或其他LLM服务·一个向量数据库本例使用Chroma3.2 定义核心组件Step 1: 创建向量检索器[python]from langchain_openai import OpenAIEmbeddings, ChatOpenAIfrom langchain_community.vectorstores import Chromafrom langchain.text_splitter import RecursiveCharacterTextSplitterfrom langchain_community.document_loaders import TextLoader加载文档loader TextLoader(“your_knowledge_base.txt”)documents loader.load()分割文档text_splitter RecursiveCharacterTextSplitter(chunk_size1000,chunk_overlap200)splits text_splitter.split_documents(documents)创建向量存储vectorstore Chroma.from_documents(documentssplits,embeddingOpenAIEmbeddings())retriever vectorstore.as_retriever(search_kwargs{“k”: 5})Step 2: 定义检索工具[python]from langchain.tools import tooltooldef retrieve_documents(query: str) - str:“”“从知识库中检索相关文档”“”docs retriever.invoke(query)return “\n\n”.join([doc.page_content for doc in docs])retriever_tool retrieve_documentsStep 3: 初始化LLM[python]from langchain.chat_models import init_chat_modelllm init_chat_model(“gpt-4.1”, temperature0)3.3 构建状态图LangGraph的核心是StateGraph我们用它来定义整个工作流[python]from langgraph.graph import StateGraph, MessagesState, START, ENDfrom langgraph.prebuilt import ToolNode, tools_conditionfrom langchain.messages import HumanMessage创建状态图workflow StateGraph(MessagesState)MessagesState 是LangGraph预置的状态类它包含一个messages列表用于存储对话历史。3.4 实现核心节点节点1: 查询生成/响应决策[python]def generate_query_or_respond(state: MessagesState):“”“让LLM决定是否需要检索”“”response llm.bind_tools([retriever_tool]).invoke(state[“messages”])return {“messages”: [response]}节点2: 查询改写[python]REWRITE_PROMPT “”分析用户问题的语义意图生成一个更适合检索的改写版本。原始问题{question}请生成一个改进后的问题要求保留核心意图添加可能的同义词消除歧义改写后的问题“”def rewrite_question(state: MessagesState):“”“改写原始问题以提高检索质量”“”question state[“messages”][0].contentprompt REWRITE_PROMPT.format(questionquestion)response llm.invoke([{“role”: “user”, “content”: prompt}])return {“messages”: [HumanMessage(contentresponse.content)]}节点3: 文档评分[python]from pydantic import BaseModel, Fieldfrom typing import Literalclass GradeDocuments(BaseModel):“”“文档相关性评分”“”binary_score: str Field(description“‘yes’ 如果文档相关‘no’ 如果不相关”)GRADE_PROMPT “”你是一个文档相关性评估器。检索到的文档{context}用户问题{question}判断文档是否包含与问题相关的信息只回答 ‘yes’ 或 ‘no’。“”def grade_documents(state: MessagesState) - Literal[“generate_answer”, “rewrite_question”]:“”“评估检索文档的相关性”“”question state[“messages”][0].contentcontext state[“messages”][-1].contentprompt GRADE\_PROMPT.format(questionquestion, contextcontext) grader llm.with\_structured\_output(GradeDocuments) response grader.invoke([{role: user, content: prompt}]) if response.binary\_score yes: return generate\_answer else: return rewrite\_question节点4: 生成回答[python]GENERATE_PROMPT “”基于以下上下文回答用户问题。如果上下文中没有足够信息请明确说明。上下文{context}问题{question}回答请标注信息来源“”def generate_answer(state: MessagesState):“”“基于检索到的上下文生成回答”“”question state[“messages”][0].contentcontext state[“messages”][-1].contentprompt GENERATE\_PROMPT.format(questionquestion, contextcontext) response llm.invoke([{role: user, content: prompt}]) return {messages: [response]}3.5 组装完整工作流[python]添加节点workflow.add_node(generate_query_or_respond)workflow.add_node(“retrieve”, ToolNode([retriever_tool]))workflow.add_node(rewrite_question)workflow.add_node(generate_answer)设置入口workflow.add_edge(START, “generate_query_or_respond”)条件边决定是否检索workflow.add_conditional_edges(“generate_query_or_respond”,tools_condition, # 自动判断是否调用了工具{“tools”: “retrieve”,END: END # 如果不需要检索直接结束})条件边评估文档质量workflow.add_conditional_edges(“retrieve”,grade_documents # 返回 “generate_answer” 或 “rewrite_question”)回答后结束workflow.add_edge(“generate_answer”, END)改写后重新开始检索workflow.add_edge(“rewrite_question”, “generate_query_or_respond”)编译图graph workflow.compile()3.6 执行和调试[python]执行查询for chunk in graph.stream({“messages”: [{“role”: “user”,“content”: “Lilian Weng关于reward hacking的类型有什么观点”}]}):for node, update in chunk.items():print(f {node} )update[“messages”][-1].pretty_print()print()输出示例 generate_query_or_respond [调用工具: retrieve_documents] retrieve [检索到3篇文档] grade_documents [评分: no - 文档不相关] rewrite_question [改写为: “Lilian Weng reward hacking types classification categorization”] generate_query_or_respond [调用工具: retrieve_documents] retrieve [检索到5篇文档] grade_documents [评分: yes - 文档相关] generate_answer 根据Lilian Weng的文章reward hacking主要有三种类型…四、进阶技巧与最佳实践4.1 多检索器协同复杂场景下你可能需要多个专业检索器[python]tooldef search_product_docs(query: str) - str:“”“搜索产品文档”“”return product_retriever.invoke(query)tooldef search_api_docs(query: str) - str:“”“搜索API文档”“”return api_retriever.invoke(query)tooldef search_troubleshooting(query: str) - str:“”“搜索故障排除指南”“”return troubleshoot_retriever.invoke(query)tools [search_product_docs, search_api_docs, search_troubleshooting]LLM会根据问题自动选择合适的工具。4.2 添加记忆能力使用LangGraph的checkpointer实现会话记忆[python]from langgraph.checkpoint.memory import InMemorySavercheckpointer InMemorySaver()graph workflow.compile(checkpointercheckpointer)执行时指定thread_idconfig {“configurable”: {“thread_id”: “user_123”}}result graph.invoke({“messages”: […]}, config)4.3 流式输出对于长回答使用流式输出提升用户体验[python]async for event in graph.astream({“messages”: […]},stream_mode“values”):if “messages” in event:print(event[“messages”][-1].content, end“”, flushTrue)4.4 错误处理与重试为关键节点添加错误处理[python]from tenacity import retry, stop_after_attempt, wait_exponentialretry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10))def retrieve_with_retry(state: MessagesState):try:return retrieve_documents(state)except Exception as e:print(f检索失败: {e}, 正在重试…)raise4.5 性能优化缓存检索结果[python]from functools import lru_cachelru_cache(maxsize1000)def cached_retrieve(query: str) - str:return retriever.invoke(query)并行检索[python]import asyncioasync def parallel_retrieve(queries: list[str]) - list[str]:tasks [async_retrieve(q) for q in queries]return await asyncio.gather(*tasks)批量处理对于大量文档使用批量嵌入[python]批量嵌入比逐个嵌入快10倍以上embeddings OpenAIEmbeddings()batch_embeddings embeddings.embed_documents([doc.page_content for doc in documents])五、常见问题与避坑指南5.1 检索质量差怎么办症状检索到的文档与问题无关解决方案10.检查分块策略chunk_size是否合适11.尝试混合检索向量关键词12.添加re-ranker重排序13.优化查询改写prompt5.2 响应太慢怎么办症状用户等待时间过长解决方案14.减少循环迭代次数设置最大重试次数15.使用更快的模型如GPT-4o-mini16.实现流式输出17.缓存常见查询5.3 幻觉仍然存在症状LLM生成错误信息解决方案18.强化文档评分逻辑19.在prompt中明确不知道就说不知道20.添加事实核查节点21.使用结构化输出强制引用来源5.4 状态管理混乱症状多轮对话中上下文丢失解决方案22.使用checkpointer持久化状态23.定期总结对话历史24.为每个用户使用独立的thread_idAI行业迎来前所未有的爆发式增长从DeepSeek百万年薪招聘AI研究员到百度、阿里、腾讯等大厂疯狂布局AI Agent再到国家政策大力扶持数字经济和AI人才培养所有信号都在告诉我们AI的黄金十年真的来了在行业火爆之下AI人才争夺战也日趋白热化其就业前景一片蓝海我给大家准备了一份全套的《AI大模型零基础入门进阶学习资源包》包括AI大模型入门学习思维导图、精品AI大模型学习书籍手册、视频教程、实战学习等录播视频免费分享出来。有需要的小伙伴可以V扫描下方二维码免费领取人才缺口巨大人力资源社会保障部有关报告显示据测算当前****我国人工智能人才缺口超过500万****供求比例达1∶10。脉脉最新数据也显示AI新发岗位量较去年初暴增29倍超1000家AI企业释放7.2万岗位……单拿今年的秋招来说各互联网大厂释放出来的招聘信息中我们就能感受到AI浪潮比如百度90%的技术岗都与AI相关就业薪资超高在旺盛的市场需求下AI岗位不仅招聘量大薪资待遇更是“一骑绝尘”。企业为抢AI核心人才薪资给的非常慷慨过去一年懂AI的人才普遍涨薪40%脉脉高聘发布的《2025年度人才迁徙报告》显示在2025年1月-10月的高薪岗位Top20排行中AI相关岗位占了绝大多数并且平均薪资月薪都超过6w在去年的秋招中小红书给算法相关岗位的薪资为50k起字节开出228万元的超高年薪据《2025年秋季校园招聘白皮书》AI算法类平均年薪达36.9万遥遥领先其他行业总结来说当前人工智能岗位需求多薪资高前景好。在职场里选对赛道就能赢在起跑线。抓住AI风口轻松实现高薪就业但现实却是仍有很多同学不知道如何抓住AI机遇会遇到很多就业难题比如❌ 技术过时只会CRUD的开发者在AI浪潮中沦为“职场裸奔者”❌ 薪资停滞初级岗位内卷到白菜价传统开发3年经验薪资涨幅不足15%❌ 转型无门想学AI却找不到系统路径83%自学党中途放弃。他们的就业难题解决问题的关键在于不仅要选对赛道更要跟对老师我给大家准备了一份全套的《AI大模型零基础入门进阶学习资源包》包括AI大模型入门学习思维导图、精品AI大模型学习书籍手册、视频教程、实战学习等录播视频免费分享出来。有需要的小伙伴可以V扫描下方二维码免费领取
RAG进化论：掌握LangGraph，构建2026年最智能检索系统！
发布时间：2026/5/22 9:18:40
当RAG学会思考用LangGraph构建Agentic RAG系统完全指南2024年我们还在为能用的RAG系统欢呼2025年RAG已经成为企业AI的标配但到了2026年如果你还在用传统的线性RAG——检索、增强、生成——你可能正在浪费用户的时间。根据2025年的生产环境数据高达30%的RAG响应存在事实性错误而这些错误的根源几乎都指向同一个问题检索质量。这篇文章将带你从传统RAG的困境出发深入理解Agentic RAG的设计哲学并通过LangGraph构建一个真正会思考的检索系统。一、传统RAG的致命缺陷1.1 线性管道的脆弱性传统RAG的工作流程非常直观用户提问 → 向量检索 → 拼接上下文 → LLM生成回答这个流程看起来完美但实际生产中却处处踩坑问题一检索器装聋作哑当用户问Lilian Weng关于reward hacking的类型有什么观点时检索器可能只返回了reward hacking相关的片段却漏掉了types这个关键限定词。结果LLM基于不完整的信息生成了一个看似合理但完全偏离重点的回答。问题二查询歧义无法处理用户问那个产品怎么样——哪个产品传统RAG没有任何机制来澄清或细化查询只能硬着头皮检索然后生成一个模糊的回答。问题三错误无法自纠正当检索到的文档与问题无关时传统RAG没有意识到错误的能力。它只会忠实地把无关内容喂给LLM然后LLM会产生一个自信但错误的回答——这就是所谓的幻觉问题。1.2 为什么简单方案不够用你可能会想“那我们加个re-ranker或者调大top-k”这些都是治标不治本。真正的问题在于传统RAG是被动的而复杂问题需要主动的推理。就像一个只会按图索骥的图书管理员传统RAG从不质疑用户的查询从不反思检索结果的质量也从不主动寻找更多信息。而Agentic RAG就是要让系统具备这种主动性。二、什么是Agentic RAG2.1 从管道到循环范式的根本转变Agentic RAG的核心转变是从线性管道到推理循环。这不是简单的技术升级而是思维方式的根本改变。传统RAG的工作模式就像一个只会执行命令的流水线工人1.收到问题2.去仓库找资料3.把资料交给写手4.写手生成答案问题出在哪里每个环节都是盲执行。工人不会质疑问题是否清晰不会检查找来的资料是否相关更不会在发现问题后主动修正。Agentic RAG的工作模式则像一个有经验的咨询顾问5.先理解你的问题必要时追问澄清6.判断需要查哪些资料去哪里查7.拿到资料后评估是否真的有用8.如果资料不对换个角度再查9.确认信息可靠后才给出专业答案这就是从被动执行到主动推理的质变。2.2 Think-Act-Observe循环Agentic RAG的核心是Think-Act-Observe思考-行动-观察循环这个循环会持续进行直到系统认为已经获得足够信息来回答问题。Think思考LLM分析当前状态决定下一步做什么·这个问题需要查资料吗·用户真正想问的是什么·我现有的信息足够吗Act行动执行具体的操作·调用检索工具·改写查询词·选择合适的数据源Observe观察评估行动结果·检索到的文档相关吗·信息是否完整·需要进一步行动吗2.3 五大核心能力详解一个成熟的Agentic RAG系统具备以下五种核心能力它们共同构成了系统的智能能力1智能查询理解系统能够深入理解用户问题的真实意图而不是简单地匹配关键词。工作原理·意图识别分析问题是问什么、“怎么问”、“为什么问”·实体抽取识别问题中涉及的关键实体人名、产品名、概念等·复杂度评估判断问题是否需要多步推理或简单查询就能解决实际例子用户问“那个产品怎么样”→ 意图识别用户想了解某个产品→ 实体抽取那个没有指明具体产品→ 系统回复“您指的是哪个产品能提供更具体的名称吗”能力2动态检索策略系统不会用固定的检索策略而是根据问题类型动态选择。支持的策略·向量检索基于语义相似度适合概念性问题·关键词检索精确匹配术语适合事实查询·混合检索结合向量和关键词取长补短·多跳检索当结果不足时自动扩展搜索范围能力3自我反思与纠错当检索结果不理想时系统能够意识到问题并主动纠正。纠错机制·相关性评分判断检索到的内容是否真的有用·质量评估检查信息是否过时、是否有偏见·改写决策当文档不相关时决定如何改写查询能力4多工具协作系统不局限于向量数据库可以调用多种外部工具。常用工具·向量数据库存储文档的检索器·Web搜索API搜索实时信息·数据库查询查询结构化数据·计算工具执行数学运算·外部API调用第三方服务能力5记忆与上下文系统记住对话历史在后续交互中保持连贯性。记忆类型·短期记忆当前会话的上下文·长期记忆跨会话的历史信息·工作记忆中间推理步骤的保存2.4 Agentic RAG vs 传统RAG特性传统RAGAgentic RAG执行模式线性管道循环推理检索方式被动检索主动决策错误处理无法纠错自我反思工具支持单一向量库多工具协作幻觉率~30%5%响应延迟低略高什么时候选择Agentic RAG·检索失败时需要自动重试·答案不在知识库中时需要追问·需要多轮交互·对质量要求高2.5 核心架构组件一个完整的Agentic RAG系统通常包含以下组件查询分析器Query Analyzer·判断是否需要检索·识别查询意图和实体·生成检索策略检索器Retriever·向量相似度检索·关键词匹配·混合检索策略查询改写器Query Rewriter·扩展查询词·消除歧义·语义增强文档评分器Document Grader·评估文档相关性·过滤无关内容·生成质量信号回答生成器Answer Generator·基于有效上下文生成回答·标注引用来源·处理不确定情况2.6 为什么选择LangGraph在众多Agent框架中LangGraph有几个独特优势状态管理内置的状态机制让复杂的多步推理变得可控。每一步的状态变化都有迹可循便于调试和优化。可视化可以清晰地看到每一步的决策过程。出了问题一眼就能看出卡在哪一步。持久化支持会话记忆和断点恢复。长对话不会丢失上下文用户体验更连贯。灵活性既可以用预置模板快速开始也可以完全自定义。从简单到复杂平滑过渡。生产就绪支持流式输出、错误重试、并发控制等企业级特性。最重要的是LangGraph的设计哲学与Agentic RAG高度契合——用图来表示状态流转用节点来封装逻辑用边来定义决策。三、用LangGraph构建Agentic RAG3.1 环境准备首先安装必要的依赖[bash]pip install langgraph langchain langchain-openai chromadb你需要准备·OpenAI API Key或其他LLM服务·一个向量数据库本例使用Chroma3.2 定义核心组件Step 1: 创建向量检索器[python]from langchain_openai import OpenAIEmbeddings, ChatOpenAIfrom langchain_community.vectorstores import Chromafrom langchain.text_splitter import RecursiveCharacterTextSplitterfrom langchain_community.document_loaders import TextLoader加载文档loader TextLoader(“your_knowledge_base.txt”)documents loader.load()分割文档text_splitter RecursiveCharacterTextSplitter(chunk_size1000,chunk_overlap200)splits text_splitter.split_documents(documents)创建向量存储vectorstore Chroma.from_documents(documentssplits,embeddingOpenAIEmbeddings())retriever vectorstore.as_retriever(search_kwargs{“k”: 5})Step 2: 定义检索工具[python]from langchain.tools import tooltooldef retrieve_documents(query: str) - str:“”“从知识库中检索相关文档”“”docs retriever.invoke(query)return “\n\n”.join([doc.page_content for doc in docs])retriever_tool retrieve_documentsStep 3: 初始化LLM[python]from langchain.chat_models import init_chat_modelllm init_chat_model(“gpt-4.1”, temperature0)3.3 构建状态图LangGraph的核心是StateGraph我们用它来定义整个工作流[python]from langgraph.graph import StateGraph, MessagesState, START, ENDfrom langgraph.prebuilt import ToolNode, tools_conditionfrom langchain.messages import HumanMessage创建状态图workflow StateGraph(MessagesState)MessagesState 是LangGraph预置的状态类它包含一个messages列表用于存储对话历史。3.4 实现核心节点节点1: 查询生成/响应决策[python]def generate_query_or_respond(state: MessagesState):“”“让LLM决定是否需要检索”“”response llm.bind_tools([retriever_tool]).invoke(state[“messages”])return {“messages”: [response]}节点2: 查询改写[python]REWRITE_PROMPT “”分析用户问题的语义意图生成一个更适合检索的改写版本。原始问题{question}请生成一个改进后的问题要求保留核心意图添加可能的同义词消除歧义改写后的问题“”def rewrite_question(state: MessagesState):“”“改写原始问题以提高检索质量”“”question state[“messages”][0].contentprompt REWRITE_PROMPT.format(questionquestion)response llm.invoke([{“role”: “user”, “content”: prompt}])return {“messages”: [HumanMessage(contentresponse.content)]}节点3: 文档评分[python]from pydantic import BaseModel, Fieldfrom typing import Literalclass GradeDocuments(BaseModel):“”“文档相关性评分”“”binary_score: str Field(description“‘yes’ 如果文档相关‘no’ 如果不相关”)GRADE_PROMPT “”你是一个文档相关性评估器。检索到的文档{context}用户问题{question}判断文档是否包含与问题相关的信息只回答 ‘yes’ 或 ‘no’。“”def grade_documents(state: MessagesState) - Literal[“generate_answer”, “rewrite_question”]:“”“评估检索文档的相关性”“”question state[“messages”][0].contentcontext state[“messages”][-1].contentprompt GRADE\_PROMPT.format(questionquestion, contextcontext) grader llm.with\_structured\_output(GradeDocuments) response grader.invoke([{role: user, content: prompt}]) if response.binary\_score yes: return generate\_answer else: return rewrite\_question节点4: 生成回答[python]GENERATE_PROMPT “”基于以下上下文回答用户问题。如果上下文中没有足够信息请明确说明。上下文{context}问题{question}回答请标注信息来源“”def generate_answer(state: MessagesState):“”“基于检索到的上下文生成回答”“”question state[“messages”][0].contentcontext state[“messages”][-1].contentprompt GENERATE\_PROMPT.format(questionquestion, contextcontext) response llm.invoke([{role: user, content: prompt}]) return {messages: [response]}3.5 组装完整工作流[python]添加节点workflow.add_node(generate_query_or_respond)workflow.add_node(“retrieve”, ToolNode([retriever_tool]))workflow.add_node(rewrite_question)workflow.add_node(generate_answer)设置入口workflow.add_edge(START, “generate_query_or_respond”)条件边决定是否检索workflow.add_conditional_edges(“generate_query_or_respond”,tools_condition, # 自动判断是否调用了工具{“tools”: “retrieve”,END: END # 如果不需要检索直接结束})条件边评估文档质量workflow.add_conditional_edges(“retrieve”,grade_documents # 返回 “generate_answer” 或 “rewrite_question”)回答后结束workflow.add_edge(“generate_answer”, END)改写后重新开始检索workflow.add_edge(“rewrite_question”, “generate_query_or_respond”)编译图graph workflow.compile()3.6 执行和调试[python]执行查询for chunk in graph.stream({“messages”: [{“role”: “user”,“content”: “Lilian Weng关于reward hacking的类型有什么观点”}]}):for node, update in chunk.items():print(f {node} )update[“messages”][-1].pretty_print()print()输出示例 generate_query_or_respond [调用工具: retrieve_documents] retrieve [检索到3篇文档] grade_documents [评分: no - 文档不相关] rewrite_question [改写为: “Lilian Weng reward hacking types classification categorization”] generate_query_or_respond [调用工具: retrieve_documents] retrieve [检索到5篇文档] grade_documents [评分: yes - 文档相关] generate_answer 根据Lilian Weng的文章reward hacking主要有三种类型…四、进阶技巧与最佳实践4.1 多检索器协同复杂场景下你可能需要多个专业检索器[python]tooldef search_product_docs(query: str) - str:“”“搜索产品文档”“”return product_retriever.invoke(query)tooldef search_api_docs(query: str) - str:“”“搜索API文档”“”return api_retriever.invoke(query)tooldef search_troubleshooting(query: str) - str:“”“搜索故障排除指南”“”return troubleshoot_retriever.invoke(query)tools [search_product_docs, search_api_docs, search_troubleshooting]LLM会根据问题自动选择合适的工具。4.2 添加记忆能力使用LangGraph的checkpointer实现会话记忆[python]from langgraph.checkpoint.memory import InMemorySavercheckpointer InMemorySaver()graph workflow.compile(checkpointercheckpointer)执行时指定thread_idconfig {“configurable”: {“thread_id”: “user_123”}}result graph.invoke({“messages”: […]}, config)4.3 流式输出对于长回答使用流式输出提升用户体验[python]async for event in graph.astream({“messages”: […]},stream_mode“values”):if “messages” in event:print(event[“messages”][-1].content, end“”, flushTrue)4.4 错误处理与重试为关键节点添加错误处理[python]from tenacity import retry, stop_after_attempt, wait_exponentialretry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10))def retrieve_with_retry(state: MessagesState):try:return retrieve_documents(state)except Exception as e:print(f检索失败: {e}, 正在重试…)raise4.5 性能优化缓存检索结果[python]from functools import lru_cachelru_cache(maxsize1000)def cached_retrieve(query: str) - str:return retriever.invoke(query)并行检索[python]import asyncioasync def parallel_retrieve(queries: list[str]) - list[str]:tasks [async_retrieve(q) for q in queries]return await asyncio.gather(*tasks)批量处理对于大量文档使用批量嵌入[python]批量嵌入比逐个嵌入快10倍以上embeddings OpenAIEmbeddings()batch_embeddings embeddings.embed_documents([doc.page_content for doc in documents])五、常见问题与避坑指南5.1 检索质量差怎么办症状检索到的文档与问题无关解决方案10.检查分块策略chunk_size是否合适11.尝试混合检索向量关键词12.添加re-ranker重排序13.优化查询改写prompt5.2 响应太慢怎么办症状用户等待时间过长解决方案14.减少循环迭代次数设置最大重试次数15.使用更快的模型如GPT-4o-mini16.实现流式输出17.缓存常见查询5.3 幻觉仍然存在症状LLM生成错误信息解决方案18.强化文档评分逻辑19.在prompt中明确不知道就说不知道20.添加事实核查节点21.使用结构化输出强制引用来源5.4 状态管理混乱症状多轮对话中上下文丢失解决方案22.使用checkpointer持久化状态23.定期总结对话历史24.为每个用户使用独立的thread_idAI行业迎来前所未有的爆发式增长从DeepSeek百万年薪招聘AI研究员到百度、阿里、腾讯等大厂疯狂布局AI Agent再到国家政策大力扶持数字经济和AI人才培养所有信号都在告诉我们AI的黄金十年真的来了在行业火爆之下AI人才争夺战也日趋白热化其就业前景一片蓝海我给大家准备了一份全套的《AI大模型零基础入门进阶学习资源包》包括AI大模型入门学习思维导图、精品AI大模型学习书籍手册、视频教程、实战学习等录播视频免费分享出来。有需要的小伙伴可以V扫描下方二维码免费领取人才缺口巨大人力资源社会保障部有关报告显示据测算当前****我国人工智能人才缺口超过500万****供求比例达1∶10。脉脉最新数据也显示AI新发岗位量较去年初暴增29倍超1000家AI企业释放7.2万岗位……单拿今年的秋招来说各互联网大厂释放出来的招聘信息中我们就能感受到AI浪潮比如百度90%的技术岗都与AI相关就业薪资超高在旺盛的市场需求下AI岗位不仅招聘量大薪资待遇更是“一骑绝尘”。企业为抢AI核心人才薪资给的非常慷慨过去一年懂AI的人才普遍涨薪40%脉脉高聘发布的《2025年度人才迁徙报告》显示在2025年1月-10月的高薪岗位Top20排行中AI相关岗位占了绝大多数并且平均薪资月薪都超过6w在去年的秋招中小红书给算法相关岗位的薪资为50k起字节开出228万元的超高年薪据《2025年秋季校园招聘白皮书》AI算法类平均年薪达36.9万遥遥领先其他行业总结来说当前人工智能岗位需求多薪资高前景好。在职场里选对赛道就能赢在起跑线。抓住AI风口轻松实现高薪就业但现实却是仍有很多同学不知道如何抓住AI机遇会遇到很多就业难题比如❌ 技术过时只会CRUD的开发者在AI浪潮中沦为“职场裸奔者”❌ 薪资停滞初级岗位内卷到白菜价传统开发3年经验薪资涨幅不足15%❌ 转型无门想学AI却找不到系统路径83%自学党中途放弃。他们的就业难题解决问题的关键在于不仅要选对赛道更要跟对老师我给大家准备了一份全套的《AI大模型零基础入门进阶学习资源包》包括AI大模型入门学习思维导图、精品AI大模型学习书籍手册、视频教程、实战学习等录播视频免费分享出来。有需要的小伙伴可以V扫描下方二维码免费领取
