企业官网建设流程全解析

1. 检索主链路到底在解决什么问题

很多人做RAG项目，前面文档解析、切块、向量化都跑通了，一到"用户提问→系统回答"这条链路就卡住。要么是检索回来的内容跟问题八竿子打不着，要么是拼出来的Prompt超了模型上下文窗口，要么是前端等半天一个字都不吐。这一章要干的事，就是把这条链路从头到尾串起来，让它真正跑通一次完整的问答闭环。

所谓"检索主链路"，说白了就是四个动作的串联：用户提问进来 → 检索相关文档片段 → 组装上下文喂给模型 → 流式返回答案。听起来简单，但每一步都有坑。检索环节涉及向量召回和关键词召回的配合，上下文组装涉及预算分配和父窗口回填，流式返回涉及SSE协议和断连处理。这四个环节任何一个出问题，整个闭环就断了。

我见过太多项目，检索用的是最朴素的向量相似度Top-K，结果用户问"上季度的销售数据"，召回来的全是"销售部门组织架构"这种语义相近但答非所问的片段。也见过上下文直接无脑拼接，把检索到的十个片段全塞进去，结果超出模型窗口，要么报错要么被截断，答案缺胳膊少腿。还有SSE流式接口写完，本地测试好好的，一上生产环境就频繁断连，前端收到一半就没了。

这一章的目标很明确：让第一次问答闭环真正跑起来，并且跑得稳。我会把LangGraph的图结构怎么设计、检索节点怎么写、上下文预算怎么算、父窗口回填怎么做、SSE流式接口怎么封装，全部拆开讲清楚。每个决策背后的理由我都会说明，每个参数我都会给出计算依据。你跟着走一遍，就能得到一个可运行、可调试、可扩展的检索主链路。

适合谁看？如果你已经完成了文档入库和向量化，正准备把问答接口串起来，这一章就是为你写的。如果你还在纠结切块策略，建议先回去把上一章的内容消化掉，因为检索主链路的效果很大程度上取决于切块质量。如果你已经有一套能跑的RAG系统，但效果不稳定、流式输出老出问题，这一章里的排查思路和参数调优经验同样适用。

2. LangGraph图结构的设计取舍

2.1 为什么不用简单的函数调用链

最朴素的RAG实现就是一个函数串到底：retrieve(query) → build_prompt(docs, query) → call_llm(prompt) → return。这种写法在Demo阶段没问题，但一旦要加多路召回、重排序、上下文压缩、多轮对话状态管理，函数链就会变成一团乱麻。你没法在中途插入条件分支，没法做并行检索，没法在某个节点失败时优雅降级。

LangGraph的价值在于把这条链路显式地建模成一张有向图。每个节点是一个处理单元，每条边定义了流转方向，状态在节点之间传递。这样做的好处是：检索和生成解耦了，你可以在检索节点后面挂一个重排序节点，也可以并行跑向量召回和关键词召回再合并，还可以在生成节点前加一个上下文预算检查节点，超预算就触发压缩或截断。

我实际项目里的图结构是这样的：入口节点接收用户query，然后分两条边并行走到向量检索节点和关键词检索节点，两个节点的结果汇合到融合排序节点，排序后的文档进入上下文组装节点，组装时做预算检查和父窗口回填，最后进入生成节点调用LLM并流式输出。整张图的状态用一个TypedDict管理，包含query、vector_docs、keyword_docs、merged_docs、context、answer等字段。

注意：LangGraph的节点函数必须是纯函数或者只通过状态通信，不要在节点里直接操作全局变量。我踩过这个坑，多个请求并发时状态串了，排查了半天才发现是节点里用了模块级变量。

2.2 状态字段的设计与并发安全

状态设计是LangGraph里最容易埋雷的地方。我的建议是：状态字段尽量扁平，避免嵌套过深。比如不要设计成{"retrieval": {"vector": [...], "keyword": [...]}}，而是直接{"vector_docs": [...], "keyword_docs": [...]}。扁平结构在节点间传递时更清晰，调试时打印状态也一目了然。

