直接切入正题。最近接手了一个内部知识库问答的项目需求很直接用户问一句话系统去企业文档库里检索相关内容再交给大模型组织语言回答。但问题就出在“回答”这一步——大模型的生成速度再快也得让人盯着“正在输入”的转圈等好几秒甚至十几秒。产品经理天天在我耳边念叨“要丝滑”后来我把Python异步编程和RAG流式回答结合起来用SSE把模型的输出一段段实时推给前端体验才真正有了质变。这篇文章就围绕这个项目复盘展开。你会看到RAG从文档加载、切块、向量化到检索、流式生成的全链路是怎么设计的也会看到Python异步编程在中间扮演什么角色——尤其是asyncio怎么解决“多个检索任务并发执行”“生成过程中客户端中途断开怎么截断”这些实际问题。适合正在做知识库问答、想把大模型回答做成流式输出、或者单纯想搞懂async/await怎么落地到真实项目的读者。1. 为什么RAG流式回答必须上异步1.1 同步调用带来的真实痛点假设你用一个同步接口去做RAG问答流程大概是先等文档检索结果再等大模型把所有token全部生成完最后把一整个字符串返回给前端。这段时间里客户端什么都拿不到只能转圈。如果检索花了1秒生成花了5秒用户就要傻等6秒。更麻烦的是后端线程池里的线程被占住了如果同时来几十个请求线程池直接被打满后面的请求排队等待接口响应时间雪上加霜。我最初用Flask写过一版同步接口压测时50个并发就把服务卡到几乎不可用。原因不复杂Flask默认的WSGI服务器是同步多线程一个worker同一时间处理一个请求IO操作查库、调模型都在阻塞线程。后来我把服务迁移到FastAPI配合async def路由才真正体会到异步IO的威力——IO等待期间事件循环腾出手来处理其他请求线程数不用开很多吞吐量反而翻了几倍。1.2 流式输出为什么天然是异步的活流式回答的本质是“边生成边发送”。大模型一次吐一个token或者几个token后端在拿到第一个token后就要立刻把它包装成事件推给前端。这个过程中检索、生成、网络传输都是典型的IO密集型操作线程在等待的时候干不了别的事只能用异步来把等待时间省出来。再往深一层说SSEServer-Sent Events协议本身就是长连接、单向推送。服务端要一直hold住连接一段一段写数据。这种长连接型的响应如果用同步线程处理server的线程资源非常容易被耗尽。异步生成器配StreamingResponse正是FastAPI对这种场景的标准解法事件循环里维护一个异步生成器生成一个块就yield一个块框架帮你把块以SSE格式刷到socket上整个过程不额外占用线程。说个生活化类比同步就是食堂阿姨每炒好一盘菜后面一排人全都等着不能同时招呼下一个。异步就是取餐号叫号阿姨炒好一份喊一个号码不用等所有人吃完再接待新客人。流式回答更像是把菜品分成了小吃、主食、甜点三道菜每做完一道就先端出来让顾客开吃而不是全做完了再一起上。1.3 异步不是银弹哪些地方仍然要小心异步编程解决的是IO密集问题不是CPU密集问题。在RAG链路里文档切块、向量化、重排序这些操作如果做的是大量本地计算尤其在CPU上跑embedding模型它们会阻塞事件循环。一个常用的补救办法是asyncio.to_thread把这些重计算丢到线程池里跑不阻塞事件循环主线。我实际项目中就是这么干的用sentence-transformers做本地embedding调用模型推理时放在to_thread里检索向量库用异步客户端最后大模型API调用走异步HTTP。这样事件循环里跑的都是真正异步的IO操作重计算全部在线程池里排队两者都不互相拖后腿。如果你不处理这种混合负载流式接口可能刚开始响应很快跑到中间突然卡一下就是因为阻塞计算把事件循环卡住了。2. RAG系统整体架构与关键选型2.1 RAG四大核心环节拆解RAG不是某一个大模型API而是一条完整的数据处理链路。我在项目里把它拆成四个环节每一环都有不同的技术和坑。第一环是文档加载与解析。企业知识库里什么格式都有PDF、Word、Markdown、数据库导出的文本。这一步要做的就是把非结构化数据变成纯文本。PDF解析是个重灾区扫描版PDF要OCR表格要还原结构处理不好后面检索质量直接崩。我建议先处理Markdown和HTML类文档因为它们自带标题和段落结构切块时能利用这些语义信息。第二环是文本切块Chunking。这是很多人忽略但影响极大的环节。大模型的上下文窗口有限你不能把整本手册一次性塞进去所以要把文档切成多个小块检索时只召回最相关的几块。切块有一个硬约束块大小和重叠度。块太小语义不完整块太大召回结果不够精确还浪费上下文空间。常用的策略是按标题、段落、句子边界切块大小在300-800个token之间带一定重叠。企业知识库场景里我习惯先按Markdown标题切出章节再在章节内做滑动窗口切块这样既保留章节语义又不会遗漏跨段信息。第三环是向量化与存储。把切好的文本块用embedding模型转成向量存进向量数据库。选向量库时我主要看三件事是否支持异步客户端、是否有本地免部署版本、是否支持元数据过滤。ChromaDB、Qdrant、Milvus都是常见选择。小项目可以直接上ChromaDB零配置进程内就能跑数据量上来以后可以考虑Qdrant或者Milvus。第四环是召回与重排。用户问题也做embedding在向量库里做近似搜索找出最相似的几个文本块。这里有个容易被忽略的点纯粹向量召回对关键词不敏感有些专业术语和缩写会导致召回不准。所以我现在都会加一层“混合检索”BM25关键词检索 向量检索再做融合排序。如果有精力还可以训练一个重排模型cross-encoder例如bge-reranker对候选结果二次排序。这个对答案质量的提升非常明显。2.2 技术栈选型FastAPI asyncio 流式响应基于上面的需求我最终选的技术栈是环节选型理由Web框架FastAPI原生支持async路由和StreamingResponse异步运行时asyncio uvicorn事件循环驱动支持高并发长连接向量存储ChromaDB本地开发/ Qdrant生产支持异步部署简单Embeddingsentence-transformers本地模型离线可用微调方便LLM调用各家云厂商的异步SDK或OpenAI兼容接口异步流式天然支持流式协议SSE (text/event-stream)单向文本推送足够满足需求在这个技术栈里最核心的选型逻辑是FastAPI把HTTP层搞定asyncio把并发搞定StreamingResponse把流式搞定。三者配合就用很少的代码串起了从收到请求到推送完整答案的全过程。2.3 RAG和MCP、Agentic RAG的区别这个项目做完以后经常有朋友来问我RAG和MCP到底是什么关系还有现在流行说的Agentic RAG跟普通RAG有什么不同。MCPModel Context Protocol本质是模型和工具之间的通信协议。如果说RAG解决的是“怎么把私有知识喂给模型”那MCP解决的是“怎么让模型安全地使用外部工具”。它们解决的问题面不一样但可以组合使用RAG做内部知识检索MCP把检索能力、企业API封装成工具让智能体统一调度。Agentic RAG则在RAG基础上加入了“规划与反思”能力。普通RAG是单轮套路用户问一句检索一次生成一次。Agentic RAG会让模型先判断“这个问题要不要检索如果检索结果不够需要再换检索词或类型吗”。简单说普通RAG是一锤子买卖Agentic RAG是一个循环决策流程。如果知识库查询条件比较复杂比如“找出上季度所有退货率超过10%且售后评分低于4分的产品明细”普通RAG大概率搞不定这个时候有必要上Agentic RAG。3. Python异步编程核心要点从语法到落地3.1 async/await与事件循环之间的关系Python异步编程的基础概念不难很多教程一上来就讲async/await但很多人没搞懂事件循环是什么。把事件循环想象成一个调度员它维护一个任务队列轮询每个任务的IO状态哪个任务的数据准备好了就继续执行它没准备好的就挂起来。这个调度员不会因为等某一个IO而停下它永远在处理那些“已经就绪”的任务。async def定义的是一个协程函数调用它会返回一个协程对象这个对象不会立即执行必须被放进事件循环里才会运行。await的作用就是“让出CPU”当一个协程执行到await时它告诉事件循环我要等一个IO或者另一个协程的结果这段时间你可以去处理别的任务。等IO完成事件循环回来继续执行。实际写代码时99%的人不需要直接接触事件循环API只要用await调用异步函数就好。但如果你看到async def func()在调用时忘了await调试器只会告诉你“coroutine was never awaited”——这个错误几乎每个写asyncio的人都碰到过。记住一个口诀调用异步函数必须加await除非你想拿协程对象去注册任务。3.2 并发检索必备asyncio.gather与超时控制RAG流式回答里最典型的并发场景是同时发起多路召回一路向量检索一路BM25关键词检索一路查数据库获取用户历史记录。这三路互不依赖理论上应该一起跑。我用asyncio.gather把它们包起来等全部完成后把结果合并重排。import asyncio async def recall_vector(query: str): await asyncio.sleep(0.3) return [向量检索结果1, 向量检索结果2] async def recall_bm25(query: str): await asyncio.sleep(0.2) return [关键词检索结果1] async def recall_history(user_id: str): await asyncio.sleep(0.1) return [用户历史问题记录] async def run_multi_recall(query: str, user_id: str): results await asyncio.gather( recall_vector(query), recall_bm25(query), recall_history(user_id), ) return resultsgather有一个隐藏行为需要小心默认情况下如果其中任何一个任务抛异常gather会立刻把异常抛给调用方但其他任务并不会被取消它们会继续在后台运行。这在某些场景会导致资源泄漏。所以我通常给gather加return_exceptionsTrue先拿到所有结果再在本地逐个处理异常。超时控制也是必须做的。检索服务或者外部API可能出现异常慢的情况用户和前端不可能会无限等待。我习惯用asyncio.wait_for给每个外部依赖单独设置超时比如检索最多等2秒大模型首包最多等5秒整体生成最多等30秒。超时之后该任务会被取消返回一个兜底结果或者错误事件前端可以立刻提示用户重试。3.3 asyncio.Queue怎么用在做流式生成的任务编排中有些场景下gather不够用因为流式生成不是一次性拿回所有结果而是逐步产生结果。举个例子前端除了要拿大模型的输出还要拿检索到了哪些文档、每一步处理的状态。这时候可以用asyncio.Queue做一个生产者-消费者模型生产者协程把“检索完成”“开始生成”“第一个token”“第N个token”这些事件依次put进队列消费端协程从队列里读事件写进SSE响应。我当时设计了一个简单的状态管道代码逻辑类似下面这样import asyncio import json async def chatbot_pipeline(query: str): # 事件队列不限制大小保证生成过程不会因为队列写满而阻塞 queue: asyncio.Queue asyncio.Queue() async def producer(): await queue.put({event: status, data: 检索中...}) docs await recall_documents(query) await queue.put({event: docs, data: docs}) await queue.put({event: status, data: 生成中...}) async for token in llm_stream(docs): await queue.put({event: token, data: token}) await queue.put({event: done, data: }) # 消费者直接return queue在另一个异步生成器里消费 return queue实际开发中我更多是直接在路由函数里写一个异步生成器把所有事件yield出去逻辑更线性、更好读。队列更适合复杂任务编排的场景比如同时多个生产者协作、需要缓冲、需要按优先级输出等。这里提它是为了给读者一个方向异步编程不只是await队列和条件变量也是重要工具。4. 流式回答与SSE实现详解4.1 SSE和WebSocket怎么选流式回答有两种主流方案SSE和WebSocket。我之前调研时也纠结过一阵子两者都能做服务端推送但方向不一样。SSE是HTTP协议之上的单向长连接客户端用EventSource接口或者fetch的ReadableStream读数据服务端可以一直往连接里写内容。因为走的是HTTP天然兼容各种网关、负载均衡、防火墙不需要额外建立WebSocket连接。缺点很明显单向客户端只能发一次请求服务端持续推送如果做多轮对话需要前端每次新建一个SSE连接。WebSocket是双向全双工协议客户端和服务端都能随时发消息。适合聊天室、在线协同、实时游戏这种高交互场景。但WebSocket在基础设施层面更复杂网关要配置连接保活、心跳都要自己考虑调试成本更高。对于大模型流式回答我最终选SSE核心原因是这个场景是“一问一答”服务端不需要主动给客户端推送除回答之外的实时消息HTTP的简单可靠足够而且市面上大模型API的流式接口也都是SSE格式整个链路统一。如果你需要在一个连接里既允许用户输入消息又允许推送模型回复那再考虑WebSocket也不迟。4.2 FastAPI如何实现SSE流式输出FastAPI的StreamingResponse是流式输出的入口。它的media_type设成text/event-stream返回一个异步生成器生成器每次yield一段SSE格式的文本FastAPI就把它刷给客户端。SSE格式有严格约定。每个事件以空行分隔事件内可以有data: 数据内容多行data就表示多行数据event: 自定义事件名id: 事件ID用于断线重连retry: 重连间隔一个最简单的事件大概是这样的data: {content: 你} data: {content: 好}两个事件之间用空行分隔客户端就能逐次读取。注意SSE必须设置连接的Content-Type为text/event-stream并且最好关闭缓冲否则服务端或中间件可能会把多个事件攒在一起一次性发出去前端就会看到一段一段的延迟。FastAPI代码写起来很直白import json from fastapi import FastAPI from fastapi.responses import StreamingResponse app FastAPI() async def generate_answer() - str: # 模拟大模型流式返回实际换成真正的异步LLM调用 for chunk in [你, 好, , 这, 是, 一个, 流式, 回答]: yield chunk await asyncio.sleep(0.1) async def sse_generator(): async for chunk in generate_answer(): yield fdata: {json.dumps({content: chunk}, ensure_asciiFalse)}\n\n yield event: done\ndata: \n\n app.post(/chat/stream) async def chat_stream(): return StreamingResponse( sse_generator(), media_typetext/event-stream, headers{ Cache-Control: no-cache, X-Accel-Buffering: no, # 关闭Nginx缓冲 } )这里面有个容易被坑的点如果服务部署在Nginx后面Nginx默认会对代理响应做缓冲导致流式输出变成一大坨一起到达。解决方式是在响应头里加X-Accel-Buffering: no。如果是本地开发用uvicorn通常不需要这个头。4.3 前端怎么消费SSE流前端消费SSE有两条路子。简单场景用EventSource但它的局限是不支持自定义请求头、不支持POST。很多业务场景需要带token验证或者传很长的查询参数EventSource不方便这时候用fetch ReadableStream更灵活。EventSource写法const eventSource new EventSource(/api/chat/stream?query你好); eventSource.onmessage (event) { const data JSON.parse(event.data); // 实时渲染 data.content renderContent(data.content); }; eventSource.onerror () { eventSource.close(); };fetch流写法const response await fetch(/api/chat/stream, { method: POST, headers: { Content-Type: application/json, Authorization: Bearer xxx }, body: JSON.stringify({ query: 你好 }), }); const reader response.body.getReader(); const decoder new TextDecoder(utf-8); let buffer ; while (true) { const { done, value } await reader.read(); if (done) break; buffer decoder.decode(value, { stream: true }); // 按 SSE 空行拆包 const events buffer.split(\n\n); buffer events.pop(); for (const event of events) { const dataLine event.split(\n).find(line line.startsWith(data: )); if (dataLine) { const data JSON.parse(dataLine.slice(6)); renderContent(data.content); } } }注意fetch流有个细节SSE协议规定事件之间必须以空行分隔但网络层可能把多个事件打包在一个chunk里也可能一个事件被拆到两个chunk里。所以前端一定要维护buffer积累到完整事件再处理不能每次拿到chunk就直接当单条事件解析。4.4 abort与取消客户端断开后怎么停住生成流式回答最让人头痛的问题是用户等得不耐烦关了页面或点了“停止生成”按钮此时HTTP连接已经断开但服务端的大模型API调用还在继续跑。如果不处理就是在白白消耗token费用还可能把生成结果堆积在不会再有人读取的socket上。FastAPI里处理这个问题的标准做法是检查请求的断开状态。在路由函数里拿到Request对象用一个循环不断检查await request.is_disconnected()一旦发现连接断开立刻break出流式生成循环。from fastapi import Request app.post(/chat/stream) async def chat_stream(request: Request): async def event_stream(): async for token in llm_stream(query): if await request.is_disconnected(): # 这里可以写日志统计中断率 break yield fdata: {json.dumps({content: token})}\n\n return StreamingResponse(event_stream(), media_typetext/event-stream)如果你的底层大模型调用不是直接循环token而是封装成了一个异步任务要取消它可以用asyncio.Task的cancel方法或者给调用包一层asyncio.wait_for设置超时。我有一个经验不要把abort逻辑写在生成器的finally块里做太重的清理工作因为finally块在服务器关闭或异常退出时也可能触发重清理会导致其他请求被拖慢。轻量清理记日志、更新状态可以放重量清理比如重新建立索引之类最好放到独立的后台任务里。5. 实战从零搭一个RAG流式问答接口5.1 项目目录和环境准备做这个项目前先把环境准备好。我建议用Python 3.10以上版本Windows和Linux都可以重点是venv隔离环境。这里不展开讲Python安装网上教程很多。核心依赖如下pip install fastapi pip install uvicorn[standard] pip install httpx pip install langchain-text-splitters # 或者手动实现切块 pip install chromadb pip install sentence-transformers项目目录我习惯按功能模块组织rag_stream_demo/ ├── main.py # FastAPI入口与路由 ├── rag/ │ ├── loader.py # 文档加载 │ ├── chunker.py # 文本切块 │ ├── embedder.py # 向量化 │ ├── retriever.py # 检索与重排 │ └── generator.py # 大模型流式生成 ├── data/ # 待检索的文档 └── requirements.txt5.2 核心模块实现细节第一步文档切块。这里为了演示先手工实现一个基于段落和滑动窗口的切块器不依赖大框架。切块策略是按段落切分如果段落长度超过max_chunk_size再按句子边界滑动切。import re def split_sentences(text: str): return re.split(r(?[。.!?])\s*, text) def chunk_text(text: str, chunk_size: int 300, overlap: int 50): paragraphs re.split(r\n\s*\n, text) chunks [] buffer for para in paragraphs: sentences split_sentences(para) for sent in sentences: sent sent.strip() if not sent: continue if len(buffer) len(sent) chunk_size: # 先把buffer里的内容收成chunk再开始新的buffer if buffer: chunks.append(buffer) # 保留重叠部分取buffer末尾的overlap长度 buffer buffer[-overlap:] sent else: buffer sent if buffer: chunks.append(buffer) return chunks切块时overlap为什么重要因为语义可能会横跨两个相邻块。比如一个结论句里提到“这种做法”四个字但对应的具体方案在上一段的末尾。如果不做重叠检索到包含“这种做法”的块时上下文缺失模型回答就容易出现问题。overlap大小我一般取chunk_size的15%-20%——太小没用太大又造成冗余。第二步向量检索。这里先定义向量化的接口方便后续切换不同的embedding模型from sentence_transformers import SentenceTransformer model SentenceTransformer(BAAI/bge-small-zh-v1.5) def embed_texts(texts: list[str]): # 同步阻塞调用放asyncio.to_thread里跑 return model.encode(texts, normalize_embeddingsTrue).tolist()检索时把query向量化和文档向量做余弦相似度计算。生产环境请用向量数据库本地测试直接内存计算就行import numpy as np from sklearn.preprocessing import normalize class LocalVectorStore: def __init__(self): self.chunks [] self.vectors [] def add_documents(self, chunks: list[str]): vectors embed_texts(chunks) self.chunks.extend(chunks) self.vectors.extend(vectors) async def search(self, query: str, top_k: int 3): query_vec embed_texts([query])[0] # 向量检索是CPU密集计算丢到线程池 def _search(): arr np.asarray(self.vectors) scores arr np.asarray(query_vec) top_indices np.argsort(scores)[::-1][:top_k] return [(self.chunks[i], float(scores[i])) for i in top_indices] return await asyncio.to_thread(_search)第三步大模型流式生成。这里做了一个可替换的接口如果你有OpenAI兼容API用对应的异步SDK如果没有用一个模拟生成器模拟流式。我建议动手练习时先跑模拟器把整体链路打通了再接真实API。import asyncio async def llm_stream(prompt: str, context: list[str]): # 模拟大模型流式输出 text f根据检索到的{len(context)}个文档片段我给出如下回答\n text 企业知识库中的相关内容显示该项目主要涉及流程规范、权限管理和数据安全三个层面。 for chunk in text: await asyncio.sleep(0.05) yield chunk5.3 多轮对话的设计思路多轮对话是RAG问答里避不开的需求。要在流式回答场景下支持多轮核心要设计好两件事一是历史消息怎么传给大模型二是检索时怎么利用历史信息。历史消息的传递比较简单大模型的接口一般支持messages数组把history拼接进去即可。但要注意上下文长度不能无限回溯。我的策略是保留最近3-5轮对话同时截断过长的历史。更难的是检索阶段怎么用历史信息。用户提问“那它的有效期是多久”这句话单独拿去检索库根本没有内容能匹配上。理想做法是对用户query做改写结合最近的对话把指代词展开成完整句。比如“它”指的是上一轮提到的《员工报销管理制度》改写后就是“《员工报销管理制度》的有效期是多久”再拿改写后的query去检索。这个改写一般让大模型来做可以用一次轻量的LLM调用也可以用一个基于规则的模板我项目里先用了规则效果已经能覆盖大部分常见场景。5.4 接口联调与效果验证写好代码后在终端启动服务uvicorn main:app --host 0.0.0.0 --port 8000 --reload用curl测试流式接口能看到内容一段段输出curl -N --no-buffer -X POST http://localhost:8000/chat/stream \ -H Content-Type: application/json \ -d {query: 员工报销的流程是什么}验证重点有三项第一项是首包时间从用户发出请求到第一个token推给前端我实测在0.5秒以内才算合格主要耗时来自检索和网络RTT第二项是内容准确性抽几个典型问题人工看答案质量第三项是并发稳定性用apache bench或jmeter做简单压测模拟50个并发观察响应延迟有没有明显波动服务端有没有大量错误日志。6. 常见问题与排查技巧实录6.1 事件循环被阻塞服务“假死”症状流式接口表现一段快一段慢偶尔完全卡住不动压测时整体吞吐骤降。排查后发现是vector search里用了同步代码直接跑在async函数里导致事件循环被长时间占用。修复方式就是文章前面提的asyncio.to_thread包裹重计算或者换用原生支持异步的向量数据库客户端。判断依据也可以用这个简单测试起两个并发请求如果其中一个完全执行完之前另一个接口没有任何响应说明事件循环被阻塞了。6.2 客户端断开后底层调用还在运行症状数据库和模型服务的日志里经常有响应时间异常长的调用但前端早已经不显示内容了。原因是没有检查request.is_disconnected或者取消逻辑只取消了SSE生成器没有取消底层LLM任务。我的做法是给底层LLM调用封装成task主生成器退出时在finally块中对task执行cancel同时捕获CancelledError避免异常污染日志。6.3 SSE事件被缓冲成一坨症状前端一次性收到全部内容中间没有任何延迟流式效果形同虚设。可能原因有两个层次应用层没关闭输出缓冲或者反向代理开启了缓冲。先在本地直接用uvicorn测如果本地正常、上代理后不正常那就是代理问题。Nginx加X-Accel-Buffering头、或者关闭proxy_buffering即可具体配置如下location /chat/stream { proxy_pass http://backend; proxy_buffering off; proxy_cache off; }6.4 常见问题速查表现象可能原因解决方案Token一个都不吐最后返回完整文本输出被缓冲检查响应头和代理配置首包时间太长检索链路慢、embedding推理慢预加载模型、优化检索数量多轮对话中后续提问回答不准没有query改写增加历史信息改写环节调用API报错“coroutine never awaited”漏写await检查所有async函数调用处并发高时响应超时同步代码阻塞事件循环重计算丢to_thread、确认数据库连接池够用SSE在前端解析乱码中文编码处理不当JSON序列化时ensure_asciiFalse6.5 关于切块和检索质量的独家电线秘诀最后再说一个很多RAG项目都会踩的坑不是检索到的内容越多越好。有人为了“让模型多掌握点信息”一次召回10个文档块结果模型反而被无关内容带偏。实践下来企业问答场景top_k取3-5比较合适。如果想要更精准可以在召回后做一步重排序过滤设定最低相似度阈值相似度低于0.4的块直接丢弃。这个方法简单有效能显著降低幻觉概率。还有一个容易被忽视的点切块时的元数据保留。给每个chunk打上来源文档标题、章节路径、页码检索结果返回时把这些信息带上。前端既能展示引用来源提高回答可信度后端排查回答问题时也能快速定位是哪段文档内容把模型带偏了。我现在做RAG项目必定会给每一条检索结果带元数据这个习惯帮我在调优时省了非常多时间。做这个项目的过程中我最深的一个体会是流式回答的流畅度往往不是大模型决定的而是整条异步链路的健壮性决定的。你可以在代码里用上所有异步技巧但只要有一个环节忘了处理阻塞或取消用户就能立刻感知到卡顿。另一个体会是别小看切块和召回质量RAG的上限由检索决定下限由模型决定——检索拉胯了模型再强也答不对。这篇实战内容到这里就完整了。如果你也在做类似的知识库问答建议先把异步流式链路用模拟器跑通再接真实模型和真实知识库。链路通顺后再去打磨切块策略和query改写效果会比一上来就堆各种大模型参数好得多。