簡(jiǎn)介面向?qū)崟r(shí)數(shù)據(jù)處理開(kāi)發(fā)者與AI應(yīng)用工程師的技術(shù)方案型PDF聚焦DeepSeek流式響應(yīng)機(jī)制與長(zhǎng)文本分塊處理兩大核心難題。文檔從實(shí)時(shí)數(shù)據(jù)處理的定義、特點(diǎn)與應(yīng)用場(chǎng)景切入系統(tǒng)梳理流式響應(yīng)的技術(shù)原理、長(zhǎng)文本分塊的必要性與挑戰(zhàn)、分塊策略選擇固定長(zhǎng)度、語(yǔ)義單元、混合分塊、上下文信息保留及結(jié)果整合方法并附有Python代碼實(shí)現(xiàn)涵蓋分塊函數(shù)定義、流式響應(yīng)編寫、完整示例與錯(cuò)誤處理、性能優(yōu)化技巧具有較強(qiáng)的工程參考價(jià)值。包體共1個(gè)PDF文件22頁(yè)正文約1.8MB目錄結(jié)構(gòu)完整章節(jié)按技術(shù)原理、方案設(shè)計(jì)、代碼實(shí)現(xiàn)、案例實(shí)踐、趨勢(shì)展望有序編排閱讀體驗(yàn)順暢。已有111人學(xué)習(xí)下載適合希望掌握DeepSeek落地調(diào)用方式、解決長(zhǎng)文本與實(shí)時(shí)數(shù)據(jù)處理場(chǎng)景難題的NLP工程師、數(shù)據(jù)開(kāi)發(fā)者及技術(shù)學(xué)習(xí)者。1. 實(shí)時(shí)數(shù)據(jù)處理不是“點(diǎn)一次等一遍”分塊和流式是一對(duì)必須同時(shí)落地的組合拳做實(shí)時(shí)數(shù)據(jù)處理的人多半被同一個(gè)場(chǎng)景折磨過(guò)一份六千字的合同或一天的服務(wù)日志直接拼進(jìn)DeepSeek的prompt里點(diǎn)發(fā)送等結(jié)果那幾十秒夠倒一杯水生成一旦超過(guò)max_tokens或撞到窗口邊界還會(huì)被攔腰截?cái)唷＝夥ň蛢蓚€(gè)全部掛在標(biāo)題里長(zhǎng)文本分塊處理把輸入切成小而完整的片段流式響應(yīng)讓結(jié)果按token一點(diǎn)點(diǎn)推出來(lái)首字延遲從幾十秒壓到幾百毫秒。這條鏈路立住后用戶看到的是文字像打字機(jī)一樣跳出不是對(duì)著加載圈干瞪眼。這篇筆記按后端和數(shù)據(jù)工程師的落地路徑走覆蓋SSE接入、增量拼接、分塊與overlap參數(shù)最后組合成一條實(shí)時(shí)處理管道并給出排障和驗(yàn)收方法。寫過(guò)流式的人可以直接跳到第四章以后新手從第二章開(kāi)始跟。2. DeepSeek流式響應(yīng)從SSE協(xié)議到增量token的接入全流程2.1 SSE不是WebSocket它是一條“單行道”第一次用DeepSeek的流式接口時(shí)最容易誤解的是響應(yīng)體形狀。它不是在請(qǐng)求完成后給你一個(gè)超大JSON而是你在請(qǐng)求頭里把stream字段設(shè)為true后服務(wù)端把HTTP響應(yīng)變成持續(xù)推送的字節(jié)流邊生成邊發(fā)。協(xié)議層面走的是SSEServer-Sent Events響應(yīng)頭的Content-Type是text/event-stream多個(gè)事件之間用空行分開(kāi)每個(gè)事件以data:前綴開(kāi)始全部推完后服務(wù)端發(fā)送一個(gè)data: [DONE]事件再關(guān)閉連接。SSE和WebSocket常被拿來(lái)對(duì)比但定位完全不同。WebSocket是一次握手后雙向自由收發(fā)適合聊天、協(xié)作編輯這類雙方都要開(kāi)口的場(chǎng)景SSE是純單向的服務(wù)端到客戶端鏈路薄、兼容性好也沒(méi)有WebSocket那種升級(jí)握手的負(fù)擔(dān)。對(duì)大模型生成這個(gè)場(chǎng)景本來(lái)就只有服務(wù)端在說(shuō)話SSE就是最便宜的選擇。很多團(tuán)隊(duì)一上來(lái)就上WebSocket結(jié)果發(fā)現(xiàn)客戶端根本沒(méi)有上行需求白付了一筆心跳和連接維護(hù)的成本。SSE有一個(gè)弱點(diǎn)極其共性每個(gè)事件里只裝增量但多數(shù)HTTP客戶端默認(rèn)會(huì)等整個(gè)響應(yīng)體收完再返回。用requests.get(url).json()這種寫法去接流式優(yōu)化全部白費(fèi)拿到的還是“等了全套生成完”的結(jié)果。真正要吃到流式紅利要么用支持增量迭代的SDK要么手動(dòng)按行解析。這兩種方式本篇文章都會(huì)給到。2.2 最小接入用OpenAI兼容SDK發(fā)起流式請(qǐng)求DeepSeek的接口兼容OpenAI的/chat/completions格式這讓我們不用重學(xué)一套客戶端。base_url指到DeepSeek開(kāi)放平臺(tái)api_key從控制臺(tái)生成模型名寫deepseek-chat下面這段代碼我把它當(dāng)成腳手架凡是需要流式輸出的地方邏輯都從它展開(kāi)from openai import OpenAI client OpenAI( api_keysk-your-key, base_urlhttps://api.deepseek.com/, timeout30.0, ) resp client.chat.completions.create( modeldeepseek-chat, messages[ {role: system, content: 你是數(shù)據(jù)管道助手只輸出結(jié)構(gòu)化內(nèi)容。}, {role: user, content: 把下面這段日志按異常類型歸類每類給兩個(gè)例子。\n chunk_text}, ], streamTrue, stream_options{include_usage: True}, )代碼很短但三個(gè)細(xì)節(jié)值得留意。第一個(gè)是streamTrue這是流式響應(yīng)的開(kāi)關(guān)不傳它resp會(huì)一直阻塞直到完整文本生成完傳了它resp變成一個(gè)可迭代對(duì)象每次迭代拿出一個(gè)事件。第二個(gè)是stream_options{include_usage: True}它讓流結(jié)束前最后一個(gè)事件帶上token用量方便做成本核算。很多教程不寫這個(gè)參數(shù)導(dǎo)致最后為了統(tǒng)計(jì)又要再調(diào)一次計(jì)數(shù)接口。第三個(gè)是timeout30.0它約束的是連接建立和每次接收數(shù)據(jù)之間的等待上限不是整個(gè)生成周期的硬時(shí)限設(shè)得太小公網(wǎng)環(huán)境下一慢就會(huì)在第一個(gè)token到達(dá)前被判死。2.3 增量拼接的核心delta、reasoning_content與[DONE]的判定拿到resp之后要做的其實(shí)很機(jī)械把增量一段段接起來(lái)?！霸隽俊痹贠penAI協(xié)議里就是每個(gè)事件里的choices[0].delta.content。注意命名——是delta不是message它裝的是“剛生成的那一小段”不是“現(xiàn)在累計(jì)的全部”。要想得到完整回答必須自己維護(hù)一個(gè)字符串變量累加。這里還要區(qū)分DeepSeek的兩種模型。deepseek-chat每個(gè)事件的delta里只有contentdeepseek-reasoner則額外帶reasoning_content那是模型暴露出來(lái)的思維鏈。這兩段用途完全不同一定要分開(kāi)累加、分開(kāi)存儲(chǔ)否則用戶會(huì)看到大段推理過(guò)程被當(dāng)成答案。full_text reasoning_text for chunk in resp: # 最后一個(gè)usage事件choices為空直接跳過(guò)。 if chunk.choices is None or len(chunk.choices) 0: continue delta chunk.choices[0].delta if delta is None: continue if delta.reasoning_content: reasoning_text delta.reasoning_content if delta.content: full_text delta.content # 處理業(yè)務(wù)時(shí)只用full_textreasoning_text僅作審計(jì)記錄 print(full_text)這里有個(gè)血的教訓(xùn)最后一個(gè)攜帶usage的事件choices是空的如果代碼里不加判空直接訪問(wèn)delta.content會(huì)拋AttributeError。我第一次跑通時(shí)就把這個(gè)異常當(dāng)網(wǎng)絡(luò)問(wèn)題排查了半天實(shí)際只是少了一個(gè)if chunk.choices的判斷。如果不用SDK而是用requests裸接SSE——比如你要寫一個(gè)轉(zhuǎn)發(fā)網(wǎng)關(guān)把DeepSeek的流原樣轉(zhuǎn)給下游——結(jié)束判斷必須手動(dòng)做。常見(jiàn)寫法是逐行讀行以data:開(kāi)頭時(shí)取后面內(nèi)容遇到data: [DONE]退循環(huán)import json import requests payload { model: deepseek-chat, messages: [{role: user, content: 寫一段300字的實(shí)施方案}], stream: True, } resp requests.post( https://api.deepseek.com/chat/completions, jsonpayload, headers{Authorization: Bearer sk-your-key, Content-Type: application/json}, streamTrue, timeout30, ) for line in resp.iter_lines(decode_unicodeTrue): if not line or not line.startswith(data:): continue data line[5:].strip() if data [DONE]: break event json.loads(data) if event.get(choices): delta event[choices][0].get(delta, {}) if delta.get(content): # 在這里把增量推給下游 print(delta[content], end)用SDK時(shí)[DONE]由SDK內(nèi)部消化用裸HTTP時(shí)它就是你循環(huán)的出口。兩種寫法都正確差異在于是不是需要控制每個(gè)事件的去向。2.4 三個(gè)影響成敗的參數(shù)max_tokens、temperature、重試邊界max_tokens在DeepSeek接口里指的是“生成部分的最大token數(shù)”不是“請(qǐng)求響應(yīng)的總預(yù)算”。如果這個(gè)值給得太小流會(huì)在生成中途被掐斷且可能等不到[DONE]事件客戶端邏輯卡在讀取尾端。我一般按預(yù)期輸出長(zhǎng)度乘1.3給寧可多留余量。temperature控制采樣隨機(jī)性。日志歸類、摘要提取這類強(qiáng)格式任務(wù)給0最穩(wěn)開(kāi)放問(wèn)答給0.7以上才有味道。有一個(gè)容易忽略的連帶效果temperature0時(shí)同樣的問(wèn)題流式輸出基本一字不差溫度高了以后每次用詞都有細(xì)小變化。如果下游做答案全文比對(duì)這個(gè)差異可能被誤判成異常。重試邊界是常年翻車的點(diǎn)。如果把整個(gè)create()調(diào)用包在一個(gè)通用重試裝飾器里一旦服務(wù)端已經(jīng)生成了一部分才斷連重試會(huì)把同一個(gè)回答按計(jì)費(fèi)生成兩遍。正確的思路是重試只放在“連接建立、首個(gè)token到達(dá)前”這個(gè)階段進(jìn)入流式讀取期后不再自動(dòng)重發(fā)。中途中止的流讓業(yè)務(wù)層用冪等ID去判斷結(jié)果是否已存在而不是盲目重放。3. 長(zhǎng)文本分塊讓每個(gè)請(qǐng)求只帶最可能被用到的上下文3.1 為什么長(zhǎng)文本不能全文塞進(jìn)prompt上下文窗口能裝下不代表應(yīng)該裝下。有兩個(gè)現(xiàn)實(shí)壓力。第一個(gè)是成本DeepSeek按輸入token計(jì)費(fèi)塞一萬(wàn)和塞四萬(wàn)同樣的輸出費(fèi)用差四倍。第二個(gè)是首token延遲輸入越長(zhǎng)prefill階段計(jì)算越重TTFT肉眼可見(jiàn)地變長(zhǎng)。目標(biāo)是做實(shí)時(shí)數(shù)據(jù)處理這兩點(diǎn)都不能接受。更隱蔽的是注意力稀釋。關(guān)鍵信息可能埋在第9000個(gè)字符的位置前面一大段合同模板、版權(quán)聲明、重復(fù)日志會(huì)占走注意力的相當(dāng)比重。模型不是每字都讀它按權(quán)重采樣有效信息被噪音擠掉之后回答質(zhì)量明顯下滑。這也是很多團(tuán)隊(duì)即便上下文窗口翻倍文檔問(wèn)答準(zhǔn)確率也沒(méi)有跟上來(lái)的原因。所以“長(zhǎng)文本分塊處理”的本質(zhì)不是把文本切小而是把“全文在上下文中”改成“相關(guān)部分在上下文中”。讓每個(gè)請(qǐng)求里的每一個(gè)token都在貢獻(xiàn)信息量而不是在湊數(shù)。3.2 三種分塊策略和它們的適用邊界固定長(zhǎng)度分塊按字符數(shù)或token數(shù)硬切。實(shí)現(xiàn)最省事一個(gè)循環(huán)就能寫完但自然語(yǔ)言會(huì)被攔腰截?cái)嘁痪湓捘阋话胛乙话肽Ｐ秃蜋z索都容易誤解。它適合代碼、JSON、固定寬度日志這類本身有明確行邊界的內(nèi)容。遞歸字符分塊先按段落分隔符\n\n切切不動(dòng)了再降級(jí)到句子分隔符。再不行才到詞和字符。它保證每個(gè)分塊盡量是完整語(yǔ)義單元。LangChain里的RecursiveCharacterTextSplitter是這個(gè)思路的通用實(shí)現(xiàn)大部分中文文檔場(chǎng)景默認(rèn)選它都不會(huì)錯(cuò)。語(yǔ)義分塊按文檔標(biāo)題、章節(jié)、頁(yè)簽結(jié)構(gòu)或embedding相似度來(lái)確定邊界。質(zhì)量最好但需要額外解析和可能多跑一次向量計(jì)算。適合論文、合規(guī)合同、法律條文這類強(qiáng)結(jié)構(gòu)文本。策略優(yōu)點(diǎn)缺點(diǎn)適用場(chǎng)景固定長(zhǎng)度快、無(wú)依賴截?cái)嗾Z(yǔ)義、檢索質(zhì)量低日志、代碼、JSON遞歸字符語(yǔ)義完整、參數(shù)少對(duì)強(qiáng)結(jié)構(gòu)文檔不夠聰明新聞、合同、說(shuō)明文檔語(yǔ)義分塊邊界貼合語(yǔ)義要額外算力和模型調(diào)用論文、法律條文、長(zhǎng)報(bào)告3.3 chunk_size與chunk_overlap參數(shù)怎么定不算玄學(xué)用遞歸字符分塊時(shí)最常見(jiàn)的起步組合是chunk_size2000、chunk_overlap200。這里的單位是字符不是token。中文場(chǎng)景下一個(gè)漢字大體對(duì)應(yīng)0.7到1.2個(gè)token2000字符折算下來(lái)約800到1000個(gè)token落在多數(shù)任務(wù)的舒適區(qū)。如果文本是英文或中英混合同樣的2000字符對(duì)應(yīng)的token會(huì)變多需要按實(shí)際usage事件里的數(shù)字反推校準(zhǔn)。from langchain_text_splitters import RecursiveCharacterTextSplitter splitter RecursiveCharacterTextSplitter( chunk_size2000, # 字符粒度的塊大小上限 chunk_overlap200, # 相鄰塊之間的重復(fù)字符數(shù) separators[\n\n, \n, 。, , , , , , ], ) chunks splitter.split_text(long_document) for idx, c in enumerate(chunks): print(idx, len(c), c[:40], ...)chunk_overlap解決的是“關(guān)鍵信息正好卡在切縫上”的問(wèn)題。如果不重疊跨切縫的關(guān)鍵句會(huì)被劈成兩半前后兩個(gè)塊都缺信息下游再聰明也補(bǔ)不齊。overlap一般取chunk_size的10%上下2000配2001500配150。超過(guò)20%重復(fù)內(nèi)容變多同一段信息在多個(gè)塊里重復(fù)出現(xiàn)做摘要時(shí)容易被重復(fù)統(tǒng)計(jì)計(jì)費(fèi)成本也跟著漲。日志、代碼這類結(jié)構(gòu)化內(nèi)容字符密度大chunk_size可以放寬到2500到3000敘事、合同這類語(yǔ)義耦合度高的文本縮到1200到1500。這不是玄學(xué)是根據(jù)兩類文本里“一句話的平均長(zhǎng)度”推導(dǎo)的——句子越長(zhǎng)塊內(nèi)語(yǔ)義耦合越高塊就該越小。3.4 分塊之后索引、召回、再進(jìn)流式接口分塊只是生產(chǎn)管線的第一個(gè)環(huán)節(jié)。實(shí)用管線通常是文檔進(jìn)入系統(tǒng) → 分塊 → 存入帶索引的存儲(chǔ)向量庫(kù)、Elasticsearch或者一張帶關(guān)鍵詞的SQL表都行→ 遇到用戶查詢時(shí)召回TopK塊 → 拼進(jìn)prompt → 調(diào)用DeepSeek流式返回。召回環(huán)節(jié)有一個(gè)容易被忽略的聯(lián)動(dòng)塊數(shù)不要貪多。一次查詢帶兩三個(gè)塊足夠每塊2000字符總上下文兩三千token。塞五個(gè)塊以上輸入長(zhǎng)度翻倍流式TTFT顯著變長(zhǎng)用戶看到的“打字機(jī)”就變“卡帶機(jī)”。如果問(wèn)題確實(shí)橫跨多個(gè)塊寧可多輪對(duì)話逐塊追問(wèn)也別一次全堆進(jìn)去。4. 實(shí)時(shí)鏈路排障流式斷連、分塊斷裂與并發(fā)串號(hào)的五個(gè)高發(fā)坑4.1 連接被重置流走到一半再?zèng)]有下一個(gè)事件現(xiàn)象前幾個(gè)delta正常收到跑一會(huì)兒整個(gè)循環(huán)卡住或者直接拋Connection reset。原因多數(shù)是空閑超時(shí)。服務(wù)端期望客戶端持續(xù)消費(fèi)請(qǐng)求兩次事件間隔較長(zhǎng)時(shí)中間網(wǎng)關(guān)設(shè)備會(huì)認(rèn)為連接閑置直接把鏈路斷開(kāi)。另一個(gè)常見(jiàn)原因是網(wǎng)絡(luò)中間層把SSE響應(yīng)緩沖住了事件沒(méi)有及時(shí)落到客戶端。解決先排查網(wǎng)絡(luò)中間層。部署在nginx后面時(shí)把對(duì)應(yīng)路徑的proxy_buffering off加上自建網(wǎng)關(guān)要確認(rèn)它沒(méi)有把text/event-stream當(dāng)普通文本做整包緩存。再看應(yīng)用側(cè)代碼必須邊收邊處理不能再包一層“等全部讀取完”的邏輯那樣事件積壓很快觸發(fā)超時(shí)。4.2 分塊交界處語(yǔ)義斷裂模型回答前后矛盾現(xiàn)象按文檔順序逐塊喂給DeepSeek上一塊的回答和下一塊的回答對(duì)同一事實(shí)描述不一致甚至說(shuō)“沒(méi)看到相關(guān)上下文”。原因分塊做成了按行硬切句子被攔腰截?cái)?。前一個(gè)塊里有后半句沒(méi)有前半句后一個(gè)塊有后半句沒(méi)有前半句單獨(dú)看都缺主語(yǔ)模型只能靠猜。解決換成遞歸字符分塊分隔符里把。都帶上在成本允許的范圍內(nèi)調(diào)大chunk_overlap。更穩(wěn)的做法是檢索時(shí)把上一塊的末尾作為補(bǔ)充上下文拼進(jìn)prompt直接補(bǔ)上切縫兩側(cè)缺失的線索。4.3 并發(fā)請(qǐng)求一多輸出內(nèi)容互相串現(xiàn)象本地單請(qǐng)求測(cè)得好好的上線后兩個(gè)用戶同時(shí)問(wèn)把A的答案拼到了B的回復(fù)里。原因拼接緩沖區(qū)寫成了全局變量多個(gè)線程同時(shí)往同一個(gè)字符串里追加互相覆蓋。這不是DeepSeek的問(wèn)題是并發(fā)數(shù)據(jù)隔離沒(méi)做對(duì)。解決拼接緩沖區(qū)只存在于單次請(qǐng)求的函數(shù)局部作用域。如果代碼里出現(xiàn)“把同一個(gè)list或str傳到多個(gè)線程里共用”先改成每個(gè)請(qǐng)求獨(dú)立創(chuàng)建對(duì)象。需要跨線程匯總時(shí)用threading.local()或者在線程內(nèi)部算完再提交結(jié)果。這個(gè)改動(dòng)代碼量很少但能避免最隱蔽的生產(chǎn)環(huán)境事故。4.4 思維鏈被當(dāng)成最終回復(fù)推給用戶現(xiàn)象用了deepseek-reasoner用戶看到大段“推理過(guò)程”被當(dāng)成回答正主content反而在后面。原因推理模型把思考過(guò)程作為reasoning_content流式推送和最終回答是兩個(gè)字段。前端展示時(shí)把兩個(gè)字段拼在一起或者后端把reasoning_content誤當(dāng)成content處理。解決后端分別存儲(chǔ)兩個(gè)字段只把content拼給用戶reasoning_content作為審計(jì)記錄或成本分析留底。如果產(chǎn)品不展示思考過(guò)程直接選deepseek-chat模型連字段都省了。4.5 流式中途斷掉后同一段回答被計(jì)費(fèi)兩次現(xiàn)象日志顯示某次請(qǐng)求觸發(fā)了重試重試后用戶收到兩遍重復(fù)回復(fù)賬單上的completion_tokens是預(yù)期的一倍多。原因重試裝飾器包住了整個(gè)create()調(diào)用鏈路一斷整個(gè)流從頭重放。大模型服務(wù)端在沒(méi)有收到中止信號(hào)時(shí)可能已經(jīng)把之前的生成算費(fèi)了。解決重試只放在連接建立和首包返回前流進(jìn)入讀取階段后放棄自動(dòng)重發(fā)。業(yè)務(wù)層用請(qǐng)求ID或內(nèi)容哈希做冪等判斷重復(fù)消費(fèi)直接跳過(guò)。把“重試”的邊界畫在鏈路前段而不是整個(gè)請(qǐng)求外圈。5. 組合落地分塊、檢索、流式返回放進(jìn)同一條實(shí)時(shí)管道5.1 管道分四個(gè)節(jié)點(diǎn)兩個(gè)離線、兩個(gè)在線拿服務(wù)日志分析的場(chǎng)景來(lái)設(shè)計(jì)管道。原始日志每天幾百M(fèi)B全部交給DeepSeek分類不現(xiàn)實(shí)必須先入庫(kù)分塊再把塊按關(guān)鍵詞或向量建立索引。這兩步是離線預(yù)處理可以放在文檔上傳、日志落盤后的定時(shí)任務(wù)里執(zhí)行。查詢階段只做兩件事第一按用戶提問(wèn)召回最可能相關(guān)的塊第二把召回塊拼接成prompt走DeepSeek流式接口把增量結(jié)果邊收邊往展示端推。用戶的等待時(shí)間從“全文處理完”縮短到“召回完成加首token返回”塊數(shù)量少時(shí)通常一秒以內(nèi)。5.2 一個(gè)能跑通的最小實(shí)現(xiàn)下面這段代碼把四個(gè)節(jié)點(diǎn)的核心邏輯壓在一個(gè)文件里。離線部分是一次分塊在線部分是關(guān)鍵詞召回加流式生成增量用生成器函數(shù)逐段吐出。import re from openai import OpenAI from langchain_text_splitters import RecursiveCharacterTextSplitter client OpenAI(api_keysk-your-key, base_urlhttps://api.deepseek.com, timeout30.0) def pre_chunk(doc: str) - list[str]: splitter RecursiveCharacterTextSplitter( chunk_size2000, chunk_overlap200, separators[\n\n, \n, 。, , , , , , ], ) return splitter.split_text(doc) def pick_blocks(chunks: list[str], query: str, top_k: int 2) - list[str]: scores [] for c in chunks: score sum(1 for kw in re.split(r[\s,。], query) if kw and kw in c) scores.append((score, c)) scores.sort(keylambda x: x[0], reverseTrue) picked [c for _, c in scores[:top_k] if _ 0] return picked or chunks[:1] def stream_answer(blocks: list[str], query: str): context \n\n.join(blocks) resp client.chat.completions.create( modeldeepseek-chat, messages[ {role: system, content: 嚴(yán)格依據(jù)給定上下文回答找不到就明確說(shuō)不知道。}, {role: user, content: f上下文\n{context}\n\n問(wèn)題{query}}, ], streamTrue, stream_options{include_usage: True}, ) for chunk in resp: if chunk.choices and chunk.choices[0].delta.content: yield chunk.choices[0].delta.contentpick_blocks用的是關(guān)鍵詞打分召回對(duì)內(nèi)部工具體量夠用換成向量檢索只需替換這個(gè)函數(shù)的實(shí)現(xiàn)接口保持不變。stream_answer是生成器調(diào)用方用for sentence in stream_answer(...)拿到逐段增量直接作為事件推給前端。5.3 邊收邊推給前端或企微機(jī)器人別等整篇生成完拿到增量后最常見(jiàn)的錯(cuò)誤是先“收完整篇”再統(tǒng)一推送。這樣流式優(yōu)化又白做了。正確的姿勢(shì)是接收增量時(shí)就同步推送網(wǎng)頁(yè)端用SSE轉(zhuǎn)發(fā)給瀏覽器企業(yè)微信機(jī)器人按批次攢一小段再發(fā)都能讓用戶獲得秒回體驗(yàn)。并發(fā)控制放在推送層之前。同時(shí)運(yùn)行多個(gè)流式請(qǐng)求時(shí)用信號(hào)量限制同時(shí)調(diào)用的連接數(shù)避免接口被限流。import threading from concurrent.futures import ThreadPoolExecutor sem threading.Semaphore(4) def run_query(blocks, query, sink): with sem: for piece in stream_answer(blocks, query): sink.append(piece) sink [] with ThreadPoolExecutor(max_workers8) as pool: pool.submit(run_query, problem_blocks, query, sink)Semaphore(4)把在途的DeepSeek并發(fā)請(qǐng)求控制在4個(gè)以內(nèi)ThreadPoolExecutor(8)讓線程池略大于信號(hào)量即使個(gè)別請(qǐng)求在排隊(duì)其他連接也能繼續(xù)拉增量。sink示例里是list實(shí)際項(xiàng)目換成消息隊(duì)列或WebSocket通道都行。6. 進(jìn)階驗(yàn)收量TTFT、設(shè)雙超時(shí)、支持用戶中途“反悔”先說(shuō)兩個(gè)硬指標(biāo)能不能叫“實(shí)時(shí)”不能靠感覺(jué)。TTFT即從發(fā)起請(qǐng)求到收到第一個(gè)delta.content的時(shí)間。上下文兩三千token的請(qǐng)求正常公網(wǎng)環(huán)境下TTFT在一秒偏上如果穩(wěn)定到兩三秒以上檢查是不是召回塊太多、網(wǎng)絡(luò)中間有緩沖、或者并發(fā)信號(hào)量把連接卡死了。第二個(gè)指標(biāo)是token吞吐率用最終usage事件里的completion_tokens除以總耗時(shí)。重點(diǎn)不是絕對(duì)數(shù)值而是它在你本地和線上是否接近跨環(huán)境斷崖式下跌八成是網(wǎng)絡(luò)或網(wǎng)關(guān)問(wèn)題。超時(shí)設(shè)置也值得單獨(dú)說(shuō)。我給每個(gè)流式請(qǐng)求設(shè)兩個(gè)上限首token超時(shí)3秒全流超時(shí)按max_tokens乘0.1秒估算再留10%余量。前者攔“連接建立成功但沒(méi)有增量到達(dá)”的假死后者攔“生成了但鏈路遲遲不推”的靜默掛起。兩個(gè)超時(shí)都落在鏈路讀取階段不會(huì)誤傷正常慢響應(yīng)。還要支持用戶中途“反悔”。流式請(qǐng)求一旦發(fā)起用戶可能在答案方向不對(duì)時(shí)取消前端需要能通知后端停止消費(fèi)resp并立刻釋放連接而不是等整條流讀完再?zèng)Q定扔不扔。實(shí)現(xiàn)方式就是用一個(gè)取消標(biāo)志位打斷for循環(huán)再用finally關(guān)掉響應(yīng)體。這不只省token更重要的是空出的并發(fā)槽位能馬上接下一個(gè)請(qǐng)求。我自己寫流式代碼有個(gè)血淚習(xí)慣所有拼接緩沖只放在函數(shù)局部絕不跨線程共享所有重試只保連接、不保生成所有分塊都帶overlap。這三條加在一起讓我那個(gè)內(nèi)部日志巡檢工具從單請(qǐng)求等兩分鐘縮到日常平均首包八百毫秒幾千個(gè)請(qǐng)求沒(méi)再出過(guò)錯(cuò)亂。希望這套方案幫到你動(dòng)手時(shí)別學(xué)我把全局變量當(dāng)局部用的壞習(xí)慣。本文還有配套的精品資源點(diǎn)擊獲取