1. 检索主链路到底在解决什么问题做企业级智能问答系统最怕的不是模型不够聪明而是模型“知道但找不到”。检索主链路就是把用户问题变成向量、去知识库里捞出最相关的片段、再交给大模型组织成答案的完整闭环。这一章的核心目标很明确让第一次问答真正跑通从用户敲下回车到看到流式输出的每一个环节都能被观测、被调试、被替换。我见过太多项目卡在“Demo能跑上线就崩”的阶段。原因往往不是LangGraph或Milvus本身有多难而是检索链路里的每个组件都被当成黑盒出了问题只能靠猜。所以这一章我会把链路拆到最细Ollama负责本地推理Milvus负责向量存储与检索LangGraph负责编排状态流转SSE负责把结果实时推给前端。四个组件各司其职任何一个环节的延迟或错误都会在最终答案里被放大。适合谁看如果你已经跑通过Ollama的基本对话也大致知道RAG是什么但还没把“检索→增强→生成”串成一条稳定的流水线这一章就是为你写的。如果你正在用FastAPI做后端、Vue做前端想搞清楚SSE流式接口怎么封装、Milvus的余弦相似度怎么调、LangGraph的节点怎么设计下面的内容可以直接抄作业。2. 整体架构设计与技术选型逻辑2.1 为什么是LangGraph而不是普通ChainLangChain的Chain适合线性流程但企业级问答系统往往需要条件分支检索到的文档相关性不够时要不要重试用户问题涉及多个子问题时要不要拆解这些场景用Chain写会变成一堆if-else嵌套而LangGraph把每个步骤抽象成节点用边定义流转条件状态在节点间传递和修改。我选LangGraph的另一个原因是可观测性。每个节点的输入输出都可以单独打日志出问题时能精确定位是检索节点召回率低还是生成节点提示词没写好。普通Chain一旦跑起来就是个黑盒调试成本高得多。2.2 Milvus standalone模式够不够用企业级不等于一上来就上集群。Milvus standalone模式在单机上能支撑百万级向量的检索对于大多数内部知识库场景完全够用。它的优势是部署简单用Docker一条命令就能起来数据持久化到本地磁盘重启不丢。真正需要分布式集群的场景是向量规模超过千万级、QPS要求上千、或者需要多租户隔离。如果只是给几十个内部员工用standalone模式配合合理的索引参数响应时间可以稳定在几十毫秒。我实测下来十万条向量用IVF_FLAT索引TopK5的检索耗时在15ms左右完全不会成为瓶颈。2.3 Ollama做本地推理的取舍Ollama最大的好处是离线可用、数据不出内网。企业知识库往往涉及内部文档用云端API会有合规风险。Ollama支持GGUF格式的量化模型7B参数的模型在16GB内存的机器上就能跑生成速度大约20-40 tokens/秒对于问答场景足够。但Ollama也有坑默认的模型存储路径在系统盘模型文件动辄几个GB很容易把盘撑满。另外国内下载模型速度慢需要配置镜像源或者手动导入离线包。这些细节后面会展开。2.4 SSE为什么比WebSocket更适合这个场景SSE是服务器单向推送WebSocket是双向通信。问答系统的交互模式是“用户问一次系统流式返回答案”不需要客户端在生成过程中持续发送数据。SSE基于HTTP协议实现简单浏览器原生支持EventSourceNginx配置也成熟。更重要的是SSE的断线重连机制。网络抖动导致连接断开时浏览器会自动重连配合服务端的事件ID可以做到续传。WebSocket需要自己实现心跳和重连逻辑复杂度高不少。当然SSE也有缺点只能传文本二进制数据需要Base64编码不过问答场景都是文本这个问题不存在。3. 核心细节解析与实操要点3.1 Ollama环境准备与模型拉取第一步是安装Ollama。Linux上直接用官方脚本curl -fsSL https://ollama.com/install.sh | sh安装完成后修改模型存储路径避免占满系统盘。编辑systemd服务文件sudo systemctl edit ollama.service加入环境变量[Service] EnvironmentOLLAMA_MODELS/data/ollama/models然后重载并重启sudo systemctl daemon-reload sudo systemctl restart ollama拉取模型时如果速度慢可以配置国内镜像源。设置环境变量OLLAMA_HOST指向镜像地址或者手动下载GGUF文件后用ollama create导入。我一般选qwen2.5:7b或llama3.1:8b中文场景下qwen表现更稳。注意Ollama默认监听127.0.0.1:11434如果后端服务和Ollama不在同一台机器需要设置OLLAMA_HOST0.0.0.0并配置防火墙规则。生产环境建议加一层Nginx做鉴权和限流。3.2 Milvus安装与集合设计用Docker安装Milvus standalonewget https://github.com/milvus-io/milvus/releases/download/v2.4.0/milvus-standalone-docker-compose.yml -O docker-compose.yml docker compose up -d启动后默认端口19530。Python端连接from pymilvus import connections, Collection, FieldSchema, CollectionSchema, DataType connections.connect(hostlocalhost, port19530)集合的Schema设计是关键。我一般用四个字段id主键自增、vector稠密向量维度跟Embedding模型对齐、text原始文本块、metadataJSON存来源、页码等。索引选IVF_FLATnlist设为128度量方式用COSINE。fields [ FieldSchema(nameid, dtypeDataType.INT64, is_primaryTrue, auto_idTrue), FieldSchema(namevector, dtypeDataType.FLOAT_VECTOR, dim1024), FieldSchema(nametext, dtypeDataType.VARCHAR, max_length65535), FieldSchema(namemetadata, dtypeDataType.JSON), ] schema CollectionSchema(fields, descriptionknowledge base) collection Collection(namekb_chunks, schemaschema) index_params {index_type: IVF_FLAT, metric_type: COSINE, params: {nlist: 128}} collection.create_index(field_namevector, index_paramsindex_params)提示Milvus的余弦相似度返回的是相似度分数越大越相似。如果用的是L2距离则是越小越相似。检索时metric_type必须和索引一致否则会报错。3.3 文本切分与向量化策略文本切分直接影响召回质量。我试过固定长度切分、按段落切分、按语义切分三种方案。固定长度最简单但容易切断句子按段落切分保留了语义完整性但块大小不均。最终我采用折中方案先按段落分超过500字的段落再按句号切分相邻块之间保留50字重叠。向量化用BGE-M3或text-embedding-3-small。BGE-M3支持多语言1024维本地部署免费。Ollama也支持Embedding模型ollama pull bge-m3调用方式import requests resp requests.post(http://localhost:11434/api/embeddings, json{ model: bge-m3, prompt: 待向量化的文本 }) vector resp.json()[embedding]注意Ollama的Embedding接口每次只处理一条文本批量向量化需要自己写循环并发。如果知识库有几十万条建议用专门的Embedding服务或者GPU加速。3.4 LangGraph状态图设计LangGraph的核心是StateGraph。定义状态类型from typing import TypedDict, List class QAState(TypedDict): question: str rewritten_query: str retrieved_docs: List[dict] answer: str sources: List[str]然后定义节点函数。检索节点负责调Milvus生成节点负责调Ollama重写节点负责把用户口语化的问题转成更适合检索的查询。from langgraph.graph import StateGraph, END def rewrite_node(state: QAState): # 调用LLM重写查询 return {rewritten_query: rewritten} def retrieve_node(state: QAState): # 调Milvus检索 return {retrieved_docs: docs} def generate_node(state: QAState): # 调Ollama生成答案 return {answer: answer, sources: sources} graph StateGraph(QAState) graph.add_node(rewrite, rewrite_node) graph.add_node(retrieve, retrieve_node) graph.add_node(generate, generate_node) graph.set_entry_point(rewrite) graph.add_edge(rewrite, retrieve) graph.add_edge(retrieve, generate) graph.add_edge(generate, END) app graph.compile()这个图看起来简单但每个节点内部都有讲究。重写节点不是必须的如果用户问题本身就很规范可以跳过。检索节点要处理Milvus连接失败、返回空结果等情况。生成节点要控制提示词长度避免超出模型上下文窗口。4. 实操过程与核心环节实现4.1 从零启动Milvus并验证连接先确认Docker环境正常docker --version docker compose version下载compose文件并启动mkdir -p /opt/milvus cd /opt/milvus wget https://github.com/milvus-io/milvus/releases/download/v2.4.0/milvus-standalone-docker-compose.yml -O docker-compose.yml docker compose up -d检查容器状态docker compose ps应该看到三个容器milvus-standalone、milvus-etcd、milvus-minio。等30秒左右让服务完全启动然后用Python验证from pymilvus import connections, utility connections.connect(hostlocalhost, port19530) print(utility.list_collections())如果返回空列表说明连接成功。如果报连接超时检查防火墙和Docker端口映射。4.2 知识库入库完整流程假设我们有一批Markdown格式的内部文档入库流程分四步读取文件、切分文本、向量化、写入Milvus。import os import glob from pymilvus import Collection collection Collection(kb_chunks) def chunk_text(text, max_len500, overlap50): paragraphs text.split(\n\n) chunks [] for para in paragraphs: if len(para) max_len: chunks.append(para) else: sentences para.split(。) current for sent in sentences: if len(current) len(sent) max_len: current sent 。 else: chunks.append(current) current sent 。 if current: chunks.append(current) return chunks def embed(text): resp requests.post(http://localhost:11434/api/embeddings, json{ model: bge-m3, prompt: text }) return resp.json()[embedding] for filepath in glob.glob(/data/docs/**/*.md, recursiveTrue): with open(filepath, r, encodingutf-8) as f: content f.read() chunks chunk_text(content) for chunk in chunks: vector embed(chunk) collection.insert([{ vector: vector, text: chunk, metadata: {source: filepath} }]) collection.flush()入库完成后建索引并加载collection.create_index(field_namevector, index_paramsindex_params) collection.load()实操心得入库时批量插入比逐条插入快很多。可以把chunk攒到100条再一次性insert减少网络往返。另外embedding接口如果并发太高会被Ollama限流建议用信号量控制并发数在4-8之间。4.3 检索节点实现与参数调优检索节点的核心是构造搜索向量并调Milvus的search接口def retrieve(query, top_k5, threshold0.5): query_vector embed(query) results collection.search( data[query_vector], anns_fieldvector, param{metric_type: COSINE, params: {nprobe: 16}}, limittop_k, output_fields[text, metadata] ) docs [] for hits in results: for hit in hits: if hit.score threshold: docs.append({ text: hit.entity.get(text), source: hit.entity.get(metadata)[source], score: hit.score }) return docsnprobe参数控制搜索的精度和速度。nlist128时nprobe16意味着扫描约1/8的聚类中心召回率在95%以上耗时增加不明显。如果对精度要求极高可以把nprobe调到32甚至64但延迟会线性增长。threshold是相似度阈值低于这个分数的文档直接丢弃。设太低会引入无关内容干扰生成设太高可能召回为空。我一般从0.5开始调根据实际效果微调。4.4 生成节点与提示词工程生成节点的提示词模板直接决定答案质量。我用的模板PROMPT_TEMPLATE 你是一个企业知识库助手。请根据以下参考资料回答用户问题。 如果参考资料中没有相关信息请明确说“根据现有资料无法回答”不要编造。 参考资料 {context} 用户问题{question} 回答要求 1. 只使用参考资料中的信息 2. 回答要简洁准确 3. 如果引用了具体内容标注来源文件名 调用Ollama生成def generate(question, docs): context \n\n.join([f[{d[source]}] {d[text]} for d in docs]) prompt PROMPT_TEMPLATE.format(contextcontext, questionquestion) resp requests.post(http://localhost:11434/api/generate, json{ model: qwen2.5:7b, prompt: prompt, stream: False, options: {temperature: 0.1, num_predict: 1024} }) return resp.json()[response]temperature设0.1是为了让答案更确定减少发挥。num_predict限制最大生成长度防止模型啰嗦。4.5 SSE流式接口封装FastAPI端实现SSEfrom fastapi import FastAPI from fastapi.responses import StreamingResponse import json app FastAPI() async def event_generator(question: str): # 先发一个事件告知开始 yield fevent: start\ndata: {json.dumps({status: retrieving})}\n\n docs retrieve(question) yield fevent: sources\ndata: {json.dumps({sources: [d[source] for d in docs]})}\n\n # 流式生成 context \n\n.join([d[text] for d in docs]) prompt PROMPT_TEMPLATE.format(contextcontext, questionquestion) with requests.post(http://localhost:11434/api/generate, json{ model: qwen2.5:7b, prompt: prompt, stream: True, options: {temperature: 0.1} }, streamTrue) as r: for line in r.iter_lines(): if line: chunk json.loads(line) if response in chunk: yield fevent: token\ndata: {json.dumps({text: chunk[response]})}\n\n yield fevent: done\ndata: {json.dumps({status: finished})}\n\n app.get(/api/qa) async def qa(question: str): return StreamingResponse( event_generator(question), media_typetext/event-stream, headers{Cache-Control: no-cache, X-Accel-Buffering: no} )前端Vue端用EventSource接收const es new EventSource(/api/qa?question${encodeURIComponent(q)}) es.addEventListener(token, (e) { const data JSON.parse(e.data) answer.value data.text }) es.addEventListener(done, () { es.close() }) es.onerror () { es.close() // 重连逻辑 }注意Nginx默认会缓冲SSE响应导致前端收不到实时数据。必须在location配置里加proxy_buffering off;和proxy_cache off;。另外X-Accel-Buffering: no响应头也能让Nginx不缓冲。5. 常见问题与排查技巧实录5.1 SSE连接中断问题最常见的报错是stream disconnected before completion: idle timeout waiting for sse。原因是Nginx或负载均衡器有默认的超时时间通常是60秒。如果生成时间超过这个值连接会被切断。解决方案有三层第一在Nginx配置里加大超时location /api/qa { proxy_pass http://backend; proxy_buffering off; proxy_cache off; proxy_read_timeout 300s; proxy_send_timeout 300s; chunked_transfer_encoding on; }第二在SSE流里定期发送心跳注释行保持连接活跃yield : heartbeat\n\n第三前端实现自动重连记录最后收到的事件ID重连时带上Last-Event-ID头服务端从断点继续。5.2 Milvus检索返回空结果检索为空通常有三个原因集合没加载、向量维度不匹配、相似度阈值设太高。先检查集合是否loadfrom pymilvus import utility print(utility.load_state(kb_chunks))如果是NotLoad状态执行collection.load()。向量维度不匹配会在search时报错检查Embedding模型输出维度和Schema里的dim是否一致。阈值问题最好排查把threshold临时设为0看是否有结果返回。还有一个隐蔽的坑Milvus的COSINE相似度范围是-1到1但实际文本向量相似度通常在0.3到0.9之间。如果threshold设0.8很多相关文档会被过滤掉。建议先用0.3试再逐步提高。5.3 Ollama生成速度慢7B模型在CPU上生成速度可能只有5-10 tokens/秒体验很差。优化方向换更小的模型3B、用量化版本Q4_K_M、上GPU。如果必须用CPU可以调整Ollama的并行参数OLLAMA_NUM_PARALLEL2 OLLAMA_NUM_THREADS8num_threads设为CPU物理核心数num_parallel控制同时处理的请求数。另外num_predict限制生成长度也能减少等待时间。5.4 常见问题速查表问题现象可能原因排查方法解决方案SSE连接60秒断开Nginx超时查看Nginx错误日志加大proxy_read_timeout检索结果为空集合未加载utility.load_statecollection.load()向量维度报错Embedding维度不匹配打印向量长度统一模型和Schema维度生成答案编造提示词约束不够检查prompt模板加强“不知道就说不知道”Ollama连接拒绝服务未启动或端口不对curl localhost:11434检查systemd状态和防火墙入库速度慢逐条插入观察insert耗时批量插入并发embedding5.5 独家避坑技巧第一个坑Milvus的collection.insert不会立即持久化必须调flush()才能被检索到。我一开始忘了flush查了半天以为索引没建好。第二个坑Ollama的/api/embeddings接口在模型未加载时会自动拉取但如果模型名写错会静默返回空向量。一定要检查返回的embedding长度是否等于预期维度。第三个坑LangGraph的节点函数如果抛异常整个图会中断。建议在每个节点内部用try-except包裹把错误信息写入state让流程能继续走到错误处理节点。第四个坑SSE的data字段如果包含换行符会被解析成多个data行。发送JSON时用json.dumps确保没有裸换行或者用data:前缀逐行发送。6. 检索链路的性能观测与迭代方向链路跑通只是第一步真正让它稳定服务需要持续观测。我在每个节点加了耗时打点重写耗时、检索耗时、生成首token耗时、生成总耗时。这些指标写入日志后用简单的脚本就能分析瓶颈在哪。实测下来检索耗时通常占10%以内生成耗时占80%以上。所以优化重点应该放在生成环节换更快的模型、用GPU加速、或者对常见问题做缓存。检索环节的优化空间有限除非向量规模特别大需要调索引参数。另一个迭代方向是混合检索。纯向量检索对关键词匹配不敏感比如用户搜“Ch07”这种编号向量模型可能召回不准。可以加一路BM25关键词检索两路结果用RRF融合。Milvus 2.4支持稀疏向量可以直接在同一个集合里做混合检索不需要额外维护ES。还有一个容易忽略的点是查询重写。用户问“那个部署文档在哪”直接向量化效果很差。先用LLM把问题重写成“Milvus部署文档 安装步骤”检索召回率会明显提升。LangGraph里加一个rewrite节点就能实现成本很低但收益很大。最后再分享一个小技巧Milvus的search接口支持output_fields指定返回字段只取需要的字段能减少网络传输。如果metadata很大可以只返回source不返回全文全文等生成时再按id查。这个优化在向量量大时效果明显。