并发安全方面，LangGraph本身对状态更新是做了隔离的，每个请求有独立的状态副本。但如果你在节点里调用了外部服务（比如向量数据库客户端），要确保这个客户端是线程安全的或者每个请求独立创建。我用的Milvus客户端就是全局单例，但它的查询方法是线程安全的，所以没问题。如果你用的是某些非线程安全的SDK，建议在节点内部按需创建连接。

另一个容易忽略的点是状态字段的默认值。LangGraph在初始化状态时，如果你没给某个字段赋值，它可能是None。节点函数里如果直接对这个字段做操作，就会报NoneType错误。我的做法是在图的入口节点里把所有字段都初始化一遍，该给空列表的给空列表，该给空字符串的给空字符串。

from typing import TypedDict, List class RAGState(TypedDict): query: str vector_docs: List[dict] keyword_docs: List[dict] merged_docs: List[dict] context: str answer: str token_budget: int used_tokens: int

这个状态定义里，token_budget和used_tokens是给上下文预算控制用的，后面会详细讲。

2.3 条件边与降级路径

图结构里必须考虑失败降级。向量检索超时了怎么办？关键词检索返回空了怎么办？融合排序后文档数量不够怎么办？这些都需要条件边来处理。

我的做法是：在向量检索节点后面加一个条件判断，如果返回结果为空或超时，就走一条边到"仅关键词检索"的降级路径；如果两个检索都失败，就走"无检索直接生成"的兜底路径，同时给用户返回一个提示，说明当前检索服务不可用，答案可能不准确。

条件边的实现用LangGraph的add_conditional_edges方法，传入一个判断函数，根据状态里的字段决定走哪条边。判断函数要尽量简单，只做路由决策，不要在里面做复杂计算。

def route_after_vector_search(state: RAGState) -> str: if not state["vector_docs"]: return "keyword_only" return "merge" graph.add_conditional_edges( "vector_search", route_after_vector_search, {"keyword_only": "keyword_search", "merge": "merge_node"} )

这种设计的好处是，即使某一路检索挂了，整个问答链路还能继续跑，用户体验不会完全中断。生产环境里，可用性比完美答案更重要。

3. 向量召回与关键词召回的配合逻辑

3.1 纯向量检索的盲区在哪里

向量检索擅长语义匹配，用户问"怎么提升团队效率"，它能召回"团队协作方法""效率工具推荐"这类语义相近的文档。但它有个致命盲区：对精确匹配不敏感。用户问"OKR和KPI的区别"，向量检索可能召回一堆"绩效管理"的文档，但真正讲OKR和KPI对比的那篇可能排在后面。

另一个盲区是专有名词和缩写。比如用户问"SSE接口怎么封装"，向量模型可能把SSE理解成"服务器发送事件"的语义，但如果你文档里SSE是作为技术缩写出现的，向量检索的召回率会明显下降。还有数字、日期、版本号这类信息，向量检索基本无能为力。

我实测过一个案例：知识库里有篇文档标题是"Ch07 检索主链路"，用户问"第七章讲什么"，纯向量检索召回的却是"主链路设计原则"这种语义相近但章节不对的文档。后来加了关键词检索，用"Ch07"做精确匹配，才把正确的文档捞回来。

3.2 关键词检索的补位策略

关键词检索我用的是BM25算法，配合jieba分词做中文处理。BM25的核心思想是：一个词在文档中出现频率越高、在整个语料中出现频率越低，它的权重就越大。这正好补上了向量检索对精确匹配不敏感的短板。

具体实现上，我用Elasticsearch做关键词检索的存储和查询。文档入库时，除了存向量，也把原文和分词后的term存进去。查询时，对用户query做同样的分词，然后用BM25算分。这里有个细节：中文分词要考虑领域词典。比如"检索主链路"这个词，通用分词器可能切成"检索/主/链路"，但如果你把"检索主链路"加到自定义词典里，它就会作为一个整体term，匹配精度会高很多。

import jieba jieba.add_word("检索主链路") jieba.add_word("父窗口回填") jieba.add_word("上下文预算") def tokenize(text: str) -> List[str]: return list(jieba.cut(text))

自定义词典这个事，我建议在项目初期就建起来，把领域内的专有名词、产品名、技术缩写都加进去。后期文档多了再补，成本会高很多。

3.3 两路结果的融合排序

