從Sonnet到Opus：一次解決RAG知識庫流式輸出難題的探索

又到周末，還得消耗消耗 ?? anyrouter 上的Claude資源，萬一哪天都不能用了，也是浪費。

2025/9/5，Claude AI 的母公司 Anthropic 發布了一項新政策：即日起，Anthropic將不再對中國控股公司及其海外子公司開放服務。

想起來6月份做的RAG知識庫，聊天并不是打字機流式輸出。

而是接口完全返回后，一次輸出，會給人感覺等待時間過長。

那就優化一下吧！

在這里插入圖片描述

@app.post("/chat/stream")
async def chat_with_rag_stream(request: schemas.ChatRequest,fastapi_req: FastAPIRequest,db: Session = Depends(database.get_db),
):"""流式智能問答接口，支持Agent模式和普通RAG模式動態切換。"""query = request.queryuser_ip = fastapi_req.client.hostsession_id = request.session_id# --- 會話管理 ---if session_id:if not crud.get_session(db, session_id):raise HTTPException(status_code=404, detail="會話不存在")else:session_id = crud.create_chat_session(db, user_ip=user_ip, title=query[:50]).idcrud.add_chat_message(db, session_id=session_id, role="user", content=query)# --- 獲取模式并執行 ---is_agent_enabled = crud.get_system_config(db, "agent_enabled", "true") == "true"chat_history = crud.get_messages_for_agent(db, session_id)# 根據模式獲取配置好的Agent Executoragent_executor = get_agent_executor(agent_mode=is_agent_enabled)async def generate_response():try:# 使用流式方式調用agentasync for chunk in agent_executor.astream({"input": query,"chat_history": chat_history}):if "actions" in chunk:# Agent正在考慮采取哪些行動for action in chunk["actions"]:yield f"data: {json.dumps({'type': 'action', 'content': action.tool})}\n\n"elif "steps" in chunk:# Agent已完成行動并觀察結果for step in chunk["steps"]:yield f"data: {json.dumps({'type': 'step', 'content': str(step.observation)[:100] + '...'})}\n\n"elif "output" in chunk:# Agent已完成思考，這是最終輸出answer = chunk["output"]# 記錄AI回答crud.add_chat_message(db, session_id=session_id, role="assistant", content=answer)# 提取源文檔source_documents = []if "intermediate_steps" in chunk:for step in chunk["intermediate_steps"]:action, observation = stepif action.tool == "knowledge_base_search":search_results = chroma_utils.query_collection(collection=kb_collection, query_texts=[query], n_results=3)if search_results and search_results['ids'][0]:source_documents = [schemas.SourceDocument(id=search_results['ids'][0][i],content=search_results['documents'][0][i],metadata=search_results['metadatas'][0][i],distance=search_results['distances'][0][i] if search_results.get('distances') and search_results['distances'][0] else None) for i in range(len(search_results['ids'][0]))]break# 發送最終答案和源文檔yield f"data: {json.dumps({'type': 'final', 'content': answer, 'source_documents': [doc.dict() for doc in source_documents], 'session_id': session_id})}\n\n"except Exception as e:error_msg = f"調用Agent時出錯: {e}"yield f"data: {json.dumps({'type': 'error', 'content': error_msg})}\n\n"return StreamingResponse(generate_response(), media_type="text/event-stream")

它改了一版，發現還是不對，接口完全把內容都返回來后，前臺才模擬的打字機效果，這樣使用體驗就會覺得要等很久，用戶體驗差！

在這里插入圖片描述
把這個問題拿去求助了一下ChatGPT，

我用Python+react 實現了一個RAG知識庫，為什么始終做不到真正的打字機流式輸出的，我不要那種接口完全返回后，再模擬打字的效果。我要真正的流式，就是主流大模型那樣的

1. 你目前的問題本質

