基于LangGraph构建Agentic RAG系统：从金融问答机器人实战理解AI智能体开发

30款热门AI模型一站整合DeepSeek/GLM/Qwen 随心用限时 5 折。 点击领海量免费额度在实际 AI 大模型应用开发面试中面试官考察的远不止你是否知道 Agent、RAG、LangChain 这些名词。他们更关心你能否将这些技术栈有机地组合起来设计出一个能解决真实业务问题的、具备决策和推理能力的智能系统。一个只会调用 API 的开发者和一个能清晰阐述从数据准备、工具定义、流程编排到最终生成答案全链路的工程师在面试中的表现是天壤之别的。本文将以一个“金融大模型问答机器人”项目为蓝本带你从零开始手把手构建一个基于 LangGraph 的 Agentic RAG 系统。通过这个项目你将不仅掌握 LangGraph 的核心概念和 API更能理解如何将 Agent 的决策能力与 RAG 的精准检索相结合打造出超越传统问答机器人的智能体。无论你是准备面试还是希望将 AI 大模型能力落地到具体业务这篇文章都将为你提供一条清晰的实践路径。1. 理解 Agentic RAG从被动检索到主动决策在深入代码之前我们必须先厘清几个核心概念以及它们如何协同工作。这是面试中回答“项目设计思路”问题的关键。1.1 RAG、Agent 与 LangGraph 的角色定位RAG的核心是“检索-增强生成”。它像一个勤奋但被动的图书管理员用户问一个问题它就去知识库向量数据库里查找最相关的文档片段然后把片段和问题一起交给大模型生成答案。它的流程是线性的提问 - 检索 - 生成。这种模式的缺点是如果第一次检索的结果不相关它无法自我修正只能基于错误信息生成可能不准确的答案。Agent的核心是“感知-决策-行动”。它像一个有主观能动性的侦探接收到任务后它会思考“我需要什么信息”然后决定调用哪个“工具”比如搜索、计算、查询数据库去获取信息再根据获取的结果决定下一步是继续调查还是给出结论。Agent 具备推理和决策链的能力。Agentic RAG则是两者的结合。它让 RAG 过程变得“智能”。一个典型的 Agentic RAG 流程可能是理解问题LLM 分析用户问题。决策是否检索LLM 判断是否需要从知识库获取信息来回答。如果问题很通用如“你好”则直接回答如果问题涉及特定知识如“贵公司理财产品A的收益率是多少”则决定调用检索工具。评估检索质量检索到文档后另一个 LLM 或同一个 LLM 会评估这些文档是否真的与问题相关。决策下一步如果文档相关则基于文档生成最终答案如果文档不相关则尝试重写问题然后回到第2步用新问题再次检索。生成最终答案。LangChain是一个用于构建 LLM 应用的框架它提供了大量的组件Models, Prompts, Chains, Agents, Retrievers等像乐高积木一样你可以用它们快速搭建起一个应用。它的Agent模块封装了上述的“思考-行动”循环。LangGraph是 LangChain 的一个子库它专注于用图Graph的方式来描述和运行复杂、有状态的工作流。如果说 LangChain 的 Chain 是线性的管道那么 LangGraph 的 Graph 就是可以包含条件分支、循环、并行节点的流程图。它非常适合用来构建上述那种包含“判断-循环-重试”逻辑的 Agentic RAG 系统。在面试中能说清楚 LangGraph 适用于有状态、多步骤、带条件分支的复杂 Agent 场景是一个重要的加分项。1.2 项目案例金融大模型问答机器人假设你在一次面试中被要求描述一个相关项目你可以这样组织你的回答项目公司某头部互联网金融科技公司。项目职责作为 AI 大模型应用开发工程师我负责设计并实现核心的智能问答引擎以提升客服机器人在金融产品咨询场景下的准确率和可靠性。项目设计痛点传统检索式问答在用户问题模糊或表述不专业时容易检索到不相关的产品文档导致回答错误在金融领域可能引发客诉。方案采用 Agentic RAG 架构。核心是一个由 LangGraph 驱动的智能体其工作流包括问题意图识别 - 智能检索决策 - 检索结果相关性评估 - 问题重写与重试 - 基于可靠上下文的答案生成。数据知识库来源于公司内部的金融产品说明书、合规文档、历史问答对经过清洗、分段和向量化处理。项目实现使用LangChain的文档加载器、文本分割器、OpenAI Embeddings 和 Chroma 向量数据库构建知识库。使用LangGraph定义工作流图Graph节点包括generate_query_or_respond决策节点、retrieve检索工具节点、grade_documents评估节点、rewrite_question重写节点、generate_answer生成节点。使用条件边Conditional Edge来控制流程走向例如根据评估节点的结果决定是跳转到生成答案还是重写问题。使用 Pydantic 模型来约束评估节点的结构化输出确保流程的稳定性。项目业绩上线后在金融产品咨询场景下问答的准确率从传统 RAG 的 78% 提升至 94%无效或错误回答率下降 70%。系统具备了初步的“追问”和“澄清”能力用户体验显著提升。项目采用的技术Python, LangChain, LangGraph, OpenAI GPT-4 API, OpenAI Embeddings, Chroma, FastAPI用于封装服务Docker。接下来我们将抛开抽象描述进入实战环节一步步还原这个系统的核心构建过程。2. 环境准备与核心依赖配置在开始构建之前我们需要一个干净的 Python 环境。强烈建议使用 Conda 或 venv 创建虚拟环境以避免包冲突。2.1 创建虚拟环境与安装依赖# 创建并激活虚拟环境 (以 conda 为例) conda create -n agentic-rag python3.11 conda activate agentic-rag # 安装核心依赖 pip install -U langgraph langchain langchain-openai langchain-text-splitters langchain-community # 安装用于网页抓取的库用于构建示例知识库 pip install beautifulsoup4 requests # 安装向量数据库这里以轻量级的Chroma为例 pip install chromadb关键依赖说明langgraph: 用于构建和运行工作流图。langchain: 提供基础组件如模型调用、提示词模板、文档处理工具。langchain-openai: OpenAI 模型的官方 LangChain 集成。langchain-text-splitters: 文本分割工具用于处理长文档。chromadb: 一个轻量级、嵌入式的向量数据库适合本地开发和测试。2.2 设置 API 密钥本项目需要调用 OpenAI 的 API 来获取 Embeddings 和调用大模型。请将你的 OpenAI API Key 设置为环境变量。# Linux/Mac export OPENAI_API_KEYyour-api-key-here # Windows (PowerShell) $env:OPENAI_API_KEYyour-api-key-here或者在 Python 代码中直接设置import os os.environ[“OPENAI_API_KEY”] “your-api-key-here”注意在生产环境中绝对不要将 API Key 硬编码在代码中。应使用环境变量、密钥管理服务或配置文件来管理。3. 构建知识库RAG 的基石Agentic RAG 的“R”依然依赖于一个高质量的知识库。我们首先模拟一个金融知识库的构建过程。3.1 加载与处理文档假设我们有一些金融产品说明书的文本文件。这里我们以加载本地文件为例。from langchain_community.document_loaders import TextLoader from langchain_text_splitters import RecursiveCharacterTextSplitter # 1. 加载文档 (假设文档在 ./docs 目录下) documents [] import glob for file_path in glob.glob(“./docs/*.txt”): # 加载所有txt文件 loader TextLoader(file_path, encoding“utf-8”) documents.extend(loader.load()) # 如果没有本地文件我们可以创建一些模拟数据 if not documents: print(“未找到本地文档使用模拟金融知识片段...”) from langchain_core.documents import Document documents [ Document( page_content“”” 理财产品A稳健型起投金额1万元预期年化收益率3.5%-4.2%投资周期为6个月。主要投资于高信用等级债券和货币市场工具风险等级PR2中低风险。 “””, metadata{“source”: “product_a.txt”, “product”: “A”} ), Document( page_content“”” 理财产品B成长型起投金额5万元预期年化收益率5.0%-8.0%投资周期为1年。主要投资于混合型基金和优质上市公司股票风险等级PR3中等风险。 “””, metadata{“source”: “product_b.txt”, “product”: “B”} ), Document( page_content“”” 根据监管要求投资者在购买理财产品前需完成风险承受能力评估。评估结果有效期为一年。PR2等级产品适合稳健型及以上投资者PR3等级产品适合平衡型及以上投资者。 “””, metadata{“source”: “regulation.txt”, “type”: “compliance”} ), ] # 2. 分割文档 # 金融文档可能较长需要分割成适合检索的片段。 text_splitter RecursiveCharacterTextSplitter( chunk_size500, # 每个片段大约500字符 chunk_overlap100, # 片段间重叠100字符保持上下文 separators[“\n\n”, “\n”, “。”, “”, “”, “ “, “”] # 中文分隔符 ) doc_splits text_splitter.split_documents(documents) print(f“原始文档数{len(documents)}分割后片段数{len(doc_splits)}”)3.2 创建向量数据库与检索器我们将分割后的文本片段转化为向量并存入向量数据库同时创建一个检索器Retriever。from langchain_openai import OpenAIEmbeddings from langchain_community.vectorstores import Chroma # 1. 初始化 Embedding 模型 embeddings OpenAIEmbeddings(model“text-embedding-3-small”) # 使用较小、成本更低的模型 # 2. 创建向量数据库并持久化可选 # persist_directory 指定持久化目录下次可以直接加载无需重新计算向量 persist_directory “./chroma_finance_db” vectorstore Chroma.from_documents( documentsdoc_splits, embeddingembeddings, persist_directorypersist_directory ) vectorstore.persist() # 保存到磁盘 print(f“向量数据库已创建并保存至 {persist_directory}”) # 3. 从已持久化的数据库加载如果之前已经创建过 # vectorstore Chroma(persist_directorypersist_directory, embedding_functionembeddings) # 4. 创建检索器 # search_kwargs 可以控制返回结果的数量和相似度阈值 retriever vectorstore.as_retriever( search_type“similarity”, # 相似度搜索 search_kwargs{“k”: 3} # 返回最相关的3个片段 ) # 测试检索器 test_query “理财产品A的起投金额是多少” retrieved_docs retriever.invoke(test_query) print(f“查询: ‘{test_query}’“) print(f“检索到 {len(retrieved_docs)} 个相关片段:”) for i, doc in enumerate(retrieved_docs): print(f”[{i1}] {doc.page_content[:150]}... (来源: {doc.metadata.get(‘source’, ‘N/A’)})“)至此一个简单的金融知识库就搭建完成了。它是我们后续智能体进行检索操作的数据来源。4. 定义智能体工具与 LangGraph 状态Agentic RAG 的核心是让 LLM 能够主动使用工具。我们首先需要将上一步的检索器封装成一个“工具”。4.1 创建检索工具在 LangChain/LangGraph 中工具是一个可以被 LLM 调用的函数需要清晰的描述。from langchain.tools import tool from typing import Any tool def retrieve_financial_knowledge(query: str) - str: “”” 根据用户问题从金融产品知识库中检索相关信息。 只有当用户的问题涉及具体的理财产品、投资规则、风险说明、合规要求时才使用此工具。 输入应为清晰、简洁的查询语句。 “”” # 这里直接使用前面创建好的 retriever # 注意在实际项目中retriever 可能需要通过上下文或全局变量传递这里为简化示例。 # 更健壮的做法是使用依赖注入或类封装。 try: docs retriever.invoke(query) if not docs: return “未在知识库中找到相关信息。” # 将检索到的文档内容拼接成一个字符串返回 return “\n\n”.join([f”内容片段 {i1}: {doc.page_content}“ for i, doc in enumerate(docs)]) except Exception as e: return f“检索过程中发生错误{str(e)}” # 将工具放入列表供后续的智能体使用 tools [retrieve_financial_knowledge]4.2 定义 LangGraph 的状态LangGraph 的工作流围绕一个“状态”State对象运行。我们需要定义这个状态包含哪些信息。对于聊天式 Agent最常用的状态是MessagesState它主要包含一个消息列表。from typing import Annotated, List from typing_extensions import TypedDict from langgraph.graph.message import add_messages from langchain_core.messages import BaseMessage # 定义状态结构 class AgentState(TypedDict): “””定义智能体工作流的状态。“”” # 核心消息列表记录用户输入、AI回复、工具调用及结果 messages: Annotated[List[BaseMessage], add_messages] # 你可以在这里添加其他状态例如 # turn_count: int # 记录对话轮次防止无限循环 # retrieved_context: str # 专门存储检索到的上下文Annotated[List[BaseMessage], add_messages]是一个高级用法。它告诉 LangGraphmessages字段是一个列表当有新的消息需要添加时应该使用add_messages函数来合并而不是直接覆盖。这简化了状态更新逻辑。5. 构建 LangGraph 工作流节点现在我们来构建工作流中的各个功能节点Node。每个节点都是一个函数它接收当前状态执行一些操作并返回更新后的状态。5.1 决策节点判断是否需要检索这是工作流的入口和核心决策点。LLM 会分析当前对话历史主要是最新的用户问题决定是直接回答还是调用检索工具。from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, AIMessage, ToolMessage # 初始化聊天模型 llm ChatOpenAI(model“gpt-4o-mini”, temperature0) # temperature0 使输出更确定 def should_retrieve_node(state: AgentState) - AgentState: “”” 决策节点分析用户问题决定是否需要调用检索工具。 如果问题涉及具体金融知识则调用工具如果是问候或通用问题则直接生成回复。 “”” # 1. 获取最新的用户消息 last_message state[“messages”][-1] if not isinstance(last_message, HumanMessage): # 如果不是用户消息理论上不应该进入此节点这里直接返回 return state # 2. 将工具绑定给模型并调用模型 # bind_tools 让模型知道它可以调用哪些工具 llm_with_tools llm.bind_tools(tools) # 调用模型传入当前所有消息历史 ai_message llm_with_tools.invoke(state[“messages”]) # 3. 将模型的回复可能包含工具调用添加到状态中 # LangGraph 的 add_messages 注解会自动处理列表合并 return {“messages”: [ai_message]} # 测试决策节点 test_state { “messages”: [HumanMessage(content“你好”)] } result should_retrieve_node(test_state) last_msg result[“messages”][-1] print(“测试1 - 通用问候”) print(f”AI 回复: {last_msg.content}“) print(f”是否调用了工具 {hasattr(last_msg, ‘tool_calls’) and last_msg.tool_calls}“) print(“-” * 50) test_state_2 { “messages”: [HumanMessage(content“请介绍一下理财产品A的风险等级和收益率。”)] } result2 should_retrieve_node(test_state_2) last_msg2 result2[“messages”][-1] print(“测试2 - 具体金融问题”) print(f”AI 回复内容: {last_msg2.content}“) print(f”是否调用了工具 {hasattr(last_msg2, ‘tool_calls’) and last_msg2.tool_calls}“) if hasattr(last_msg2, ‘tool_calls’) and last_msg2.tool_calls: for tc in last_msg2.tool_calls: print(f” 工具调用: {tc[‘name’]}参数: {tc[‘args’]}“)运行测试你会看到对于“你好”模型直接生成了回复对于具体金融问题模型生成了一个tool_calls对象指示需要调用retrieve_financial_knowledge工具。5.2 工具执行节点当决策节点决定调用工具后我们需要一个节点来实际执行这些工具调用。LangGraph 提供了ToolNode来简化这个过程。from langgraph.prebuilt import ToolNode # 创建工具执行节点 tool_node ToolNode(tools) # 这个节点会自动识别状态中最后一个消息AIMessage里的 tool_calls # 然后依次执行对应的工具函数并将结果封装成 ToolMessage 添加到状态中。5.3 评估节点判断检索结果是否相关这是实现“智能”的关键。我们不能盲目相信检索到的内容需要让另一个 LLM或同一个来评估检索到的文档是否真的能回答用户的问题。from pydantic import BaseModel, Field # 1. 定义一个 Pydantic 模型来约束 LLM 的输出格式 class GradeDocuments(BaseModel): “””文档相关性评估结果。“”” binary_score: str Field( description“相关性评分‘yes’ 表示相关‘no’ 表示不相关”, enum[“yes”, “no”] ) reasoning: str Field(description“做出该评分的简要理由”) # 2. 初始化一个用于评估的模型可以与主模型相同 grader_llm ChatOpenAI(model“gpt-4o-mini”, temperature0).with_structured_output(GradeDocuments) def grade_documents_node(state: AgentState) - dict: “”” 评估节点判断工具返回的检索结果是否与原始问题相关。 返回一个字典指示下一步应该去哪个节点。 “”” # 获取原始用户问题和工具返回的结果 user_question state[“messages”][0].content # 假设第一个消息是用户问题 # 最后一个消息应该是 ToolMessage包含检索结果 last_message state[“messages”][-1] if not isinstance(last_message, ToolMessage): # 如果不是 ToolMessage默认视为不相关进入重写环节 return {“next”: “rewrite_question”} retrieved_context last_message.content # 构建评估提示词 grade_prompt f“”” 你是一个严格的评估员。请评估以下检索到的文档内容是否足以回答用户的问题。 仅根据提供的内容进行评估不要使用外部知识。 用户问题{user_question} 检索到的内容 “““ {retrieved_context} ”““ 请按以下格式输出 - 如果内容明确包含与问题直接相关的信息或者能基于内容进行合理推断来回答问题则评分为 ‘yes’。 - 如果内容与问题无关、信息不足或模糊不清则评分为 ‘no’。 同时请提供一句简短的推理。 “”” # 调用评估模型 try: grade_result: GradeDocuments grader_llm.invoke(grade_prompt) print(f“评估结果{grade_result.binary_score}理由{grade_result.reasoning}”) if grade_result.binary_score “yes”: return {“next”: “generate_answer”} else: return {“next”: “rewrite_question”} except Exception as e: print(f“评估过程出错{e}默认进入重写环节”) return {“next”: “rewrite_question”}5.4 问题重写节点如果评估认为检索结果不相关可能是用户问题表述不清或检索词不佳。这个节点会尝试重写问题以便进行下一轮检索。def rewrite_question_node(state: AgentState) - AgentState: “”” 重写问题节点基于原始问题和失败的检索生成一个更清晰、更可能检索到相关信息的新问题。 “”” user_question state[“messages”][0].content # 可以加入一些逻辑来防止无限重写例如检查重写次数需要扩展状态 # current_turn state.get(“turn_count”, 0) rewrite_prompt f“”” 原始的用户问题是“{user_question}” 根据这个问题的第一次检索未能找到足够相关的信息。 你的任务是将这个问题重写成一个更清晰、更具体、更可能从金融产品知识库中检索到答案的形式。 重写时请考虑 1. 使用更标准的产品名称或术语如“理财产品A”。 2. 将模糊的描述具体化如“收益高” - “年化收益率”。 3. 如果问题包含多个子问题尝试拆解或聚焦于核心一点。 只输出重写后的问题不要输出任何解释。 “”” rewritten_question llm.invoke(rewrite_prompt).content print(f“问题重写 ‘{user_question}’ - ‘{rewritten_question}’”) # 将重写后的问题作为新的人类消息触发新一轮的决策流程 return {“messages”: [HumanMessage(contentrewritten_question)]}5.5 答案生成节点如果评估认为检索结果相关则进入此节点基于问题和检索到的上下文生成最终答案。def generate_answer_node(state: AgentState) - AgentState: “”” 答案生成节点基于原始问题和相关的检索上下文生成最终答案。 “”” user_question state[“messages”][0].content # 找到最后一个 ToolMessage即相关的检索结果 context “” for msg in reversed(state[“messages”]): if isinstance(msg, ToolMessage): context msg.content break if not context: # 如果没有找到上下文直接基于问题生成一个保守答案 answer llm.invoke(f“用户问题{user_question}\n\n我无法从知识库中找到相关信息请根据你的通用知识谨慎回答。”).content return {“messages”: [AIMessage(contentanswer)]} answer_prompt f“”” 你是一个专业的金融客服助手。请严格根据以下提供的上下文信息来回答用户的问题。 如果上下文信息不足以完全回答问题请明确指出哪些信息是已知的哪些是未知的。 不要编造上下文以外的信息保持回答简洁、准确。 上下文信息 “““ {context} ”““ 用户问题{user_question} 请生成最终答案 “”” final_answer llm.invoke(answer_prompt).content return {“messages”: [AIMessage(contentfinal_answer)]}6. 组装与运行 LangGraph 工作流有了所有节点现在我们需要用 LangGraph 的图GraphAPI 将它们连接起来定义完整的流程。6.1 定义图结构from langgraph.graph import StateGraph, END # 1. 创建图并指定状态类型 workflow StateGraph(AgentState) # 2. 添加节点 workflow.add_node(“should_retrieve”, should_retrieve_node) # 决策节点 workflow.add_node(“retrieve”, tool_node) # 工具执行节点 workflow.add_node(“grade_documents”, grade_documents_node) # 评估节点 workflow.add_node(“rewrite_question”, rewrite_question_node) # 重写节点 workflow.add_node(“generate_answer”, generate_answer_node) # 答案生成节点 # 3. 设置入口点 workflow.set_entry_point(“should_retrieve”) # 4. 添加条件边和普通边 # 4.1 从决策节点出发判断是否调用了工具 def route_after_decision(state: AgentState) - str: last_message state[“messages”][-1] if hasattr(last_message, “tool_calls”) and last_message.tool_calls: return “retrieve” # 调用了工具去执行工具 else: return END # 没调用工具直接结束模型已生成回复 workflow.add_conditional_edges( “should_retrieve”, route_after_decision, { “retrieve”: “retrieve”, END: END } ) # 4.2 从工具执行节点出发必须去评估检索结果 workflow.add_edge(“retrieve”, “grade_documents”) # 4.3 从评估节点出发根据评估结果路由 # grade_documents_node 返回的是 {“next”: “generate_answer”} 或 {“next”: “rewrite_question”} workflow.add_conditional_edges( “grade_documents”, # 这个 lambda 函数从状态中提取评估节点返回的 “next” 值 # 注意这里需要根据 grade_documents_node 的返回值来设计。 # 一种更清晰的方式是让 grade_documents_node 直接返回下一个节点的名称字符串。 # 我们修改一下 grade_documents_node 的返回值为字符串。 lambda state: state.get(“next”, “rewrite_question”), # 假设我们将结果存在 state[“next”] 中这需要修改节点逻辑。 # 为了简化我们采用另一种常见模式让评估节点返回一个路由字典并在条件边函数中解析。 # 下面我们重构 grade_documents_node 并调整这里的逻辑。 ) # 由于条件边函数需要基于状态返回下一个节点名我们调整评估节点的逻辑 # 我们不修改状态而是让评估节点返回一个结果并由条件边函数处理。 # 让我们重新定义评估节点和路由逻辑。 print(“正在构建图...此部分代码需要根据路由逻辑调整下文将给出完整示例”)6.2 完整的工作流组装与运行示例为了让路由逻辑更清晰我们稍微调整一下设计。评估节点 (grade_documents_node) 不再修改状态而是将判断逻辑移到条件边函数中。下面是一个更完整的、可运行的组装示例from langgraph.graph import StateGraph, END from langchain_core.messages import HumanMessage, AIMessage import json # --- 重新定义评估节点简化版只评估并返回结果 --- def grade_documents_node_simple(state: AgentState) - AgentState: “””评估节点将评估结果‘relevant’ 或 ‘irrelevant’存入状态。“”” user_question state[“messages”][0].content last_message state[“messages”][-1] if not isinstance(last_message, ToolMessage): return {“evaluation”: “irrelevant”} retrieved_context last_message.content # 简化的评估逻辑实际应用中应使用更可靠的LLM评估 grade_prompt f“问题‘{user_question}’\n上下文‘{retrieved_context[:300]}...’\n上下文是否直接回答了问题只回答‘是’或‘否’。” response llm.invoke(grade_prompt).content.lower() is_relevant “是” in response or “yes” in response or “relevant” in response evaluation “relevant” if is_relevant else “irrelevant” print(f“文档评估{evaluation}”) return {“evaluation”: evaluation} # 将评估结果存入状态 # --- 重新组装图 --- workflow StateGraph(AgentState) workflow.add_node(“should_retrieve”, should_retrieve_node) workflow.add_node(“retrieve”, tool_node) workflow.add_node(“grade_documents”, grade_documents_node_simple) # 使用新版评估节点 workflow.add_node(“rewrite_question”, rewrite_question_node) workflow.add_node(“generate_answer”, generate_answer_node) workflow.set_entry_point(“should_retrieve”) # 边1决策后路由 def route_after_decision(state: AgentState) - str: last_message state[“messages”][-1] if hasattr(last_message, “tool_calls”) and last_message.tool_calls: return “retrieve” else: return END workflow.add_conditional_edges(“should_retrieve”, route_after_decision, {“retrieve”: “retrieve”, END: END}) # 边2工具执行后必然进入评估 workflow.add_edge(“retrieve”, “grade_documents”) # 边3根据评估结果路由 def route_after_grade(state: AgentState) - str: # 从状态中读取评估结果 eval_result state.get(“evaluation”, “irrelevant”) if eval_result “relevant”: return “generate_answer” else: return “rewrite_question” workflow.add_conditional_edges(“grade_documents”, route_after_grade, {“generate_answer”: “generate_answer”, “rewrite_question”: “rewrite_question”}) # 边4重写问题后应回到决策节点用新问题重新开始流程 workflow.add_edge(“rewrite_question”, “should_retrieve”) # 边5生成答案后流程结束 workflow.add_edge(“generate_answer”, END) # 编译图 graph workflow.compile() print(“LangGraph 工作流编译成功”)6.3 运行智能体并分析结果现在我们可以用这个图来处理用户问题了。from langchain_core.messages import HumanMessage def run_agent(query: str): “””运行智能体处理单个查询。“”” print(f“\n{*60}”) print(f“用户问题{query}”) print(f“{‘’*60}”) # 初始化状态 initial_state {“messages”: [HumanMessage(contentquery)]} # 运行图 final_state graph.invoke(initial_state) # 获取最终答案 final_messages final_state[“messages”] for msg in final_messages: if isinstance(msg, AIMessage) and not hasattr(msg, ‘tool_calls’): print(f“\n最终答案\n{msg.content}”) break else: print(“\n未生成最终答案。”) # 打印完整交互历史可选 print(f“\n{‘-’*40} 完整消息流 {-’*40}”) for i, msg in enumerate(final_messages): print(f”[{i}] {type(msg).__name__}: {msg.content[:200] if msg.content else ‘N/A’}“) if hasattr(msg, ‘tool_calls’): print(f” 工具调用: {msg.tool_calls}“) # 测试案例 run_agent(“理财产品A的风险等级是多少”) # 应能直接检索并回答 run_agent(“有什么高收益的产品推荐吗”) # 问题较模糊可能触发重写 run_agent(“你好”) # 通用问候应直接回复不检索运行上述代码你将看到智能体完整的决策流程对于明确的问题它检索、评估、生成答案对于模糊的问题它可能重写后再检索对于问候它直接回复。7. 常见问题排查与优化策略构建这样一个系统时你会遇到各种问题。以下是典型问题及其排查思路。7.1 问题工具从未被调用现象无论问什么智能体都直接回答不进行检索。排查检查工具描述tool装饰器内的函数文档字符串docstring是 LLM 决定是否调用该工具的主要依据。确保描述清晰说明了工具的用途和调用时机。例如“当用户询问具体的理财产品信息、规则或数据时使用此工具”。检查模型绑定在should_retrieve_node中是否使用了llm.bind_tools(tools)只有绑定了工具模型才知道它们的存在。检查提示词/系统消息如果你为模型设置了系统消息确保它没有过度限制模型调用工具例如系统消息说“你是一个直接回答问题的助手”。测试决策节点单独运行should_retrieve_node函数打印出模型的响应查看tool_calls属性是否存在。7.2 问题检索结果总是不相关导致无限重写循环现象智能体陷入“检索 - 评估为不相关 - 重写 - 再检索 - 仍不相关”的死循环。排查与优化评估逻辑过于严格检查grade_documents_node中的提示词。它可能把“部分相关”或“需要推理”的内容误判为“不相关”。可以调整提示词例如“如果内容为回答问题提供了任何有用的线索或信息则评分为‘yes’”。向量检索效果差Embedding 模型尝试不同的 Embedding 模型如text-embedding-3-large。检索参数调整retriever的search_kwargs例如增加k返回更多结果或尝试search_type“mmr”最大边际相关性来平衡相关性和多样性。文本分割检查RecursiveCharacterTextSplitter的chunk_size和chunk_overlap。块太大可能包含无关信息太小可能丢失关键上下文。对于金融文档chunk_size300-500可能比较合适。知识库数据质量确保原始文档清洗干净没有无关的页眉页脚、广告等。金融文档应聚焦于产品条款、规则说明等核心内容。设置循环上限在状态中增加一个计数器如retry_count在rewrite_question_node中检查如果超过一定次数如3次则强制跳转到generate_answer节点并给出“无法找到确切信息”的回复。7.3 问题答案生成时胡编乱造幻觉现象即使提供了正确的上下文模型生成的答案仍包含知识库中不存在的信息。排查与优化强化提示词约束在generate_answer_node的提示词中使用更强烈的指令如“必须严格依据提供的上下文生成答案”“禁止添加任何上下文以外的信息”“如果上下文没有提到请明确回答‘根据提供的信息无法确定’”。使用支持“引用”的模型一些更新的模型如 GPT-4o在 API 调用时可以指定response_format{ “type”: “json_object” }并要求模型以结构化格式输出答案和引用来源的片段。后处理验证可以添加一个“答案验证”节点将生成的答案与检索到的上下文再次对比检查一致性。7.4 关键参数配置表组件参数建议值/选项说明文本分割器chunk_size300-500 (字符)取决于文档密度。金融条款文档可稍小长篇文章可稍大。chunk_overlap50-100 (字符)保持块间上下文连贯防止关键信息被割裂。向量检索器search_type“similarity”或“mmr”similarity纯按相似度排序mmr在相关性和多样性间平衡。k3-5返回的文档片段数量。太少可能信息不足太多可能引入噪声。大模型 (Chat)modelgpt-4o-mini,gpt-4omini成本低、速度快适合决策和评估4o能力更强适合最终答案生成。temperature0 (决策/评估) 0.1-0.3 (生成)决策和评估需要确定性生成答案可稍有创造性但需控制。工作流重试次数限制2-3 次在状态中维护计数器防止模糊问题导致无限循环。8. 生产环境最佳实践与扩展方向将上述原型部署到生产环境还需要考虑更多工程化因素。8.1 生产环境考量配置管理将 API Keys、模型名称、温度参数、向量数据库连接信息等抽取到配置文件如config.yaml或环境变量中。错误处理与重试在网络调用、模型 API 调用、数据库操作等环节添加完善的try-except和重试机制如使用tenacity库。日志与监控集成日志系统如structlog记录每个节点的输入、输出、耗时和错误。使用 LangSmith 可以可视化跟踪整个 Graph 的执行过程是调试和监控的利器。性能优化缓存对 Embedding 结果、频繁的相似查询结果进行缓存。异步如果节点间没有严格顺序依赖可以考虑使用 LangGraph 的异步支持或asyncio提升吞吐量。安全性输入过滤对用户输入进行清洗防止 Prompt 注入攻击。输出审查对模型生成的答案进行合规性审查尤其是金融场景可以添加一个审查节点。权限控制确保知识库检索范围受用户权限约束。8.2 项目扩展方向多工具 Agent除了检索工具可以集成计算器计算收益、数据库查询工具查询用户持仓、外部 API 调用工具获取实时行情等让智能体能力更强。记忆与多轮对话当前是单轮问答。可以引入langgraph.checkpoint来实现对话记忆让智能体能够引用之前的对话上下文。流式输出使用graph.stream()接口实现答案的逐词或逐句流式返回提升用户体验。更复杂的评估体系不仅评估相关性还可以评估答案的完整性、准确性、与内部知识的一致性等形成多级质检流程。与业务系统集成将整个 LangGraph 工作流封装成 gRPC 或 HTTP 服务如使用 FastAPI供前端或业务系统调用。构建一个成熟的 Agentic RAG 系统是一个迭代过程。从本文的最小可行产品开始逐步加入错误处理、监控、缓存和更复杂的业务逻辑你就能搭建起一个真正智能、可靠且可用于生产的金融问答引擎。理解每个组件的作用和它们之间的数据流是你在面试中展现技术深度的关键也是在实际工作中解决复杂问题的基础。 30款热门AI模型一站整合DeepSeek/GLM/Qwen 随心用限时 5 折。 点击领海量免费额度