先说说这个项目从哪来的。上个月公司内部知识库还是老一套关键词检索同事问“报销流程最长几天”系统返回的是行政制度PDF的第14页气得产品经理直接把需求扔到我桌上“搞个能真正理解问题的助手”。于是就有了这套基于LangGraph FastAPI 混元大模型的企业智能助手RAG知识引擎系统迭代到21.5这个小版本终于把“能回答问题”升级成了“能像老员工一样思考和找人求助”。这套系统的核心价值不复杂把企业散落的制度文档、操作手册、历史FAQ变成可检索、可推理、可追溯的知识资产用Agent去理解用户真实意图先多路召回再生成答案而不是拿个向量库糊一脸相似文本。如果你正在搭个人RAG、想做智能客服、或者苦恼于LangChain的Chain写不出复杂逻辑这篇实践记录你应该能直接用上。1. 技术选型为什么偏偏是LangGraph FastAPI 混元这套组合1.1 传统RAG的瓶颈与Agentic RAG的破局点先泼一盆冷水裸RAG在企业场景里大概率翻车。我刚做第一版的时候就是“用户问题 - embedding - 向量检索 top5 - 拼接上下文 - 直接问混元”看起来挺顺但实际用起来问题一大堆。第一用户问“报销单填错了财务会不会打回”文档里写的是“发票粘贴不规范将退回”这俩表述语义差着十万八千里向量检索根本召不回“退回”这个关键动作第二多轮对话场景下“那要怎么改”这种指代问题裸RAG没有任何记忆能力第三企业复杂问题往往需要“查制度 看流程 问历史工单”多步操作裸RAG一次只能做一次检索。所以要上Agentic RAG也就是让大模型学会“规划、召回、判断、再行动”。LangGraph正好是把这种思维方式落地的靠谱框架它的核心优势不是“能调大模型”而是把整个推理过程变成了有状态、可回退、可控制的图。系统能做到什么样的上限很大程度上取决于你把Agent的“思考动作”拆得多细。1.2 LangGraph比裸LangChain好在哪网上关于“LangGraph和LangChain的区别”的讨论特别多我用大白话给你讲清楚。LangChain的核心抽象是Chain一条链子从A到B到C顺序固定中间要加个分支、加个循环就得靠包装各种Chain来硬怼维护成本直接起飞。LangGraph的核心抽象是Graph节点Node就是处理函数边Edge就是流转规则还支持条件路由Conditional Edge和循环Cycle这就相当于把“链式调用”升级成了“状态机工作流”。打个比方LangChain是一条笔直的流水线产品从这头进去那头出来你要在中间加一道“质检不合格就送回上一站”的工序得把整条流水线拆了重装。LangGraph是一个有分岔路、有返工线的工厂车间你只要在图纸上多画一个节点、加一条带条件的传送带产品就能自己走到该去的地方。实际做Agent的时候要的就是这种控制力比如先检索一次发现置信度不够就改写关键词再检索一次比如用户说“你理解错了”Agent要能回到上一步重新规划。这些逻辑用LangGraph写起来就是几个if判断加几条边的事用LangChain写那就是另一个项目了。1.3 FastAPI做服务层的三个硬理由后端服务层选择FastAPI不是我追新是被逼出来的。第一我们需要给前端Vue3提供后端API而且对话功能必须支持流式输出用户提问之后一个字一个字往外蹦体验和整体等待完全不一样。FastAPI天然支持StreamingResponse结合async generator能轻松实现SSEServer-Sent Events推送这是Flask和Django要费不少功夫才能搞利索的。第二FastAPI基于Pydantic做类型校验请求参数、响应模型都是强类型定义前端联调时看到Swagger文档自动生成接口长什么样一目了然省了维护接口文档的时间。第三性能上FastAPI是异步框架LangGraph的astream异步流式方法正好能和它无缝配合一路async走到底不会出现同步阻塞把事件循环卡死的问题。还有一点很实际就是FastAPI生态对比经典Python服务端框架更现代部署时用uvicorn跑起来配个Docker文件就能放到K8s里不折腾。项目里我们用FastAPI同时暴露了REST接口和SSE接口REST用于普通查询SSE用于对话流前端一个axios一个EventSource分工清爽。1.4 混元大模型在企业场景的取舍逻辑大模型选型时我们认真比过开源私有化方案和云API方案最终选了混元大模型。原因很朴素企业数据不能出内网但纯私有化部署一个高质量模型GPU成本第一年就能让预算报表很难看混元的API接入成本低中文理解能力在企业文档场景实测表现稳定而且它支持function calling和较长上下文刚好满足LangGraph里Agent需要“判断是否调用检索工具”的需求。另一个重要点是合规同量级的国内模型在数据安全责任边界上更清楚省去很多法务来回沟通。接入方式我们走了OpenAI兼容协议这样后续如果换模型供应商改一个base_url和api_key就可以替换模型层不用重构。这个设计决策在后面换模型做A/B测试时帮了大忙。2. 整体架构一个企业知识引擎的四个核心层次2.1 接入层详解API怎么设计才能让前端和第三方都好接这一层是给前端和第三方系统用的设计原则是“能简单就不复杂”。我们对外暴露了三个主要接口POST /api/chat对话式问答返回流式SSE用于智能助手主界面。POST /api/query一次性问答普通JSON返回用于第三方系统集成场景。POST /api/knowledge/upload知识库文档上传与管理由管理后台调用。接口设计上有个经验不要在一个接口里混太多语义。早期我把“查知识库”和“闲聊”都放在/chat里由Agent内部判断结果前端很难处理“这个问题走知识库但Agent没触发检索”的情况。后来改成query带一个require_knowledge参数前端根据场景显式告诉后端“这次我看要看知识库”准确率立刻上升了几个点。同理上传接口一定要支持异步任务文件先上传返回file_id后台异步解析切块入库前端轮询任务状态这样大PDF也不会把请求超时打爆。2.2 Agent编排层LangGraph的状态机如何管住多轮对话这是整个系统的大脑。LangGraph的State是贯穿全局的数据载体每一轮Agent运行所有的中间结果都会写进State里原始问题、改写后的问题、检索到的文档片段、重排后的结果、最终生成的答案、历史对话记录。状态设计决定了Agent能做什么、不能做什么。我们的核心State大概是这样的结构class AgentState(TypedDict, totalFalse): question: str # 用户原始问题 rewritten_question: str # 改写后的检索式 chat_history: list # 近三轮对话历史 retrieved_docs: list # 多路召回重排后的文档 agent_scratchpad: list # Agent思考过程记录 answer: str # 最终答案 need_followup: bool # 是否需要追问澄清 confidence: float # 本次回答置信度图结构里我们设计了五个核心节点意图识别节点、检索规划节点、多路召回节点、置信度判断节点、答案生成节点。节点之间的边有普通边和条件边两种比如置信度判断节点会根据confidence值决定走向“直接生成答案”还是“重新改写检索式再试一轮”意图识别节点会判断用户是想查知识库、聊闲天还是需要调用外部工具然后走向完全不同的子图。这种编排方式让整个Agent的行为完全可控出了问题知道是哪个节点的事而不是面对一段黑盒代码无从下手。2.3 检索层多路召回加重排序的落地姿势“RAG多路召回”这个词看着高级实际落地就是三个通道并行检索最后合并排序。第一路是向量检索用embedding模型把用户问题和文档都转成高维向量在向量数据库里找余弦相似度最高的片段这路擅长找语义相近但用词不同的内容第二路是BM25关键字检索基于词频和逆文档频率匹配擅长找“工号”“报销单号”这类精确术语和唯一标识第三路是元数据过滤检索比如“2024年考勤制度”“销售部差旅标准”先按文档标签过滤再语义检索减少干扰。三路召回结果合并后必须过重排序模型Reranker这是很多RAG项目忽略的一步。向量检索召回的top20里可能有一大半是“沾点边但没用”的内容Reranker会逐条计算与问句的真实相关度并重新排序。我们实际测试下来加一个cross-encoder重排最终答案的用户接受度能提升20%左右这个钱和时间绝对不能省。2.4 模型层混元接入与统一模型网关模型层我们做了一个很薄的网关模块统一封装对混元API的调用。网关做的事情有三件一是管理API密钥和访问配置环境变量隔离二是统一请求格式把LangGraph节点里传进来的消息列表转成混元能识别的格式三是限流和重试混元API有QPS限制网关里用信号量控制并发对5xx错误做指数退避重试。网关的核心就是不让业务代码感知底层的模型品牌和接口细节。LangGraph节点里只需要一句await llm_gateway.chat(messages, tools...)$返回的就是解析好的字符串或工具调用结果。这个抽象层在后面接入别的大模型做对比评测时非常有用。3. 知识库构建实操从原始文档到可检索的切片3.1 文档预处理不干净的数据做不了RAG知识库构建的第一步不是切块是清洗。企业文档什么妖孽都有从飞书导出的PDF页眉页脚全是公司名、Word转的PDF有大段空白页、扫描件OCR出来中文乱码、表格被转成图片。我踩过最大的坑是拿原始PDF直接切块结果向量库里塞满了“机密”字样页眉检索时凡是相关文档第一条永远是不相关的页眉文字。所以在预处理阶段我们写了一个PipelinePDF先做OCR如果扫描件然后用正则和版面分析去掉页眉页脚、页码、目录、水印最后把表格区域单独提取出来转成Markdown格式。这里有个小经验表格千万别转成纯文本把“姓名|部门|报销金额”这种结构拍平成一行语义就毁了。转成Markdown的表格语法embedding效果和后续大模型理解效果都会好很多。清洗规则维护上建议不要把规则写死在代码里而是做成一个可配置的规则文件YAML或数据库表每个知识域维护自己的清洗规则否则运营一个文档子库加两条规则还要发版效率太低。3.2 切块策略的取舍固定窗口、父子块还是语义切块网上关于“RAG文档怎么切块”的教程特别多我直接说结论没有银弹要看文档类型。固定窗口切块最简单但会在关键句子中间拦腰斩断父子块方案好用父块保证语义完整子块提高检索精准度语义切块最优雅但对资源要求高。我们的做法是混合使用。对于制度、流程类文档默认用固定窗口参数是chunk_size500字符、overlap50但加了一个硬规则切块边界不能落在句子中间。实现上先按段落分段落太长再在句子边界处切这样很少出现半个句子入库的尴尬。对于操作手册、FAQ类文档用父子块方案子块设定300字符用于检索父块是子块所在的完整章节用于生成上下文检索时命中子块、返回父块保证答案完整。有个参数要特别留意overlap不是越多越好。overlap太多同一段话被重复存进多个chunk检索时会重复召回相似内容反而稀释了有效信息。50到100字符是个合理区间但还需要结合你自己测试集的结果微调。3.3 向量化与混合检索的细节实现embedding模型选择上我们项目用的是BGE系列的embedding模型中英文效果平均维度768足够企业内部文档用。向量库用的Milvus因为支持后续扩展到千万级向量。入库时每个chunk还会附带元数据文档编号、标题、章节路径、更新时间、权限范围这些对后面的元数据过滤检索特别关键。多路召回和重排在代码里的结构大致是这样async def hybrid_retrieve(state: AgentState) - dict: query state.get(rewritten_question) or state[question] # 向量检索取top20 vector_results await vector_search(query, top_k20) # BM25检索取top20 bm25_results await bm25_search(query, top_k20) # 元数据过滤检索 if state.get(doc_filter): filter_results await filtered_search(query, state[doc_filter], top_k10) # RRF(Reciprocal Rank Fusion)融合三路结果 fused rrf_fusion([vector_results, bm25_results, filter_results]) # 重排序 reranked await rerank(query, fused[:30]) return {retrieved_docs: reranked[:5]}融合排序用的是RRF方法虽然朴素但非常有效每条文档在多路结果里的排名取倒数求和再按总分排。三路里哪怕只有一个通道召回了正确答案RRF也能把它抬到顶部比简单拼接再让大模型挑要稳定得多。4. LangGraph编排开发从零搭一个可用的智能助手Agent4.1 定义图结构状态、节点、边的设计要点LangGraph开发最核心的是“先把图想清楚再写代码”。我的建议是拿到需求后先在白板上画图把每个节点、每条边、每个条件都画出来确认不会出现死循环再动手。我们一个典型的知识问答流程的图逻辑如下from langgraph.graph import StateGraph, END def route_after_intent(state: AgentState) - str: intent state.get(intent, knowledge) if intent chitchat: return chat elif intent knowledge: return hybrid_retrieve return clarify graph StateGraph(AgentState) graph.add_node(intent, intent_node) graph.add_node(hybrid_retrieve, hybrid_retrieve) graph.add_node(judge_confidence, judge_confidence) graph.add_node(rewrite_query, rewrite_query) graph.add_node(generate, generate_node) graph.add_node(chat, chitchat_node) graph.set_entry_point(intent) graph.add_conditional_edges(intent, route_after_intent, {chat: chat, knowledge: hybrid_retrieve, clarify: generate}) graph.add_edge(hybrid_retrieve, judge_confidence) graph.add_conditional_edges(judge_confidence, is_confidence_enough, {True: generate, False: rewrite_query}) graph.add_edge(rewrite_query, hybrid_retrieve) graph.add_edge(generate, END) graph.add_edge(chat, END)注意confidence判断到rewrite_query再到hybrid_retrieve这个循环最多重试两轮第二轮还是低置信度就直接进generate用已有上下文生成答案并提示“结果可能不太准确”。这个循环上限必须有否则遇到一个彻底无法检索的问题Agent会无限循环烧钱。4.2 多Agent协作主管-工人模式在企业场景的裁剪遇到“人力制度里没有的东西能不能问财务知识库”这类跨域问题单Agent会乱。我们在LangGraph里实现了主管-工人模式主管Agent负责理解用户意图然后把任务分配给HR、财务、IT、行政等子Agent每个子Agent对应不同的知识库和工具。这个模式在企业场景下要裁剪着用。第一没必要给每个部门都搞一个复杂子Agent我们按知识域聚合成了三个子Agent人事行政域、财务报销域、IT支持域每人负责自己的知识库集合。第二子Agent之间不通信所有交互都通过主管。企业场景里跨Agent通信容易造成信息混乱主管统一协调反而更好维护。第三分配规则不是全让大模型自由判断而是加入了规则预判问题里出现“报销”“发票”“差旅费”等关键词直接路由到财务子Agent减轻主管模型的判断压力也减少走错库的概率。4.3 与FastAPI集成从同步接口到SSE流式输出的改造LangGraph的图本身可以同步调用但在FastAPI里一定要用异步方式否则一个长回答会阻塞整个服务。我们用graph.astream()取出每一步的状态流然后通过SSE推给前端。from fastapi import APIRouter, Request from fastapi.responses import StreamingResponse import json router APIRouter() async def event_generator(graph, initial_state: dict): async for event in graph.astream(initial_state, stream_modeupdates): # 只把关键节点的中间状态发给前端 for node_name, node_output in event.items(): if node_name generate and node_output.get(answer): yield fdata: {json.dumps({type: answer, content: node_output[answer]}, ensure_asciiFalse)}\n\n yield fdata: {json.dumps({type: done})}\n\n router.post(/api/chat) async def chat(request: Request, payload: ChatRequest): state { question: payload.question, chat_history: payload.chat_history[-6:], # 保留近三轮 } return StreamingResponse(event_generator(graph, state), media_typetext/event-stream)这里有个很重要的性能细节如果用了SSE流式大模型生成阶段本身就是流式的建议把混元API的stream参数打开让混元生成的每个token通过生成器不断yield给前端。否则用户看到的效果就是“转圈10秒然后整段文字突然出现”体验大打折扣。但要注意混元API的流式和你graph.astream()的流是两层东西需要看清楚LangGraph在哪里能拿到内部模型输出。前端Vue3侧的处理也不复杂用EventSource或fetch配合ReadableStream读取SSE流按data:前缀解析JSON逐步渲染。这里不建议用WebSocketSSE的自动重连机制对对话场景更省心。5. 工程化落地性能、监控与上线注意事项5.1 异步与并发FastAPI的协程坑和混元API的限流FastAPI异步是把双刃剑用得不好反而更糟。最大的坑是在async函数里调同步第三库比如有人图省事直接在节点里用requests.get()这会在协程内部发起同步阻塞网络请求整个事件循环被卡住所有并发请求都跟着排队。解决方法是统一用httpx.AsyncClient或者在节点函数里用await asyncio.to_thread(func, args)把同步操作丢到线程池但优先方案还是全链路异步。混元API的限流是另一个高频翻车点。企业内部上了几十个知识库和多个Agent节点都要并发调用LLM时API的QPS限流很快触发。我们在模型网关里加了信号量import asyncio semaphore asyncio.Semaphore(5) # 控制最大并发5个请求 async def chat_with_rate_limit(messages, **kwargs): async with semaphore: try: return await chat_completion(messages, **kwargs) except RateLimitError: await asyncio.sleep(1) return await chat_completion(messages, **kwargs)注意重试策略不能每失败一次就立刻重试要加退避。但重试逻辑要放在网关的统一出口不是每个业务节点都自己try-except不然维护起来很乱。5.2 缓存与检索优化把能省的算力全省掉RAG系统的成本大头在哪一是大模型生成二是embedding调用。这两个都有办法缓存。热门问题的对话结果我们会做语义缓存用户问完“试用期是多久”第一个人问过之后第二个人问“试用期多长时间”虽然字面不同但语义几乎一样就没必要再走一遍完整的Agent流程了。做法是对用户问题做embedding和Redis里缓存的问题向量算相似度超过0.95就直接返回上一次的答案。这个优化实测能挡住30%到40%的相似重复提问省下的token费用相当可观。文档embedding缓存更暴力切块入库时如果检测到文档哈希没变就直接从缓存读向量不重新调用embedding服务。我们嵌入模型的并发标准是每秒最多20次请求重新embedding一批500块的文档要好几分钟有缓存之后秒级完成。5.3 监控、日志与灰度发布Agent系统比普通CRUD接口难定位问题得多因为一次回答可能经过5个节点、3次外部调用任何一个环节出问题都会导致最终结果异常。所以我们从第一天就做了全链路日志每个节点的输入、输出、耗时、调用的工具、LLM消耗的token数都结构化打印。线上出问题后按request_id串起整条链路的日志一眼就能看出是“检索没召回到内容”还是“模型生成阶段崩溃”。监控指标重点看三个检索召回率、重排后首条命中率、最终答案平均长度和token消耗。召回率低说明知识库切块有问题首条命中率低说明重排模型或融合策略需要调token消耗异常说明Agent在循环里反复调用模型大概率是条件路由写岔了。部署我们用的是Docker Compose起步FastAPI服务一个容器、LangGraph作为内置服务跑在同一个进程里、Redis一个容器、Milvus独立部署。等服务稳定了再平滑迁移K8s。这里要特别提醒LangGraph的图在每次请求时build一次会浪费初始化时间建议进程启动时build好图对象放在模块级变量里复用。6. 实战中踩过的坑问题排查与避坑速查表6.1 检索结果差的5个常见原因检索结果差是RAG项目里被问得最多的问题。我把这一个月踩过的坑整理成表格你照着排查基本能解决80%的问题现象可能原因排查方式召回内容明显不相关切块太大一个chunk混了多个主题检查切块后chunk的主题一致性调整chunk_size相关文档在top5里找不到只有向量检索没有BM25精确词被忽略打开多路召回日志确认三路召回各自的命中情况重排后结果反而变差重排模型和领域不匹配用已有的测试集对重排结果抽检不匹配就换模型或用GPT做伪标签自训练多轮对话越问越偏对话历史拼接到了检索query里确认检索时用的query是独立改写后的而不是拼接历史新文档入库检索不到嵌入缓存命中旧文档检查文档哈希缓存是否失效强制刷新缓存6.2 LangGraph状态更新的经典报错LangGraph有几个报错频率极高全是状态Schema没定义好的问题。最常见的是“Node X tried to update key Y, but it is not in the state schema”就是节点返回的字典里带了State里没定义的键。这个坑在调试初期几乎天天踩。解决办法是给State统一加上totalFalse并预先定义好所有可能用到的字段否则Agent节点临时想往State里塞个中间量就会直接崩。还有个隐蔽的问题在条件路由函数里读状态但路由函数返回的key和目标节点对不上运行时提示找不到边。这属于图结构定义错误排查方法是把整个图逻辑检查一遍确认每个条件边的映射都完整建议把路由映射写在代码里而不是让模型自由发挥稳定得多。6.3 FastAPI异步代码的隐蔽陷阱有段时间系统一有并发压力就全部请求超时排查了老半天发现是某个工具函数里用了time.sleep(2)等待外部接口响应。这在普通同步框架里没问题但在FastAPI的异步环境里time.sleep会阻塞整个事件循环所有并发请求都排队等这2秒。换成await asyncio.sleep(2)之后立刻恢复正常。所以建议写一个硬性规范凡是async函数里禁止出现time.sleep、requests.get这类同步阻塞调用。代码审查时就把这条卡死避免生产环境出大面积故障。6.4 安装环境相关的坑从pycharm装失败到uv包管理器很多人在PyCharm里装FastAPI失败报错一会儿是“Could not find a version that satisfies the requirement fastapi”一会儿是pip超时。原因不外乎两个Python版本太低FastAPI新版本要求3.8、默认的PyPI源访问不稳定。解决方案很简单换国内镜像源或者直接用Python官方推荐的uv工具。uv这个包管理器确实好用创建虚拟环境和安装依赖的速度比pip快好几倍。我们项目现在的标准流程是# 创建项目目录并初始化虚拟环境 uv venv .venv source .venv/bin/activate # 安装核心依赖 uv pip install fastapi uvicorn langgraph langchain-openai milvus-client redis用uv还有个好处它的依赖解析能力比较强多个库之间的版本冲突很少出现。之前手动用pip装的时候langchain和pydantic版本冲突能让人折腾一晚上换uv之后这种问题基本消失了。7. 一些最后想分享的体会这套系统做下来经常有朋友问“RAG框架选哪个”“要不要上Agent”我的答案始终没变先想清楚你有没有一个足够干净、足够结构化的知识库。LangGraph再强、混元再聪明喂给它一堆乱七八糟没切好的PDF它也只能一本正经地胡说八道。切块和清洗是RAG的地基这个功夫花不到位后面所有优化都是浪费。另外提醒一句不要为了Agent而Agent。如果业务流程本来就是固定顺序查库回答案那就不要硬加意图判断和循环路由简单的RAG反而更稳。我们的21.5版本里有很多复杂的图结构但都是业务逼出来的“我们真的需要这个分支”而不是为了展示技术做的摆设。判断一个RAG系统好不好我个人的标准很简单你愿意把公司的日常问答交给它并且出了问题你能在十分钟内定位到是切块、检索还是生成环节的锅。这套系统现在基本达到了这个标准希望这篇实践记录也能让你少踩几个我踩过的坑。