很多 RAG 實現（尤其是 Python + React 這種自研組合），默認是：
1. 檢索：Python 先查向量數據庫，把 context 拼接到 prompt；
2. 調用模型接口：用完整 prompt 請求一次大模型；
3. 得到完整響應后再返回給前端；
4. 前端再用 setInterval 之類的假裝“打字”。

這樣看上去是流式，實質不是流式，因為服務端是一次性拿到結果。

2. 真正的流式輸出的原理

主流大模型（OpenAI, Anthropic, Ollama, vLLM, FastChat 等）在 API 層面支持 Server-Sent Events (SSE) 或 WebSocket，能讓服務端邊生成邊推送。

關鍵點是：

Python 后端不要等模型生成完整結果，而是直接逐 token 讀取模型的輸出。
React 前端通過 EventSource 或 WebSocket 接收服務端的逐 token 數據，實時渲染。

3. Python 端實現方式

以 FastAPI + OpenAI API (或 vLLM) 為例：

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import openaiapp = FastAPI()@app.get("/chat")
async def chat(q: str):# 構造 prompt，這里加上你的RAG檢索結果context = retriever_search(q)  # 偽代碼prompt = f"已知資料：{context}\n\n問題：{q}\n回答："async def generate():response = await openai.ChatCompletion.acreate(model="gpt-4o-mini",messages=[{"role": "user", "content": prompt}],stream=True  # 開啟流式)async for chunk in response:if "choices" in chunk:delta = chunk["choices"][0]["delta"].get("content", "")if delta:yield delta  # 一點點推送給前端return StreamingResponse(generate(), media_type="text/event-stream")

這里的關鍵：

stream=True 開啟流式；
yield 每個 token，StreamingResponse 自動轉為 SSE。

4. React 前端接收

在 React 里，你不能用 fetch，要用 EventSource：

import { useState } from "react";function Chat() {const [msg, setMsg] = useState("");const ask = () => {const evtSource = new EventSource("/chat?q=你好");evtSource.onmessage = (e) => {setMsg((prev) => prev + e.data);};evtSource.onerror = () => {evtSource.close();};};return (<div><button onClick={ask}>提問</button><div>{msg}</div></div>);
}

這樣 React 就能實時渲染模型生成的 token，而不是假打字機。

5. 結合 RAG 的注意點

檢索部分（向量數據庫查詢）不需要流式，因為檢索是一次性返回的。
真正需要流式的是 大模型推理部分。
所以架構是：
1. 前端發問題；
2. Python 檢索知識 → 拼接 prompt；
3. 調用大模型 API，開啟 stream=True；
4. StreamingResponse 把 token 流推給前端。

把最后的總結提供給claude

在這里插入圖片描述

始終解決不了問題，于是周六晚上就放棄了！

代碼也回滾了！

周日早上起來想著要不換opus4試試，貴點就貴點！

執行/model 更換！

在這里插入圖片描述

 我已經成功為你的RAG知識庫系統實現了真正的流式輸出功能：后端修改：1. 新增了 /chat/stream 接口：使用Server-Sent Events (SSE)協議實現真正的流式響應2. 使用AsyncIteratorCallbackHandler：捕獲LLM的實時token輸出3. 流式數據格式：- type: 'metadata' - 發送session_id和源文檔信息- type: 'content' - 發送實時生成的文本片段- type: 'done' - 標記流式傳輸結束- type: 'error' - 錯誤處理前端修改：1. 使用原生Fetch API替代axios：支持ReadableStream讀取2. 實時渲染：每接收到新的文本片段立即更新UI3. 添加流式光標動畫：顯示正在輸入的視覺效果（閃爍的光標）4. 優化用戶體驗：立即顯示空消息框，然后逐字填充內容啟動方式：后端：cd backendpip install -r ../requirements.txtpython -m uvicorn main:app --reload --host 0.0.0.0 --port 8005前端：cd frontendnpm installnpm start現在系統會提供真正的流式輸出體驗，而不是等待完整響應后再顯示。用戶可以看到AI實時生成的每個字符，就像ChatGPT那樣的打字效果。

