1. 检索主链路到底在解决什么问题做企业级智能问答系统最怕的不是模型不够聪明而是模型答非所问、胡编乱造。检索主链路的核心任务就是在用户提问和模型生成之间插入一道“查资料”的工序先从企业私有知识库里把相关片段捞出来再让模型基于这些片段组织答案。这一章要落地的就是这条链路从零到一的第一次完整闭环。我见过太多团队在这一步翻车。有人把向量库当万能药结果召回的全是无关内容有人把流式输出做成“假流式”前端等半天才蹦出一个字还有人本地调试一切正常一上服务器就报连接超时。这些坑我在实际项目里基本都踩过一遍所以这一章不讲虚的直接把 LangGraph 编排、Milvus 检索、Ollama 生成、SSE 推送这四个环节串起来给你一条能跑通、能观测、能排错的主链路。适合谁看如果你已经搭好了文档解析和向量化入库现在卡在“怎么把检索和生成接起来”这一章就是为你写的。如果你还在纠结 Milvus 装 standalone 还是集群、Ollama 模型放哪个盘文中也会顺带把工程决策讲清楚。整条链路的技术选型是 LangGraph RAG Milvus Ollama SSE这套组合在私有化部署场景里性价比很高下面逐个拆。2. 整体链路设计与技术选型考量2.1 为什么用 LangGraph 而不是裸写流程第一次做问答闭环很多人的直觉是写一个函数接收问题、查向量库、拼 prompt、调模型、返回结果。简单场景确实够用但企业级系统很快会遇到几个问题检索结果要不要重排召回为空要不要走兜底多轮对话怎么带上下文这些分支一旦多起来裸写函数就会变成一坨 if-else。LangGraph 的价值在于把流程显式建模成图。节点是处理步骤边是流转条件状态在节点间传递。我第一次用它的时候觉得有点重但当我需要加一个“检索置信度低于阈值就触发追问”的分支时改起来非常清爽——加一个条件边就行不用动主流程。这就是它比 LangChain 的 Chain 更适合生产的原因Chain 是线性的Graph 是有状态的、可分支的。具体到本章我设计的图有三个核心节点retrieve检索、generate生成、fallback兜底。入口是用户问题retrieve去 Milvus 捞文档如果捞到了就进generate捞不到或者分数太低就进fallback返回“知识库暂无相关内容”。这个结构简单但已经覆盖了 80% 的问答场景。2.2 Milvus 选 standalone 还是集群Milvus 的部署模式直接决定你后面调优的空间。standalone 模式把所有组件塞进一个进程适合开发和小规模生产集群模式拆分了 coordinator、proxy、query node 等角色适合高并发。我个人的建议是日活低于一万、知识库规模在百万级向量以内standalone 完全够用别一上来就上集群运维成本差好几倍。安装上Mac 用户用 Docker 最省事官方 compose 文件拉下来改改端口就能跑。Linux 服务器上我习惯用milvus_uri: str ./data/milvus.db这种本地文件模式做轻量测试但要注意这只是 Milvus Lite 的用法生产环境还是得连真正的服务端。有个细节很多人忽略Milvus 的余弦相似度检索需要你在建集合时指定metric_typeCOSINE建完再改就得重建索引所以一开始就要想清楚。2.3 Ollama 的模型存储与离线部署Ollama 下载慢是高频吐槽点尤其在国内网络环境下。我的做法是提前把模型文件下好通过环境变量OLLAMA_MODELS指向自定义目录比如挂载一块大容量数据盘。Linux 上修改模型存储路径就是改这个变量然后重启服务Mac 上默认在~/.ollama/models想换盘得手动迁移。离线安装包这块Ollama 官方提供了各平台的二进制包内网环境直接拷贝安装即可。模型文件也可以从一台已下载的机器上整体拷贝models目录过去省去重复下载。至于“如何关闭 gemma 系列模型的思考过程”那是模型层面的 prompt 控制跟链路本身关系不大后面生成节点会提到怎么在系统提示里约束输出格式。2.4 SSE 为什么是流式输出的首选流式输出有三种常见方案WebSocket、SSE、轮询。WebSocket 双向通信能力强但实现复杂轮询延迟高体验差SSE 基于 HTTP 单向推送实现简单、浏览器原生支持、天然适配“服务端持续吐字”的场景。企业问答系统里用户只需要看答案一个字一个字出来不需要双向交互所以 SSE 是最优解。但 SSE 有个经典报错stream disconnected before completion: idle timeout waiting for sse。这个坑我在 Nginx 反代后面踩过原因是 Nginx 默认的proxy_read_timeout是 60 秒而大模型生成慢的时候超过这个时间连接就被掐了。解决办法是在 Nginx 配置里把这个值调大同时后端要定期发送心跳注释行保持连接活跃。Vue 前端那边用EventSource接收注意它不支持自定义请求头所以鉴权信息一般放在 URL 参数或者 cookie 里。3. 核心细节解析与实操要点3.1 检索节点的参数设计检索节点看起来只是“查一下向量库”但参数设计直接决定召回质量。核心参数有三个top_k、相似度阈值、是否开启重排。top_k控制返回多少条候选。设太小可能漏掉关键信息设太大则噪声多、拖慢生成。我的经验值是先用 10 做初筛如果接了重排模型再压到 3 到 5 条喂给大模型。相似度阈值是过滤低质量召回的闸门Milvus 返回的 cosine 分数在 0 到 1 之间我一般把阈值设在 0.5 到 0.6低于这个值就认为“没查到相关内容”走兜底分支。这里有个容易忽略的点查询文本的向量化必须和入库时用同一个 embedding 模型。我见过有人入库用 A 模型、查询用 B 模型结果召回全是乱的排查了半天才发现是模型不一致。所以工程上要把 embedding 模型名写进配置检索和入库共用。# 检索节点核心逻辑示意 def retrieve_node(state): query state[question] query_vector embed_model.encode(query) results milvus_client.search( collection_nameknowledge_base, data[query_vector], limit10, output_fields[content, source], search_params{metric_type: COSINE, params: {nprobe: 10}} ) docs [hit[entity][content] for hit in results[0] if hit[distance] 0.55] return {documents: docs, has_context: len(docs) 0}3.2 生成节点的 prompt 组织生成节点的关键是把检索到的文档和用户问题组装成一个清晰的 prompt。我的模板结构是系统角色说明 参考资料 用户问题 输出约束。系统角色里明确告诉模型“只依据参考资料回答资料里没有就说不知道”这一句能大幅降低幻觉。参考资料部分要给每条文档编号方便模型引用也方便前端做溯源展示。输出约束里可以要求模型“用简洁的中文回答不要重复问题”如果用的是带思考过程的模型还要在提示里明确“直接给出答案不要输出推理过程”否则前端会看到一堆思考文字。提示prompt 里的参考资料不要无脑全塞超过模型上下文窗口会被截断。一般 3 到 5 条、每条 300 字以内比较稳妥具体看模型支持的上下文长度。3.3 SSE 流式接口的封装后端用 FastAPI 实现 SSE核心是返回一个StreamingResponse媒体类型设为text/event-stream。生成器函数里每拿到一个 token 就 yield 一行data: xxx\n\n格式的数据。这里要注意两点一是每个事件必须以两个换行结尾否则前端解析不出来二是要在流结束时发送一个特殊标记比如data: [DONE]\n\n让前端知道可以关闭连接了。from fastapi.responses import StreamingResponse async def sse_generator(question: str): async for token in graph.astream({question: question}): if generate in token: content token[generate].get(token, ) if content: yield fdata: {json.dumps({text: content})}\n\n yield data: [DONE]\n\n app.get(/chat/stream) async def chat_stream(q: str): return StreamingResponse( sse_generator(q), media_typetext/event-stream, headers{Cache-Control: no-cache, X-Accel-Buffering: no} )那个X-Accel-Buffering: no响应头很关键它告诉 Nginx 不要缓冲这个响应否则流式效果会被 Nginx 攒着一起发用户看到的还是“一次性蹦出来”。3.4 状态在节点间的传递LangGraph 的状态是一个字典节点函数接收状态、返回要更新的字段。我定义的状态包含question、documents、answer、has_context几个键。retrieve节点写入documents和has_context条件边根据has_context决定走generate还是fallbackgenerate节点写入answer。这里有个实践技巧状态里不要塞大对象。比如不要把整个 Milvus 返回结果塞进去只提取需要的文本字段。状态越大图执行时的序列化开销越大流式场景下会明显感觉到首字延迟变高。4. 实操过程与核心环节实现4.1 环境准备与依赖安装先把基础环境搭起来。Python 建议 3.10 以上依赖主要包括langgraph、langchain、pymilvus、fastapi、uvicorn、ollama的 Python 客户端。Milvus 用 Docker 起 standaloneOllama 装好后拉一个中文能力还行的模型比如 qwen 系列。# 启动 Milvus standalone docker compose -f milvus-standalone-docker-compose.yml up -d # 拉取模型提前配好镜像源或离线导入 ollama pull qwen2.5:7b # 安装 Python 依赖 pip install langgraph langchain pymilvus fastapi uvicorn ollamaMilvus 起来后默认端口是 19530用pymilvus连接时填http://localhost:19530。如果你在 Mac 上用 Docker注意 Docker 的网络模式和宿主机不完全互通容器内访问宿主机服务要用host.docker.internal。4.2 构建 LangGraph 问答图图的构建分三步定义状态、注册节点、连边。我用StateGraph来组织条件边用add_conditional_edges实现分支。from langgraph.graph import StateGraph, END from typing import TypedDict, List class QAState(TypedDict): question: str documents: List[str] answer: str has_context: bool def build_graph(): graph StateGraph(QAState) graph.add_node(retrieve, retrieve_node) graph.add_node(generate, generate_node) graph.add_node(fallback, fallback_node) graph.set_entry_point(retrieve) graph.add_conditional_edges( retrieve, lambda s: generate if s[has_context] else fallback, {generate: generate, fallback: fallback} ) graph.add_edge(generate, END) graph.add_edge(fallback, END) return graph.compile()编译后的图对象支持invoke同步和astream异步流式两种调用方式。SSE 场景必须用astream它会在每个节点产出时 yield 中间状态我们从中提取生成节点的 token。4.3 生成节点的流式实现生成节点要调用 Ollama 的流式接口把 token 逐个吐出来。Ollama 的 Python 客户端支持streamTrue返回一个生成器。我在节点里遍历这个生成器把每个 token 累积到完整答案里同时通过 LangGraph 的流式机制往外传。import ollama def generate_node(state): context \n\n.join( f[{i1}] {doc} for i, doc in enumerate(state[documents]) ) prompt f你是企业知识助手只依据以下资料回答。 资料 {context} 问题{state[question]} 要求直接给出答案不要输出推理过程。资料中没有的信息就说不知道。 full_answer stream ollama.chat( modelqwen2.5:7b, messages[{role: user, content: prompt}], streamTrue ) for chunk in stream: token chunk[message][content] full_answer token # 通过自定义机制把 token 推给 SSE 生成器 yield_token(token) return {answer: full_answer}这里有个工程细节LangGraph 节点函数默认是同步的而 SSE 生成器是异步的两者要打通需要一点技巧。我的做法是用一个队列做中转节点里把 token 放进队列SSE 生成器从队列里取。也可以用asyncio把节点改成异步但要注意 Ollama 客户端的异步支持情况。4.4 前端 SSE 接收与渲染Vue 前端用EventSource接收流。注意EventSource只能发 GET 请求所以问题参数要拼在 URL 上。接收到的每条消息是 JSON 字符串解析后把text字段追加到答案变量里Vue 的响应式会自动更新视图。const startChat (question) { const url /chat/stream?q${encodeURIComponent(question)} const es new EventSource(url) answer.value es.onmessage (event) { if (event.data [DONE]) { es.close() return } const data JSON.parse(event.data) answer.value data.text } es.onerror () { es.close() // 触发重连或提示用户 } }实测下来这套前端渲染的体验很顺用户能看到答案逐字出现首字延迟通常在 1 到 2 秒取决于检索和模型加载速度。5. 常见问题与排查技巧实录5.1 检索召回为空或全是无关内容这是最高频的问题。排查顺序是先确认查询向量和入库向量是否同模型再检查 Milvus 的metric_type是否一致最后看阈值是不是设太高。我遇到过一次召回全空最后发现是入库时用了归一化向量、查询时没归一化余弦值算出来全偏低。解决办法是统一在 embedding 后做归一化。另一个隐蔽问题是集合的索引类型。Milvus 默认可能用 FLAT 或 IVF 索引不同索引对召回率有影响。小数据量用 FLAT 最准大数据量用 IVF 要调nprobe参数值越大越准但越慢。5.2 SSE 连接中途断开stream disconnected before completion这个报错基本就是超时问题。排查清单如下排查点检查方法解决方式Nginx 读超时看 nginx.conf 的 proxy_read_timeout调到 300s 以上Nginx 缓冲检查是否配置了 proxy_buffering设为 off后端心跳看生成器是否定期发注释行每 15s 发一次: keepalive客户端超时看 EventSource 是否被浏览器掐断加自动重连逻辑我一般会在 SSE 生成器里加一个心跳机制即使模型还没开始输出也定期发一个空注释保持连接。这个技巧在模型冷启动慢的时候特别有用。5.3 Ollama 生成速度慢生成慢的原因可能是模型太大、显存不够、或者没走 GPU。先确认 Ollama 是否识别到了 GPU用ollama ps看模型加载情况。如果显存不足模型会部分跑在 CPU 上速度断崖式下降。7B 模型量化后大概需要 6 到 8G 显存13B 需要 12G 以上。另一个提速手段是控制输出长度。在 prompt 里明确要求“回答不超过 200 字”能显著减少生成时间。流式场景下用户对首字延迟敏感、对总时长相对宽容所以优先优化首字延迟比如把检索和模型预热并行做。5.4 多轮对话上下文丢失第一次闭环通常只处理单轮问答但用户很快会问“那第二点呢”这种依赖上下文的问题。解决办法是在状态里加一个history字段把前几轮的问答对带上。但要注意历史不能无限增长一般保留最近 3 轮即可否则 prompt 会超长。LangGraph 天然支持这种状态累积只要在状态定义里加字段、在节点里读写就行。我建议把历史做摘要压缩比如用一个小模型把前几轮浓缩成一句话既保留上下文又控制长度。5.5 排查问题的通用思路遇到链路问题我的排查顺序是“分段隔离”先单独测检索确认能召回再单独测生成确认模型能答最后测 SSE确认能流式推送。哪一段出问题就集中查那一段不要一上来就怀疑整个链路。日志要打全每个节点的输入输出都记下来出问题时能快速定位是哪个环节的数据不对。6. 链路观测与后续扩展方向第一次闭环跑通后别急着加功能先把观测做起来。我在每个节点里都加了耗时统计检索花了多少毫秒、生成首字延迟多少、总时长多少这些指标直接决定后续优化方向。如果检索占了大部分时间就优化索引和nprobe如果生成慢就换更小的模型或者加缓存。缓存是个被低估的优化点。相同或相似的问题在企业场景里重复率很高把“问题向量 答案”缓存起来命中时直接返回能省掉整条链路的开销。Milvus 本身也能当缓存用把历史问答对存进去检索时先查缓存集合。后续扩展可以往几个方向走加一个重排节点提升召回精度加一个查询改写节点处理口语化提问加一个引用溯源节点把答案和原文片段对应起来。这些都是在现有图结构上加节点、加边的事LangGraph 的扩展性在这里体现得很明显。我个人在实际操作中的体会是第一次闭环不要追求完美能跑通、能观测、能排错就是胜利。很多团队卡在“想一次做对”结果迟迟上不了线。先把主链路打通让真实用户用起来再根据反馈迭代这才是企业级系统落地的正常节奏。