向量检索返回的是余弦相似度分数，关键词检索返回的是BM25分数，这两个分数不在一个量纲上，不能直接相加。我的做法是先各自归一化，再加权融合。

归一化用Min-Max归一化，把每路检索的分数映射到0到1之间。然后给两路分配权重，我默认是向量0.7、关键词0.3。这个权重不是拍脑袋定的，是根据实际效果调的。如果你的场景里用户提问偏口语化、语义化，向量权重要高一些；如果偏精确查询、术语查询，关键词权重要高一些。

融合的时候还要考虑去重。同一篇文档可能同时被两路检索召回，这时候要合并成一条，分数取加权后的最大值或者平均值。我用的是加权求和后再除以出现次数，避免同一文档因为被两路都召回而分数虚高。

我实际用的是加权求和，然后在融合排序节点后面加了一个可选的交叉编码重排节点。重排用的小模型是bge-reranker-base，对Top-20的文档做精排，取Top-5进入上下文组装。这一步能把最终答案的准确率提升10%到15%，代价是增加约200毫秒的延迟。如果你的场景对延迟敏感，可以跳过重排，直接取融合后的Top-5。

4. 上下文预算与父窗口回填的实操细节

4.1 上下文预算到底怎么算

上下文预算这个词听起来很玄，其实就是一个简单的算术题：模型的最大上下文窗口减去系统提示词占用的token数，再减去预留的输出token数，剩下的就是检索文档能用的token数。

以GPT-4o为例，最大上下文窗口是128K token。系统提示词我写的是约500 token，预留输出2000 token，那么检索文档可用的预算是128000 - 500 - 2000 = 125500 token。但实际使用中我不会把预算用满，因为token计数有误差，而且留一些余量能让模型有更多空间做推理。我通常把实际使用量控制在预算的80%左右。

token计数用tiktoken库，这是OpenAI官方的计数工具，准确度很高。但要注意，不同模型的tokenizer不一样，如果你用的是Claude或者国产模型，要用对应的计数工具。我项目里做了个抽象层，根据模型名称选择对应的计数器。

import tiktoken def count_tokens(text: str, model: str = "gpt-4o") -> int: encoding = tiktoken.encoding_for_model(model) return len(encoding.encode(text))

预算分配上，我的策略是：按文档相关性分数从高到低依次加入上下文，直到预算用尽。每加入一篇文档，就计算一次累计token数，超过预算就停止。这样能保证最相关的文档一定在上下文里。

4.2 父窗口回填解决碎片化问题

切块策略有个天然矛盾：块切得小，检索精度高，但上下文碎片化；块切得大，上下文完整，但检索精度下降。父窗口回填就是解决这个矛盾的。

具体做法是：检索时用小块做向量匹配，命中后不直接返回小块，而是返回这个小块所属的父块（通常是更大的段落或整节内容）。这样既保证了检索精度，又保证了上下文的完整性。

实现上，我在文档入库时给每个小块记录一个parent_id，父块单独存储。检索命中小块后，根据parent_id去取父块内容。父块的大小我控制在1000到1500 token之间，太小了上下文不够，太大了浪费预算。

提示：父窗口回填会增加一次数据库查询，如果父块和子块存在同一个集合里，查询会很快；如果分集合存储，要注意加索引。我一开始没加索引，回填时查一次要200毫秒，后来在parent_id上建了索引，降到5毫秒以内。

回填之后还要做去重。多个小块可能属于同一个父块，回填后会出现重复的父块内容。我的做法是用父块的ID做去重，同一个父块只保留一次，分数取该父块下所有命中子块中的最高分。

4.3 预算超限时的截断与压缩

即使做了预算控制，有时候单篇父块就超了预算，或者多篇父块加起来超了。这时候需要截断或压缩。

截断的策略是：从文档末尾开始截，因为大部分文档的核心信息在前面。但这不是绝对的，有些文档结论在后面，截断会丢关键信息。我的做法是给每篇文档算一个"信息密度"分数，密度高的文档优先保留完整，密度低的文档从末尾截断。

压缩的策略是用一个小模型对超长文档做摘要，把摘要放进上下文。这个方案效果好但增加延迟和成本。我一般只在预算超限严重（超过20%）时才启用压缩，轻微超限直接截断。