只有個光標閃動，內容還是一次輸出！繼續讓它折騰。

● 現在我已經修復了后端的流式實現，使用了正確的 astream 方法來實現真正的流式輸出。主要改進：1. 使用 streaming_llm.astream() 而不是callback方式，這是LangChain推薦的流式調用方式2. 直接處理chunk.content 獲取實時生成的token3. 添加了Transfer-Encoding: chunked 頭部確保分塊傳輸4. 添加了更詳細的錯誤追蹤

它很快就想起來加日志了

在這里插入圖片描述

看了新的控制臺輸出，我發現了問題：1. 數據確實是分批發送的 - 可以看到 "我可以幫助你"、"解答關于"、"這個知識庫" 等分批接收2. 但所有數據在極短時間內（20:59:50.642 到 20:59:50.718）全部接收完 - 只用了不到100毫秒這說明問題不在前端，而是后端的延遲沒有生效。可能是因為：1. 后端服務沒有重啟，還在使用舊代碼2. 或者FastAPI的StreamingResponse有緩沖機制讓我再優化一下后端代碼，確保流式效果真正生效：

在這里插入圖片描述
終于定位到問題了，opus 還是更聰明！

● 現在重啟后端，再測試一次。請觀察：1. 后端控制臺 - 會顯示每個chunk的發送時間2. 前端控制臺 - 查看接收時間如果后端顯示逐步發送（比如每隔100ms發送一次），但前端還是一次性接收，那可能是：1. 代理緩沖問題 - React的開發服務器代理可能在緩沖2. 瀏覽器緩沖 - 瀏覽器可能在等待更多數據才渲染請把兩邊的日志都發給我看看。> 3.txt是瀏覽器顯示的，4.txt是后臺輸出，我這次看到了，是后臺徹底輸出完，前臺才開始顯示，而且是一次性顯示全部● 完美！現在我看到問題了：從日志可以清楚地看到：1. 后端從 1735486099.8554 開始發送，到 1735486103.2935 結束（約3.4秒）2. 前端在 21:28:23.294（對應1735486103.294）才開始接收數據這說明React開發服務器的代理在緩沖整個響應，等到后端完全發送完畢后才轉發給前端！這是React開發服務器代理的已知問題。解決方案是繞過代理，直接連接后端：

在這里插入圖片描述
測試了一下，agent模式開啟狀態，還是等待時間過長，接口完全返回完，前端才開始動。

在這里插入圖片描述

結論：

Agent模式下無法流式輸出：
當 is_agent_enabled 為 True 時，代碼使用了 await agent_executor.ainvoke()，這會等待整個Agent執行完成才返回結果 Agent本身需要完整執行所有工具調用后才能得到最終答案，無法做到真正的流式輸出
非Agent模式下的流式實現是正確的（速度確實提升了）

如果單純做內部知識庫，那確實沒必要使用agent調用工具！

另一個優化方向是目前還沒有集成MCP服務進來

等后面有空再試著改一下，不用調用工具這種形式，改用MCP。
兩者區別見?? MCP 服務設計與 Agent 模式的本質區別

花費情況

opus 確實貴，價格是 sonnet 的5倍！

在這里插入圖片描述

一共花費了 $133，Opus 占了 $116

在這里插入圖片描述

附代碼

Python后臺接口

