1. 检索主链路到底在搭什么从“能聊”到“能查”的分水岭很多人做企业级问答系统卡在第三章、第四章就停了——模型接上了Prompt 调通了单轮对话也能跑但一旦问它“我们公司去年Q3的差旅报销标准是多少”它就开始一本正经地胡说八道。这不是模型不行是你还没把检索主链路接上。Ch07 这一章要干的事就是把前面几章攒下来的零件——文档入库、向量化、LangGraph 编排、SSE 推流——串成一条完整的“用户提问 → 检索 → 生成 → 流式返回”的闭环。跑通这一条链路你的系统才算真正从“聊天玩具”变成“知识问答工具”。这条链路的核心关键词是RAG检索增强生成。说白了RAG 就是让模型在回答之前先去“翻资料”。你可以把它理解成一个开卷考试模型是考生Milvus 是书架Ollama 跑的大模型是那个会答题但记性不好的学生LangGraph 是监考老师负责安排“先翻书、再答题、答完交卷”的流程SSE 则是把答案一个字一个字念给用户听的喇叭。这套组合在当下企业私有化部署场景里非常主流原因很直接数据不出内网、模型可换、检索可控、成本可算。适合谁来读这篇如果你已经跟着前几章把 Milvus 跑起来了、Ollama 也拉好了模型、LangGraph 的基本节点概念也清楚了那这一章就是你的“合龙”时刻。如果你还没搭好环境也别急着关页面我会在关键节点补上环境确认的检查项你对照着排查就行。整条链路我实测下来在单机 Docker 环境下从提问到首字返回控制在 1.5 秒以内是完全可行的后面会讲怎么调。先说清楚这一章最终要交付的东西一个 FastAPI 接口接收用户问题走 LangGraph 编排的检索-生成流程通过 SSE 把答案流式吐给前端前端用 Vue 或者任何能消费 EventSource 的框架都能接。链路里每一个环节我都会给出可复制的代码和参数说明以及我踩过的坑。2. 整体链路设计与技术选型为什么是这套组合2.1 链路全景一次问答到底经过了哪些环节在动手写代码之前先把这条链路的“地图”画清楚。用户在前端输入一个问题比如“报销标准是多少”这个请求打到 FastAPI 的/chat接口接口内部启动一个 LangGraph 图图的执行顺序大致是这样接收问题节点拿到原始 query做基础清洗去首尾空格、过滤超长输入。检索节点把 query 用 Embedding 模型转成向量去 Milvus 里做相似度检索召回 Top-K 相关文档片段。组装上下文节点把召回片段拼成一段 context和原始问题一起塞进 Prompt 模板。生成节点把组装好的 Prompt 交给 Ollama 跑的本地大模型流式生成答案。推流节点每生成一个 token就通过 SSE 推给前端。这五步里第 2 步和第 4 步是性能大头第 5 步是体验关键。LangGraph 的价值在于它把这五步变成了一个有状态、可中断、可回溯的图而不是一坨写死的函数调用。为什么这点重要因为企业级场景里你迟早要加“多路召回”“重排序”“意图识别分支”“人工审核卡点”这些东西用函数堆砌会越写越乱用图编排则每个能力都是一个节点加节点不影响已有逻辑。2.2 为什么检索用 Milvus 而不是别的向量库的选择上Milvus 在企业私有化场景里几乎是默认答案。原因有三第一它能扛住千万级甚至亿级的向量规模你前期用几百条文档测试感觉不出来但企业知识库动辄几十万份文档Milvus 的分布式能力是刚需第二它的索引类型丰富IVF_FLAT、HNSW、DiskANN 各有适用场景调优空间大第三社区活跃Docker 部署文档齐全standalone 模式一条命令就能起。这里要特别提醒一个热词里反复出现的坑Milvus 的 URI 配置。很多人在本地开发时用milvus_uri: str ./data/milvus.db这种本地文件模式Milvus Lite跑得好好的一上服务器就报错。原因是 Milvus Lite 只支持本地嵌入式场景不支持多客户端并发也不支持完整的索引能力。服务器 Linux 环境上正确做法是用 Docker 起 standalone 模式URI 写成http://localhost:19530。这个切换点我在项目里踩过一次本地测试全绿部署后检索一直返回空排查了半天才发现是 URI 模式不对。2.3 为什么生成用 OllamaSSE 为什么不能省Ollama 的价值在于“离线可控”。企业内网环境往往不允许调用外部 APIOllama 让你把模型权重下载到本地用一条ollama run就能起一个推理服务。热词里“ollama 下载慢”“ollama 离线安装包”“ollama 国内镜像源”这些搜索量很高说明大家最头疼的就是模型拉取。我的经验是生产环境一定要提前把模型文件准备好用离线方式导入别指望部署时现场拉。SSEServer-Sent Events则是流式体验的载体。为什么不用 WebSocket因为问答场景是单向推流——服务端推、客户端收不需要双向通信。SSE 基于 HTTP实现简单浏览器原生支持 EventSourceNginx 配置也简单。热词里“stream disconnected before completion: idle timeout waiting for sse”这个报错非常典型本质是 SSE 连接空闲超时被中间层掐断了后面排查章节会专门讲。2.4 选型对照表环节选型备选选择理由编排LangGraph手写函数链有状态、可加节点、支持中断恢复向量库Milvus standaloneFAISS、Chroma支持大规模、索引丰富、企业级生成模型Ollama 本地模型云端 API数据不出内网、成本可控推流SSEWebSocket单向推流够用、实现简单服务框架FastAPIFlask原生 async、SSE 支持好这张表不是让你照抄而是让你理解每个选择的“为什么”。如果你的场景是几百条文档的小工具FAISS 完全够用上 Milvus 是杀鸡用牛刀如果你允许调外部 API那 Ollama 也不是必须的。选型永远服务于场景。3. 核心细节拆解检索节点与生成节点的实现要点3.1 检索节点向量化与相似度检索的关键参数检索节点的核心动作是“把问题变成向量去库里找最像的”。这里有两个参数直接决定召回质量Embedding 模型和Top-K。Embedding 模型的选择上中文场景我建议用 BGE 系列或者 M3E它们在中文语义相似度任务上表现稳定。Ollama 本身也能跑 embedding 模型比如nomic-embed-text好处是统一用 Ollama 管理坏处是中文效果一般。我的做法是embedding 用专门的模型服务生成用 Ollama各司其职。Top-K 的选择是个权衡。K 太小召回不全模型答不出来K 太大上下文塞太多噪声模型反而被干扰而且 token 消耗暴涨。我的经验值是K5 到 K8配合一个相似度阈值比如 0.6做过滤。实测下来K5 加阈值过滤在技术文档类知识库上召回准确率能到 85% 以上。Milvus 的相似度度量方式也要注意。热词里“milvus 余弦值”说明有人踩过这个坑。Milvus 支持 L2、IP、COSINE 三种度量用 COSINE 时记得把向量归一化否则结果会偏。如果你用 IP内积且向量已归一化那 IP 和 COSINE 等价。我一般直接用 COSINE省心。# 检索节点核心逻辑示意 from pymilvus import Collection import numpy as np def retrieve_node(state): query state[question] # 1. 问题向量化 query_vec embed_model.encode(query, normalize_embeddingsTrue) # 2. Milvus 检索 collection Collection(knowledge_base) collection.load() results collection.search( data[query_vec.tolist()], anns_fieldembedding, param{metric_type: COSINE, params: {nprobe: 16}}, limit5, output_fields[content, source] ) # 3. 阈值过滤 组装 docs [] for hit in results[0]: if hit.score 0.6: docs.append({content: hit.entity.get(content), score: hit.score}) return {retrieved_docs: docs}这段代码里nprobe是 IVF 索引的搜索参数值越大越准但越慢。16 是个平衡点数据量小的时候可以调到 32。3.2 生成节点Prompt 模板与流式输出的配合生成节点的关键在 Prompt 模板。RAG 的 Prompt 有个经典结构系统指令 检索上下文 用户问题 输出约束。我常用的模板长这样你是一个企业知识助手。请严格根据以下参考资料回答问题。 如果参考资料中没有相关信息直接回答“根据现有资料无法回答”不要编造。 参考资料 {context} 用户问题{question} 请用简洁的中文回答这个模板里最重要的那句是“如果参考资料中没有相关信息直接回答无法回答”。没有这句话模型会习惯性地用它的预训练知识去补这在企业场景里是致命的——用户要的是“我们公司的规定”不是“一般情况下的规定”。流式输出这块Ollama 的 Python SDK 支持streamTrue返回一个生成器每 yield 一个 chunk 就是一个 token。你要做的是把这个生成器的输出转成 SSE 格式推出去。这里有个细节Ollama 返回的 chunk 结构是{message: {content: ...}}你要提取content字段然后包装成data: {...}\n\n的格式。3.3 LangGraph 图的状态设计LangGraph 的图是围绕“状态”转的。状态就是一个字典在各个节点之间传递和更新。我这条链路的状态设计如下from typing import TypedDict, List class QAState(TypedDict): question: str # 用户原始问题 retrieved_docs: List # 检索到的文档 context: str # 组装后的上下文 answer: str # 生成的答案 sources: List # 引用来源状态设计的原则是只放节点间需要传递的数据。不要把数据库连接、模型实例这些放进去那些应该作为节点函数的闭包变量或者依赖注入。状态越干净图越好调试。图的构建用StateGraph加节点、加边、设入口和出口from langgraph.graph import StateGraph, END workflow StateGraph(QAState) workflow.add_node(retrieve, retrieve_node) workflow.add_node(assemble, assemble_node) workflow.add_node(generate, generate_node) workflow.set_entry_point(retrieve) workflow.add_edge(retrieve, assemble) workflow.add_edge(assemble, generate) workflow.add_edge(generate, END) app workflow.compile()这个图现在是线性的但它的价值在于扩展性。等你需要加“意图识别”分支时只要在 retrieve 前面加一个判断节点用条件边分流就行已有节点完全不用动。4. 实操过程从 FastAPI 接口到 SSE 推流的完整实现4.1 环境确认清单动手之前先确认这几样东西是好的。我列个清单你逐项打勾Milvus standalone 已启动http://localhost:19530能连通collection 已建好且有数据。Ollama 服务已启动ollama list能看到你要用的模型比如qwen2:7b。Python 环境装好了fastapi、uvicorn、pymilvus、langgraph、ollama、sse-starlette。前端或测试工具能发 HTTP 请求并消费 SSE。这四项任何一项没通后面的代码都跑不起来。特别是 Milvus很多人卡在 collection 没 load 就检索返回空结果还不报错非常隐蔽。4.2 FastAPI 接口与 SSE 响应FastAPI 里做 SSE我推荐用sse-starlette这个库它封装了EventSourceResponse比手写StreamingResponse省事。核心代码如下from fastapi import FastAPI from sse_starlette.sse import EventSourceResponse from pydantic import BaseModel import json app FastAPI() class ChatRequest(BaseModel): question: str app.post(/chat) async def chat(req: ChatRequest): async def event_generator(): # 走 LangGraph 图 async for event in app_graph.astream({question: req.question}): # event 是 {节点名: 状态更新} for node_name, state_update in event.items(): if node_name generate and answer in state_update: yield { event: message, data: json.dumps({content: state_update[answer]}, ensure_asciiFalse) } # 结束信号 yield {event: done, data: [DONE]} return EventSourceResponse(event_generator())这里有几个关键点。第一用astream而不是invoke因为流式场景要边执行边拿结果。第二SSE 的每条消息格式是event: xxx\ndata: xxx\n\nsse-starlette帮你处理了换行。第三结束时要发一个明确的done事件前端收到这个才知道流结束了否则会一直挂着等。4.3 生成节点的流式改造上面接口里generate 节点如果是一次性返回完整答案那 SSE 就失去意义了。所以 generate 节点本身要改成流式。但这里有个矛盾LangGraph 的节点函数默认是返回一个状态更新字典不是生成器。怎么让节点内部流式我的做法是在节点内部把流式 chunk 累积同时通过一个队列往外推。或者更简单的做法把生成逻辑从图里拆出来图只负责检索和组装生成单独用一个流式函数处理。这样图的状态更新到 context 就结束生成阶段直接调 Ollama 的流式接口。import ollama async def generate_stream(context: str, question: str): prompt PROMPT_TEMPLATE.format(contextcontext, questionquestion) stream ollama.chat( modelqwen2:7b, messages[{role: user, content: prompt}], streamTrue ) for chunk in stream: content chunk[message][content] if content: yield content然后在接口里先跑图拿到 context再调这个流式函数app.post(/chat) async def chat(req: ChatRequest): async def event_generator(): # 第一步跑图拿 context result await app_graph.ainvoke({question: req.question}) context result[context] sources result.get(sources, []) # 先推来源 yield {event: sources, data: json.dumps(sources, ensure_asciiFalse)} # 第二步流式生成 async for token in generate_stream(context, req.question): yield {event: message, data: json.dumps({content: token}, ensure_asciiFalse)} yield {event: done, data: [DONE]} return EventSourceResponse(event_generator())这个结构清晰很多图负责“查”流式函数负责“答”接口负责“推”。职责分离调试也方便。4.4 前端消费 SSE 的写法前端用原生 EventSource 就能接但注意 EventSource 只支持 GET 请求。如果你要用 POST 传问题得用fetch加ReadableStream手动解析。Vue 项目里我一般封装一个 composableasync function askQuestion(question, onToken, onDone) { const response await fetch(/chat, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ question }) }); const reader response.body.getReader(); const decoder new TextDecoder(); let buffer ; while (true) { const { done, value } await reader.read(); if (done) break; buffer decoder.decode(value, { stream: true }); const lines buffer.split(\n\n); buffer lines.pop(); for (const line of lines) { if (line.startsWith(data:)) { const data line.slice(5).trim(); if (data [DONE]) { onDone(); return; } try { const parsed JSON.parse(data); if (parsed.content) onToken(parsed.content); } catch (e) { /* 忽略解析错误 */ } } } } }这段代码的关键是buffer的处理。SSE 的消息以\n\n分隔但网络传输可能把一个消息拆成多个 chunk所以要用 buffer 累积按\n\n切分最后一段留在 buffer 里等下一个 chunk。这个细节不处理好会出现消息截断或者 JSON 解析失败。4.5 完整链路的时序与性能实测我把整条链路跑了一遍记录各阶段耗时单机 DockerMilvus 10 万条向量Ollama qwen2:7bGPU 推理阶段耗时说明问题向量化30-50msEmbedding 模型前向Milvus 检索20-40msnprobe1610 万级上下文组装5ms字符串拼接首 token 生成300-800ms取决于模型和硬件后续 token20-50ms/token流式输出端到端首字约 1.2s从请求到首字返回首字延迟是体验的关键指标。用户能接受 1-2 秒的首字延迟但接受不了 5 秒的白屏。所以优化重点应该放在“尽快出第一个字”上而不是“尽快出完整答案”。这也是流式输出的意义。5. 常见问题与排查技巧实录5.1 SSE 连接中断idle timeout 的三种成因热词里“stream disconnected before completion: idle timeout waiting for sse”这个报错我遇到过三次成因各不相同。第一种Nginx 的 proxy_read_timeout 太短。Nginx 默认 60 秒没数据传输就断连接。如果你的模型生成慢两个 token 之间超过 60 秒连接就断了。解决方法是把proxy_read_timeout调到 300 秒以上并且加proxy_buffering off让 Nginx 不缓冲 SSE 数据。第二种模型生成卡住。有时候 Ollama 处理长上下文会卡住几十秒不出 token。这种情况要在服务端加心跳每隔 15 秒发一个注释行: heartbeat\n\n保持连接活跃。第三种客户端超时。有些 HTTP 客户端默认超时时间短比如 30 秒。前端 fetch 要设置足够的 timeout或者干脆不设靠服务端的 done 事件来结束。排查顺序建议先看 Nginx 日志有没有 499客户端断开或 504网关超时再看服务端日志有没有异常最后看模型推理日志。5.2 Milvus 检索返回空结果的排查路径检索返回空是 RAG 里最高频的问题。我整理了一个排查清单现象可能原因排查方法完全返回空collection 没 load调collection.load()确认返回空向量维度不匹配对比 embedding 维度和 collection schema返回结果但 score 很低度量方式不对确认用 COSINE 且向量归一化返回结果但内容不对数据没入库成功查 collection.num_entities本地正常服务器空URI 模式不对确认用 http 而非本地文件其中“向量维度不匹配”最隐蔽。比如你建 collection 时用的是 768 维后来换了 embedding 模型变成 1024 维插入和检索都会报错或者返回空。建库时一定要把维度定死换模型就重建库。5.3 Ollama 模型加载慢与显存不足Ollama 首次加载模型会从磁盘读权重到显存7B 模型大概要 5-10 秒。如果你发现每次请求都很慢可能是模型被卸载了。Ollama 有个keep_alive参数默认 5 分钟没请求就卸载模型。生产环境建议设成-1常驻或者一个较大的值# 请求时带上 keep_alive ollama.chat(modelqwen2:7b, messages[...], keep_alive30m)显存不足的话Ollama 会自动把部分层放到 CPU速度会断崖式下降。7B 模型 FP16 大概要 14GB 显存量化到 Q4 只要 4-5GB。如果显存紧张优先用量化版本。5.4 上下文组装的两个坑第一个坑是上下文超长。召回 5 个片段每个 500 字加上 Prompt 模板轻松超过模型的上下文窗口。解决办法是给每个片段设长度上限或者用重排序只保留最相关的 3 个。第二个坑是片段之间没有分隔。直接把多个片段拼在一起模型分不清哪段是哪段。正确做法是给每个片段加编号和来源标记[片段1] 来源员工手册.pdf 内容... [片段2] 来源财务制度.docx 内容...这样模型引用时也能准确说出“根据员工手册”。5.5 独家避坑技巧汇总Milvus 的 nprobe 别设太大nprobe 从 16 调到 64召回率提升有限但延迟翻倍。先用 16 测不够再调。SSE 的 data 里别放换行JSON 序列化时ensure_asciiFalse保留中文但内容里的换行要转义成\n否则会破坏 SSE 格式。Ollama 的 stream 要显式关闭如果客户端提前断开服务端的生成器要能感知并停止否则会一直占着 GPU。用try/finally包住生成逻辑。LangGraph 的节点函数别做重活节点函数应该轻量重活比如模型推理放到单独的服务或异步任务里否则图执行会阻塞。测试时先用小数据量10 条文档跑通链路再灌 10 万条。数据量大了之后索引构建、检索延迟的问题才会暴露。6. 链路跑通之后下一步能往哪扩链路跑通只是起点。这条“检索-生成-推流”的主干搭好之后往上加东西就顺了。我列几个我实际做过的扩展方向供你参考。多路召回现在只有向量检索一路可以加一路关键词检索BM25两路结果合并去重。向量检索擅长语义相似关键词检索擅长精确匹配两者互补。LangGraph 里加一个并行节点就能实现。重排序召回 Top-20用一个小的重排序模型比如 BGE-Reranker精排取 Top-5 给生成。这一步能把召回准确率再提 10-15 个百分点代价是增加 100-200ms 延迟。意图识别分支在检索前加一个节点判断用户问题是“知识问答”还是“闲聊”还是“操作指令”。不同意图走不同分支闲聊直接生成知识问答走 RAG操作指令走工具调用。这就是 LangGraph 条件边的用武之地。引用溯源生成答案时让模型标注每个结论来自哪个片段前端把来源高亮显示。企业场景里用户对“这个答案哪来的”非常在意溯源能大幅提升信任度。多轮对话现在的链路是单轮的加一个历史消息管理把前几轮问答拼进 Prompt就能支持追问。注意历史不能无限拼要设个轮数上限或者做摘要压缩。我个人在实际操作中的体会是RAG 系统的效果七分靠检索三分靠生成。很多人把精力花在换更大的模型上但真正该花时间的是文档切分策略、Embedding 模型选择、召回参数调优这些“脏活”。我见过用 7B 模型加精细检索效果吊打 70B 模型加粗糙检索的案例。所以链路跑通之后别急着换模型先把检索质量打磨到位。最后分享一个小技巧建一个“评测集”准备 50-100 个真实问题和标准答案每次调整检索参数或 Prompt都跑一遍评测集看准确率变化。没有评测集的调优就是盲调今天调好了明天可能又坏了。这个评测集不用多复杂一个 JSON 文件加一个脚本就够但它是你系统持续迭代的锚点。