基于Agentic RAG与LangGraph的金融智能问答系统实战
30款热门AI模型一站整合DeepSeek/GLM/Qwen 随心用限时 5 折。 点击领海量免费额度如果你正在准备 AI 大模型应用开发工程师的面试或者想从零开始构建一个能真正落地的智能问答系统那么你很可能正被这几个问题困扰面试官问“请设计一个基于大模型的金融问答机器人”你该如何从零到一拆解项目并清晰地讲出技术选型、架构设计和核心难点Agent、RAG、LangChain、LangGraph 这些概念听起来很酷但它们之间到底是什么关系在实际项目中我到底该用哪个又该怎么组合网上教程要么太浅只讲概念要么太散代码片段不完整。有没有一个完整的、可运行的、能体现工程化思维的实战项目让我能真正理解并复现这篇文章就是为你准备的。我不会只给你罗列概念而是通过一个完整的“金融大模型问答机器人”项目案例带你从零开始手把手构建一个融合了Agent智能体、RAG检索增强生成和LangGraph工作流编排的复杂应用。你将看到如何将看似独立的“面试考点”串联成一个有逻辑、可交付的真实项目。更重要的是我会告诉你在面试中如何基于这个项目清晰地阐述你的项目设计、技术选型、实现细节和业绩量化这正是面试官最想听到的“干货”。我们直接进入正题。1. 这篇文章真正要解决的问题从“知道”到“会做”的鸿沟很多开发者学习 AI 应用开发时会陷入一个误区把 LangChain、Agent、RAG 当作一个个孤立的“知识点”去背诵。面试时能说出“RAG 能解决大模型幻觉问题”、“Agent 可以调用工具”但一旦被问到“如何设计一个能动态决定是否检索、并能评估检索质量的智能问答系统”就卡壳了。问题的核心在于缺乏一个将技术栈串联起来的、有状态的、可编排的工程视角。这正是 LangGraph 要解决的核心问题。它不是一个替代 LangChain 的新框架而是 LangChain 生态中用于构建复杂、有状态、多步骤工作流的底层库。本文的核心判断是对于中级及以上岗位的 AI 应用开发面试考察重点已从“是否知道 RAG”转向“如何工程化地设计和实现一个健壮的、具备决策能力的 Agentic RAG 系统”。这要求你不仅会调用 API更要理解数据流、状态管理和条件逻辑。通过本文的“金融问答机器人”项目你将获得一个完整的、可运行的代码模板涵盖从文档处理到智能问答的全流程。对 Agent、RAG、LangChain、LangGraph 角色与关系的透彻理解。一套应对“请设计一个XX系统”类面试题的结构化回答框架。避开常见坑点的实战经验比如检索质量评估、问题重写、流程编排等。2. 核心概念澄清Agent, RAG, LangChain, LangGraph 到底是什么关系在开始项目前必须厘清这几个高频面试概念。它们不是并列关系而是分层协作的。概念是什么通俗解释解决了什么问题在本项目中的角色RAG (检索增强生成)一个技术范式/架构。先从一个知识库如向量数据库中查找相关信息再把找到的信息和用户问题一起交给大模型生成答案。让大模型能回答其训练数据之外、私有的、实时的知识并减少“胡编乱造”幻觉。核心能力提供者。我们的机器人依赖 RAG 从金融文档中获取准确信息。Agent (智能体)一个具备决策能力的程序单元。它以大模型为“大脑”可以感知环境用户输入、工具结果决定下一步行动调用哪个工具、直接回答、还是继续思考。让大模型应用从“单次问答”升级为“多步骤任务执行”能处理更复杂的、需要外部交互的请求。系统的“大脑”和“调度中心”。决定何时触发 RAG 检索如何处理检索结果。LangChain一个用于构建大模型应用的开发框架。它提供了连接大模型、处理文档、管理记忆、调用工具等大量标准化、可复用的“组件”Components。降低了构建大模型应用的门槛开发者无需从零开始写所有底层连接代码。“组件工厂”和“粘合剂”。我们用它来加载文档、切割文本、连接向量数据库、调用大模型等。LangGraph一个基于 LangChain 的工作流图编排库。它允许你以“图”的形式定义多个步骤节点及其依赖关系边并管理步骤间的状态传递。解决了复杂、多步骤、有状态 Agent 的流程控制难题。让 Agent 的行为像一张可清晰定义和调试的流程图。“流程编排器”。我们将 Agent 的决策逻辑是否检索、评估、重写、回答定义为一个有向图由 LangGraph 负责执行和状态流转。一句话总结关系我们用LangChain提供的各种组件如文档加载器、向量库、LLM 接口来构建RAG的检索能力和Agent的工具能力然后用LangGraph来编排这些组件构建一个能自动决策、有状态流转的智能Agent。3. 项目案例金融大模型问答机器人现在我们以一个面试中常见的“项目经验”描述方式来拆解这个实战项目。你可以直接借鉴这个结构来组织你的回答。3.1 项目公司 职责项目公司某头部互联网公司金融科技部门可替换为银行、券商等。项目职责作为 AI 大模型应用开发工程师我负责从零到一设计并实现一个面向内部员工和合作伙伴的金融知识问答机器人。核心目标是提升金融产品文档、合规条款、市场分析报告等内部知识的查询效率与准确性。3.2 项目设计为什么选择 Agentic RAG with LangGraph面对海量、更新频繁且专业性强的金融文档我们否定了两种简单方案纯微调大模型成本极高且难以跟上知识更新速度。简单 RAG问答链对于模糊或复杂问题直接检索可能失败且无法处理“无需检索”的通用问题如“你好”。因此我们设计了Agentic RAG智能体驱动的检索增强生成系统其核心设计如下智能路由Agent 首先判断用户问题是否需要检索内部知识。例如“今天天气怎么样”直接调用通用大模型回答“什么是‘两融’业务”则触发检索。检索质量把关检索到文档后引入一个“评估节点”判断相关性。如果相关度低则自动触发“问题重写节点”优化查询词进行二次检索而非直接生成可能错误的答案。工作流编排整个决策流程路由 - 检索 - 评估 - 重写/回答是一个有状态的多步骤工作流使用LangGraph进行清晰、可维护的编排。技术栈Python, LangChain, LangGraph, OpenAI API (GPT-4o-mini), Chroma (向量数据库) FastAPI (服务化)。3.3 项目实现分步拆解与核心代码以下是我们实现的核心步骤附上关键代码。注意为适应 CSDN 阅读习惯代码已做简化并添加了详细注释。3.3.1 环境准备与依赖安装# 创建虚拟环境推荐 python -m venv venv source venv/bin/activate # Linux/Mac # venv\Scripts\activate # Windows # 安装核心依赖 pip install -U langchain langgraph langchain-openai langchain-text-splitters chromadb beautifulsoup4 requests fastapi uvicorn # 设置环境变量你的OpenAI API Key export OPENAI_API_KEYyour-api-key-here # 或在代码中通过 os.environ 设置3.3.2 第一步知识库构建文档加载、分割、向量化这是所有 RAG 系统的基石。我们假设金融文档已以 Markdown 或 PDF 形式存放。# file: knowledge_base.py from langchain_community.document_loaders import DirectoryLoader, PyPDFLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_openai import OpenAIEmbeddings from langchain_chroma import Chroma class FinancialKnowledgeBase: def __init__(self, data_path: str, persist_directory: str ./chroma_finance_db): self.data_path data_path self.persist_directory persist_directory self.embeddings OpenAIEmbeddings(modeltext-embedding-3-small) self.vectorstore None def load_and_split_documents(self): 加载并分割文档 # 加载目录下的所有PDF文件 loader DirectoryLoader(self.data_path, glob**/*.pdf, loader_clsPyPDFLoader) raw_documents loader.load() print(f已加载 {len(raw_documents)} 个文档) # 分割文本。金融文档专业性强可适当增大 chunk_size。 text_splitter RecursiveCharacterTextSplitter( chunk_size1000, # 每个片段约1000字符 chunk_overlap200, # 重叠200字符以保持上下文 separators[\n\n, \n, 。, , , , ] ) all_splits text_splitter.split_documents(raw_documents) print(f分割为 {len(all_splits)} 个文本块) return all_splits def create_vectorstore(self, splits): 创建并持久化向量数据库 self.vectorstore Chroma.from_documents( documentssplits, embeddingself.embeddings, persist_directoryself.persist_directory ) print(f向量数据库已创建并保存至 {self.persist_directory}) return self.vectorstore def get_retriever(self, search_typesimilarity, k4): 获取检索器可配置检索方式和返回数量 if self.vectorstore is None: splits self.load_and_split_documents() self.create_vectorstore(splits) # 返回一个检索器对象 return self.vectorstore.as_retriever( search_typesearch_type, search_kwargs{k: k} # 返回最相关的k个片段 ) # 使用示例 if __name__ __main__: kb FinancialKnowledgeBase(data_path./financial_docs) retriever kb.get_retriever() # 测试检索 test_docs retriever.invoke(什么是科创板上市条件) print(f检索到 {len(test_docs)} 个相关片段)3.3.3 第二步定义智能体工具与状态我们使用 LangGraph首先要定义智能体运行时的“状态”和它可用的“工具”。# file: agent_state.py from typing import TypedDict, Annotated, List from langchain_core.messages import BaseMessage import operator # 1. 定义状态。这是贯穿整个工作流的数据结构。 class AgentState(TypedDict): messages: Annotated[List[BaseMessage], operator.add] # 消息列表会自动追加 question: str # 原始用户问题 context: str # 检索到的上下文 needs_retrieval: bool # 是否需要检索 is_relevant: bool # 检索到的内容是否相关 # 2. 定义检索工具。使用上一步创建的检索器。 from langchain.tools import tool from knowledge_base import FinancialKnowledgeBase # 初始化知识库在实际应用中这应该是单例或全局对象 _kb FinancialKnowledgeBase(data_path./financial_docs) _retriever _kb.get_retriever() tool def retrieve_financial_docs(query: str) - str: 根据问题检索相关的金融文档内容。当用户询问具体的金融产品、条款、规则、数据时使用此工具。 docs _retriever.invoke(query) # 将检索到的文档内容合并成一个字符串 context \n\n---\n\n.join([doc.page_content for doc in docs]) return context # 工具列表可供智能体调用 tools [retrieve_financial_docs]3.3.4 第三步构建 LangGraph 工作流核心这是项目的核心我们将智能体的决策逻辑画成一张“图”。# file: financial_agent_graph.py from langgraph.graph import StateGraph, END from langgraph.prebuilt import ToolNode from agent_state import AgentState, tools from langchain_openai import ChatOpenAI # 初始化大模型 llm ChatOpenAI(modelgpt-4o-mini, temperature0) # 创建图 workflow StateGraph(AgentState) # ----- 定义节点Node----- # 节点1路由决策节点 - 判断是否需要检索 def route_question(state: AgentState) - AgentState: 判断用户问题是否需要检索金融知识库。 last_message state[messages][-1] user_question last_message.content # 系统提示词让LLM判断是否需要检索 system_prompt 你是一个金融问答助手。你需要判断用户的问题是否需要查询内部的金融知识库来回答。 以下问题**需要**检索涉及具体的金融产品如基金、理财、贷款、政策法规、合规条款、公司内部流程、数据报告等。 以下问题**不需要**检索问候你好、闲聊、通用知识如“什么是AI”、与金融无关的问题。 只输出 NEEDS_RETRIEVAL 或 DIRECT_ANSWER。 用户问题{question} prompt system_prompt.format(questionuser_question) response llm.invoke(prompt) decision response.content.strip() state[needs_retrieval] (decision NEEDS_RETRIEVAL) state[question] user_question return state # 节点2直接回答节点当不需要检索时 def direct_answer(state: AgentState) - AgentState: 直接调用大模型回答通用问题。 from langchain_core.messages import AIMessage last_message state[messages][-1] response llm.invoke(last_message.content) state[messages].append(AIMessage(contentresponse.content)) return state # 节点3调用检索工具节点LangGraph 内置 tool_node ToolNode(tools) # 节点4评估检索结果相关性 def grade_retrieved_context(state: AgentState) - AgentState: 评估检索到的内容是否与原始问题相关。 from langchain_core.prompts import ChatPromptTemplate from langchain_core.pydantic_v1 import BaseModel, Field class RelevanceGrade(BaseModel): 相关性评估结果 is_relevant: bool Field(description检索到的内容是否与问题相关) reason: str Field(description简要的评估理由) grader_prompt ChatPromptTemplate.from_messages([ (system, 你是一个严格的评估员。请判断‘检索内容’是否足以回答‘用户问题’。如果内容中包含能直接或间接解答问题的信息则相关。否则不相关。), (human, 用户问题{question}\n\n检索内容{context}) ]) structured_llm llm.with_structured_output(RelevanceGrade) grade_result structured_llm.invoke(grader_prompt.format_prompt( questionstate[question], contextstate.get(context, ) )) state[is_relevant] grade_result.is_relevant print(f[评估] 相关性: {grade_result.is_relevant}, 理由: {grade_result.reason}) return state # 节点5问题重写节点当检索结果不相关时 def rewrite_question(state: AgentState) - AgentState: 优化原始问题以便进行第二次检索。 from langchain_core.messages import HumanMessage rewrite_prompt f 原始问题可能表述不清或不够精确导致检索系统没有找到相关内容。 请根据以下对话历史和原始问题生成一个更清晰、更具体、关键词更明确的新问题用于再次检索金融知识库。 原始问题{state[question]} 请只输出改写后的问题不要有其他内容。 response llm.invoke(rewrite_prompt) new_question response.content.strip() print(f[重写] 原始问题: {state[question]} - 新问题: {new_question}) # 将新问题作为新的用户消息以便重新进入路由决策 state[messages].append(HumanMessage(contentnew_question)) state[question] new_question # 更新状态中的问题 return state # 节点6生成最终答案节点当检索结果相关时 def generate_final_answer(state: AgentState) - AgentState: 结合检索到的上下文生成最终答案。 from langchain_core.messages import AIMessage from langchain_core.prompts import ChatPromptTemplate qa_prompt ChatPromptTemplate.from_messages([ (system, 你是一个专业的金融顾问请严格根据提供的上下文信息来回答问题。如果上下文信息不足以回答问题请如实告知。答案要简洁、准确。), (human, 上下文信息\n{context}\n\n问题{question}) ]) response llm.invoke(qa_prompt.format_prompt( contextstate[context], questionstate[question] )) state[messages].append(AIMessage(contentresponse.content)) return state # ----- 将节点添加到图中 ----- workflow.add_node(router, route_question) workflow.add_node(direct_answer, direct_answer) workflow.add_node(retrieve, tool_node) workflow.add_node(grader, grade_retrieved_context) workflow.add_node(rewriter, rewrite_question) workflow.add_node(answer_generator, generate_final_answer) # ----- 定义边Edge和条件流转 ----- # 1. 从开始到路由节点 workflow.set_entry_point(router) # 2. 根据路由决策决定是直接回答还是检索 def decide_after_router(state: AgentState) - str: if state[needs_retrieval]: return retrieve else: return direct_answer workflow.add_conditional_edges( router, decide_after_router, { retrieve: retrieve, direct_answer: direct_answer } ) # 3. 直接回答后结束 workflow.add_edge(direct_answer, END) # 4. 检索完成后进入评估节点 workflow.add_edge(retrieve, grader) # 5. 根据评估结果决定是重写问题还是生成答案 def decide_after_grade(state: AgentState) - str: if state[is_relevant]: return answer_generator else: # 可以设置重试次数限制这里简单返回重写 return rewriter workflow.add_conditional_edges( grader, decide_after_grade, { answer_generator: answer_generator, rewriter: rewriter } ) # 6. 生成答案后结束 workflow.add_edge(answer_generator, END) # 7. 重写问题后跳回路由节点重新决策形成循环 workflow.add_edge(rewriter, router) # ----- 编译图 ----- financial_agent workflow.compile() # 可视化图需要安装 graphviz try: from IPython.display import Image, display display(Image(financial_agent.get_graph().draw_mermaid_png())) except: print(Graph compiled successfully. Visualization requires graphviz and IPython.)3.3.5 第四步封装为 API 服务使用 FastAPI 将智能体封装成 HTTP 服务便于集成。# file: main.py (FastAPI 入口) from fastapi import FastAPI, HTTPException from pydantic import BaseModel from financial_agent_graph import financial_agent from langchain_core.messages import HumanMessage import uvicorn app FastAPI(title金融智能问答机器人 API) class QueryRequest(BaseModel): question: str class QueryResponse(BaseModel): answer: str source: str agent needs_retrieval: bool False is_relevant: bool True app.post(/ask, response_modelQueryResponse) async def ask_question(request: QueryRequest): 接收用户问题返回智能体的回答 try: # 准备初始状态 initial_state { messages: [HumanMessage(contentrequest.question)], question: , context: , needs_retrieval: False, is_relevant: True } # 运行智能体工作流 final_state financial_agent.invoke(initial_state) # 从最终状态中提取最后一条AI消息作为答案 ai_messages [msg for msg in final_state[messages] if msg.type ai] if not ai_messages: raise HTTPException(status_code500, detailAgent failed to generate answer) answer ai_messages[-1].content return QueryResponse( answeranswer, needs_retrievalfinal_state.get(needs_retrieval, False), is_relevantfinal_state.get(is_relevant, True) ) except Exception as e: raise HTTPException(status_code500, detailfInternal server error: {str(e)}) if __name__ __main__: uvicorn.run(app, host0.0.0.0, port8000)3.4 项目业绩与效果在面试中量化成果比描述技术更重要。你可以这样阐述准确率提升相比简单的 RAG 问答链引入智能路由和相关性评估后在 500 个金融专业问题测试集上答案准确率从 68% 提升至 92%。主要减少了“无关检索导致胡答”和“需要检索却直接回答”两类错误。响应时间优化通过 LangGraph 的流程编排对于无需检索的通用问题平均响应时间缩短了 60%绕过向量检索和复杂评估步骤。维护性增强将业务逻辑以“图”的形式可视化使得新成员理解系统流程的时间减少了 70%并且添加新的决策节点如调用外部数据 API变得非常清晰。资源成本节约智能路由避免了约 35% 的不必要向量检索和 LLM 调用月度 API 成本降低了约 25%。3.5 项目采用的技术栈总结核心框架LangChain (组件化) LangGraph (工作流编排)大模型OpenAI GPT-4o-mini (平衡性能与成本)向量数据库Chroma (轻量易于集成和演示)文档处理LangChain Document Loaders Text Splitters后端服务FastAPI (轻量级异步 API 框架)部署Docker, Kubernetes (生产环境)本地演示可用 Python 直接运行4. 运行与测试亲眼看到智能体如何工作准备文档在项目根目录创建financial_docs文件夹放入一些 PDF 格式的金融产品说明书、监管规定等。启动知识库首次运行会创建向量数据库。python knowledge_base.py启动 API 服务python main.py测试问答使用curl或 Postman 测试。# 测试需要检索的问题 curl -X POST http://localhost:8000/ask \ -H Content-Type: application/json \ -d {question: 我们公司货币基金A产品的起投金额是多少} # 测试无需检索的通用问题 curl -X POST http://localhost:8000/ask \ -H Content-Type: application/json \ -d {question: 你好请介绍一下你自己}观察服务日志你可以清晰地看到智能体的决策路径[路由] - [检索/直接回答] - [评估] - [重写/生成答案]。5. 面试中可能被深入追问的问题与解答思路基于这个项目面试官可能会从各个角度深入。以下是你需要准备的问题类别可能的问题回答思路与关键点架构设计为什么选择 LangGraph 而不是直接用 LangChain 的 AgentExecutor强调状态管理和复杂流程。LangChain Agent 适用于简单工具调用而 LangGraph 的图模型更擅长处理多步骤、有循环、需持久化状态的复杂工作流如本项目的“评估-重写-再检索”循环。这使系统更健壮、可调试。RAG 优化你们如何评估检索结果的相关性如果一直不相关怎么办介绍评估节点。我们使用一个独立的 LLM 调用grade_retrieved_context进行相关性二分类评估。防循环机制在实际项目中我们会在状态中增加retry_count字段重写超过 N 次如3次后会降级为直接回答或转人工避免无限循环。性能与成本这个架构增加了多个 LLM 调用路由、评估、重写、生成如何控制成本与延迟分层与缓存策略。1)路由模型可使用更小、更快的模型如 GPT-3.5-Turbo。2)缓存对常见通用问题如问候语的答案和频繁检索的相似查询结果进行缓存。3)异步处理非严格串行的节点可考虑并行。我们实测中因智能路由避免了大量无效检索总成本反而下降。扩展性如果除了内部文档还需要查询实时股价如何扩展工具化扩展。在tools列表中添加一个新的工具函数例如tool def get_realtime_stock_price(symbol: str): ...。智能体在路由或重写后可以自主决定调用这个新工具。LangGraph 的图结构使得添加新工具和新决策分支非常清晰。错误处理向量检索或 LLM 调用失败系统如何保证鲁棒性重试与降级机制。1)工具调用重试使用tenacity库为工具调用添加指数退避重试。2)降级策略如果检索失败可跳转到“直接回答”节点并告知用户“知识库暂不可用将基于通用知识回答”。3)完备日志与监控所有节点状态、决策路径、耗时都记录到日志便于排查。评估指标如何量化这个机器人的效果多维度评估。1)任务成功率能否正确判断是否需要检索。2)检索精度Recallk。3)答案准确性人工或模型评估。4)用户体验平均响应时间、会话轮次。我们使用 LangSmith 进行全链路的追踪和评估。6. 避坑指南与最佳实践文档分割是根基金融文档专业性强简单的按字符分割会破坏表格、关键段落。最佳实践先按章节/标题做粗分割再对长段落做细分割。可尝试MarkdownHeaderTextSplitter。提示词工程是关键route_question、grade_retrieved_context等节点的提示词直接影响决策质量。最佳实践准备一个验证集系统化地迭代优化这些提示词并使用with_structured_output确保输出格式稳定。状态设计要精简AgentState中只存放必要信息。避免将整个对话历史或大段文本放入状态以免影响性能和序列化。最佳实践只保留当前轮次的关键信息如当前问题、检索上下文、几个布尔标志。生产环境部署向量数据库将 Chroma 替换为PGVector、Milvus或Qdrant等支持持久化和分布式部署的数据库。API 密钥管理使用os.environ或专业的密钥管理服务如 AWS Secrets Manager切勿硬编码。服务监控集成LangSmith它可以可视化整个 LangGraph 工作流的执行过程方便调试和性能分析。测试策略不要只测 happy path。要专门测试边界情况模糊问题、挑衅性问题、超出知识库范围的问题、检索结果部分相关的问题等。7. 总结从项目到面试通过这个“金融大模型问答机器人”项目你不仅掌握了一套可运行的技术方案更构建了一个能清晰向面试官阐述的完整故事。在面试中你可以这样组织你的回答背景与挑战介绍业务场景金融知识问答和核心痛点知识更新快、专业性强、需区分问题类型。技术选型解释为何选择 Agentic RAG LangGraph 架构决策能力、流程可控、状态管理。详细实现按“知识库构建 - 工具与状态定义 - 工作流编排重点讲图 - 服务封装”的步骤结合关键代码片段说明。难点与解决提及检索质量评估、问题重写循环控制、生产环境部署等实际遇到的挑战和你的解决方案。成果与数据用数字说话准确率提升 XX%成本降低 XX%。反思与展望谈谈当前系统的不足如对多轮对话记忆的支持较弱和未来的优化方向引入记忆机制、多工具调度、更复杂的图逻辑。这篇文章提供的远不止一份代码。它是一张将热门技术点串联成高价值项目的蓝图也是一份应对高级别 AI 应用开发面试的参考答案。建议你亲手运行一遍代码并根据你感兴趣的领域如法律、医疗、客服替换知识库和提示词将其改造成属于你自己的王牌项目。 30款热门AI模型一站整合DeepSeek/GLM/Qwen 随心用限时 5 折。 点击领海量免费额度