向量搜索、图遍历还是网络搜索 – 本文介绍如何用 LangGraph 让智能体为每个问题选择合适的工具。语义搜索、向量数据库迁移、Graph RAG。这些系统有一个共同缺陷——都是 Pipeline固定执行序列问题进、答案出不论输入是什么。真实问题并不符合这个假设。“我们的退款政策是什么”需要向量搜索。“顶级客户和这家供应商之间有什么关联”需要图遍历。“美联储今早宣布了什么”需要网络搜索。Pipeline 只能处理其中一种。Agent 三种都能做而且能自己决定用哪种。思维转变传统 RAG 是一个函数——输入 → 输出。Agentic RAG 是一个状态机——它有记忆、能做决策、可以回溯只有在对答案有把握时才会终止。什么让它变得AgenticAgent 与 Pipeline 的区别体现在四种能力上路由实际效果演示“公司的数据留存政策是什么”——Vector向量搜索“排名前 3 的客户与哪些供应商有关联”——Graph图搜索“美联储今天关于利率的最新公告是什么”——Web网络搜索“340 的 15% 是多少”——DirectLLM直接调用 LLMLangGraph 状态机LangGraph 将 Agent 建模为有向图节点是函数边是决策。与 LangChain 的关键差异在于边可以形成循环。Agent 能不断重试、改写、自我纠错直到生成一个有把握的答案。Router入口节点Router 是 Agent 的决策中枢。它接收原始查询通过结构化 LLM 调用with_structured_output将其分为四类vector、graph、web 或 direct。分类结果在 LangGraph 中成为一条条件边决定接下来运行哪个检索节点。每次查询、每次改写后都会经过 Router。RetrieverVector / Graph / Web三个检索节点占各自槽位VectorFAISS/pgvector 余弦相似度、GraphNeo4j 自动生成 Cypher、WebTavily API。每个节点返回一组 Document 对象路由决策决定哪个节点被激活。在改写循环中Router 可能切换到不同的检索器。Grader相关性评估Grader 逐一评估每个检索到的文档相关或不相关每份文档各触发一次结构化 LLM 调用。判断逻辑如下至少 1 个文档相关 → 路由至 Generator全部不相关且rewrite_count 3→ 路由至 Rewriter全部不相关且已改写 3 次 → 路由至 web 兜底。这是整个自我纠错机制的核心。Rewriter检索失败时的查询修复检索失败后 Rewriter 启动接收原始查询或之前的改写版本要求 LLM 重新表述——更具体的词语、不同的同义词、更清晰的范围。每次改写后rewrite_count递增上限 3 次防止无限循环。改写完成后重新路由回 Router。GeneratorLLM 生成答案Generator 只接收被评为相关的文档用严格的 Prompt 综合生成有依据的答案仅根据以下上下文作答。答案和来源文档存入 Agent 状态供后续幻觉检测使用不相关的文档完全隔离在外。Hallucination Checker最终验证Hallucination Checker 拿到生成的答案和来源文档判断答案中的每一个论断是否都有上下文支撑。如果是 → 返回答案END如果否 → 以更严格的 Prompt 路由回 Generator 重新生成。这个节点专门拦截那些表面自信、实则错误的输出。完整架构——节点如何连接Router 节点对查询进行分类LLM 将输入问题分类为四类之一vector_search文档事实查询、graph_search关系查询、web_search实时信息或directLLM 已知的内容。这是 LangGraph 中的条件边输出结果决定下一个运行的节点。Retriever 节点执行对应工具三个工具作为 LangGraph 节点接入FAISS/pgvector 用于语义相似度Neo4j Cypher 用于图遍历Tavily/SerpAPI 用于实时网络结果。每个节点返回带有相关性元数据的文档列表。Grader 节点文档是否真正相关LLM 对每个检索到的文档打分相关或不相关。若所有文档均不通过边路由至 Rewriter若足够多的文档通过则路由至 Generator。Rewriter 节点修复查询重新尝试检索失败时LLM 将原始查询改写得更具体或切换检索策略然后循环回第 2 步。最多重试 3 次之后回退到网络搜索。Generator 节点综合生成答案LLM 基于检索到的文档生成有依据的答案答案和来源文档存入 Agent 状态供幻觉检测使用。Hallucination Checker最终关卡LLM 验证答案中的每个论断是否都有检索文档支撑如果是 → 返回答案如果否 → 以更严格的 Prompt 循环回 Generator。这一机制消除了自信却错误的答案。代码实现——LangGraph 混合 RAG Agent第 1 步——定义 Agent 状态from typing import TypedDict, List, Literal from langchain_core.documents import Document from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain_community.vectorstores import FAISS from langchain_openai import OpenAIEmbeddings from langchain_community.tools.tavily_search import TavilySearchResults from langchain_community.graphs import Neo4jGraph from pydantic import BaseModel llm ChatOpenAI(modelgpt-4o-mini, temperature0) # ── Agent 状态——在所有节点间持久保存 ──────────── class AgentState(TypedDict): question: str route: str # vector | graph | web | direct documents: List[Document] generation: str rewrite_count: int grade_results: List[str] # ── 工具初始化 ────────────────────────────────────────────── embeddings OpenAIEmbeddings() vector_store FAISS.load_local(faiss_index, embeddings) vector_retriever vector_store.as_retriever(search_kwargs{k: 4}) neo4j_graph Neo4jGraph(urlbolt://localhost:7687, usernameneo4j, passwordpassword) web_search TavilySearchResults(max_results3)第 2 步——Router 节点class RouteDecision(BaseModel): route: Literal[vector, graph, web, direct] reasoning: str router_llm llm.with_structured_output(RouteDecision) ROUTER_PROMPT You are a query router for a hybrid RAG system. Classify the query into ONE category: - vector: factual questions answerable from internal documents (policies, reports, product info, static knowledge) - graph: questions about relationships between entities (how X connects to Y, who knows whom, supplier chains) - web: requires real-time or current information (news, stock prices, todays events, recent releases) - direct: simple calculations or general knowledge the LLM already knows Query: {question} def router_node(state: AgentState) - AgentState: decision router_llm.invoke( ROUTER_PROMPT.format(questionstate[question]) ) print(f→ Router: {decision.route} | {decision.reasoning}) return {**state, route: decision.route} # 条件边——路由到对应的检索节点 def route_edge(state: AgentState) - str: return state[route] # vector | graph | web | direct第 3 步——三个 Retriever 节点# ── Vector 检索器 ──────────────────────────────────── def vector_node(state: AgentState) - AgentState: docs vector_retriever.invoke(state[question]) return {**state, documents: docs} # ── Graph 检索器Neo4jLLM 自动生成 Cypher────────────── from langchain.chains import GraphCypherQAChain graph_chain GraphCypherQAChain.from_llm( llmllm, graphneo4j_graph, verboseTrue, allow_dangerous_requestsTrue ) def graph_node(state: AgentState) - AgentState: result graph_chain.invoke(state[question]) doc Document(page_contentresult[result], metadata{source: neo4j_graph}) return {**state, documents: [doc]} # ── Web 搜索检索器 ───────────────────────────────── def web_node(state: AgentState) - AgentState: results web_search.invoke(state[question]) docs [Document(page_contentr[content], metadata{source: r[url]}) for r in results] return {**state, documents: docs} # ── 直接调用 LLM无需检索──────────────────────────────── def direct_node(state: AgentState) - AgentState: answer llm.invoke(state[question]).content return {**state, generation: answer, documents: []}第 4 步——Grader、Rewriter、Generator 和 Hallucination Checker# ── Grader ─────────────────────────────────────────────── class GradeDoc(BaseModel): score: Literal[relevant, irrelevant] grader_llm llm.with_structured_output(GradeDoc) def grader_node(state: AgentState) - AgentState: grades [] for doc in state[documents]: result grader_llm.invoke( fQuestion: {state[question]}\nDocument: {doc.page_content[:400]}\n Is this document relevant to answering the question? Score: relevant/irrelevant ) grades.append(result.score) return {**state, grade_results: grades} def grade_edge(state: AgentState) - str: relevant sum(1 for g in state[grade_results] if g relevant) if relevant 0: return generate elif state[rewrite_count] 3: return rewrite else: return web_fallback # 三次失败后的最终兜底 # ── Rewriter ───────────────────────────────────────────── def rewriter_node(state: AgentState) - AgentState: rewritten llm.invoke( fThe query {state[question]} returned no relevant results. Rewrite it to be more specific and searchable. Return only the rewritten query. ).content print(f→ Rewriter: {state[question]} → {rewritten}) return {**state, question: rewritten, rewrite_count: state[rewrite_count] 1} # ── Generator ──────────────────────────────────────────── def generator_node(state: AgentState) - AgentState: context \n\n.join(d.page_content for d in state[documents] if relevant in state.get(grade_results,[])) answer llm.invoke( fAnswer using only the context below.\n\nContext:\n{context}\n\nQuestion: {state[question]} ).content return {**state, generation: answer} # ── Hallucination Checker ──────────────────────────────── class HallucinationCheck(BaseModel): grounded: Literal[yes, no] halluc_llm llm.with_structured_output(HallucinationCheck) def hallucination_node(state: AgentState) - AgentState: context \n\n.join(d.page_content for d in state[documents]) result halluc_llm.invoke( fContext:\n{context}\n\nAnswer:\n{state[generation]}\n\n Is the answer fully supported by the context? grounded: yes/no ) return {**state, hallucination_check: result.grounded} def halluc_edge(state: AgentState) - str: return end if state.get(hallucination_check) yes else regenerate第 5 步——连接图结构并运行# ── 构建 LangGraph 状态机 ──────────────────────────────── workflow StateGraph(AgentState) # 添加节点 workflow.add_node(router, router_node) workflow.add_node(vector, vector_node) workflow.add_node(graph, graph_node) workflow.add_node(web, web_node) workflow.add_node(direct, direct_node) workflow.add_node(grader, grader_node) workflow.add_node(rewriter, rewriter_node) workflow.add_node(generator, generator_node) workflow.add_node(hallucination,hallucination_node) # 入口节点 workflow.set_entry_point(router) # router → 检索器条件边 workflow.add_conditional_edges(router, route_edge, { vector: vector, graph: graph, web: web, direct: direct, }) # 检索器 → grader for node in [vector, graph, web]: workflow.add_edge(node, grader) # grader → 生成 / 改写 / web 兜底 workflow.add_conditional_edges(grader, grade_edge, { generate: generator, rewrite: rewriter, web_fallback: web, }) # rewriter → 回到 router workflow.add_edge(rewriter, router) # generator → hallucination 检测 workflow.add_edge(generator, hallucination) # hallucination 检测 → 结束或重新生成 workflow.add_conditional_edges(hallucination, halluc_edge, { end: END, regenerate: generator, }) workflow.add_edge(direct, END) # 编译并运行 agent workflow.compile() result agent.invoke({ question: What is our data retention policy for EU customers?, rewrite_count: 0, documents: [], grade_results: [], generation: , route: , }) print(result[generation])总结通过引入 LangGraph 构建状态机我们将传统的“单向直通车”式 RAG 成功升级为了具备记忆、决策和自我纠错能力的智能体系统。这种混合 Agent 架构彻底打破了单一 Pipeline 的局限它不仅能通过 Router 扮演决策中枢针对不同问题智能分发最匹配的检索工具向量相似度、图谱关联、网络实时数据或直接调用大模型更重要的是它建立了一套严密的自校验闭环。借助 Grader 的相关性过滤和 Rewriter 的查询修复系统在面对检索失败时可以主动寻找替代方案而最后的 Hallucination Checker 拦截一切缺乏上下文支撑的“自信妄想”。这种架构让 AI 像人一样在不确定中不断试错、交叉验证直到确信无疑才交付结果。学AI大模型的正确顺序千万不要搞错了2026年AI风口已来各行各业的AI渗透肉眼可见超多公司要么转型做AI相关产品要么高薪挖AI技术人才机遇直接摆在眼前有往AI方向发展或者本身有后端编程基础的朋友直接冲AI大模型应用开发转岗超合适就算暂时不打算转岗了解大模型、RAG、Prompt、Agent这些热门概念能上手做简单项目也绝对是求职加分王给大家整理了超全最新的AI大模型应用开发学习清单和资料手把手帮你快速入门学习路线:✅大模型基础认知—大模型核心原理、发展历程、主流模型GPT、文心一言等特点解析✅核心技术模块—RAG检索增强生成、Prompt工程实战、Agent智能体开发逻辑✅开发基础能力—Python进阶、API接口调用、大模型开发框架LangChain等实操✅应用场景开发—智能问答系统、企业知识库、AIGC内容生成工具、行业定制化大模型应用✅项目落地流程—需求拆解、技术选型、模型调优、测试上线、运维迭代✅面试求职冲刺—岗位JD解析、简历AI项目包装、高频面试题汇总、模拟面经以上6大模块看似清晰好上手实则每个部分都有扎实的核心内容需要吃透我把大模型的学习全流程已经整理好了抓住AI时代风口轻松解锁职业新可能希望大家都能把握机遇实现薪资/职业跃迁这份完整版的大模型 AI 学习资料已经上传CSDN朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】