def truncate_to_budget(docs: List[dict], budget: int) -> List[dict]: result = [] used = 0 for doc in sorted(docs, key=lambda x: x["score"], reverse=True): doc_tokens = count_tokens(doc["content"]) if used + doc_tokens <= budget: result.append(doc) used += doc_tokens else: remaining = budget - used if remaining > 100: doc["content"] = truncate_text(doc["content"], remaining) result.append(doc) break return result

这段代码里有个细节：remaining > 100这个判断，意思是如果剩余预算不到100 token，就不加了，因为太短的片段对答案没帮助，反而可能干扰模型。

5. SSE流式接口的封装与断连处理

5.1 为什么选SSE而不是WebSocket

流式输出方案有SSE和WebSocket两种。SSE是单向的服务器推送，WebSocket是双向通信。RAG问答场景里，客户端只需要接收服务器推送的token流，不需要向服务器发消息，所以SSE足够了。

SSE的优势是：基于HTTP协议，不需要额外的协议升级，兼容性好，实现简单。浏览器原生支持EventSource，后端用FastAPI的StreamingResponse就能实现。WebSocket虽然功能更强，但要处理连接管理、心跳、重连，复杂度高很多。

我项目里用的是FastAPI + SSE。后端定义一个/chat/stream接口，返回StreamingResponse，媒体类型是text/event-stream。每次LLM生成一个token，就通过SSE推送给前端。

from fastapi import FastAPI from fastapi.responses import StreamingResponse app = FastAPI() async def event_generator(query: str): async for token in rag_pipeline.astream(query): yield f"data: {json.dumps({'token': token})}\n\n" yield "data: [DONE]\n\n" @app.get("/chat/stream") async def chat_stream(query: str): return StreamingResponse( event_generator(query), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "Connection": "keep-alive", "X-Accel-Buffering": "no" } )

这里有几个关键点：Cache-Control: no-cache防止中间层缓存，X-Accel-Buffering: no防止Nginx缓冲（这个坑我踩过，不加这个头，Nginx会把整个响应缓冲完再一次性发给客户端，流式效果就没了），[DONE]标记用来告诉前端流结束了。

5.2 前端EventSource的封装

前端用EventSource接收SSE流。但EventSource有个限制：只支持GET请求，不支持POST。如果你的query很长，或者需要传复杂的参数，GET的URL长度可能不够。解决方案是用fetch + ReadableStream手动解析SSE流。

async function streamChat(query, onToken, onDone) { const response = await fetch('/chat/stream?query=' + encodeURIComponent(query)); 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(6); if (data === '[DONE]') { onDone(); return; } try { const parsed = JSON.parse(data); onToken(parsed.token); } catch (e) { console.warn('解析SSE数据失败', e); } } } } }

这段代码里，buffer的作用是处理跨chunk的数据。SSE的消息以\n\n分隔，但网络传输时一个消息可能被拆到两个chunk里。用buffer暂存不完整的部分，等下一个chunk到了再拼接，这是处理流式数据的标准做法。

5.3 断连排查与超时配置

SSE最常见的报错就是stream disconnected before completion: idle timeout waiting for sse。这个报错的意思是：连接空闲超时，服务器或中间层主动断开了连接。

排查思路分三层：客户端、中间层、服务端。

客户端层面，检查EventSource或fetch有没有设置超时。浏览器默认的fetch没有超时，但如果你用了axios之类的库，默认超时可能是60秒。LLM生成慢的时候，60秒可能不够。

中间层层面，Nginx默认的proxy_read_timeout是60秒，超过这个时间没有数据传输就会断开。解决方案是在Nginx配置里把这个值调大，比如300秒。同时要确保proxy_buffering off，否则Nginx会缓冲响应。

服务端层面，FastAPI的StreamingResponse本身没有超时限制，但如果你用了Uvicorn，它的--timeout-keep-alive参数默认是5秒，这个参数控制的是连接保持时间，不是流式传输时间，一般不用改。真正要关注的是LLM调用本身的超时设置。

