1. 检索主链路到底在解决什么问题很多人做智能问答系统前面文档解析、切块、向量化都跑通了一到“用户提问到返回答案”这条链路就卡住。要么是检索出来的内容太碎模型拿到一堆断句拼不出完整语义要么是流式输出到一半连接断了前端一直转圈要么是答案看着像那么回事但根本不知道是从哪份文档哪一页来的业务方不敢用。这一章要干的事就是把检索主链路真正串起来完成第一次问答闭环。所谓闭环不是“能返回一段文字”就算完而是从用户输入问题开始经过查询改写、向量召回、父窗口回填、重排、拼装上下文、调用大模型、SSE流式推送到前端、最后带上出处标注整条链路一个环节都不能少而且要能稳定跑通、能排查、能复现。我先把这条链路的整体骨架摆出来后面再逐段拆用户提问 → 查询预处理改写/扩展向量检索召回子块 → 通过父子映射回填父窗口重排Rerank取TopN → 拼装上下文Prompt调用大模型 → SSE流式返回前端逐token渲染 → 附带出处标注这套东西适合谁参考如果你正在做企业知识库问答、客服机器人、内部文档助手或者你已经在用LangChain、LlamaIndex这类框架但总觉得“检索结果不对劲”那这篇内容基本能对上你的痛点。哪怕你用的是自研链路思路也是通用的。我踩过的最大一个坑就是一开始只做了子块检索没做父窗口回填结果模型拿到的上下文全是半截话回答质量惨不忍睹。后来加上父窗口回填同样的向量库、同样的模型答案完整度直接上了一个台阶。所以这条链路里父窗口回填不是可选项是必选项。2. 检索主链路的核心设计与选型考量2.1 为什么是“子块检索 父窗口回填”而不是直接检索大块这是整个检索链路里最关键的架构决策我展开讲。最朴素的做法是把文档按固定长度切块比如每500字一块然后直接拿这些块做向量检索。问题是块切得小语义完整度差块切得大向量表征会被稀释召回精度下降。这是一个两难。我试过几种方案对比方案召回精度上下文完整度实现复杂度直接检索大块1000字低高低直接检索小块200字高低低子块检索父窗口回填高高中子块检索父窗口回填的思路是用小块子块做向量索引保证召回精度每个子块记录它所属的父块ID召回子块后根据父块ID把整个父块内容取出来送给模型。这样既保住了检索精度又保住了上下文完整度。具体参数上我的经验值是子块200-300字父块800-1200字父块之间可以有重叠。子块负责“被搜到”父块负责“被读懂”。注意父窗口回填时一定要去重。多个子块可能属于同一个父块回填时要按父块ID去重否则上下文里会出现重复内容浪费token还干扰模型。2.2 SSE流式输出为什么是必选项企业级问答系统里用户等10秒才看到一整段答案体验是很差的。SSEServer-Sent Events能让模型生成一个token就推一个token前端逐字渲染用户感知到的响应时间从“10秒”变成“1秒内开始出字”。SSE的本质是HTTP长连接上的单向事件流服务端持续往客户端推data:格式的消息。相比WebSocketSSE更轻不需要额外协议升级浏览器原生EventSource就能接。对于问答这种“服务端推、客户端收”的单向场景SSE是更合适的选择。但SSE有个绕不开的问题idle timeout。如果模型生成慢或者中间某个环节卡住连接可能因为空闲超时被断开前端就会报stream disconnected before completion: idle timeout waiting for sse。这个问题的排查和处理我在第4章会详细讲。2.3 出处标注为什么不能省企业场景下答案的可信度和可追溯性和答案本身一样重要。业务方看到一段回答第一反应是“这是从哪来的”。如果系统不能给出处这段回答就没法被采信。出处标注的实现方式是在拼装上下文时给每个父块打上来源标记文档名、页码、段落ID模型生成答案时被要求引用这些标记前端再把标记渲染成可点击的引用链接。这里的关键是上下文里必须带出处信息否则模型无从引用。3. 核心环节拆解与实操要点3.1 查询预处理别把用户原话直接丢给向量库用户提问往往是口语化的、有歧义的、甚至带错别字的。直接拿原话去检索召回质量很不稳定。我一般会做两层处理第一层是查询改写。用一个小模型或者规则把用户问题改写成更适合检索的形式。比如用户问“报销流程是啥”改写成“费用报销流程 步骤 所需材料”。这一步不需要太复杂哪怕只是做同义词扩展和关键词提取效果也很明显。第二层是多路召回。同一问题用不同表述分别检索结果合并去重。比如原问题检索一路改写后的问题检索一路关键词BM25检索一路。多路结果做RRFReciprocal Rank Fusion融合比单路召回稳得多。实操上查询改写可以用便宜的小模型做成本很低。我实测下来加了查询改写之后Top5召回命中率大概能提升15-20个百分点。3.2 父窗口回填的实现细节父窗口回填的逻辑不复杂但有几个细节容易翻车。数据结构上每个子块需要存child_id、parent_id、child_text、vector。父块单独存一张表parent_id、parent_text、source_info文档名、页码等。检索流程是用查询向量在子块索引里搜TopK比如Top20拿到子块的parent_id列表按parent_id去重根据parent_id取父块内容按子块相似度分数给父块排序一个父块下多个子块命中取最高分取TopN父块比如Top5拼装上下文# 父窗口回填核心逻辑示意 def retrieve_with_parent_backfill(query_vector, top_k20, top_n5): # 1. 子块检索 child_hits vector_store.search(query_vector, top_ktop_k) # 2. 按parent_id聚合取最高分 parent_scores {} for hit in child_hits: pid hit.metadata[parent_id] if pid not in parent_scores or hit.score parent_scores[pid]: parent_scores[pid] hit.score # 3. 按分数排序取TopN sorted_parents sorted(parent_scores.items(), keylambda x: x[1], reverseTrue)[:top_n] # 4. 回填父块内容 results [] for pid, score in sorted_parents: parent parent_store.get(pid) results.append({ text: parent[text], source: parent[source_info], score: score }) return results注意TopK和TopN的比例关系要调。TopK太小父块覆盖不够TopK太大检索延迟上去了。我一般用TopK20、TopN5起步再根据实际效果微调。3.3 上下文拼装与出处标注的配合拼装上下文时每个父块前面加上出处标记格式大概是[来源1: 员工手册.pdf 第3页] 父块内容... [来源2: 报销制度.docx 第1页] 父块内容...然后在System Prompt里明确要求模型引用来源编号。比如“回答时请引用相关来源编号格式为[来源N]。”这样模型生成的答案里就会自然带上引用标记前端解析后渲染成可点击的引用。实测下来只要Prompt写清楚模型引用来源的准确率能到90%以上。3.4 SSE流式接口的封装要点SSE接口封装有几个关键点响应头设置。必须设置Content-Type: text/event-stream、Cache-Control: no-cache、Connection: keep-alive。少一个都可能导致流式不生效。心跳机制。为了防止idle timeout服务端要定期发送注释行以:开头作为心跳保持连接活跃。间隔一般设15-30秒。错误处理。流式过程中如果模型调用失败要发送一个错误事件前端收到后停止渲染并提示用户而不是让连接一直挂着。结束标记。流结束时发送一个[DONE]事件前端收到后关闭连接。# SSE接口封装示意Python FastAPI风格 from fastapi.responses import StreamingResponse import asyncio async def sse_generator(query: str): try: # 检索 contexts retrieve_with_parent_backfill(encode(query)) prompt build_prompt(query, contexts) # 流式调用大模型 async for token in llm.stream(prompt): yield fdata: {json.dumps({token: token})}\n\n # 发送出处信息 sources [c[source] for c in contexts] yield fdata: {json.dumps({sources: sources})}\n\n # 结束标记 yield data: [DONE]\n\n except Exception as e: yield fdata: {json.dumps({error: str(e)})}\n\n app.get(/chat/stream) async def chat_stream(query: str): return StreamingResponse( sse_generator(query), media_typetext/event-stream, headers{ Cache-Control: no-cache, Connection: keep-alive, X-Accel-Buffering: no # 防止Nginx缓冲 } )注意如果前面有Nginx一定要关掉代理缓冲proxy_buffering off否则SSE消息会被攒着一起发流式效果就没了。这个坑我踩过排查了半天才发现是Nginx的问题。4. 完整实操流程从提问到答案落地4.1 环境与依赖准备先把基础环境列一下这套链路我用的技术栈是向量库Milvus或Qdrant都行Milvus生态更全Qdrant更轻嵌入模型BGE-M3或类似的中文嵌入模型重排模型BGE-Reranker大模型任意支持流式输出的模型后端Python FastAPI前端Vue3 EventSource依赖安装不展开重点说配置。向量库的索引参数里nlist和nprobe对召回影响很大。nlist是聚类数一般设4*sqrt(向量总数)nprobe是搜索时探查的聚类数设nlist的10%-20%。这两个参数直接决定检索速度和召回率的平衡。4.2 检索链路的完整代码走一遍我把检索主链路的核心代码串起来走一遍方便你直接参考。class RetrievalPipeline: def __init__(self, vector_store, parent_store, embedder, reranker): self.vector_store vector_store self.parent_store parent_store self.embedder embedder self.reranker reranker def search(self, query: str, top_k20, top_n5): # 1. 查询改写 rewritten self.rewrite_query(query) # 2. 多路召回 all_hits [] for q in [query, rewritten]: vec self.embedder.encode(q) hits self.vector_store.search(vec, top_ktop_k) all_hits.extend(hits) # 3. 去重 父窗口回填 parent_map {} for hit in all_hits: pid hit.metadata[parent_id] if pid not in parent_map or hit.score parent_map[pid][score]: parent_map[pid] {score: hit.score, parent_id: pid} # 4. 重排 candidates [] for pid, info in parent_map.items(): parent self.parent_store.get(pid) candidates.append({ text: parent[text], source: parent[source_info], score: info[score] }) reranked self.reranker.rerank(query, candidates, top_ntop_n) return reranked def rewrite_query(self, query: str) - str: # 简单实现关键词提取 同义词扩展 # 生产环境建议用小模型做 return query # 占位这段代码里重排这一步很关键。向量检索是粗筛重排是精排。BGE-Reranker这类交叉编码器会把query和每个候选父块一起编码算相关性分数比向量相似度准得多。加了重排之后Top5的准确率提升非常明显。4.3 SSE流式输出的前后端联调后端SSE接口写好后前端用EventSource接// Vue3中封装SSE调用 function useChatStream() { const answer ref() const sources ref([]) const loading ref(false) function ask(query) { answer.value sources.value [] loading.value true const es new EventSource(/chat/stream?query${encodeURIComponent(query)}) es.onmessage (event) { if (event.data [DONE]) { es.close() loading.value false return } const data JSON.parse(event.data) if (data.token) { answer.value data.token } if (data.sources) { sources.value data.sources } if (data.error) { es.close() loading.value false console.error(SSE error:, data.error) } } es.onerror (err) { es.close() loading.value false console.error(SSE connection error:, err) } } return { answer, sources, loading, ask } }联调时最容易出的问题是跨域和缓冲。跨域要在后端加CORS头缓冲要检查Nginx和中间件的配置。这两个问题解决后流式输出基本就稳了。4.4 出处标注的前端渲染后端在流结束时推送sources数组前端拿到后渲染成引用列表。如果模型在答案里带了[来源N]标记可以用正则替换成可点击的角标点击滚动到对应来源。// 把答案中的[来源N]替换成可点击角标 function renderAnswer(text, sources) { return text.replace(/\[来源(\d)\]/g, (match, num) { const idx parseInt(num) - 1 if (idx 0 idx sources.length) { return sup classcitation>async def sse_with_heartbeat(generator): while True: try: item await asyncio.wait_for(generator.__anext__(), timeout20) yield item except asyncio.TimeoutError: yield : heartbeat\n\n # 注释行前端忽略 except StopAsyncIteration: break注意心跳间隔要小于服务端和中间件的超时时间。如果Nginx的proxy_read_timeout是60秒心跳间隔设20秒就够安全。5.2 检索结果不相关的排查思路检索结果不相关按这个顺序排查看嵌入模型。嵌入模型和你的语料领域不匹配召回肯定差。中文场景建议用BGE系列或M3E系列。看切块策略。子块太大或太小都会影响召回。200-300字是比较稳的区间。看是否加了重排。没加重排的话向量相似度排序经常不准加重排能救回来很多。看查询改写。用户原话直接检索效果差加改写后明显改善。看TopK。TopK太小可能漏掉正确结果适当调大再重排。我一般会准备一组测试问题每次调整参数后跑一遍看Top5命中率的变化。没有量化指标调参就是瞎调。5.3 父窗口回填后上下文超长的处理父块回填后上下文长度可能超出模型窗口。处理方式有几种限制TopN。控制回填的父块数量比如最多5个。父块截断。如果单个父块太长按句子边界截断到合理长度。动态调整。根据模型窗口大小和问题复杂度动态决定回填几个父块。我一般用TopN5 单父块不超过1500字的组合大部分场景够用。如果模型窗口小就减到TopN3。5.4 模型不引用出处怎么办模型不引用出处通常是Prompt没写清楚。几个改进点System Prompt里明确要求引用并给出格式示例上下文里的出处标记要显眼用[来源N]这种明确格式可以在Few-shot里给一个带引用的示例如果还是不引用可以在后处理阶段用规则匹配把答案里和出处高度重合的句子自动关联实测下来Prompt写清楚 上下文标记规范引用率能到90%以上。剩下10%用后处理兜底。6. 一些实操心得和后续扩展方向这套检索主链路跑通后我最大的体会是闭环比完美重要。不要一开始就追求召回率99%、引用准确率100%先把“提问→检索→生成→流式返回→出处标注”这条链路完整跑通哪怕效果只有70分你也有了可观测、可迭代的基础。后面每一步优化都有明确的对比基准。另一个心得是日志要打全。检索链路的每个环节——查询改写结果、召回的子块ID和分数、回填的父块ID、重排后的排序、最终拼装的上下文、模型输出——都要打日志。出问题时你能快速定位是哪一环掉了链子。我吃过亏一开始日志打得不全出了问题只能靠猜效率极低。后续可以扩展的方向多轮对话的上下文管理、检索结果的时效性加权、基于用户反馈的检索调优、多知识库的路由。这些都是在闭环跑通之后自然延伸出来的需求。但前提是你得先有那个闭环。