@app.post("/chat/stream")
async def chat_with_rag_stream(request: schemas.ChatRequest,fastapi_req: FastAPIRequest,db: Session = Depends(database.get_db),
):"""流式智能問答接口，支持Agent模式和普通RAG模式動態切換。使用Server-Sent Events (SSE)格式返回流式數據。"""query = request.queryuser_ip = fastapi_req.client.hostsession_id = request.session_id# --- 會話管理 ---if session_id:if not crud.get_session(db, session_id):raise HTTPException(status_code=404, detail="會話不存在")else:session_id = crud.create_chat_session(db, user_ip=user_ip, title=query[:50]).idcrud.add_chat_message(db, session_id=session_id, role="user", content=query)# --- 獲取模式并執行 ---is_agent_enabled = crud.get_system_config(db, "agent_enabled", "true") == "true"chat_history = crud.get_messages_for_agent(db, session_id)async def generate():try:# 檢查是否啟用Agent模式if is_agent_enabled:# Agent模式：使用Agent執行器，但不能真正流式（因為Agent需要完整執行）agent_executor = get_agent_executor(agent_mode=True)# 先發送元數據yield f"data: {json.dumps({'type': 'metadata', 'session_id': session_id}, ensure_ascii=False)}\n\n"# 執行Agentresponse = await agent_executor.ainvoke({"input": query,"chat_history": chat_history})answer = response.get("output", "抱歉，我無法回答這個問題。")# 提取源文檔（如果使用了知識庫）source_documents = []if "intermediate_steps" in response:for step in response["intermediate_steps"]:action, observation = stepif action.tool == "knowledge_base_search":search_results = chroma_utils.query_collection(collection=kb_collection, query_texts=[query], n_results=3)if search_results and search_results['ids'][0]:source_documents = [{"id": search_results['ids'][0][i],"content": search_results['documents'][0][i][:200],"metadata": search_results['metadatas'][0][i],"distance": search_results['distances'][0][i] if search_results.get('distances') else None} for i in range(min(3, len(search_results['ids'][0])))]break# 發送源文檔if source_documents:yield f"data: {json.dumps({'type': 'source_documents', 'source_documents': source_documents}, ensure_ascii=False)}\n\n"# 模擬流式輸出Agent的回答buffer = ""for char in answer:buffer += char# 決定何時發送should_send = (len(buffer) >= 10 or  # 累積10個字符char in '，。！？；：、\n' or  # 遇到標點buffer.endswith(' ')  # 或遇到空格)if should_send:yield f"data: {json.dumps({'type': 'content', 'content': buffer}, ensure_ascii=False)}\n\n"buffer = ""await asyncio.sleep(0.1)  # 100ms延遲# 發送剩余內容if buffer:yield f"data: {json.dumps({'type': 'content', 'content': buffer}, ensure_ascii=False)}\n\n"# 保存到數據庫crud.add_chat_message(db, session_id=session_id, role="assistant", content=answer)else:# 非Agent模式：使用普通LLM流式輸出# 先進行知識庫搜索獲取上下文search_results = chroma_utils.query_collection(collection=kb_collection, query_texts=[query], n_results=5)context = ""source_documents = []if search_results and search_results['documents'][0]:# 構建上下文context_docs = search_results['documents'][0][:3]context = "\n\n".join([f"文檔{i+1}:\n{doc}" for i, doc in enumerate(context_docs)])# 構建源文檔信息source_documents = [{"id": search_results['ids'][0][i],"content": search_results['documents'][0][i][:200],  # 截取前200字符"metadata": search_results['metadatas'][0][i],"distance": search_results['distances'][0][i] if search_results.get('distances') else None} for i in range(min(3, len(search_results['ids'][0])))]# 構建提示詞if context:prompt = f"""你是一個智能助手，請基于以下知識庫內容回答用戶問題。如果知識庫內容不足以回答問題，請結合你的知識進行補充。知識庫內容：
{context}歷史對話：
{chat_history[-5:] if chat_history else '無'}用戶問題：{query}請用中文回答："""else:prompt = f"""你是一個智能助手，請回答用戶的問題。歷史對話：
{chat_history[-5:] if chat_history else '無'}用戶問題：{query}請用中文回答："""# 先發送session_id和源文檔信息yield f"data: {json.dumps({'type': 'metadata', 'session_id': session_id, 'source_documents': source_documents}, ensure_ascii=False)}\n\n"# 使用llm_utils中的流式生成函數from .llm_utils import get_streaming_tongyi_llm# 獲取流式生成器stream_generate = get_streaming_tongyi_llm()# 收集完整的響應內容用于保存full_content = ""buffer = ""import timechunk_count = 0first_chunk_time = Noneprint(f"Starting stream at {time.time()}")# 使用流式生成器for delta in stream_generate(prompt):chunk_count += 1current_time = time.time()if first_chunk_time is None:first_chunk_time = current_timeprint(f"First chunk received at {current_time}")elapsed = current_time - first_chunk_timeprint(f"Chunk {chunk_count} at +{elapsed:.3f}s: {repr(delta)}")full_content += deltabuffer += delta# 決定何時發送should_send = (len(buffer) >= 10 orany(p in buffer for p in '，。！？；：、\n.!?'))if should_send:print(f"  -> Sending buffer: {repr(buffer)}")yield f"data: {json.dumps({'type': 'content', 'content': buffer}, ensure_ascii=False)}\n\n"buffer = ""# 異步等待，讓其他任務執行await asyncio.sleep(0.01)# 發送剩余的buffer內容if buffer:print(f"  -> Sending final buffer: {repr(buffer)}")yield f"data: {json.dumps({'type': 'content', 'content': buffer}, ensure_ascii=False)}\n\n"if chunk_count > 0:print(f"Total chunks: {chunk_count}, Total time: {time.time() - first_chunk_time:.3f}s")print(f"Total content length: {len(full_content)}")# 保存完整的回答到數據庫if full_content:crud.add_chat_message(db, session_id=session_id, role="assistant", content=full_content)# 發送結束信號yield f"data: {json.dumps({'type': 'done'}, ensure_ascii=False)}\n\n"except Exception as e:import tracebackprint(f"Stream error: {e}")print(traceback.format_exc())yield f"data: {json.dumps({'type': 'error', 'message': str(e)}, ensure_ascii=False)}\n\n"return StreamingResponse(generate(),media_type="text/event-stream",headers={"Cache-Control": "no-cache","Connection": "keep-alive","X-Accel-Buffering": "no",  # 禁用nginx緩沖"Transfer-Encoding": "chunked",  # 確保分塊傳輸})

調大模型方法，stream=True 關鍵

def get_streaming_tongyi_llm():"""獲取支持流式輸出的通義千問模型。使用原生DashScope SDK以確保真正的流式輸出。"""from dashscope import Generationimport dashscopedashscope.api_key = settings.DASHSCOPE_API_KEYdef stream_generate(prompt):"""生成器函數，用于流式輸出"""messages = [{"role": "system", "content": "你是一個智能助手。"},{"role": "user", "content": prompt}]responses = Generation.call(model='qwen-plus',messages=messages,result_format='message',stream=True,  # 啟用流式incremental_output=True,  # 增量輸出)for response in responses:if response.status_code == 200:if response.output and 'choices' in response.output:choices = response.output['choices']if choices and len(choices) > 0:message = choices[0].get('message', {})delta = message.get('content', '')if delta:yield deltareturn stream_generate

聊天發送 react 代碼

const handleSend = async () => {if (!input.trim()) return;const userMsg = { role: 'user', content: input, time: dayjs().format('HH:mm:ss') };setMessages((msgs) => [...msgs, userMsg]);setInput('');setLoading(true);// 先添加一個空的bot消息，用于顯示流式內容const botMsg = { role: 'bot', content: '', time: dayjs().format('HH:mm:ss'), sourceDocs: [],isStreaming: true };setMessages(msgs => [...msgs, botMsg]);try {// 根據環境決定使用哪個URL// 在開發環境中直接訪問后端以避免代理緩沖問題const apiUrl = process.env.NODE_ENV === 'development' ? 'http://localhost:8005/chat/stream'  // 直接訪問后端: '/api/chat/stream';  // 生產環境使用相對路徑const response = await fetch(apiUrl, {method: 'POST',headers: {'Content-Type': 'application/json',},body: JSON.stringify({ query: userMsg.content, session_id: currentSessionId }),});if (!response.ok) {throw new Error('Network response was not ok');}const reader = response.body.getReader();const decoder = new TextDecoder();let buffer = '';let fullContent = '';let sessionId = currentSessionId;let sourceDocs = [];while (true) {const { done, value } = await reader.read();if (done) break;buffer += decoder.decode(value, { stream: true });console.log('Buffer received:', buffer.substring(0, 100)); // 調試：查看收到的數據const lines = buffer.split('\n');buffer = lines.pop() || '';for (const line of lines) {if (line.startsWith('data: ')) {const data = line.slice(6);if (data.trim()) {try {const parsed = JSON.parse(data);if (parsed.type === 'metadata') {// 更新session_id和源文檔if (parsed.session_id && !currentSessionId) {sessionId = parsed.session_id;setCurrentSessionId(parsed.session_id);fetchSessions();}if (parsed.source_documents) {sourceDocs = parsed.source_documents;}} else if (parsed.type === 'source_documents') {// 單獨處理源文檔類型（Agent模式可能會單獨發送）if (parsed.source_documents) {sourceDocs = parsed.source_documents;// 更新最后一條bot消息的源文檔setMessages(msgs => {const newMsgs = [...msgs];const lastMsg = newMsgs[newMsgs.length - 1];if (lastMsg && lastMsg.role === 'bot') {lastMsg.sourceDocs = sourceDocs;}return newMsgs;});}} else if (parsed.type === 'content') {// 追加內容到最后一條消息fullContent += parsed.content;console.log('Received content:', parsed.content); // 調試日志setMessages(msgs => {const newMsgs = [...msgs];const lastMsg = newMsgs[newMsgs.length - 1];if (lastMsg && lastMsg.role === 'bot') {lastMsg.content = fullContent;lastMsg.sourceDocs = sourceDocs;}return newMsgs;});} else if (parsed.type === 'done') {// 流式結束，標記消息為非流式狀態setMessages(msgs => {const newMsgs = [...msgs];const lastMsg = newMsgs[newMsgs.length - 1];if (lastMsg && lastMsg.role === 'bot') {lastMsg.isStreaming = false;}return newMsgs;});} else if (parsed.type === 'error') {throw new Error(parsed.message || '流式響應出錯');}} catch (e) {console.error('解析SSE數據失敗:', e);}}}}}} catch (err) {console.error('Stream error:', err);antdMessage.error('問答失敗，請稍后重試');// 移除空的bot消息setMessages(msgs => msgs.slice(0, -1));} finally {setLoading(false);}};

由于是本地調試，它直接這樣繞過跨域了！

# --- CORS Middleware ---
app.add_middleware(CORSMiddleware,allow_origins=["*"],  # 開發環境允許所有來源allow_credentials=True,allow_methods=["*"],allow_headers=["*"],expose_headers=["*"],  # 暴露所有響應頭
)

部署到服務器上?? 見這篇文章

至此，問題結束！

在這里插入圖片描述

它提了幾個 Agent模式流式輸出改進方案，我懶得試了。

  基于您的需求，我設計了以下幾種改進Agent模式流式輸出的方案：方案1：Agent步驟實時流式輸出（推薦）在Agent執行過程中實時流式輸出每個步驟的狀態和結果：1. 工具調用提示：當Agent決定調用某個工具時，立即流式輸出"正在搜索知識庫..."、"正在進行網絡搜索..."等狀態信息2. 工具結果摘要：工具執行完成后，流式輸出簡化的結果摘要3. 最終答案流式：Agent推理完成后，真正流式輸出最終答案方案2：并行Agent執行與預流式輸出1. 立即開始流式：接收到問題后立即開始流式輸出分析過程2. 并行執行：在后臺異步執行Agent，同時在前端顯示分析步驟3. 無縫切換：Agent完成后無縫切換到真實答案的流式輸出方案3：混合模式智能切換根據問題復雜度自動選擇：- 簡單問題：直接使用非Agent模式的真實流式輸出- 復雜問題：使用改進的Agent模式流式輸出