location /chat/stream { proxy_pass http://backend; proxy_buffering off; proxy_cache off; proxy_read_timeout 300s; proxy_send_timeout 300s; chunked_transfer_encoding on; }

除了超时，还要处理心跳。如果LLM生成过程中有较长的停顿（比如检索阶段耗时较长），中间层可能误判为空闲连接而断开。解决方案是定期发送注释行作为心跳。

async def event_generator(query: str): yield ": heartbeat\n\n" async for token in rag_pipeline.astream(query): yield f"data: {json.dumps({'token': token})}\n\n" yield "data: [DONE]\n\n"

以冒号开头的行是SSE的注释，客户端会忽略，但能保持连接活跃。我一般每15秒发一次心跳，这个间隔要小于中间层的超时时间。

6. 第一次闭环跑通后的验证与调优

6.1 怎么判断闭环真的通了

闭环跑通的标志不是"接口返回了200"，而是从提问到答案的完整链路每个环节都有输出且合理。我一般用三个检查点来验证：

第一个检查点是检索结果。打印出向量检索和关键词检索各自返回的Top-5文档，人工看一眼是否跟问题相关。如果检索结果明显不相关，后面的生成再流畅也没用。

第二个检查点是上下文组装。打印出最终拼进Prompt的上下文内容，检查是否包含了回答问题的关键信息，token数是否在预算内，父窗口回填是否生效。

第三个检查点是流式输出。用curl或者浏览器开发者工具看SSE流，确认token是逐个到达的，不是一次性到达的。如果是一次性到达，说明中间层有缓冲。

curl -N "http://localhost:8000/chat/stream?query=检索主链路是什么"

-N参数关闭curl的缓冲，能实时看到SSE输出。如果输出是逐行出现的，说明流式没问题。

6.2 检索效果不好的常见原因

闭环跑通后，最常见的问题就是"检索效果不好"。我总结了几类原因和对应的排查方法：

切块粒度问题。块太大，检索精度低；块太小，上下文碎片化。排查方法是打印命中块的原文，看是否包含完整语义。如果块被切断了句子，说明切块策略有问题。

嵌入模型不匹配。中文场景用英文嵌入模型，效果会打折扣。排查方法是拿几个典型query，看Top-1的相似度分数。如果普遍低于0.7，可能是模型不匹配。

查询改写缺失。用户提问往往很口语化，直接拿去做向量检索效果不好。加一个查询改写节点，用LLM把口语化问题改写成检索友好的形式，能明显提升召回率。

元数据过滤缺失。如果知识库有分类、时间等元数据，检索时应该加过滤条件。比如用户问"最新的政策"，应该按时间倒序过滤，而不是纯向量检索。

6.3 延迟优化的几个着力点

检索主链路的延迟主要来自四块：向量检索、关键词检索、重排序、LLM生成。前三个是检索阶段，通常占总延迟的20%到30%；LLM生成占大头。

向量检索的优化：用HNSW索引代替IVF，查询速度能快3到5倍，代价是内存占用增加。如果向量库支持GPU加速，开启GPU能再快一个数量级。

关键词检索的优化：Elasticsearch的BM25查询本身很快，瓶颈通常在分词。把分词结果缓存起来，同样的query不用重复分词。

重排序的优化：交叉编码重排是延迟大户。如果延迟敏感，可以只对Top-10做重排，或者用更小的重排模型。我实测bge-reranker-base在CPU上跑Top-20要300毫秒，换成bge-reranker-small能降到100毫秒以内，精度损失约3%。

LLM生成的优化：用流式输出本身就是一种延迟优化，用户感知到的首token时间比完整响应时间重要得多。另外，Prompt尽量精简，系统提示词不要写太长，能减少首token延迟。

提示：我习惯在检索主链路的每个节点加耗时打点，用Python的time.perf_counter()记录每个节点的开始和结束时间，最后汇总打印。这样一眼就能看出瓶颈在哪个环节。生产环境可以用OpenTelemetry做分布式追踪，但开发阶段打点足够了。

7. 几个容易忽略的工程细节

7.1 空检索结果的兜底话术

检索返回空结果时，不要让LLM硬编。我见过有的实现，检索为空还是把空上下文拼进Prompt，结果LLM开始胡编。正确的做法是：检索为空时，直接返回预设的兜底话术，比如"抱歉，我没有找到相关信息，您可以换个说法再问一次"。

如果一定要让LLM参与，也要在Prompt里明确说明"如果没有找到相关信息，请直接告知用户，不要编造"。但实测下来，直接返回兜底话术更可控。

7.2 多轮对话的状态传递

第一次闭环通常只处理单轮问答。如果要支持多轮，需要把历史对话也纳入状态管理。我的做法是在状态里加一个history字段，存最近N轮的问答对。检索时，把当前query和上一轮query拼接起来做检索，能提升指代消解的效果。

但history不能无限增长，否则会挤占上下文预算。我一般只保留最近3轮，超过的截断。另外，history里的内容也要计入token预算。

7.3 日志与可观测性

检索主链路的日志要记录：query原文、改写后的query、向量检索Top-5的文档ID和分数、关键词检索Top-5的文档ID和分数、融合后的Top-5、最终上下文的token数、LLM的首token延迟和总延迟。这些数据是后续调优的依据。

我用的日志格式是JSON Lines，每行一个JSON对象，方便用jq或者ELK分析。关键字段加索引，方便按query或者文档ID检索。

import logging import json logger = logging.getLogger("rag_pipeline") def log_retrieval(query, vector_docs, keyword_docs, merged_docs): logger.info(json.dumps({ "event": "retrieval", "query": query, "vector_top5": [d["id"] for d in vector_docs[:5]], "keyword_top5": [d["id"] for d in keyword_docs[:5]], "merged_top5": [d["id"] for d in merged_docs[:5]], "vector_scores": [d["score"] for d in vector_docs[:5]], "keyword_scores": [d["score"] for d in keyword_docs[:5]] }, ensure_ascii=False))

这套日志跑一段时间后，你就能看出哪些query的检索效果差，针对性地优化。比如发现某类query的向量分数普遍偏低，可能是嵌入模型对这类语义不敏感，考虑换模型或者加查询改写。

7.4 配置外置与热更新

检索主链路的参数很多：Top-K、融合权重、预算大小、超时时间、重排开关。这些参数不要硬编码在代码里，要外置到配置文件或者配置中心。我用的YAML配置文件，启动时加载，支持通过接口热更新。

热更新这个功能在调参阶段特别有用。改一个权重不用重启服务，调完立即生效，效率高很多。但要注意热更新时的并发安全，用读写锁保护配置对象。

retrieval: vector_top_k: 20 keyword_top_k: 20 fusion_weights: vector: 0.7 keyword: 0.3 rerank: enabled: true top_n: 10 context: max_tokens: 8000 reserve_output: 2000 sse: heartbeat_interval: 15 timeout: 300

这份配置是我项目里的实际配置，你可以根据自己的场景调整。max_tokens设8000是因为我用的模型上下文窗口是16K，留一半给输出和系统提示词。如果你的模型窗口更大，可以相应调大。

8. 从闭环到可用的最后一步

第一次闭环跑通，意味着你的RAG系统从"能检索"进化到了"能问答"。但这离"好用"还有距离。我建议在闭环跑通后，立刻做一轮端到端测试：准备20到30个典型问题，覆盖事实查询、对比查询、多跳查询、否定查询等类型，人工评估每个问题的答案质量。

评估维度包括：检索是否召回了正确文档、答案是否准确、是否有编造、流式输出是否流畅。把不达标的case记录下来，逐个分析原因。大部分问题都能归到前面讲的几类：切块、嵌入模型、查询改写、预算、重排。

我自己的经验是，第一轮端到端测试通常有30%到40%的case不达标。经过两到三轮调优，能降到10%以内。剩下的10%往往是知识库本身缺失的内容，或者问题本身有歧义，这类case不用强求。

最后分享一个我踩过的坑：不要在检索主链路里做太多事。我一开始把查询改写、多路召回、重排、压缩全塞在一个图里，结果调试极其困难，一个环节出问题整条链路都挂。后来拆成了两个图：检索图负责召回和排序，生成图负责上下文组装和LLM调用。两个图通过状态传递衔接，调试时可以单独跑检索图看召回效果，单独跑生成图看生成质量。这个拆分让排查效率提升了很多。

检索主链路是整个RAG系统的骨架，骨架搭好了，后面加缓存、加多轮、加权限控制都是在这个骨架上挂东西。第一次闭环不用追求完美，先让它跑起来，再逐步优化。跑通比完美重要。