简介面向实时数据处理与DeepSeek应用开发者的技术方案PDF聚焦流式响应机制与长文本分块处理两大核心难题。内容从实时数据处理概述切入系统讲解DeepSeek流式响应的技术原理、分块策略选择按固定长度/语义单元/混合分块、上下文信息保留与结果整合方法并给出可运行代码实现覆盖定义分块函数、测试分块函数、实现流式响应及两者结合等步骤同时提供错误处理、性能优化GPU加速、模型量化、异步处理建议。资源还包含智能客服、新闻资讯等场景案例与实践经验总结并展望未来趋势帮助开发人员快速将方案应用到实际项目。文档结构清晰从原理到实战层层递进。文件为单个PDF共22页大小1.8MB目录完整、图文正常。已有111人学习适合正在攻克长文本实时处理、希望提升DeepSeek应用效率的工程师与算法研究人员。1. DeepSeek流式响应与长文本分块先解决“一个字一个字往外蹦却半路卡死”的问题做实时数据处理的人多半在DeepSeek API上遇到过同一个怪象开streamTrue之后首字来得很快但输出到一半连接断开、回调超时、或者拿到一段戛然而止的JSON。另一类更隐蔽——喂进去一篇文档模型还没读完就被截断答非所问。两个问题看似独立其实都指向同一件事把流式响应和长文本分块当成两个孤立功能去用没有在请求层把它们焊死在一起。这篇文章要讲的就是这套方案如何用DeepSeek的流式接口逐token接收输出如何把超长文本按token预算切成带重叠窗口的分块再让两块逻辑串成一条可复现的管线。适合正在调DeepSeek API做文档问答、日志分析、长文总结的开发者也适合想把工具调用tool calls优化进实时管线的同学。2. 流式响应与长文本分块的工作机制SSE协议、Token预检与超时控制2.1 流式为什么必须用SSE增量输出与首字延迟DeepSeek API的流式响应遵循Server-Sent EventsSSE格式即服务端把一次完整的响应拆成多帧数据逐帧推给客户端。客户端拿到的不再是一个等待3到10秒的完整JSON而是每帧几十到几百字节的文本增量。这带来的直接收益是首字延迟TTFT大幅下降用户可以感知到模型“正在工作”而不是盯着一个转圈图标焦虑等待。SSE的帧结构很简单每帧由多个字段行和一个空行组成最关键的是data:字段。DeepSeek的流式接口会在每个data:里放一个JSON对象其中choices[0].delta.content只在增量帧里有值choices[0].delta.tool_calls只在工具调用场景下出现。注意data: [DONE]这个终止标记它表示服务端已经发完所有帧。很多人在流式解析时翻车就是把data:后面的JSON当成了完整的响应体去json.loads()结果在中间帧直接抛异常。另一个容易被忽略的参数是stream_options。DeepSeek兼容OpenAI的协议当stream_options{include_usage: True}时最后一帧[DONE]前会带上累计的token用量。这个值在流式模式下特别重要因为分块策略要根据实际消耗来动态调整。没有这个数据你只能靠客户端自己数token误差在长文本场景下能到20%以上。2.2 长文本分块的三条边界Token窗口、单次请求限制、成本控制长文本分块的核心不是“按字符切”而是“按Token切”。DeepSeek模型的上下文窗口是固定的比如32K或64K具体以OpenAI兼容接口返回的模型信息为准但单次请求能安全传入的长度还要留出输出空间。常见做法是把窗口的50%到60%留给输入30%到40%留给输出留10%作为系统提示词和中间缓冲。分块时除了窗口上限还有两条容易被忽略的边界。第一是单次请求的最大输出token数DeepSeek的max_tokens参数是单次生成上限不是累计值所以每个分块都要预留足够的输出预算否则写到一半服务端主动截断。第二是重叠窗口overlap的设置分块之间要保留一定重叠区域否则一句完整的话被从中间劈开模型拿到的就是两段语义残缺的文本。中文场景下重叠窗口建议设置为块长的10%到15%并且重叠位置要尽量落在标点或段落边界附近而不是硬切在句子中间。成本控制是第三层考量。流式请求虽然按token计费但如果你每个分块都重新传一遍全部历史token消耗会呈平方级上涨。我一般会把历史记录压缩成两条一条是系统提示词加任务说明另一条是当前分块的内容。这样每个分块独立请求互不干扰网络中断时只需重发当前分块不用从头再来。3. 用Python实现DeepSeek流式响应与长文本分块最小可运行代码3.1 分块函数按Token估算与重叠窗口切分长文本先把最核心的分块逻辑写出来。这里不依赖第三方分词库用字符级估算函数做预算控制够用且没有额外依赖。import re def estimate_tokens(text: str) - int: 估算文本token数 中文按1个字符≈1个token英文按4个字符≈1个token 混合文本取两者加权实际偏保守。 if not text: return 0 # 统计中文和全角标点 cjk_chars len(re.findall(r[\u4e00-\u9fff\u3000-\u303f\uff00-\uffef], text)) # 其余按英文/数字/空格处理 other_chars len(text) - cjk_chars return int(cjk_chars * 1.0 other_chars / 3.5) 1 def split_long_text(text: str, max_chunk_tokens: int 3000, overlap_tokens: int 300): 按token预算切分长文本。 - max_chunk_tokens: 单个分块的token上限 - overlap_tokens: 相邻分块的重叠token数建议为max_chunk_tokens的10%~15% 返回分块列表每个分块附带起止字符位置。 if estimate_tokens(text) max_chunk_tokens: return [(text, 0, len(text))] chunks [] start 0 # 用标点位置做候选切分边界避免硬切半个句子 boundary_pattern re.compile(r[。\n.!?;]) while start len(text): # 计算当前块的可接受长度字符级别 # 通过token预算反推字符预算中英混合按1.6字符/token平均 char_budget int((max_chunk_tokens - estimate_tokens(text[start:start200])) * 1.6) 200 end min(start char_budget, len(text)) # 若end没到文本尾部尝试把边界后移到最近的分隔符 if end len(text): segment text[start:end] matches list(boundary_pattern.finditer(segment)) if matches: # 取最后一个分隔符后最多50个字符的位置保持上下文衔接 last_match matches[-1] if end - start - last_match.end() 50: end start last_match.end() 1 chunk text[start:end] chunks.append((chunk, start, end)) # 计算重叠区域从end位置向前回退overlap_tokens对应的字符数 overlap_chars int(overlap_tokens * 1.6) next_start max(end - overlap_chars, start 1) # 安全阀如果next_start没有前进强制前移 if next_start start: next_start end start next_start return chunks这段代码有两个关键设计。第一char_budget用“去头200字后的token余量”做反推因为开头200字已经计过一遍避免重复计算导致分块过小。第二切分时优先找。\n这些强分隔符如果找到就跳到分隔符后1个字符处这样句子不会被拦腰截断。如果找不到分隔符就硬切但重叠区域会兜住上下文。参数上max_chunk_tokens3000适合DeepSeek的32K窗口输入预算约16K token单个分块只占很小比例并发请求时不容易触发限流。如果文本是代码或日志建议把max_chunk_tokens调到1500到2000因为代码的token密度远高于自然语言按上面估算会低估实际消耗。3.2 流式请求封装SSE增量解析与指数退避重试分块是处理器流式是传输层。下一步封装一个流式请求函数它负责三件事正确解析SSE帧、合并增量内容、在连接中断时按指数退避重试。import json import time import requests def stream_chat(messages, api_key, base_urlhttps://api.deepseek.com/v1, modeldeepseek-chat, temperature0.3, max_tokens1024, max_retries3): 发送流式chat完成请求逐个增量累积content。 返回: - full_content: 拼接完成的完整响应文本 - usage: 最后一个数据帧里的token用量含prompt_tokens/completion_tokens - tool_calls_delta: 按index分组累积的tool_calls增量字典 headers { Authorization: fBearer {api_key}, Content-Type: application/json } payload { model: model, messages: messages, stream: True, temperature: temperature, max_tokens: max_tokens, stream_options: {include_usage: True} } full_content usage {} tool_calls_delta {} for attempt in range(max_retries): try: resp requests.post( f{base_url}/chat/completions, headersheaders, jsonpayload, streamTrue, timeout(10, 300) ) resp.raise_for_status() tool_calls_delta {} # 每次尝试都重置避免重试时累积脏数据 for raw_line in resp.iter_lines(decode_unicodeTrue): if not raw_line or not raw_line.startswith(data:): continue data_str raw_line[5:].strip() if data_str [DONE]: break try: frame json.loads(data_str) except json.JSONDecodeError: continue choices frame.get(choices, []) if not choices: # 无choices的帧通常是usage帧直接解析 if usage in frame: usage frame.get(usage, {}) continue delta choices[0].get(delta, {}) if delta.get(content): full_content delta[content] # tool_calls增量处理按index聚合后续可用于工具调用 if delta.get(tool_calls): for tc in delta[tool_calls]: idx tc.get(index, 0) if idx not in tool_calls_delta: tool_calls_delta[idx] {id: , type: , function: {name: , arguments: }} d tool_calls_delta[idx] if tc.get(id): d[id] tc[id] if tc.get(type): d[type] tc[type] if tc.get(function): if tc[function].get(name): d[function][name] tc[function][name] if tc[function].get(arguments): d[function][arguments] tc[function][arguments] # 正常结束返回结果 return full_content, usage, tool_calls_delta except (requests.exceptions.ConnectionError, requests.exceptions.ReadTimeout, requests.exceptions.ChunkedEncodingError) as e: if attempt max_retries - 1: raise RuntimeError(f流式请求在{max_retries}次重试后仍失败: {e}) wait_time 2 ** attempt 0.5 # 指数退避: 0.5s, 2.5s, 6.5s time.sleep(wait_time) # 重试前只保留已生成的完整内容后续新内容追加 full_content full_content return full_content, usage, tool_calls_delta这段封装把三个常见坑一并堵上。第一用resp.iter_lines(decode_unicodeTrue)逐行读而不是resp.json()这是SSE解析的基本功。第二tool_calls_delta按index分组合并因为工具调用的name和arguments在流式帧里是分片到达的直接拿最后几帧会得到残缺的JSON。第三指数退避重试只覆盖网络层异常不覆盖HTTP 4xx错误——参数错了重试多少次都一样反而浪费配额。timeout(10, 300)的意思是连接等待10秒读超时300秒。这个读超时比较宽容因为大模型生成长文本时单帧间隔可能超过60秒。如果设置太短会在模型思考时长较长时误杀连接。实际生产环境建议再加一个max_idle_seconds参数用last_frame_time做主动判断比requests的读超时更可控。3.3 全流程串联分块-逐块流式-拼接输出把分块和流式串起来需要处理好块间衔接。不能简单把每块的输出拼在一起因为模型在每块开头会重新理解上下文可能重复已提过的观点。我的做法是每块请求时带上“上一块最后一句”作为衔接提示并在系统提示词里明确“只回答当前片段不要复述历史”。def process_long_text(text_content, api_key, system_prompt你是一个严谨的文档分析助手。, max_chunk_tokens3000, overlap_tokens300, modeldeepseek-chat, temperature0.2, max_output_tokens800): 长文本处理主入口 1. 将长文本切分为带重叠的分块 2. 逐块发起流式请求实时打印并累积输出 3. 合并所有分块输出返回完整结果 chunks split_long_text(text_content, max_chunk_tokens, overlap_tokens) final_output [] all_usage {prompt_tokens: 0, completion_tokens: 0, total_tokens: 0} for idx, (chunk, start, end) in enumerate(chunks): print(f\n--- 处理第 {idx1}/{len(chunks)} 块 (字符 {start}-{end}) ---) # 构造messages首块用原文后续块带上衔接提示 if idx 0: user_msg chunk else: # 取上一块的末尾150字作为衔接上下文帮助模型理解前文 prev_tail chunks[idx-1][0][-150:] user_msg f以下是接续内容前面部分提到\n{prev_tail}\n\n请继续分析以下新片段不要复述前文\n{chunk} messages [ {role: system, content: system_prompt}, {role: user, content: user_msg} ] content, usage, tool_calls stream_chat( messagesmessages, api_keyapi_key, modelmodel, temperaturetemperature, max_tokensmax_output_tokens ) # 实时输出到控制台用于监控 print(content, flushTrue) final_output.append(content) # 累加token用量 if usage: for key in all_usage.keys(): all_usage[key] usage.get(key, 0) full_result \n.join(final_output) print(f\n全部处理完成。总token消耗: {all_usage}) return full_result, all_usage, len(chunks) # 示例调用 if __name__ __main__: # 读取一个长文档本地文件 with open(long_document.txt, r, encodingutf-8) as f: doc_text f.read() # 请替换为真实API Key生产环境建议从环境变量读取 api_key sk-你的key放这里 result, usage, chunk_count process_long_text( doc_text, api_key, system_prompt你是数据分析助手。请分块处理用户提供的内容每块给出要点摘要保持编号连续。, max_chunk_tokens2500, overlap_tokens300 ) # 可选将完整结果写入文件 with open(result_output.txt, w, encodingutf-8) as f: f.write(result)这段串联逻辑里有一个容易被忽略的细节prev_tail chunks[idx-1][0][-150:]取的是“上一块文本末尾的150个字符”而不是上一块的模型输出。原因在于模型的输出可能未经修改或包含重复直接用模型输出做衔接会引入噪声用原始文本则能精确告诉模型“前文讲到哪里了”。这个150字也不是拍脑袋定的太少了模型get不到上下文太多了会挤占当前块的输入预算。还有一个实用参数max_output_tokens800。不要把它设成和输入块一样大因为分块处理的目的是“分别消化”每块输出800字以内的摘要即可最终拼接时你需要的是要点而不是全文。4. DeepSeek流式与分块处理的避坑清单5个高频踩坑点4.1 现象流式响应只拿到最后一帧很多人写完resp.json()直接解析发现流式接口返回的不是全量内容而是最后一帧的碎片。原因requests在不设置streamTrue时会把整个响应体缓存到内存SSE帧被拼成一整个文本来解析choices[0].delta.content自然只有最后一段。解决必须按iter_lines逐行解析看到data:前缀才处理遇到[DONE]就退出循环。这是我见过最多的一个错误没有之一。4.2 现象长文本被截断模型说“内容超出我的处理范围”原因多半不是模型真的不处理长文本而是你忘了分块直接把全部文本塞进了messages。DeepSeek的上下文窗口是硬限制超了就报错或者静默截断。解决先跑一次estimate_tokens超过窗口安全线就走split_long_text分块每块单独请求。我一般把输入预算控制在窗口的50%以内这样即使模型输出较长也不会撞到天花板。4.3 现象工具调用tool_calls的arguments是乱码这是一个让很多人卡半天的怪问题。流式模式下delta.tool_calls里的arguments是分片到达的——第一帧可能是{lo第二帧cation:第三帧北京}。如果只在最后一个data帧里取arguments拿到的永远是残缺JSON。解决按index字段分组每帧到达时累加到对应分组的function.arguments字符串里全部流结束后再整体json.loads()。我在3.2节代码里已经实现了这个逻辑直接复用即可。4.4 现象分块后模型重复回答同一个问题原因相邻分块的重叠区域太大或者没有在提示词里强调“不要复述前文”。模型中“你不知道我不知道”的视角盲区在这里体现得特别明显。解决重叠token控制在10%到15%之间同时把上一块末尾的150字作为衔接上下文传给下一块并在用户消息里显式声明“这是接续内容”。这样模型会把前文当成已知信息而不是需要回答的新问题。4.5 现象长时间流式请求被网关断开重试后数据重复现象是重试后输出内容出现重复片段。原因重试时full_content已经累积了上一轮的产出但messages里还是旧的用户消息模型重新生成时不知道“你已经说完开头了”。解决重试逻辑里如果full_content已经有内容把这段内容作为assistant消息追加到messages里替换掉原来的用户消息里的对应部分。更简单的做法是直接放弃当前分块、重新发起请求虽然浪费一点token但逻辑清晰不出错。5. 把分块与流式升级成消息工具的实时管线验证方法与进阶配置5.1 工具调用场景下的二段流式当你想把DeepSeek接进Agent框架比如DeepSeek Harness或Claude Code接入DeepSeek API时长文本分块和流式响应会相互作用复杂度成倍增加。一个常见的模式是“二段流式”第一段是模型触发工具调用流式输出tool_calls的增量第二段是工具执行完成后把结果拼接回上下文模型继续流式生成指示或下一轮调用。这两段之间不能简单用一次stream_chat搞定——工具调用那部分不需要阻塞等待但必须等完整的arguments解析完毕才能真正执行工具。我的做法是把3.2节的stream_chat返回值拆开用。当tool_calls_delta非空时先不消费full_content而是等所有index分组的arguments都达到}闭合后再json.loads接着执行工具然后把结果作为tool消息追加到messages再次调用stream_chat。这样长文本分块、流式响应、工具调用三者形成一个闭环消息在管线里流动不会因为某一帧缺失而卡死。5.2 验证指标别只看“跑通”评估这套方案的工程质量我给你三个可量化的指标。第一个是“首字延迟”从调用stream_chat到第一帧content落地的耗时。正常网络环境应在0.5到2秒之间如果超过5秒说明请求排队或网络路径有问题优先排查API网关的限流策略。第二个是“分块有效率”total_tokens / (sum(所有分块预估token))。这个值越接近1越好如果低于0.8说明重叠区域算得太宽或者estimate_tokens高估了实际消耗。第三个是“中断恢复率”人为在网络中间kill掉连接统计重试成功的次数占比。成功率应达到95%以上达不到就把指数退避的基数从2改成3或者增加max_retries。5.3 进阶配置参数最后给两组常用配置模板按场景直接用。文档问答场景max_chunk_tokens2500overlap_tokens300temperature0.2max_output_tokens800streamTruestream_options{include_usage: True}。这个配置能在速度和精度之间取得平衡摘要型任务输出不会过长输入也不会挤占窗口。代码审查场景max_chunk_tokens1500overlap_tokens200temperature0max_output_tokens1200。代码的token密度高分块要更小防截断是第一优先级。我把这套方案在文档问答和日志归因两类任务上跑过最深的体会是“分块不是切字符串是切语义”。重叠窗口和衔接提示词看着不起眼却是决定输出质量的分水岭。每次模型输出重复内容先查重叠是不是太大每次工具调用解析失败先查增量合并是不是漏了帧每次超时先查timeout参数给的够不够宽。这些习惯帮我把流式处理稳定的时间从小时级压到分钟级希望帮到你。本文还有配套的精品资源点击获取 DeepSeek流式响应与长文本分块：Python实现与避坑指南
简介面向实时数据处理与DeepSeek应用开发者的技术方案PDF聚焦流式响应机制与长文本分块处理两大核心难题。内容从实时数据处理概述切入系统讲解DeepSeek流式响应的技术原理、分块策略选择按固定长度/语义单元/混合分块、上下文信息保留与结果整合方法并给出可运行代码实现覆盖定义分块函数、测试分块函数、实现流式响应及两者结合等步骤同时提供错误处理、性能优化GPU加速、模型量化、异步处理建议。资源还包含智能客服、新闻资讯等场景案例与实践经验总结并展望未来趋势帮助开发人员快速将方案应用到实际项目。文档结构清晰从原理到实战层层递进。文件为单个PDF共22页大小1.8MB目录完整、图文正常。已有111人学习适合正在攻克长文本实时处理、希望提升DeepSeek应用效率的工程师与算法研究人员。1. DeepSeek流式响应与长文本分块先解决“一个字一个字往外蹦却半路卡死”的问题做实时数据处理的人多半在DeepSeek API上遇到过同一个怪象开streamTrue之后首字来得很快但输出到一半连接断开、回调超时、或者拿到一段戛然而止的JSON。另一类更隐蔽——喂进去一篇文档模型还没读完就被截断答非所问。两个问题看似独立其实都指向同一件事把流式响应和长文本分块当成两个孤立功能去用没有在请求层把它们焊死在一起。这篇文章要讲的就是这套方案如何用DeepSeek的流式接口逐token接收输出如何把超长文本按token预算切成带重叠窗口的分块再让两块逻辑串成一条可复现的管线。适合正在调DeepSeek API做文档问答、日志分析、长文总结的开发者也适合想把工具调用tool calls优化进实时管线的同学。2. 流式响应与长文本分块的工作机制SSE协议、Token预检与超时控制2.1 流式为什么必须用SSE增量输出与首字延迟DeepSeek API的流式响应遵循Server-Sent EventsSSE格式即服务端把一次完整的响应拆成多帧数据逐帧推给客户端。客户端拿到的不再是一个等待3到10秒的完整JSON而是每帧几十到几百字节的文本增量。这带来的直接收益是首字延迟TTFT大幅下降用户可以感知到模型“正在工作”而不是盯着一个转圈图标焦虑等待。SSE的帧结构很简单每帧由多个字段行和一个空行组成最关键的是data:字段。DeepSeek的流式接口会在每个data:里放一个JSON对象其中choices[0].delta.content只在增量帧里有值choices[0].delta.tool_calls只在工具调用场景下出现。注意data: [DONE]这个终止标记它表示服务端已经发完所有帧。很多人在流式解析时翻车就是把data:后面的JSON当成了完整的响应体去json.loads()结果在中间帧直接抛异常。另一个容易被忽略的参数是stream_options。DeepSeek兼容OpenAI的协议当stream_options{include_usage: True}时最后一帧[DONE]前会带上累计的token用量。这个值在流式模式下特别重要因为分块策略要根据实际消耗来动态调整。没有这个数据你只能靠客户端自己数token误差在长文本场景下能到20%以上。2.2 长文本分块的三条边界Token窗口、单次请求限制、成本控制长文本分块的核心不是“按字符切”而是“按Token切”。DeepSeek模型的上下文窗口是固定的比如32K或64K具体以OpenAI兼容接口返回的模型信息为准但单次请求能安全传入的长度还要留出输出空间。常见做法是把窗口的50%到60%留给输入30%到40%留给输出留10%作为系统提示词和中间缓冲。分块时除了窗口上限还有两条容易被忽略的边界。第一是单次请求的最大输出token数DeepSeek的max_tokens参数是单次生成上限不是累计值所以每个分块都要预留足够的输出预算否则写到一半服务端主动截断。第二是重叠窗口overlap的设置分块之间要保留一定重叠区域否则一句完整的话被从中间劈开模型拿到的就是两段语义残缺的文本。中文场景下重叠窗口建议设置为块长的10%到15%并且重叠位置要尽量落在标点或段落边界附近而不是硬切在句子中间。成本控制是第三层考量。流式请求虽然按token计费但如果你每个分块都重新传一遍全部历史token消耗会呈平方级上涨。我一般会把历史记录压缩成两条一条是系统提示词加任务说明另一条是当前分块的内容。这样每个分块独立请求互不干扰网络中断时只需重发当前分块不用从头再来。3. 用Python实现DeepSeek流式响应与长文本分块最小可运行代码3.1 分块函数按Token估算与重叠窗口切分长文本先把最核心的分块逻辑写出来。这里不依赖第三方分词库用字符级估算函数做预算控制够用且没有额外依赖。import re def estimate_tokens(text: str) - int: 估算文本token数 中文按1个字符≈1个token英文按4个字符≈1个token 混合文本取两者加权实际偏保守。 if not text: return 0 # 统计中文和全角标点 cjk_chars len(re.findall(r[\u4e00-\u9fff\u3000-\u303f\uff00-\uffef], text)) # 其余按英文/数字/空格处理 other_chars len(text) - cjk_chars return int(cjk_chars * 1.0 other_chars / 3.5) 1 def split_long_text(text: str, max_chunk_tokens: int 3000, overlap_tokens: int 300): 按token预算切分长文本。 - max_chunk_tokens: 单个分块的token上限 - overlap_tokens: 相邻分块的重叠token数建议为max_chunk_tokens的10%~15% 返回分块列表每个分块附带起止字符位置。 if estimate_tokens(text) max_chunk_tokens: return [(text, 0, len(text))] chunks [] start 0 # 用标点位置做候选切分边界避免硬切半个句子 boundary_pattern re.compile(r[。\n.!?;]) while start len(text): # 计算当前块的可接受长度字符级别 # 通过token预算反推字符预算中英混合按1.6字符/token平均 char_budget int((max_chunk_tokens - estimate_tokens(text[start:start200])) * 1.6) 200 end min(start char_budget, len(text)) # 若end没到文本尾部尝试把边界后移到最近的分隔符 if end len(text): segment text[start:end] matches list(boundary_pattern.finditer(segment)) if matches: # 取最后一个分隔符后最多50个字符的位置保持上下文衔接 last_match matches[-1] if end - start - last_match.end() 50: end start last_match.end() 1 chunk text[start:end] chunks.append((chunk, start, end)) # 计算重叠区域从end位置向前回退overlap_tokens对应的字符数 overlap_chars int(overlap_tokens * 1.6) next_start max(end - overlap_chars, start 1) # 安全阀如果next_start没有前进强制前移 if next_start start: next_start end start next_start return chunks这段代码有两个关键设计。第一char_budget用“去头200字后的token余量”做反推因为开头200字已经计过一遍避免重复计算导致分块过小。第二切分时优先找。\n这些强分隔符如果找到就跳到分隔符后1个字符处这样句子不会被拦腰截断。如果找不到分隔符就硬切但重叠区域会兜住上下文。参数上max_chunk_tokens3000适合DeepSeek的32K窗口输入预算约16K token单个分块只占很小比例并发请求时不容易触发限流。如果文本是代码或日志建议把max_chunk_tokens调到1500到2000因为代码的token密度远高于自然语言按上面估算会低估实际消耗。3.2 流式请求封装SSE增量解析与指数退避重试分块是处理器流式是传输层。下一步封装一个流式请求函数它负责三件事正确解析SSE帧、合并增量内容、在连接中断时按指数退避重试。import json import time import requests def stream_chat(messages, api_key, base_urlhttps://api.deepseek.com/v1, modeldeepseek-chat, temperature0.3, max_tokens1024, max_retries3): 发送流式chat完成请求逐个增量累积content。 返回: - full_content: 拼接完成的完整响应文本 - usage: 最后一个数据帧里的token用量含prompt_tokens/completion_tokens - tool_calls_delta: 按index分组累积的tool_calls增量字典 headers { Authorization: fBearer {api_key}, Content-Type: application/json } payload { model: model, messages: messages, stream: True, temperature: temperature, max_tokens: max_tokens, stream_options: {include_usage: True} } full_content usage {} tool_calls_delta {} for attempt in range(max_retries): try: resp requests.post( f{base_url}/chat/completions, headersheaders, jsonpayload, streamTrue, timeout(10, 300) ) resp.raise_for_status() tool_calls_delta {} # 每次尝试都重置避免重试时累积脏数据 for raw_line in resp.iter_lines(decode_unicodeTrue): if not raw_line or not raw_line.startswith(data:): continue data_str raw_line[5:].strip() if data_str [DONE]: break try: frame json.loads(data_str) except json.JSONDecodeError: continue choices frame.get(choices, []) if not choices: # 无choices的帧通常是usage帧直接解析 if usage in frame: usage frame.get(usage, {}) continue delta choices[0].get(delta, {}) if delta.get(content): full_content delta[content] # tool_calls增量处理按index聚合后续可用于工具调用 if delta.get(tool_calls): for tc in delta[tool_calls]: idx tc.get(index, 0) if idx not in tool_calls_delta: tool_calls_delta[idx] {id: , type: , function: {name: , arguments: }} d tool_calls_delta[idx] if tc.get(id): d[id] tc[id] if tc.get(type): d[type] tc[type] if tc.get(function): if tc[function].get(name): d[function][name] tc[function][name] if tc[function].get(arguments): d[function][arguments] tc[function][arguments] # 正常结束返回结果 return full_content, usage, tool_calls_delta except (requests.exceptions.ConnectionError, requests.exceptions.ReadTimeout, requests.exceptions.ChunkedEncodingError) as e: if attempt max_retries - 1: raise RuntimeError(f流式请求在{max_retries}次重试后仍失败: {e}) wait_time 2 ** attempt 0.5 # 指数退避: 0.5s, 2.5s, 6.5s time.sleep(wait_time) # 重试前只保留已生成的完整内容后续新内容追加 full_content full_content return full_content, usage, tool_calls_delta这段封装把三个常见坑一并堵上。第一用resp.iter_lines(decode_unicodeTrue)逐行读而不是resp.json()这是SSE解析的基本功。第二tool_calls_delta按index分组合并因为工具调用的name和arguments在流式帧里是分片到达的直接拿最后几帧会得到残缺的JSON。第三指数退避重试只覆盖网络层异常不覆盖HTTP 4xx错误——参数错了重试多少次都一样反而浪费配额。timeout(10, 300)的意思是连接等待10秒读超时300秒。这个读超时比较宽容因为大模型生成长文本时单帧间隔可能超过60秒。如果设置太短会在模型思考时长较长时误杀连接。实际生产环境建议再加一个max_idle_seconds参数用last_frame_time做主动判断比requests的读超时更可控。3.3 全流程串联分块-逐块流式-拼接输出把分块和流式串起来需要处理好块间衔接。不能简单把每块的输出拼在一起因为模型在每块开头会重新理解上下文可能重复已提过的观点。我的做法是每块请求时带上“上一块最后一句”作为衔接提示并在系统提示词里明确“只回答当前片段不要复述历史”。def process_long_text(text_content, api_key, system_prompt你是一个严谨的文档分析助手。, max_chunk_tokens3000, overlap_tokens300, modeldeepseek-chat, temperature0.2, max_output_tokens800): 长文本处理主入口 1. 将长文本切分为带重叠的分块 2. 逐块发起流式请求实时打印并累积输出 3. 合并所有分块输出返回完整结果 chunks split_long_text(text_content, max_chunk_tokens, overlap_tokens) final_output [] all_usage {prompt_tokens: 0, completion_tokens: 0, total_tokens: 0} for idx, (chunk, start, end) in enumerate(chunks): print(f\n--- 处理第 {idx1}/{len(chunks)} 块 (字符 {start}-{end}) ---) # 构造messages首块用原文后续块带上衔接提示 if idx 0: user_msg chunk else: # 取上一块的末尾150字作为衔接上下文帮助模型理解前文 prev_tail chunks[idx-1][0][-150:] user_msg f以下是接续内容前面部分提到\n{prev_tail}\n\n请继续分析以下新片段不要复述前文\n{chunk} messages [ {role: system, content: system_prompt}, {role: user, content: user_msg} ] content, usage, tool_calls stream_chat( messagesmessages, api_keyapi_key, modelmodel, temperaturetemperature, max_tokensmax_output_tokens ) # 实时输出到控制台用于监控 print(content, flushTrue) final_output.append(content) # 累加token用量 if usage: for key in all_usage.keys(): all_usage[key] usage.get(key, 0) full_result \n.join(final_output) print(f\n全部处理完成。总token消耗: {all_usage}) return full_result, all_usage, len(chunks) # 示例调用 if __name__ __main__: # 读取一个长文档本地文件 with open(long_document.txt, r, encodingutf-8) as f: doc_text f.read() # 请替换为真实API Key生产环境建议从环境变量读取 api_key sk-你的key放这里 result, usage, chunk_count process_long_text( doc_text, api_key, system_prompt你是数据分析助手。请分块处理用户提供的内容每块给出要点摘要保持编号连续。, max_chunk_tokens2500, overlap_tokens300 ) # 可选将完整结果写入文件 with open(result_output.txt, w, encodingutf-8) as f: f.write(result)这段串联逻辑里有一个容易被忽略的细节prev_tail chunks[idx-1][0][-150:]取的是“上一块文本末尾的150个字符”而不是上一块的模型输出。原因在于模型的输出可能未经修改或包含重复直接用模型输出做衔接会引入噪声用原始文本则能精确告诉模型“前文讲到哪里了”。这个150字也不是拍脑袋定的太少了模型get不到上下文太多了会挤占当前块的输入预算。还有一个实用参数max_output_tokens800。不要把它设成和输入块一样大因为分块处理的目的是“分别消化”每块输出800字以内的摘要即可最终拼接时你需要的是要点而不是全文。4. DeepSeek流式与分块处理的避坑清单5个高频踩坑点4.1 现象流式响应只拿到最后一帧很多人写完resp.json()直接解析发现流式接口返回的不是全量内容而是最后一帧的碎片。原因requests在不设置streamTrue时会把整个响应体缓存到内存SSE帧被拼成一整个文本来解析choices[0].delta.content自然只有最后一段。解决必须按iter_lines逐行解析看到data:前缀才处理遇到[DONE]就退出循环。这是我见过最多的一个错误没有之一。4.2 现象长文本被截断模型说“内容超出我的处理范围”原因多半不是模型真的不处理长文本而是你忘了分块直接把全部文本塞进了messages。DeepSeek的上下文窗口是硬限制超了就报错或者静默截断。解决先跑一次estimate_tokens超过窗口安全线就走split_long_text分块每块单独请求。我一般把输入预算控制在窗口的50%以内这样即使模型输出较长也不会撞到天花板。4.3 现象工具调用tool_calls的arguments是乱码这是一个让很多人卡半天的怪问题。流式模式下delta.tool_calls里的arguments是分片到达的——第一帧可能是{lo第二帧cation:第三帧北京}。如果只在最后一个data帧里取arguments拿到的永远是残缺JSON。解决按index字段分组每帧到达时累加到对应分组的function.arguments字符串里全部流结束后再整体json.loads()。我在3.2节代码里已经实现了这个逻辑直接复用即可。4.4 现象分块后模型重复回答同一个问题原因相邻分块的重叠区域太大或者没有在提示词里强调“不要复述前文”。模型中“你不知道我不知道”的视角盲区在这里体现得特别明显。解决重叠token控制在10%到15%之间同时把上一块末尾的150字作为衔接上下文传给下一块并在用户消息里显式声明“这是接续内容”。这样模型会把前文当成已知信息而不是需要回答的新问题。4.5 现象长时间流式请求被网关断开重试后数据重复现象是重试后输出内容出现重复片段。原因重试时full_content已经累积了上一轮的产出但messages里还是旧的用户消息模型重新生成时不知道“你已经说完开头了”。解决重试逻辑里如果full_content已经有内容把这段内容作为assistant消息追加到messages里替换掉原来的用户消息里的对应部分。更简单的做法是直接放弃当前分块、重新发起请求虽然浪费一点token但逻辑清晰不出错。5. 把分块与流式升级成消息工具的实时管线验证方法与进阶配置5.1 工具调用场景下的二段流式当你想把DeepSeek接进Agent框架比如DeepSeek Harness或Claude Code接入DeepSeek API时长文本分块和流式响应会相互作用复杂度成倍增加。一个常见的模式是“二段流式”第一段是模型触发工具调用流式输出tool_calls的增量第二段是工具执行完成后把结果拼接回上下文模型继续流式生成指示或下一轮调用。这两段之间不能简单用一次stream_chat搞定——工具调用那部分不需要阻塞等待但必须等完整的arguments解析完毕才能真正执行工具。我的做法是把3.2节的stream_chat返回值拆开用。当tool_calls_delta非空时先不消费full_content而是等所有index分组的arguments都达到}闭合后再json.loads接着执行工具然后把结果作为tool消息追加到messages再次调用stream_chat。这样长文本分块、流式响应、工具调用三者形成一个闭环消息在管线里流动不会因为某一帧缺失而卡死。5.2 验证指标别只看“跑通”评估这套方案的工程质量我给你三个可量化的指标。第一个是“首字延迟”从调用stream_chat到第一帧content落地的耗时。正常网络环境应在0.5到2秒之间如果超过5秒说明请求排队或网络路径有问题优先排查API网关的限流策略。第二个是“分块有效率”total_tokens / (sum(所有分块预估token))。这个值越接近1越好如果低于0.8说明重叠区域算得太宽或者estimate_tokens高估了实际消耗。第三个是“中断恢复率”人为在网络中间kill掉连接统计重试成功的次数占比。成功率应达到95%以上达不到就把指数退避的基数从2改成3或者增加max_retries。5.3 进阶配置参数最后给两组常用配置模板按场景直接用。文档问答场景max_chunk_tokens2500overlap_tokens300temperature0.2max_output_tokens800streamTruestream_options{include_usage: True}。这个配置能在速度和精度之间取得平衡摘要型任务输出不会过长输入也不会挤占窗口。代码审查场景max_chunk_tokens1500overlap_tokens200temperature0max_output_tokens1200。代码的token密度高分块要更小防截断是第一优先级。我把这套方案在文档问答和日志归因两类任务上跑过最深的体会是“分块不是切字符串是切语义”。重叠窗口和衔接提示词看着不起眼却是决定输出质量的分水岭。每次模型输出重复内容先查重叠是不是太大每次工具调用解析失败先查增量合并是不是漏了帧每次超时先查timeout参数给的够不够宽。这些习惯帮我把流式处理稳定的时间从小时级压到分钟级希望帮到你。本文还有配套的精品资源点击获取