简介面向实时数据处理开发者与AI应用工程师的技术方案型PDF聚焦DeepSeek流式响应机制与长文本分块处理两大核心难题。文档从实时数据处理的定义、特点与应用场景切入系统梳理流式响应的技术原理、长文本分块的必要性与挑战、分块策略选择固定长度、语义单元、混合分块、上下文信息保留及结果整合方法并附有Python代码实现涵盖分块函数定义、流式响应编写、完整示例与错误处理、性能优化技巧具有较强的工程参考价值。包体共1个PDF文件22页正文约1.8MB目录结构完整章节按技术原理、方案设计、代码实现、案例实践、趋势展望有序编排阅读体验顺畅。已有111人学习下载适合希望掌握DeepSeek落地调用方式、解决长文本与实时数据处理场景难题的NLP工程师、数据开发者及技术学习者。1. 实时数据处理不是“点一次等一遍”分块和流式是一对必须同时落地的组合拳做实时数据处理的人多半被同一个场景折磨过一份六千字的合同或一天的服务日志直接拼进DeepSeek的prompt里点发送等结果那几十秒够倒一杯水生成一旦超过max_tokens或撞到窗口边界还会被拦腰截断。解法就两个全部挂在标题里长文本分块处理把输入切成小而完整的片段流式响应让结果按token一点点推出来首字延迟从几十秒压到几百毫秒。这条链路立住后用户看到的是文字像打字机一样跳出不是对着加载圈干瞪眼。这篇笔记按后端和数据工程师的落地路径走覆盖SSE接入、增量拼接、分块与overlap参数最后组合成一条实时处理管道并给出排障和验收方法。写过流式的人可以直接跳到第四章以后新手从第二章开始跟。2. DeepSeek流式响应从SSE协议到增量token的接入全流程2.1 SSE不是WebSocket它是一条“单行道”第一次用DeepSeek的流式接口时最容易误解的是响应体形状。它不是在请求完成后给你一个超大JSON而是你在请求头里把stream字段设为true后服务端把HTTP响应变成持续推送的字节流边生成边发。协议层面走的是SSEServer-Sent Events响应头的Content-Type是text/event-stream多个事件之间用空行分开每个事件以data:前缀开始全部推完后服务端发送一个data: [DONE]事件再关闭连接。SSE和WebSocket常被拿来对比但定位完全不同。WebSocket是一次握手后双向自由收发适合聊天、协作编辑这类双方都要开口的场景SSE是纯单向的服务端到客户端链路薄、兼容性好也没有WebSocket那种升级握手的负担。对大模型生成这个场景本来就只有服务端在说话SSE就是最便宜的选择。很多团队一上来就上WebSocket结果发现客户端根本没有上行需求白付了一笔心跳和连接维护的成本。SSE有一个弱点极其共性每个事件里只装增量但多数HTTP客户端默认会等整个响应体收完再返回。用requests.get(url).json()这种写法去接流式优化全部白费拿到的还是“等了全套生成完”的结果。真正要吃到流式红利要么用支持增量迭代的SDK要么手动按行解析。这两种方式本篇文章都会给到。2.2 最小接入用OpenAI兼容SDK发起流式请求DeepSeek的接口兼容OpenAI的/chat/completions格式这让我们不用重学一套客户端。base_url指到DeepSeek开放平台api_key从控制台生成模型名写deepseek-chat下面这段代码我把它当成脚手架凡是需要流式输出的地方逻辑都从它展开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: 你是数据管道助手只输出结构化内容。}, {role: user, content: 把下面这段日志按异常类型归类每类给两个例子。\n chunk_text}, ], streamTrue, stream_options{include_usage: True}, )代码很短但三个细节值得留意。第一个是streamTrue这是流式响应的开关不传它resp会一直阻塞直到完整文本生成完传了它resp变成一个可迭代对象每次迭代拿出一个事件。第二个是stream_options{include_usage: True}它让流结束前最后一个事件带上token用量方便做成本核算。很多教程不写这个参数导致最后为了统计又要再调一次计数接口。第三个是timeout30.0它约束的是连接建立和每次接收数据之间的等待上限不是整个生成周期的硬时限设得太小公网环境下一慢就会在第一个token到达前被判死。2.3 增量拼接的核心delta、reasoning_content与[DONE]的判定拿到resp之后要做的其实很机械把增量一段段接起来。“增量”在OpenAI协议里就是每个事件里的choices[0].delta.content。注意命名——是delta不是message它装的是“刚生成的那一小段”不是“现在累计的全部”。要想得到完整回答必须自己维护一个字符串变量累加。这里还要区分DeepSeek的两种模型。deepseek-chat每个事件的delta里只有contentdeepseek-reasoner则额外带reasoning_content那是模型暴露出来的思维链。这两段用途完全不同一定要分开累加、分开存储否则用户会看到大段推理过程被当成答案。full_text reasoning_text for chunk in resp: # 最后一个usage事件choices为空直接跳过。 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 # 处理业务时只用full_textreasoning_text仅作审计记录 print(full_text)这里有个血的教训最后一个携带usage的事件choices是空的如果代码里不加判空直接访问delta.content会抛AttributeError。我第一次跑通时就把这个异常当网络问题排查了半天实际只是少了一个if chunk.choices的判断。如果不用SDK而是用requests裸接SSE——比如你要写一个转发网关把DeepSeek的流原样转给下游——结束判断必须手动做。常见写法是逐行读行以data:开头时取后面内容遇到data: [DONE]退循环import json import requests payload { model: deepseek-chat, messages: [{role: user, content: 写一段300字的实施方案}], 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时[DONE]由SDK内部消化用裸HTTP时它就是你循环的出口。两种写法都正确差异在于是不是需要控制每个事件的去向。2.4 三个影响成败的参数max_tokens、temperature、重试边界max_tokens在DeepSeek接口里指的是“生成部分的最大token数”不是“请求响应的总预算”。如果这个值给得太小流会在生成中途被掐断且可能等不到[DONE]事件客户端逻辑卡在读取尾端。我一般按预期输出长度乘1.3给宁可多留余量。temperature控制采样随机性。日志归类、摘要提取这类强格式任务给0最稳开放问答给0.7以上才有味道。有一个容易忽略的连带效果temperature0时同样的问题流式输出基本一字不差温度高了以后每次用词都有细小变化。如果下游做答案全文比对这个差异可能被误判成异常。重试边界是常年翻车的点。如果把整个create()调用包在一个通用重试装饰器里一旦服务端已经生成了一部分才断连重试会把同一个回答按计费生成两遍。正确的思路是重试只放在“连接建立、首个token到达前”这个阶段进入流式读取期后不再自动重发。中途中止的流让业务层用幂等ID去判断结果是否已存在而不是盲目重放。3. 长文本分块让每个请求只带最可能被用到的上下文3.1 为什么长文本不能全文塞进prompt上下文窗口能装下不代表应该装下。有两个现实压力。第一个是成本DeepSeek按输入token计费塞一万和塞四万同样的输出费用差四倍。第二个是首token延迟输入越长prefill阶段计算越重TTFT肉眼可见地变长。目标是做实时数据处理这两点都不能接受。更隐蔽的是注意力稀释。关键信息可能埋在第9000个字符的位置前面一大段合同模板、版权声明、重复日志会占走注意力的相当比重。模型不是每字都读它按权重采样有效信息被噪音挤掉之后回答质量明显下滑。这也是很多团队即便上下文窗口翻倍文档问答准确率也没有跟上来的原因。所以“长文本分块处理”的本质不是把文本切小而是把“全文在上下文中”改成“相关部分在上下文中”。让每个请求里的每一个token都在贡献信息量而不是在凑数。3.2 三种分块策略和它们的适用边界固定长度分块按字符数或token数硬切。实现最省事一个循环就能写完但自然语言会被拦腰截断一句话你一半我一半模型和检索都容易误解。它适合代码、JSON、固定宽度日志这类本身有明确行边界的内容。递归字符分块先按段落分隔符\n\n切切不动了再降级到句子分隔符。再不行才到词和字符。它保证每个分块尽量是完整语义单元。LangChain里的RecursiveCharacterTextSplitter是这个思路的通用实现大部分中文文档场景默认选它都不会错。语义分块按文档标题、章节、页签结构或embedding相似度来确定边界。质量最好但需要额外解析和可能多跑一次向量计算。适合论文、合规合同、法律条文这类强结构文本。策略优点缺点适用场景固定长度快、无依赖截断语义、检索质量低日志、代码、JSON递归字符语义完整、参数少对强结构文档不够聪明新闻、合同、说明文档语义分块边界贴合语义要额外算力和模型调用论文、法律条文、长报告3.3 chunk_size与chunk_overlap参数怎么定不算玄学用递归字符分块时最常见的起步组合是chunk_size2000、chunk_overlap200。这里的单位是字符不是token。中文场景下一个汉字大体对应0.7到1.2个token2000字符折算下来约800到1000个token落在多数任务的舒适区。如果文本是英文或中英混合同样的2000字符对应的token会变多需要按实际usage事件里的数字反推校准。from langchain_text_splitters import RecursiveCharacterTextSplitter splitter RecursiveCharacterTextSplitter( chunk_size2000, # 字符粒度的块大小上限 chunk_overlap200, # 相邻块之间的重复字符数 separators[\n\n, \n, 。, , , , , , ], ) chunks splitter.split_text(long_document) for idx, c in enumerate(chunks): print(idx, len(c), c[:40], ...)chunk_overlap解决的是“关键信息正好卡在切缝上”的问题。如果不重叠跨切缝的关键句会被劈成两半前后两个块都缺信息下游再聪明也补不齐。overlap一般取chunk_size的10%上下2000配2001500配150。超过20%重复内容变多同一段信息在多个块里重复出现做摘要时容易被重复统计计费成本也跟着涨。日志、代码这类结构化内容字符密度大chunk_size可以放宽到2500到3000叙事、合同这类语义耦合度高的文本缩到1200到1500。这不是玄学是根据两类文本里“一句话的平均长度”推导的——句子越长块内语义耦合越高块就该越小。3.4 分块之后索引、召回、再进流式接口分块只是生产管线的第一个环节。实用管线通常是文档进入系统 → 分块 → 存入带索引的存储向量库、Elasticsearch或者一张带关键词的SQL表都行→ 遇到用户查询时召回TopK块 → 拼进prompt → 调用DeepSeek流式返回。召回环节有一个容易被忽略的联动块数不要贪多。一次查询带两三个块足够每块2000字符总上下文两三千token。塞五个块以上输入长度翻倍流式TTFT显著变长用户看到的“打字机”就变“卡带机”。如果问题确实横跨多个块宁可多轮对话逐块追问也别一次全堆进去。4. 实时链路排障流式断连、分块断裂与并发串号的五个高发坑4.1 连接被重置流走到一半再没有下一个事件现象前几个delta正常收到跑一会儿整个循环卡住或者直接抛Connection reset。原因多数是空闲超时。服务端期望客户端持续消费请求两次事件间隔较长时中间网关设备会认为连接闲置直接把链路断开。另一个常见原因是网络中间层把SSE响应缓冲住了事件没有及时落到客户端。解决先排查网络中间层。部署在nginx后面时把对应路径的proxy_buffering off加上自建网关要确认它没有把text/event-stream当普通文本做整包缓存。再看应用侧代码必须边收边处理不能再包一层“等全部读取完”的逻辑那样事件积压很快触发超时。4.2 分块交界处语义断裂模型回答前后矛盾现象按文档顺序逐块喂给DeepSeek上一块的回答和下一块的回答对同一事实描述不一致甚至说“没看到相关上下文”。原因分块做成了按行硬切句子被拦腰截断。前一个块里有后半句没有前半句后一个块有后半句没有前半句单独看都缺主语模型只能靠猜。解决换成递归字符分块分隔符里把。都带上在成本允许的范围内调大chunk_overlap。更稳的做法是检索时把上一块的末尾作为补充上下文拼进prompt直接补上切缝两侧缺失的线索。4.3 并发请求一多输出内容互相串现象本地单请求测得好好的上线后两个用户同时问把A的答案拼到了B的回复里。原因拼接缓冲区写成了全局变量多个线程同时往同一个字符串里追加互相覆盖。这不是DeepSeek的问题是并发数据隔离没做对。解决拼接缓冲区只存在于单次请求的函数局部作用域。如果代码里出现“把同一个list或str传到多个线程里共用”先改成每个请求独立创建对象。需要跨线程汇总时用threading.local()或者在线程内部算完再提交结果。这个改动代码量很少但能避免最隐蔽的生产环境事故。4.4 思维链被当成最终回复推给用户现象用了deepseek-reasoner用户看到大段“推理过程”被当成回答正主content反而在后面。原因推理模型把思考过程作为reasoning_content流式推送和最终回答是两个字段。前端展示时把两个字段拼在一起或者后端把reasoning_content误当成content处理。解决后端分别存储两个字段只把content拼给用户reasoning_content作为审计记录或成本分析留底。如果产品不展示思考过程直接选deepseek-chat模型连字段都省了。4.5 流式中途断掉后同一段回答被计费两次现象日志显示某次请求触发了重试重试后用户收到两遍重复回复账单上的completion_tokens是预期的一倍多。原因重试装饰器包住了整个create()调用链路一断整个流从头重放。大模型服务端在没有收到中止信号时可能已经把之前的生成算费了。解决重试只放在连接建立和首包返回前流进入读取阶段后放弃自动重发。业务层用请求ID或内容哈希做幂等判断重复消费直接跳过。把“重试”的边界画在链路前段而不是整个请求外圈。5. 组合落地分块、检索、流式返回放进同一条实时管道5.1 管道分四个节点两个离线、两个在线拿服务日志分析的场景来设计管道。原始日志每天几百MB全部交给DeepSeek分类不现实必须先入库分块再把块按关键词或向量建立索引。这两步是离线预处理可以放在文档上传、日志落盘后的定时任务里执行。查询阶段只做两件事第一按用户提问召回最可能相关的块第二把召回块拼接成prompt走DeepSeek流式接口把增量结果边收边往展示端推。用户的等待时间从“全文处理完”缩短到“召回完成加首token返回”块数量少时通常一秒以内。5.2 一个能跑通的最小实现下面这段代码把四个节点的核心逻辑压在一个文件里。离线部分是一次分块在线部分是关键词召回加流式生成增量用生成器函数逐段吐出。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: 严格依据给定上下文回答找不到就明确说不知道。}, {role: user, content: f上下文\n{context}\n\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用的是关键词打分召回对内部工具体量够用换成向量检索只需替换这个函数的实现接口保持不变。stream_answer是生成器调用方用for sentence in stream_answer(...)拿到逐段增量直接作为事件推给前端。5.3 边收边推给前端或企微机器人别等整篇生成完拿到增量后最常见的错误是先“收完整篇”再统一推送。这样流式优化又白做了。正确的姿势是接收增量时就同步推送网页端用SSE转发给浏览器企业微信机器人按批次攒一小段再发都能让用户获得秒回体验。并发控制放在推送层之前。同时运行多个流式请求时用信号量限制同时调用的连接数避免接口被限流。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并发请求控制在4个以内ThreadPoolExecutor(8)让线程池略大于信号量即使个别请求在排队其他连接也能继续拉增量。sink示例里是list实际项目换成消息队列或WebSocket通道都行。6. 进阶验收量TTFT、设双超时、支持用户中途“反悔”先说两个硬指标能不能叫“实时”不能靠感觉。TTFT即从发起请求到收到第一个delta.content的时间。上下文两三千token的请求正常公网环境下TTFT在一秒偏上如果稳定到两三秒以上检查是不是召回块太多、网络中间有缓冲、或者并发信号量把连接卡死了。第二个指标是token吞吐率用最终usage事件里的completion_tokens除以总耗时。重点不是绝对数值而是它在你本地和线上是否接近跨环境断崖式下跌八成是网络或网关问题。超时设置也值得单独说。我给每个流式请求设两个上限首token超时3秒全流超时按max_tokens乘0.1秒估算再留10%余量。前者拦“连接建立成功但没有增量到达”的假死后者拦“生成了但链路迟迟不推”的静默挂起。两个超时都落在链路读取阶段不会误伤正常慢响应。还要支持用户中途“反悔”。流式请求一旦发起用户可能在答案方向不对时取消前端需要能通知后端停止消费resp并立刻释放连接而不是等整条流读完再决定扔不扔。实现方式就是用一个取消标志位打断for循环再用finally关掉响应体。这不只省token更重要的是空出的并发槽位能马上接下一个请求。我自己写流式代码有个血泪习惯所有拼接缓冲只放在函数局部绝不跨线程共享所有重试只保连接、不保生成所有分块都带overlap。这三条加在一起让我那个内部日志巡检工具从单请求等两分钟缩到日常平均首包八百毫秒几千个请求没再出过错乱。希望这套方案帮到你动手时别学我把全局变量当局部用的坏习惯。本文还有配套的精品资源点击获取 DeepSeek流式响应与长文本分块：从SSE到实时处理管道
简介面向实时数据处理开发者与AI应用工程师的技术方案型PDF聚焦DeepSeek流式响应机制与长文本分块处理两大核心难题。文档从实时数据处理的定义、特点与应用场景切入系统梳理流式响应的技术原理、长文本分块的必要性与挑战、分块策略选择固定长度、语义单元、混合分块、上下文信息保留及结果整合方法并附有Python代码实现涵盖分块函数定义、流式响应编写、完整示例与错误处理、性能优化技巧具有较强的工程参考价值。包体共1个PDF文件22页正文约1.8MB目录结构完整章节按技术原理、方案设计、代码实现、案例实践、趋势展望有序编排阅读体验顺畅。已有111人学习下载适合希望掌握DeepSeek落地调用方式、解决长文本与实时数据处理场景难题的NLP工程师、数据开发者及技术学习者。1. 实时数据处理不是“点一次等一遍”分块和流式是一对必须同时落地的组合拳做实时数据处理的人多半被同一个场景折磨过一份六千字的合同或一天的服务日志直接拼进DeepSeek的prompt里点发送等结果那几十秒够倒一杯水生成一旦超过max_tokens或撞到窗口边界还会被拦腰截断。解法就两个全部挂在标题里长文本分块处理把输入切成小而完整的片段流式响应让结果按token一点点推出来首字延迟从几十秒压到几百毫秒。这条链路立住后用户看到的是文字像打字机一样跳出不是对着加载圈干瞪眼。这篇笔记按后端和数据工程师的落地路径走覆盖SSE接入、增量拼接、分块与overlap参数最后组合成一条实时处理管道并给出排障和验收方法。写过流式的人可以直接跳到第四章以后新手从第二章开始跟。2. DeepSeek流式响应从SSE协议到增量token的接入全流程2.1 SSE不是WebSocket它是一条“单行道”第一次用DeepSeek的流式接口时最容易误解的是响应体形状。它不是在请求完成后给你一个超大JSON而是你在请求头里把stream字段设为true后服务端把HTTP响应变成持续推送的字节流边生成边发。协议层面走的是SSEServer-Sent Events响应头的Content-Type是text/event-stream多个事件之间用空行分开每个事件以data:前缀开始全部推完后服务端发送一个data: [DONE]事件再关闭连接。SSE和WebSocket常被拿来对比但定位完全不同。WebSocket是一次握手后双向自由收发适合聊天、协作编辑这类双方都要开口的场景SSE是纯单向的服务端到客户端链路薄、兼容性好也没有WebSocket那种升级握手的负担。对大模型生成这个场景本来就只有服务端在说话SSE就是最便宜的选择。很多团队一上来就上WebSocket结果发现客户端根本没有上行需求白付了一笔心跳和连接维护的成本。SSE有一个弱点极其共性每个事件里只装增量但多数HTTP客户端默认会等整个响应体收完再返回。用requests.get(url).json()这种写法去接流式优化全部白费拿到的还是“等了全套生成完”的结果。真正要吃到流式红利要么用支持增量迭代的SDK要么手动按行解析。这两种方式本篇文章都会给到。2.2 最小接入用OpenAI兼容SDK发起流式请求DeepSeek的接口兼容OpenAI的/chat/completions格式这让我们不用重学一套客户端。base_url指到DeepSeek开放平台api_key从控制台生成模型名写deepseek-chat下面这段代码我把它当成脚手架凡是需要流式输出的地方逻辑都从它展开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: 你是数据管道助手只输出结构化内容。}, {role: user, content: 把下面这段日志按异常类型归类每类给两个例子。\n chunk_text}, ], streamTrue, stream_options{include_usage: True}, )代码很短但三个细节值得留意。第一个是streamTrue这是流式响应的开关不传它resp会一直阻塞直到完整文本生成完传了它resp变成一个可迭代对象每次迭代拿出一个事件。第二个是stream_options{include_usage: True}它让流结束前最后一个事件带上token用量方便做成本核算。很多教程不写这个参数导致最后为了统计又要再调一次计数接口。第三个是timeout30.0它约束的是连接建立和每次接收数据之间的等待上限不是整个生成周期的硬时限设得太小公网环境下一慢就会在第一个token到达前被判死。2.3 增量拼接的核心delta、reasoning_content与[DONE]的判定拿到resp之后要做的其实很机械把增量一段段接起来。“增量”在OpenAI协议里就是每个事件里的choices[0].delta.content。注意命名——是delta不是message它装的是“刚生成的那一小段”不是“现在累计的全部”。要想得到完整回答必须自己维护一个字符串变量累加。这里还要区分DeepSeek的两种模型。deepseek-chat每个事件的delta里只有contentdeepseek-reasoner则额外带reasoning_content那是模型暴露出来的思维链。这两段用途完全不同一定要分开累加、分开存储否则用户会看到大段推理过程被当成答案。full_text reasoning_text for chunk in resp: # 最后一个usage事件choices为空直接跳过。 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 # 处理业务时只用full_textreasoning_text仅作审计记录 print(full_text)这里有个血的教训最后一个携带usage的事件choices是空的如果代码里不加判空直接访问delta.content会抛AttributeError。我第一次跑通时就把这个异常当网络问题排查了半天实际只是少了一个if chunk.choices的判断。如果不用SDK而是用requests裸接SSE——比如你要写一个转发网关把DeepSeek的流原样转给下游——结束判断必须手动做。常见写法是逐行读行以data:开头时取后面内容遇到data: [DONE]退循环import json import requests payload { model: deepseek-chat, messages: [{role: user, content: 写一段300字的实施方案}], 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时[DONE]由SDK内部消化用裸HTTP时它就是你循环的出口。两种写法都正确差异在于是不是需要控制每个事件的去向。2.4 三个影响成败的参数max_tokens、temperature、重试边界max_tokens在DeepSeek接口里指的是“生成部分的最大token数”不是“请求响应的总预算”。如果这个值给得太小流会在生成中途被掐断且可能等不到[DONE]事件客户端逻辑卡在读取尾端。我一般按预期输出长度乘1.3给宁可多留余量。temperature控制采样随机性。日志归类、摘要提取这类强格式任务给0最稳开放问答给0.7以上才有味道。有一个容易忽略的连带效果temperature0时同样的问题流式输出基本一字不差温度高了以后每次用词都有细小变化。如果下游做答案全文比对这个差异可能被误判成异常。重试边界是常年翻车的点。如果把整个create()调用包在一个通用重试装饰器里一旦服务端已经生成了一部分才断连重试会把同一个回答按计费生成两遍。正确的思路是重试只放在“连接建立、首个token到达前”这个阶段进入流式读取期后不再自动重发。中途中止的流让业务层用幂等ID去判断结果是否已存在而不是盲目重放。3. 长文本分块让每个请求只带最可能被用到的上下文3.1 为什么长文本不能全文塞进prompt上下文窗口能装下不代表应该装下。有两个现实压力。第一个是成本DeepSeek按输入token计费塞一万和塞四万同样的输出费用差四倍。第二个是首token延迟输入越长prefill阶段计算越重TTFT肉眼可见地变长。目标是做实时数据处理这两点都不能接受。更隐蔽的是注意力稀释。关键信息可能埋在第9000个字符的位置前面一大段合同模板、版权声明、重复日志会占走注意力的相当比重。模型不是每字都读它按权重采样有效信息被噪音挤掉之后回答质量明显下滑。这也是很多团队即便上下文窗口翻倍文档问答准确率也没有跟上来的原因。所以“长文本分块处理”的本质不是把文本切小而是把“全文在上下文中”改成“相关部分在上下文中”。让每个请求里的每一个token都在贡献信息量而不是在凑数。3.2 三种分块策略和它们的适用边界固定长度分块按字符数或token数硬切。实现最省事一个循环就能写完但自然语言会被拦腰截断一句话你一半我一半模型和检索都容易误解。它适合代码、JSON、固定宽度日志这类本身有明确行边界的内容。递归字符分块先按段落分隔符\n\n切切不动了再降级到句子分隔符。再不行才到词和字符。它保证每个分块尽量是完整语义单元。LangChain里的RecursiveCharacterTextSplitter是这个思路的通用实现大部分中文文档场景默认选它都不会错。语义分块按文档标题、章节、页签结构或embedding相似度来确定边界。质量最好但需要额外解析和可能多跑一次向量计算。适合论文、合规合同、法律条文这类强结构文本。策略优点缺点适用场景固定长度快、无依赖截断语义、检索质量低日志、代码、JSON递归字符语义完整、参数少对强结构文档不够聪明新闻、合同、说明文档语义分块边界贴合语义要额外算力和模型调用论文、法律条文、长报告3.3 chunk_size与chunk_overlap参数怎么定不算玄学用递归字符分块时最常见的起步组合是chunk_size2000、chunk_overlap200。这里的单位是字符不是token。中文场景下一个汉字大体对应0.7到1.2个token2000字符折算下来约800到1000个token落在多数任务的舒适区。如果文本是英文或中英混合同样的2000字符对应的token会变多需要按实际usage事件里的数字反推校准。from langchain_text_splitters import RecursiveCharacterTextSplitter splitter RecursiveCharacterTextSplitter( chunk_size2000, # 字符粒度的块大小上限 chunk_overlap200, # 相邻块之间的重复字符数 separators[\n\n, \n, 。, , , , , , ], ) chunks splitter.split_text(long_document) for idx, c in enumerate(chunks): print(idx, len(c), c[:40], ...)chunk_overlap解决的是“关键信息正好卡在切缝上”的问题。如果不重叠跨切缝的关键句会被劈成两半前后两个块都缺信息下游再聪明也补不齐。overlap一般取chunk_size的10%上下2000配2001500配150。超过20%重复内容变多同一段信息在多个块里重复出现做摘要时容易被重复统计计费成本也跟着涨。日志、代码这类结构化内容字符密度大chunk_size可以放宽到2500到3000叙事、合同这类语义耦合度高的文本缩到1200到1500。这不是玄学是根据两类文本里“一句话的平均长度”推导的——句子越长块内语义耦合越高块就该越小。3.4 分块之后索引、召回、再进流式接口分块只是生产管线的第一个环节。实用管线通常是文档进入系统 → 分块 → 存入带索引的存储向量库、Elasticsearch或者一张带关键词的SQL表都行→ 遇到用户查询时召回TopK块 → 拼进prompt → 调用DeepSeek流式返回。召回环节有一个容易被忽略的联动块数不要贪多。一次查询带两三个块足够每块2000字符总上下文两三千token。塞五个块以上输入长度翻倍流式TTFT显著变长用户看到的“打字机”就变“卡带机”。如果问题确实横跨多个块宁可多轮对话逐块追问也别一次全堆进去。4. 实时链路排障流式断连、分块断裂与并发串号的五个高发坑4.1 连接被重置流走到一半再没有下一个事件现象前几个delta正常收到跑一会儿整个循环卡住或者直接抛Connection reset。原因多数是空闲超时。服务端期望客户端持续消费请求两次事件间隔较长时中间网关设备会认为连接闲置直接把链路断开。另一个常见原因是网络中间层把SSE响应缓冲住了事件没有及时落到客户端。解决先排查网络中间层。部署在nginx后面时把对应路径的proxy_buffering off加上自建网关要确认它没有把text/event-stream当普通文本做整包缓存。再看应用侧代码必须边收边处理不能再包一层“等全部读取完”的逻辑那样事件积压很快触发超时。4.2 分块交界处语义断裂模型回答前后矛盾现象按文档顺序逐块喂给DeepSeek上一块的回答和下一块的回答对同一事实描述不一致甚至说“没看到相关上下文”。原因分块做成了按行硬切句子被拦腰截断。前一个块里有后半句没有前半句后一个块有后半句没有前半句单独看都缺主语模型只能靠猜。解决换成递归字符分块分隔符里把。都带上在成本允许的范围内调大chunk_overlap。更稳的做法是检索时把上一块的末尾作为补充上下文拼进prompt直接补上切缝两侧缺失的线索。4.3 并发请求一多输出内容互相串现象本地单请求测得好好的上线后两个用户同时问把A的答案拼到了B的回复里。原因拼接缓冲区写成了全局变量多个线程同时往同一个字符串里追加互相覆盖。这不是DeepSeek的问题是并发数据隔离没做对。解决拼接缓冲区只存在于单次请求的函数局部作用域。如果代码里出现“把同一个list或str传到多个线程里共用”先改成每个请求独立创建对象。需要跨线程汇总时用threading.local()或者在线程内部算完再提交结果。这个改动代码量很少但能避免最隐蔽的生产环境事故。4.4 思维链被当成最终回复推给用户现象用了deepseek-reasoner用户看到大段“推理过程”被当成回答正主content反而在后面。原因推理模型把思考过程作为reasoning_content流式推送和最终回答是两个字段。前端展示时把两个字段拼在一起或者后端把reasoning_content误当成content处理。解决后端分别存储两个字段只把content拼给用户reasoning_content作为审计记录或成本分析留底。如果产品不展示思考过程直接选deepseek-chat模型连字段都省了。4.5 流式中途断掉后同一段回答被计费两次现象日志显示某次请求触发了重试重试后用户收到两遍重复回复账单上的completion_tokens是预期的一倍多。原因重试装饰器包住了整个create()调用链路一断整个流从头重放。大模型服务端在没有收到中止信号时可能已经把之前的生成算费了。解决重试只放在连接建立和首包返回前流进入读取阶段后放弃自动重发。业务层用请求ID或内容哈希做幂等判断重复消费直接跳过。把“重试”的边界画在链路前段而不是整个请求外圈。5. 组合落地分块、检索、流式返回放进同一条实时管道5.1 管道分四个节点两个离线、两个在线拿服务日志分析的场景来设计管道。原始日志每天几百MB全部交给DeepSeek分类不现实必须先入库分块再把块按关键词或向量建立索引。这两步是离线预处理可以放在文档上传、日志落盘后的定时任务里执行。查询阶段只做两件事第一按用户提问召回最可能相关的块第二把召回块拼接成prompt走DeepSeek流式接口把增量结果边收边往展示端推。用户的等待时间从“全文处理完”缩短到“召回完成加首token返回”块数量少时通常一秒以内。5.2 一个能跑通的最小实现下面这段代码把四个节点的核心逻辑压在一个文件里。离线部分是一次分块在线部分是关键词召回加流式生成增量用生成器函数逐段吐出。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: 严格依据给定上下文回答找不到就明确说不知道。}, {role: user, content: f上下文\n{context}\n\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用的是关键词打分召回对内部工具体量够用换成向量检索只需替换这个函数的实现接口保持不变。stream_answer是生成器调用方用for sentence in stream_answer(...)拿到逐段增量直接作为事件推给前端。5.3 边收边推给前端或企微机器人别等整篇生成完拿到增量后最常见的错误是先“收完整篇”再统一推送。这样流式优化又白做了。正确的姿势是接收增量时就同步推送网页端用SSE转发给浏览器企业微信机器人按批次攒一小段再发都能让用户获得秒回体验。并发控制放在推送层之前。同时运行多个流式请求时用信号量限制同时调用的连接数避免接口被限流。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并发请求控制在4个以内ThreadPoolExecutor(8)让线程池略大于信号量即使个别请求在排队其他连接也能继续拉增量。sink示例里是list实际项目换成消息队列或WebSocket通道都行。6. 进阶验收量TTFT、设双超时、支持用户中途“反悔”先说两个硬指标能不能叫“实时”不能靠感觉。TTFT即从发起请求到收到第一个delta.content的时间。上下文两三千token的请求正常公网环境下TTFT在一秒偏上如果稳定到两三秒以上检查是不是召回块太多、网络中间有缓冲、或者并发信号量把连接卡死了。第二个指标是token吞吐率用最终usage事件里的completion_tokens除以总耗时。重点不是绝对数值而是它在你本地和线上是否接近跨环境断崖式下跌八成是网络或网关问题。超时设置也值得单独说。我给每个流式请求设两个上限首token超时3秒全流超时按max_tokens乘0.1秒估算再留10%余量。前者拦“连接建立成功但没有增量到达”的假死后者拦“生成了但链路迟迟不推”的静默挂起。两个超时都落在链路读取阶段不会误伤正常慢响应。还要支持用户中途“反悔”。流式请求一旦发起用户可能在答案方向不对时取消前端需要能通知后端停止消费resp并立刻释放连接而不是等整条流读完再决定扔不扔。实现方式就是用一个取消标志位打断for循环再用finally关掉响应体。这不只省token更重要的是空出的并发槽位能马上接下一个请求。我自己写流式代码有个血泪习惯所有拼接缓冲只放在函数局部绝不跨线程共享所有重试只保连接、不保生成所有分块都带overlap。这三条加在一起让我那个内部日志巡检工具从单请求等两分钟缩到日常平均首包八百毫秒几千个请求没再出过错乱。希望这套方案帮到你动手时别学我把全局变量当局部用的坏习惯。本文还有配套的精品资源点击获取