RAG检索主链路实战：LangGraph+Milvus+Ollama+SSE流式问答

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 图，图的执行顺序大致是这样：

1. 接收问题节点：拿到原始 query，做基础清洗（去首尾空格、过滤超长输入）。
2. 检索节点：把 query 用 Embedding 模型转成向量，去 Milvus 里做相似度检索，召回 Top-K 相关文档片段。
3. 组装上下文节点：把召回片段拼成一段 context，和原始问题一起塞进 Prompt 模板。
4. 生成节点：把组装好的 Prompt 交给 Ollama 跑的本地大模型，流式生成答案。
5. 推流节点：每生成一个 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 为什么生成用 Ollama，SSE 为什么不能省

Ollama 的价值在于“离线可控”。企业内网环境往往不允许调用外部 API，Ollama 让你把模型权重下载到本地，用一条ollama run就能起一个推理服务。热词里“ollama 下载慢”“ollama 离线安装包”“ollama 国内镜像源”这些搜索量很高，说明大家最头疼的就是模型拉取。我的经验是：生产环境一定要提前把模型文件准备好，用离线方式导入，别指望部署时现场拉。

SSE（Server-Sent Events）则是流式体验的载体。为什么不用 WebSocket？因为问答场景是单向推流——服务端推、客户端收，不需要双向通信。SSE 基于 HTTP，实现简单，浏览器原生支持 EventSource，Nginx 配置也简单。热词里“stream disconnected before completion: idle timeout waiting for sse”这个报错非常典型，本质是 SSE 连接空闲超时被中间层掐断了，后面排查章节会专门讲。

2.4 选型对照表

这张表不是让你照抄，而是让你理解每个选择的“为什么”。如果你的场景是几百条文档的小工具，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 消耗暴涨。我的经验值是K=5 到 K=8，配合一个相似度阈值（比如 0.6）做过滤。实测下来，K=5 加阈值过滤，在技术文档类知识库上召回准确率能到 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_embeddings=True) # 2. Milvus 检索 collection = Collection("knowledge_base") collection.load() results = collection.search( data=[query_vec.tolist()], anns_field="embedding", param={"metric_type": "COSINE", "params": {"nprobe": 16}}, limit=5, 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 支持stream=True，返回一个生成器，每 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_ascii=False) } # 结束信号 yield {"event": "done", "data": "[DONE]"} return EventSourceResponse(event_generator())

这里有几个关键点。第一，用astream而不是invoke，因为流式场景要边执行边拿结果。第二，SSE 的每条消息格式是event: xxx\ndata: xxx\n\n，sse-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(context=context, question=question) stream = ollama.chat( model="qwen2:7b", messages=[{"role": "user", "content": prompt}], stream=True ) 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_ascii=False)} # 第二步：流式生成 async for token in generate_stream(context, req.question): yield {"event": "message", "data": json.dumps({"content": token}, ensure_ascii=False)} yield {"event": "done", "data": "[DONE]"} return EventSourceResponse(event_generator())

这个结构清晰很多：图负责“查”，流式函数负责“答”，接口负责“推”。职责分离，调试也方便。

4.4 前端消费 SSE 的写法

前端用原生 EventSource 就能接，但注意 EventSource 只支持 GET 请求。如果你要用 POST 传问题，得用fetch加ReadableStream手动解析。Vue 项目里我一般封装一个 composable：

async 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 完整链路的时序与性能实测

我把整条链路跑了一遍，记录各阶段耗时（单机 Docker，Milvus 10 万条向量，Ollama qwen2:7b，GPU 推理）：

首字延迟是体验的关键指标。用户能接受 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 时用的是 768 维，后来换了 embedding 模型变成 1024 维，插入和检索都会报错或者返回空。建库时一定要把维度定死，换模型就重建库。

5.3 Ollama 模型加载慢与显存不足

Ollama 首次加载模型会从磁盘读权重到显存，7B 模型大概要 5-10 秒。如果你发现每次请求都很慢，可能是模型被卸载了。Ollama 有个keep_alive参数，默认 5 分钟没请求就卸载模型。生产环境建议设成-1（常驻）或者一个较大的值：

# 请求时带上 keep_alive ollama.chat(model="qwen2:7b", messages=[...], keep_alive="30m")

显存不足的话，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_ascii=False保留中文，但内容里的换行要转义成\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 文件加一个脚本就够，但它是你系统持续迭代的锚点。