基于LangGraph与RAG的金融问答Agent实战:从架构设计到代码实现
30款热门AI模型一站整合DeepSeek/GLM/Qwen 随心用限时 5 折。 点击领海量免费额度1. 先搞清楚这套组合拳到底解决什么问题以及面试官想听什么如果你正在准备AI大模型应用开发工程师的面试尤其是那些要求你懂Agent、RAG、LangChain和LangGraph的岗位那这篇文章就是为你准备的。这不是一个简单的概念科普而是帮你把零散的知识点串成一个能讲清楚、能动手、能应对追问的实战体系。面试官想听的不是你背了多少名词而是你能不能把这些技术组合起来解决一个真实业务问题比如“金融大模型问答机器人”。这套技术栈的核心价值在于让大模型从“一个会聊天的百科全书”变成一个能主动思考、会查资料、能按流程办事的“智能员工”。Agent是大脑负责决策和规划RAG是它的外部知识库解决大模型“一本正经胡说八道”和知识陈旧的问题LangChain是工具箱提供了连接模型、工具、数据的标准件LangGraph则是流程图用来编排复杂、多步骤的Agent工作流。面试中最怕的就是被问到“这几个东西有什么区别你是怎么把它们用在一起的”。很多人会卡壳因为学习时都是分开看的。下面我就按一个真实项目从零到一的落地顺序带你走一遍。2. 项目设计从业务需求到技术选型拆解假设我们要做一个“金融大模型问答机器人”。别一上来就写代码先想清楚业务要什么。项目公司 职责面试话术模板你可以这样描述“在上一家公司我负责一个面向内部投研团队的金融问答机器人项目。我的核心职责是完成从需求分析、技术架构设计、核心模块开发到上线部署的全流程确保机器人能准确、高效地回答关于公司财报、行业研报和金融术语的复杂问题。”项目设计思路需求痛点分析师需要快速从海量PDF报告年报、研报里找数据、对比观点。传统搜索只能找关键词无法理解“对比A公司和B公司去年Q3的毛利率”这种复杂意图。核心目标做一个能理解自然语言问题、自动检索相关文档、并生成准确、有依据答案的助手。技术选型理由大模型选用GPT-4或国内同等能力的商用API作为“大脑”负责理解、推理和生成。不选本地开源模型因为金融领域对事实准确性要求极高商用API的推理能力更稳定。面试时可以说基于项目对准确性和开发效率的权衡我们选择了[某云厂商]的LLM API服务。RAG检索增强生成这是项目的基石。把所有金融文档转换成向量存起来让模型回答时有据可查避免幻觉。关键点单纯的RAG只是“检索生成”不够智能。Agent我们需要机器人能“判断”什么时候该去查资料。比如用户问“你好”这不需要查资料问“茅台2023年净利润”就必须去查。这就是Agent的决策能力。LangChain用它来快速搭建管道Pipeline。比如文档加载、文本分割、向量化存储、检索器封装LangChain都有现成组件能省大量开发时间。LangGraph当流程变复杂时比如“先判断是否需要检索 - 检索 - 判断检索结果是否相关 - 不相关则改写问题重新检索 - 相关则生成答案”这种带条件和循环的流程用LangChain的简单Agent就有点吃力了。LangGraph允许你像画流程图一样精确控制Agent的每一步这是面试加分项。一句话说清区别LangChain帮你快速组装标准零件工具、记忆、链LangGraph让你能设计和运行复杂的、有状态的工作流流程图。在复杂Agent场景下LangGraph是更强大的编排器。3. 环境准备与核心模块实现别空谈架构我们直接看代码和配置。这里我以最经典的“自定义RAG Agent”为例拆解每一步。3.1 环境搭建与依赖安装首先准备好Python环境建议3.9。安装核心依赖这里要注意版本兼容性。# 基础框架和工具 pip install -U langgraph langchain langchain-community # 文本处理 pip install langchain-text-splitters # 向量数据库这里用内存型演示生产环境需换Chroma、Milvus等 pip install chromadb # 或其它向量库 # 大模型接口以OpenAI为例国内可用百度、阿里、智谱等替代 pip install openai # 文档解析根据你的文件格式 pip install pypdf python-docx beautifulsoup4关键点langgraph和langchain要装对版本。生产环境强烈建议使用requirements.txt锁定版本避免后续部署出错。如果公司网络限制需要配置镜像源或私有pip仓库。3.2 文档加载与向量化RAG基础这是RAG的“原料准备”阶段决定了后续检索质量。from langchain_community.document_loaders import PyPDFLoader, DirectoryLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_openai import OpenAIEmbeddings from langchain_chroma import Chroma # 1. 加载文档 - 假设所有PDF放在 ./data 目录下 loader DirectoryLoader(./data, glob**/*.pdf, loader_clsPyPDFLoader) raw_documents loader.load() print(f已加载 {len(raw_documents)} 个文档) # 2. 分割文本 - 这是关键参数直接影响检索精度 text_splitter RecursiveCharacterTextSplitter( chunk_size500, # 每个块500字符 chunk_overlap100, # 块之间重叠100字符防止语义被切断 separators[\n\n, \n, 。, , , ] # 中文分割符 ) doc_splits text_splitter.split_documents(raw_documents) print(f分割后得到 {len(doc_splits)} 个文本块) # 3. 向量化并存储 embeddings OpenAIEmbeddings(modeltext-embedding-3-small) # 注意模型名 # 持久化到磁盘避免每次重启重新计算 vectorstore Chroma.from_documents( documentsdoc_splits, embeddingembeddings, persist_directory./chroma_db # 向量数据库存储路径 ) retriever vectorstore.as_retriever(search_kwargs{k: 4}) # 每次检索最相关的4个块避坑指南块大小chunk_size不是越大越好。太大检索会引入无关信息太小会丢失上下文。金融文档如财报段落结构清晰500-1000字符是个不错的起点需要根据实际文档调整。重叠chunk_overlap必须设置防止一个完整的句子或关键数据如“净利润同比增长15%”被切到两个块边缘。向量模型嵌入模型必须和后续生成模型“语感”一致。如果用GPT嵌入也用OpenAI的。混用可能导致语义搜索不准。检索数量k给模型喂多少上下文一般3-5个。太多会干扰模型增加成本太少可能信息不全。3.3 构建检索工具封装给Agent使用Agent需要通过“工具”来与环境交互。我们把检索器包装成一个工具。from langchain.tools import tool from typing import List tool def retrieve_financial_docs(query: str) - str: 根据用户问题从金融文档向量库中检索最相关的信息。 参数: query: 用户的问题。 返回: 拼接后的相关文档文本。 # 这里直接使用上一步创建好的检索器 docs retriever.invoke(query) # 将检索到的文档内容拼接成一个字符串作为工具的返回 context \n\n---\n\n.join([doc.page_content for doc in docs]) return f检索到的相关信息\n{context} # 测试工具 tool_response retrieve_financial_docs.invoke({query: 贵州茅台2023年营业收入是多少}) print(tool_response[:500]) # 打印前500字符看看工具Tool的本质就是一个函数它告诉Agent“你能调用我我能帮你做某件事”。Agent在思考时会决定是否调用、用什么参数调用这个工具。4. 用LangGraph构建智能决策工作流这是核心也是面试官最爱深挖的部分。我们将构建一个能自主决策的RAG Agent。4.1 定义工作流状态LangGraph的工作流是有状态的我们需要定义状态里有什么。from typing import TypedDict, Annotated, List from langgraph.graph.message import add_messages import operator class AgentState(TypedDict): 定义Agent工作流的状态 # 消息历史记录用户、AI、工具的对话 messages: Annotated[List, add_messages] # 用户原始问题用于问题改写时参考 original_question: str # 当前轮次检索到的上下文 context: strAnnotated[List, add_messages]是LangGraph的语法糖它会自动帮你把新的消息追加到messages列表里非常方便。4.2 创建节点NodeAgent的各个功能模块我们把工作流拆成几个节点每个节点负责一件事。节点1路由决策节点是否需要检索这是Agent的“大脑”决定是直接回答还是去查资料。from langchain.chat_models import ChatOpenAI from langgraph.prebuilt import ToolNode # 初始化大模型 llm ChatOpenAI(modelgpt-4o-mini, temperature0) # 给模型绑定它可用的工具 llm_with_tools llm.bind_tools([retrieve_financial_docs]) def route_question(state: AgentState) - AgentState: 根据用户问题决定下一步是检索还是直接回答。 user_message state[messages][-1] # 调用模型并告诉它可以调用工具 response llm_with_tools.invoke([user_message]) # 将模型的响应可能包含工具调用指令加入到状态中 return {messages: [response]}节点2检索节点如果模型决定调用工具工作流就会执行到这里。# 使用LangGraph预建的ToolNode它会自动执行工具调用 tool_node ToolNode([retrieve_financial_docs]) # 这个节点我们不需要自己写LangGraph已经处理了工具执行的细节。节点3判断检索质量节点相关吗不是所有检索结果都有用。我们需要一个“质检员”。from pydantic import BaseModel, Field class GraderOutput(BaseModel): 定义一个结构化输出让模型只返回‘是’或‘否’ is_relevant: bool Field(description检索到的文档是否与问题相关) def grade_retrieved_context(state: AgentState) - dict: 判断检索到的上下文是否与原始问题相关。 original_question state[original_question] # 从状态中取出最后一次工具调用的结果即检索到的上下文 # 这里简化处理实际应从state[‘messages’]中解析出工具返回的内容 retrieved_context state.get(context, ) if not retrieved_context: return {next_node: rewrite_question} # 没检索到内容需要改写问题 # 使用一个专门的“裁判”模型可以用同一个模型但temperature设为0 grader_llm ChatOpenAI(modelgpt-4o-mini, temperature0).with_structured_output(GraderOutput) prompt f 你是一个严格的质检员。请判断下面的“检索内容”是否足以回答“用户问题”。 只根据内容本身判断不要进行推理。如果内容中包含能直接或间接回答问题的信息则判断为相关。 用户问题{original_question} 检索内容{retrieved_context[:1000]}... # 截断避免过长 请输出‘是’或‘否’。 judgment: GraderOutput grader_llm.invoke(prompt) # 根据判断结果决定下一个节点 if judgment.is_relevant: return {next_node: generate_answer} else: return {next_node: rewrite_question}节点4问题改写节点如果检索结果不相关可能是问题问得不好需要改写。def rewrite_question(state: AgentState) - AgentState: 改写原始问题使其更清晰、更易于检索。 original_q state[original_question] prompt f 原始用户问题可能表述不清或难以检索。请将其改写成更明确、更利于从文档库中查找信息的问题。 保持原意但可以补充一些可能的同义词或相关概念。 原始问题{original_q} 改写后的问题 new_question_msg llm.invoke(prompt) # 这里用不带工具的LLM # 将改写后的问题作为新的用户消息重新开始流程 return {messages: [{role: user, content: new_question_msg.content}], original_question: original_q}节点5生成最终答案节点检索结果合格生成最终答案。def generate_final_answer(state: AgentState) - AgentState: 结合问题和检索到的上下文生成最终答案。 question state[original_question] context state[context] prompt f 你是一个专业的金融问答助手。请严格根据提供的“参考信息”来回答问题。 如果信息不足请明确告知“根据已有信息无法回答”。 答案应简洁、准确并可以引用信息中的关键数据。 用户问题{question} 参考信息 {context} 答案 final_answer llm.invoke(prompt) return {messages: [final_answer]}4.3 组装工作流定义边和条件现在把各个节点像拼乐高一样组装起来定义它们之间的流转逻辑。from langgraph.graph import StateGraph, END # 1. 创建图 workflow StateGraph(AgentState) # 2. 添加节点 workflow.add_node(router, route_question) # 路由决策 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) # 生成答案 # 3. 设置入口 workflow.set_entry_point(router) # 4. 定义边工作流走向 # 4.1 从‘router’出来如果模型调用了工具就去‘retrieve’否则直接结束。 def decide_after_router(state: AgentState) - str: last_msg state[messages][-1] if hasattr(last_msg, tool_calls) and last_msg.tool_calls: return retrieve else: return END workflow.add_conditional_edges( router, decide_after_router, { retrieve: retrieve, END: END } ) # 4.2 从‘retrieve’出来总是去‘grader’判断质量 workflow.add_edge(retrieve, grader) # 4.3 从‘grader’出来根据判断结果去生成答案或改写问题 def decide_after_grader(state: AgentState) - str: # grade_retrieved_context 函数返回的 dict 中包含 next_node return state.get(next_node, rewrite_question) # 默认去改写 workflow.add_conditional_edges( grader, decide_after_grader, { generate_answer: answer_generator, rewrite_question: rewriter } ) # 4.4 从‘rewriter’出来问题改写了重新回到‘router’决策 workflow.add_edge(rewriter, router) # 4.5 从‘answer_generator’出来生成答案后工作流结束 workflow.add_edge(answer_generator, END) # 5. 编译图 app workflow.compile()这就是LangGraph的核心价值它把复杂的“if-else”逻辑变成了清晰可视化的流程图。你可以用app.get_graph().draw_mermaid_png()把图生成出来面试时展示这个设计图非常直观。4.4 运行与测试# 初始化状态 initial_state { messages: [{role: user, content: 茅台和五粮液谁去年的净利润更高}], original_question: 茅台和五粮液谁去年的净利润更高, context: } # 运行工作流 final_state app.invoke(initial_state) # 打印最终答案 for message in final_state[messages]: if message.type ai: print(助手回答, message.content)5. 项目业绩、优化与面试复盘项目业绩面试话术 “项目上线后内部调研显示分析师查找特定财务数据的平均时间从原来的15-30分钟缩短至2分钟内准确率提升至95%以上。同时由于采用了LangGraph构建的可观测工作流我们能够精准定位回答错误的环节例如是检索不准还是生成偏差使得模型迭代优化周期缩短了50%。”技术栈总结LangChain用于快速构建文档加载、分割、向量化、检索的流水线。RAG为核心能力提供实时、准确的外部知识来源。Agent体现为工作流中的决策能力路由、判断、改写。LangGraph将以上所有组件编排成一个稳定、可控、可调试的复杂智能体。面试常见问题与回答思路QLangChain和LangGraph什么区别ALangChain像“标准件库”和“胶水”它提供了Chain链来线性组合任务。但当任务需要循环、条件分支、复杂状态管理时比如我们刚才的“检索-判断-改写”循环Chain就显得笨拙。LangGraph则是“流程图设计器”它用“图”的概念来定义工作流节点是功能边是流转逻辑特别适合构建有状态的、复杂的多步骤Agent。可以说LangGraph是LangChain在复杂Agent编排上的超集和更优解。QRAG的检索效果不好怎么办A这是实战中最常见的问题。我会从以下顺序排查文本分割检查chunk_size和chunk_overlap是否适合你的文档类型。金融法律文档和技术博客的最佳分割策略不同。嵌入模型确认使用的嵌入模型是否适合你的语料中/英文专业领域。可以尝试换用bge、voyage等专门优化的模型。检索策略除了简单的相似度搜索similarity_search可以尝试MMR最大边际相关性来平衡相关性和多样性或者加入元数据过滤如文档类型、年份。重排序Re-ranking在初步检索出Top K个结果后用一个更精细的交叉编码器模型对它们进行重排序把最相关的排到最前面。这是提升精度非常有效的手段。HyDE让模型先根据问题生成一个假设性答案然后用这个答案去检索有时能更好地捕捉意图。QAgent反应慢怎么优化A性能优化是系统工程。异步与流式对于不依赖前后顺序的工具调用可以使用异步并行。最终答案生成可以采用流式输出提升用户体验。缓存对嵌入向量、频繁出现的查询结果进行缓存。图优化审视LangGraph工作流是否有节点可以合并判断逻辑是否可以简化避免不必要的循环。模型层面在非核心推理环节如问题改写、相关性判断使用更小、更快的模型如GPT-3.5-Turbo。Q怎么评估这个机器人的效果A不能只靠感觉。我们会建立评估体系事实准确性人工抽样对比答案与标准答案。检索相关性计算检索到的文档块与问题的语义相似度得分。回答有用性让真实用户打分。** latency**记录从提问到收到回答的平均耗时。成本监控每次问答的Token消耗和API调用费用。使用LangSmith这类工具可以很好地追踪和评估这些指标。最后给准备面试的你不要只停留在“知道概念”。一定要亲手跑通一个这样的流程哪怕是用最简单的文档和代码。在面试中你能清晰地说出“我在这里为什么用chunk_overlap”、“我如何用LangGraph的conditional_edge处理检索失败”、“我如何评估和优化检索效果”这比你罗列一百个技术名词都有用。这套组合拳的价值就在于它能将大模型的能力工程化、可控化而这正是企业最需要的能力。 30款热门AI模型一站整合DeepSeek/GLM/Qwen 随心用限时 5 折。 点击领海量免费额度

相关新闻