简介本资源为一份聚焦 DeepSeek 批量请求与异步调用的实战型 PDF 文档面向希望提升大模型接口调用效率的技术开发人员、数据工程师及自动化脚本编写者。文档从 DeepSeek 的技术特点与典型应用场景切入系统梳理了批量请求的通信开销优势和高并发适用场景并给出基于 Python asyncio、JavaScript Promise/async-await 的实现思路覆盖请求构造、响应解析、错误处理、重试机制以及并发数量控制等关键环节。同时配有性能测试指标、常见问题排查和优化策略方便读者按目录快速定位并迁移到实际项目中。压缩包共 1 个文件为 PDF 格式整体大小 1.74MB内容完整、目录清晰、排版无异常。当前已有 63 人学习浏览适合正在接入 DeepSeek API 并希望提升请求吞吐量、降低响应延迟的开发者作为参考手册使用。1. 当串行请求吃掉你一整天DeepSeek批量调用到底慢在哪用DeepSeek API做过批量任务的人大概率都遇到过这种场景拿一段Python脚本for循环里塞了100条调用跑了两个小时还剩80条。不是DeepSeek慢是你的调用方式压根没把API用起来。DeepSeek批量请求与异步调用这个方向解决的就是这类问题——把耗时从小时级压到分钟级把并发从1路顶到几十路把同样的token预算换成更多产出。这篇东西写给两类人。一类是刚接触DeepSeek API手头有几千条文本要批量打标、批量改写、批量总结正在用同步循环硬扛的开发者另一类是用LangChain、Codex、Claude Code这类工具接入了DeepSeek但总卡在请求排队上的人。我会把批量请求怎么拆、并发数怎么定、异步任务怎么排队、失败怎么补偿拆开讲清楚代码是可直接复制的Python版本参数是踩过坑之后的实测值。2. 批量请求的底层逻辑你的100次调用为什么跑了2小时2.1 一次DeepSeek API调用到底发生了什么先拆一次调用。你发送一个HTTP POST到DeepSeek的chat/completions接口请求体里带着model、messages、temperature、max_tokens这些参数。服务端收到后先做鉴权和参数校验然后进入排队。DeepSeek的推理服务是共享的高峰期队列长度会明显变长这一步消耗的时间从几百毫秒到几秒不等完全取决于你所在的时段和账号的负载等级。真正生成token的时候速度主要受max_tokens影响输出越长耗时越长。常见的误区是只盯着生成阶段算时间。实际上一次完整调用的耗时 网络往返RTT 排队等待 token生成时间。你在本机用SDK调DeepSeek和直接用curl调底层都是这个链路。很多人用requests库写for循环每发一个请求就干等返回网络往返的延迟被完全晾在串行逻辑里。假如你调100次每次往返1.5秒光是网络往返就是150秒这还没算生成时间。批量请求的核心思路就是把这150秒的串行等待压缩成并发的重叠等待。2.2 同步循环与并发请求差距不是几倍是几十倍直接看代码对比。同步写法是这样的import requests import json import time API_KEY your-deepseek-api-key URL https://api.deepseek.com/chat/completions def sync_call(text, api_keyAPI_KEY): headers { Authorization: fBearer {api_key}, Content-Type: application/json } payload { model: deepseek-chat, messages: [ {role: user, content: f请对以下文本做情感分类只输出正面或负面{text}} ], temperature: 0.3, max_tokens: 100 } resp requests.post(URL, headersheaders, jsonpayload, timeout30) resp.raise_for_status() return resp.json()[choices][0][message][content] texts [今天天气不错] * 50 # 模拟50条待处理文本 start time.time() results [sync_call(t) for t in texts] print(f同步耗时: {time.time() - start:.2f}s)这段代码的问题很典型列表推导式里逐个调用sync_call每条调用必须等上一条完全返回包括网络往返和生成时间才轮到下一条。50条数据不多但如果有2000条你就得盯着进度条干等。更麻烦的是同步代码里一旦某条请求因为超时或限流失败异常直接抛出后面的任务全部中断连断点续跑都得自己另外写。改用并发后同样的50条数据只要并发数设置合理耗时基本等于最慢的那一条请求而不是所有请求的耗时之和。实际跑下来50条文本从同步的60到90秒压到8到12秒是常态这就是「10倍效率提升」的主要来源。并发不是玄学只是把等待时间重叠了而已。2.3 并发不是无限开DeepSeek限流参数与并发上限估算并发请求不是把线程数拉到100就完事。DeepSeek的API有速率限制单位时间内的请求次数RPM和每分钟token消耗量TPM都有限制。超了会返回429状态码响应体里带着rate limit reached之类提示。我测过不同账号的默认限制普通key大概在每分钟几十次到几百次请求之间浮动具体数值你可以在DeepSeek开放平台的账号面板里查。安全起见我一般把并发数控制在10到20之间重试逻辑兜底。这样既不会触发限流又能把大部分空闲等待吃掉。如果你的任务类型是短文本、低max_tokens可以把并发拉到30如果是长文本生成每条请求要跑十几秒并发数反而要降下来否则大量请求同时到达服务端排队时间急剧上升总耗时反而变长。核心指标不是并发数本身而是「每秒实际发出的请求数」不超过RPM限制的80%。3. 用Python把DeepSeek批量请求跑起来从并发到异步3.1 第一版并发脚本ThreadPoolExecutor与请求函数改造用Python标准库里的concurrent.futures.ThreadPoolExecutor做并发是最快能落地的方案不需要引入额外的异步框架。你只需要把原来的同步调用函数包一层丢进线程池即可。改造之后main函数里用as_completed收集结果每条任务的返回顺序和提交顺序无关谁先回来谁先写盘。import requests import json import time from concurrent.futures import ThreadPoolExecutor, as_completed API_KEY your-deepseek-api-key URL https://api.deepseek.com/chat/completions def call_deepseek(content): headers { Authorization: fBearer {API_KEY}, Content-Type: application/json } payload { model: deepseek-chat, messages: [{role: user, content: content}], temperature: 0.3, max_tokens: 200 } resp requests.post(URL, headersheaders, jsonpayload, timeout60) resp.raise_for_status() return resp.json()[choices][0][message][content] def batch_process_with_threads(samples, max_workers10): results {} with ThreadPoolExecutor(max_workersmax_workers) as executor: future_map {executor.submit(call_deepseek, s): idx for idx, s in enumerate(samples)} for future in as_completed(future_map): idx future_map[future] try: results[idx] future.result() except Exception as e: results[idx] fERROR: {e} return [results[i] for i in range(len(samples))] if __name__ __main__: samples [请用一句话介绍杭州] * 30 start time.time() output batch_process_with_threads(samples, max_workers10) print(f并发耗时: {time.time() - start:.2f}s) print(json.dumps(output[:2], ensure_asciiFalse, indent2))逻辑说明executor.submit把每个请求包装成一个future对象as_completed在future完成时立刻返回不需要等所有任务结束才统一收结果。代码里我故意把results设计成按下标存储这样最终返回的列表顺序和输入样本顺序保持一致避免并发把顺序打乱之后没法对应回原文本。参数说明max_workers10是线程池最大并发数实测DeepSeek普通key稳定跑这个值基本不触发限流。timeout60建议只增不减因为长文本生成真正跑到60秒以上的情况并不罕见设短了会误杀正常请求。temperature0.3是DeepSeek比较稳妥的批量任务参数分类、打标、改写这类结构化任务不要开太高。3.2 真异步方案用asyncio aiohttp改写后的完整流程线程池方案能解决80%的问题但线程是有成本的——每个线程占内存和上下文切换开销max_workers调高之后性能提升会越来越不明显。更干净的做法是用asyncioaiohttp做真正的IO异步请求。注意asyncio不是多线程它是事件循环里跑协程用一个线程处理所有并发IO对限流的控制也更直观。import aiohttp import asyncio import json import time API_KEY your-deepseek-api-key URL https://api.deepseek.com/chat/completions MAX_CONCURRENT 15 # 信号量控制并发上限 async def call_deepseek_async(session, content, semaphore): headers { Authorization: fBearer {API_KEY}, Content-Type: application/json } payload { model: deepseek-chat, messages: [{role: user, content: content}], temperature: 0.3, max_tokens: 200 } async with semaphore: try: async with session.post(URL, headersheaders, jsonpayload, timeout60) as resp: data await resp.json() if resp.status ! 200: return fHTTP {resp.status}: {data} return data[choices][0][message][content] except asyncio.TimeoutError: return ERROR: timeout async def run_batch(samples): semaphore asyncio.Semaphore(MAX_CONCURRENT) async with aiohttp.ClientSession() as session: tasks [call_deepseek_async(session, s, semaphore) for s in samples] results await asyncio.gather(*tasks) return results if __name__ __main__: samples [用一句话描述量子计算] * 30 start time.time() results asyncio.run(run_batch(samples)) print(f异步耗时: {time.time() - start:.2f}s) print(json.dumps(results[:2], ensure_asciiFalse, indent2))逻辑说明asyncio.Semaphore(MAX_CONCURRENT)是整个异步方案里最关键的限流手段。没有信号量的话asyncio会一次性把所有请求全部发出去30条还好3000条直接把服务端打秃429全返回来。信号量保证同时活跃的请求数不超过15。asyncio.gather(*tasks)并发跑所有协程并等待全部完成返回顺序和tasks列表顺序一致天然对齐样本顺序。参数说明MAX_CONCURRENT15是我在DeepSeek上压测过比较稳的值token生成速度快的短文本任务可以到20长文本任务建议降到8。session.post里的timeout60是aiohttp的总体超时如果你的任务动不动要输出上千token放宽到120更稳妥。asyncio.run是Python 3.7的入口别在Jupyter Notebook里直接调用会报事件循环冲突。3.3 批量改写与批量打标的参数模板temperature与max_tokens怎么配同样是批量请求任务类型不同参数模板差异很大。批量分类打标任务我推荐temperature0.1把随机性压到最低输出稳定不会出现同一条文本两次分类结果不一样。批量改写或者摘要任务temperature0.3到0.5之间保留表达变化但不会跑偏。批量创意文案才建议开0.7以上但那就不是这篇要聊的效率问题了。max_tokens的坑最多。很多人以为设得越大越好其实max_tokens只限制最大输出长度不限制模型实际生成量而且它同时占着TPM限流的预算。设个500的结果是模型已经回答完了但token预算算的是你声明的最大值还是实际消耗DeepSeek按实际消耗计费限流则按实际tokens计算。我一般用200到300的max_tokens做批量总结输出意外的长文本时再单条重跑而不是把全局配成1000。还有一个提示词层面的效率技巧用system prompt固定输出格式。批量任务里最耗token的不是内容生成而是模型在自由发挥前后夹带解释。你在system里写明「只输出JSON不要多余文字」单条请求能省下20到50个token批量几千条的时候省下的总量就很可观了。4. 处理速率限制与错误重试DeepSeek批量调用避坑指南4.1 429限流报错现象、定位与指数退避重试现象批量脚本跑到某个时刻日志里开始出现429 Too Many Requests响应体里通常有rate limit reached字样。同步循环里出现一次429整个脚本就停了并发方案里429通常是一批一批出现的因为你瞬时打出的请求数超过了RPM限制。原因DeepSeek的速率限制是滑动窗口计数你在1分钟内打出了超过限额的请求数。并发数开得越高越容易撞上这个窗口。还有一个隐蔽因素如果你同时运行多个脚本进程比如本机跑一个批量任务另一个服务也在调DeepSeek共享同一个API key的配额撞限流的概率翻倍。解决用指数退避重试而不是固定间隔重试。第一次失败等1秒第二次等2秒第三次等4秒最多等10秒后放弃。代码实现里加一个带退避的重试装饰器或者直接在请求函数里做循环重试import time import random def call_with_retry(content, max_retries5): for attempt in range(max_retries): try: resp requests.post(URL, headersheaders, jsonpayload, timeout60) if resp.status_code 429: retry_after int(resp.headers.get(Retry-After, 2 ** attempt)) time.sleep(min(retry_after random.uniform(0, 1), 10)) continue resp.raise_for_status() return resp.json()[choices][0][message][content] except requests.exceptions.ConnectionError: wait min(2 ** attempt random.uniform(0, 1), 10) time.sleep(wait) return ERROR: retry exhausted重试代码的核心是把429响应头里的Retry-After作为首选等待时间如果服务端没给用2的attempt次方做退避指数增长比固定间隔更适合限流场景。重试5次之后还失败直接返回错误字符串等主流程统一处理不要让单条失败阻塞整个批次。4.2 超时误杀timeout设短了会让正常慢请求全军覆没现象并发数调到20以上后突然出现大量TimeoutError但你单独发一条同样的请求5秒就返回了。排查发现不是DeepSeek挂了你是某些请求排队时间过长你的客户端超时先到了。原因并发提高后服务端排队时间同步拉长。API在高峰期可能让请求排队十几秒才开始生成token你的timeout30把排队时间也算进去了排队时间一超就断连。解决把超时拆成连接超时和读取超时。连接超时5秒足够读取超时给到120秒。requests库的timeout参数可以传元组timeout(5, 120)。连接超时负责快速失败服务端不可达时马上暴露问题读取超时负责给模型生成留足时间。并发越高读取超时要给得越宽因为排队时间的方差会变大。4.3 顺序错乱与结果对不上并发脚本里最常见的翻车点现象并发跑完结果文件里第10条文本对应的输出内容明显是对第25条文本的回复。检查代码逻辑没错但结果就是乱了。原因线程池方案里如果你用executor.map或者不按下标收集结果任务的返回顺序和提交顺序并不保证一致。asyncio方案里gather虽然保持顺序但如果你提前把部分结果写进了文件后续结果覆盖错位同样会乱。解决统一用「输入带ID、输出带ID」的方式。每条待处理文本在一开始就分配一个唯一ID请求消息里带上这个ID返回结果按ID写回对应槽位。整理结果时不做顺序假设只按ID合并。这是我踩过最大的坑两千条数据跑了几十分钟最后因为结果对应不上全部重跑。4.4 本地部署与API调用的取舍vllm部署时并发参数要换一套如果你不是用DeepSeek官方API而是通过vllm在本地部署DeepSeek模型批量请求的并发逻辑要调整。vllm服务端本身支持高并发请求排队它的吞吐量瓶颈在GPU显存和batch size而不是RPM限制。本地部署时客户端的并发数可以调到50甚至更高但要注意max_num_seqs参数vllm默认值是256超过这个值请求会在服务端排队客户端超时设置同样要放宽。本地部署还有一层参数在服务端启动命令里--max-model-len控制最大上下文长度--gpu-memory-utilization控制显存利用比例。批量任务如果文本长度参差不齐建议把max-model-len设得比最长样本大20%左右否则长文本直接给你截断。本地部署的好处是彻底避开限流和费用问题坏处是服务端的并发能力受硬件限制参数调优的复杂度上了一个台阶。5. 把批量请求推到生产级任务队列、断点续跑与成本控制5.1 用队列替代裸并发批量任务里tokens、优先级与失败补偿怎么设计裸并发方案做几百条数据可以做几万条就不够用了。生产级做法是把待处理文本丢进任务队列消费者从队列里取任务处理完写结果。队列的好处有三个一是你可以随时暂停和恢复不丢任务二是可以按优先级插队紧急文本插到队首三是失败补偿可以集中在队列层面做——某条任务重试耗尽后把它重新塞回队列尾部标记重试次数上限累计重试超限才放弃。实现上不必上重型消息队列用Python标准库queue.PriorityQueue就够。优先级设计成小数值优先普通任务优先级设10紧急任务设1这样插队逻辑天然成立。队列里存待处理文本的元组(priority, index, text, retry_count)消费者取任务时按优先级出队。5.2 断点续跑程序被中断后怎么不从头再来批量任务最怕的不是慢是跑了一半挂了然后从头开始。断点续跑的常规做法是给每条任务加状态标记结果落盘采用「先写临时文件完成后改后缀」的策略。每条任务处理完把结果追加到一个done.jsonl文件里启动时扫描done.jsonl已处理的样本ID跳过。import os import json DONE_FILE done.jsonl def load_done_ids(file_pathDONE_FILE): if not os.path.exists(file_path): return set() done_ids set() with open(file_path, r, encodingutf-8) as f: for line in f: record json.loads(line) done_ids.add(record[id]) return done_ids def append_done(record, file_pathDONE_FILE): with open(file_path, a, encodingutf-8) as f: f.write(json.dumps(record, ensure_asciiFalse) \n) # 在消费任务前检查 done_ids load_done_ids() for item in all_samples: if item[id] in done_ids: continue result process(item[text]) append_done({id: item[id], result: result})这个做法的核心逻辑是把「进度」固化到磁盘上而不是依赖内存变量。任何时刻进程崩溃、断网、手动停止重启后load_done_ids自动跳过已完成任务。done.jsonl是追加写不会因为覆盖而丢数据。我在处理3000条文本时就是靠这一招省掉了多次返工——更早期的方案是整体结果统一写一个文件写到一半挂掉整个文件损坏全部重跑。5.3 成本翻倍不是效率翻倍并发调用如何控制token消耗并发把时间压缩了10倍不代表成本会同样增长。DeepSeek按token计费成本取决于你发送了多少输入token和接收了多少输出token。并发改变的是请求节奏不改变单条请求的token消耗。唯一可能让成本爆炸的是重试——429限流后重试的请求会重复计费输入token超时后重发的请求也会重复算。你把重试逻辑写好等于直接控制了成本上限。另一个成本控制点是输入token压缩。批量任务里提示词模板越精简越好。比如你要让DeepSeek做关键词提取不要在每条请求里重复一大段背景说明尽量把背景信息压缩进一两条system消息里短时间内重复的内容不再反复写。几千条短文本批量调用时提示词精简20%总成本可能降30%以上。5.4 长任务分割与合并一条大请求拆成多条小请求反而更快遇到超长文本处理比如几万字的长文档总结不要指望一条请求解决。DeepSeek的上下文窗口虽然支持长文本但超长输入有两个代价单条请求耗时暴涨、max_tokens输出不够用。实践做法是把长文档按段落或固定长度切片每个切片独立生成摘要最后把分摘要合并成一条新请求让它基于分摘要生成总摘要。分割粒度放在500字到2000字之间比较合适太碎了调用次数增加太粗了摘要质量下降。6. 验证压测结果与调优一次批量任务从90分钟到9分钟的完整路径把上面的方案组合起来做一次真实压测。样本3050条中文短文本每条需要DeepSeek输出100到150字的摘要。第一次跑用的是同步循环耗时94分钟。改成ThreadPoolExecutor并发10线程后耗时12分钟。换asyncio加信号量控制、并发数调整到15后稳定在9分半左右。中间穿插了三次429限流指数退避都兜住了任务没有中断。验证效率提升有个简单方法日志里给每条任务记下开始时间和结束时间最后统计整个批次的p50和p95单条耗时。p95显著低于平均值加上网络延迟时说明并发起了作用如果p95反而升高说明并发打到了限流边界降并发数或者拉大重试间隔。最后一个习惯我每次跑大批量任务之前都会先跑20条小样本验证参数。20条样本3分钟内出结果能看出有没有限流、输出格式对不对、token预算够不够。不要拿全部数据直接上生产配置后悔药只有断点续跑这一种。希望这篇实战笔记帮你把DeepSeek的批量调用效率真正拉起来。本文还有配套的精品资源点击获取 DeepSeek批量请求与异步调用实战：并发提速10倍全攻略
简介本资源为一份聚焦 DeepSeek 批量请求与异步调用的实战型 PDF 文档面向希望提升大模型接口调用效率的技术开发人员、数据工程师及自动化脚本编写者。文档从 DeepSeek 的技术特点与典型应用场景切入系统梳理了批量请求的通信开销优势和高并发适用场景并给出基于 Python asyncio、JavaScript Promise/async-await 的实现思路覆盖请求构造、响应解析、错误处理、重试机制以及并发数量控制等关键环节。同时配有性能测试指标、常见问题排查和优化策略方便读者按目录快速定位并迁移到实际项目中。压缩包共 1 个文件为 PDF 格式整体大小 1.74MB内容完整、目录清晰、排版无异常。当前已有 63 人学习浏览适合正在接入 DeepSeek API 并希望提升请求吞吐量、降低响应延迟的开发者作为参考手册使用。1. 当串行请求吃掉你一整天DeepSeek批量调用到底慢在哪用DeepSeek API做过批量任务的人大概率都遇到过这种场景拿一段Python脚本for循环里塞了100条调用跑了两个小时还剩80条。不是DeepSeek慢是你的调用方式压根没把API用起来。DeepSeek批量请求与异步调用这个方向解决的就是这类问题——把耗时从小时级压到分钟级把并发从1路顶到几十路把同样的token预算换成更多产出。这篇东西写给两类人。一类是刚接触DeepSeek API手头有几千条文本要批量打标、批量改写、批量总结正在用同步循环硬扛的开发者另一类是用LangChain、Codex、Claude Code这类工具接入了DeepSeek但总卡在请求排队上的人。我会把批量请求怎么拆、并发数怎么定、异步任务怎么排队、失败怎么补偿拆开讲清楚代码是可直接复制的Python版本参数是踩过坑之后的实测值。2. 批量请求的底层逻辑你的100次调用为什么跑了2小时2.1 一次DeepSeek API调用到底发生了什么先拆一次调用。你发送一个HTTP POST到DeepSeek的chat/completions接口请求体里带着model、messages、temperature、max_tokens这些参数。服务端收到后先做鉴权和参数校验然后进入排队。DeepSeek的推理服务是共享的高峰期队列长度会明显变长这一步消耗的时间从几百毫秒到几秒不等完全取决于你所在的时段和账号的负载等级。真正生成token的时候速度主要受max_tokens影响输出越长耗时越长。常见的误区是只盯着生成阶段算时间。实际上一次完整调用的耗时 网络往返RTT 排队等待 token生成时间。你在本机用SDK调DeepSeek和直接用curl调底层都是这个链路。很多人用requests库写for循环每发一个请求就干等返回网络往返的延迟被完全晾在串行逻辑里。假如你调100次每次往返1.5秒光是网络往返就是150秒这还没算生成时间。批量请求的核心思路就是把这150秒的串行等待压缩成并发的重叠等待。2.2 同步循环与并发请求差距不是几倍是几十倍直接看代码对比。同步写法是这样的import requests import json import time API_KEY your-deepseek-api-key URL https://api.deepseek.com/chat/completions def sync_call(text, api_keyAPI_KEY): headers { Authorization: fBearer {api_key}, Content-Type: application/json } payload { model: deepseek-chat, messages: [ {role: user, content: f请对以下文本做情感分类只输出正面或负面{text}} ], temperature: 0.3, max_tokens: 100 } resp requests.post(URL, headersheaders, jsonpayload, timeout30) resp.raise_for_status() return resp.json()[choices][0][message][content] texts [今天天气不错] * 50 # 模拟50条待处理文本 start time.time() results [sync_call(t) for t in texts] print(f同步耗时: {time.time() - start:.2f}s)这段代码的问题很典型列表推导式里逐个调用sync_call每条调用必须等上一条完全返回包括网络往返和生成时间才轮到下一条。50条数据不多但如果有2000条你就得盯着进度条干等。更麻烦的是同步代码里一旦某条请求因为超时或限流失败异常直接抛出后面的任务全部中断连断点续跑都得自己另外写。改用并发后同样的50条数据只要并发数设置合理耗时基本等于最慢的那一条请求而不是所有请求的耗时之和。实际跑下来50条文本从同步的60到90秒压到8到12秒是常态这就是「10倍效率提升」的主要来源。并发不是玄学只是把等待时间重叠了而已。2.3 并发不是无限开DeepSeek限流参数与并发上限估算并发请求不是把线程数拉到100就完事。DeepSeek的API有速率限制单位时间内的请求次数RPM和每分钟token消耗量TPM都有限制。超了会返回429状态码响应体里带着rate limit reached之类提示。我测过不同账号的默认限制普通key大概在每分钟几十次到几百次请求之间浮动具体数值你可以在DeepSeek开放平台的账号面板里查。安全起见我一般把并发数控制在10到20之间重试逻辑兜底。这样既不会触发限流又能把大部分空闲等待吃掉。如果你的任务类型是短文本、低max_tokens可以把并发拉到30如果是长文本生成每条请求要跑十几秒并发数反而要降下来否则大量请求同时到达服务端排队时间急剧上升总耗时反而变长。核心指标不是并发数本身而是「每秒实际发出的请求数」不超过RPM限制的80%。3. 用Python把DeepSeek批量请求跑起来从并发到异步3.1 第一版并发脚本ThreadPoolExecutor与请求函数改造用Python标准库里的concurrent.futures.ThreadPoolExecutor做并发是最快能落地的方案不需要引入额外的异步框架。你只需要把原来的同步调用函数包一层丢进线程池即可。改造之后main函数里用as_completed收集结果每条任务的返回顺序和提交顺序无关谁先回来谁先写盘。import requests import json import time from concurrent.futures import ThreadPoolExecutor, as_completed API_KEY your-deepseek-api-key URL https://api.deepseek.com/chat/completions def call_deepseek(content): headers { Authorization: fBearer {API_KEY}, Content-Type: application/json } payload { model: deepseek-chat, messages: [{role: user, content: content}], temperature: 0.3, max_tokens: 200 } resp requests.post(URL, headersheaders, jsonpayload, timeout60) resp.raise_for_status() return resp.json()[choices][0][message][content] def batch_process_with_threads(samples, max_workers10): results {} with ThreadPoolExecutor(max_workersmax_workers) as executor: future_map {executor.submit(call_deepseek, s): idx for idx, s in enumerate(samples)} for future in as_completed(future_map): idx future_map[future] try: results[idx] future.result() except Exception as e: results[idx] fERROR: {e} return [results[i] for i in range(len(samples))] if __name__ __main__: samples [请用一句话介绍杭州] * 30 start time.time() output batch_process_with_threads(samples, max_workers10) print(f并发耗时: {time.time() - start:.2f}s) print(json.dumps(output[:2], ensure_asciiFalse, indent2))逻辑说明executor.submit把每个请求包装成一个future对象as_completed在future完成时立刻返回不需要等所有任务结束才统一收结果。代码里我故意把results设计成按下标存储这样最终返回的列表顺序和输入样本顺序保持一致避免并发把顺序打乱之后没法对应回原文本。参数说明max_workers10是线程池最大并发数实测DeepSeek普通key稳定跑这个值基本不触发限流。timeout60建议只增不减因为长文本生成真正跑到60秒以上的情况并不罕见设短了会误杀正常请求。temperature0.3是DeepSeek比较稳妥的批量任务参数分类、打标、改写这类结构化任务不要开太高。3.2 真异步方案用asyncio aiohttp改写后的完整流程线程池方案能解决80%的问题但线程是有成本的——每个线程占内存和上下文切换开销max_workers调高之后性能提升会越来越不明显。更干净的做法是用asyncioaiohttp做真正的IO异步请求。注意asyncio不是多线程它是事件循环里跑协程用一个线程处理所有并发IO对限流的控制也更直观。import aiohttp import asyncio import json import time API_KEY your-deepseek-api-key URL https://api.deepseek.com/chat/completions MAX_CONCURRENT 15 # 信号量控制并发上限 async def call_deepseek_async(session, content, semaphore): headers { Authorization: fBearer {API_KEY}, Content-Type: application/json } payload { model: deepseek-chat, messages: [{role: user, content: content}], temperature: 0.3, max_tokens: 200 } async with semaphore: try: async with session.post(URL, headersheaders, jsonpayload, timeout60) as resp: data await resp.json() if resp.status ! 200: return fHTTP {resp.status}: {data} return data[choices][0][message][content] except asyncio.TimeoutError: return ERROR: timeout async def run_batch(samples): semaphore asyncio.Semaphore(MAX_CONCURRENT) async with aiohttp.ClientSession() as session: tasks [call_deepseek_async(session, s, semaphore) for s in samples] results await asyncio.gather(*tasks) return results if __name__ __main__: samples [用一句话描述量子计算] * 30 start time.time() results asyncio.run(run_batch(samples)) print(f异步耗时: {time.time() - start:.2f}s) print(json.dumps(results[:2], ensure_asciiFalse, indent2))逻辑说明asyncio.Semaphore(MAX_CONCURRENT)是整个异步方案里最关键的限流手段。没有信号量的话asyncio会一次性把所有请求全部发出去30条还好3000条直接把服务端打秃429全返回来。信号量保证同时活跃的请求数不超过15。asyncio.gather(*tasks)并发跑所有协程并等待全部完成返回顺序和tasks列表顺序一致天然对齐样本顺序。参数说明MAX_CONCURRENT15是我在DeepSeek上压测过比较稳的值token生成速度快的短文本任务可以到20长文本任务建议降到8。session.post里的timeout60是aiohttp的总体超时如果你的任务动不动要输出上千token放宽到120更稳妥。asyncio.run是Python 3.7的入口别在Jupyter Notebook里直接调用会报事件循环冲突。3.3 批量改写与批量打标的参数模板temperature与max_tokens怎么配同样是批量请求任务类型不同参数模板差异很大。批量分类打标任务我推荐temperature0.1把随机性压到最低输出稳定不会出现同一条文本两次分类结果不一样。批量改写或者摘要任务temperature0.3到0.5之间保留表达变化但不会跑偏。批量创意文案才建议开0.7以上但那就不是这篇要聊的效率问题了。max_tokens的坑最多。很多人以为设得越大越好其实max_tokens只限制最大输出长度不限制模型实际生成量而且它同时占着TPM限流的预算。设个500的结果是模型已经回答完了但token预算算的是你声明的最大值还是实际消耗DeepSeek按实际消耗计费限流则按实际tokens计算。我一般用200到300的max_tokens做批量总结输出意外的长文本时再单条重跑而不是把全局配成1000。还有一个提示词层面的效率技巧用system prompt固定输出格式。批量任务里最耗token的不是内容生成而是模型在自由发挥前后夹带解释。你在system里写明「只输出JSON不要多余文字」单条请求能省下20到50个token批量几千条的时候省下的总量就很可观了。4. 处理速率限制与错误重试DeepSeek批量调用避坑指南4.1 429限流报错现象、定位与指数退避重试现象批量脚本跑到某个时刻日志里开始出现429 Too Many Requests响应体里通常有rate limit reached字样。同步循环里出现一次429整个脚本就停了并发方案里429通常是一批一批出现的因为你瞬时打出的请求数超过了RPM限制。原因DeepSeek的速率限制是滑动窗口计数你在1分钟内打出了超过限额的请求数。并发数开得越高越容易撞上这个窗口。还有一个隐蔽因素如果你同时运行多个脚本进程比如本机跑一个批量任务另一个服务也在调DeepSeek共享同一个API key的配额撞限流的概率翻倍。解决用指数退避重试而不是固定间隔重试。第一次失败等1秒第二次等2秒第三次等4秒最多等10秒后放弃。代码实现里加一个带退避的重试装饰器或者直接在请求函数里做循环重试import time import random def call_with_retry(content, max_retries5): for attempt in range(max_retries): try: resp requests.post(URL, headersheaders, jsonpayload, timeout60) if resp.status_code 429: retry_after int(resp.headers.get(Retry-After, 2 ** attempt)) time.sleep(min(retry_after random.uniform(0, 1), 10)) continue resp.raise_for_status() return resp.json()[choices][0][message][content] except requests.exceptions.ConnectionError: wait min(2 ** attempt random.uniform(0, 1), 10) time.sleep(wait) return ERROR: retry exhausted重试代码的核心是把429响应头里的Retry-After作为首选等待时间如果服务端没给用2的attempt次方做退避指数增长比固定间隔更适合限流场景。重试5次之后还失败直接返回错误字符串等主流程统一处理不要让单条失败阻塞整个批次。4.2 超时误杀timeout设短了会让正常慢请求全军覆没现象并发数调到20以上后突然出现大量TimeoutError但你单独发一条同样的请求5秒就返回了。排查发现不是DeepSeek挂了你是某些请求排队时间过长你的客户端超时先到了。原因并发提高后服务端排队时间同步拉长。API在高峰期可能让请求排队十几秒才开始生成token你的timeout30把排队时间也算进去了排队时间一超就断连。解决把超时拆成连接超时和读取超时。连接超时5秒足够读取超时给到120秒。requests库的timeout参数可以传元组timeout(5, 120)。连接超时负责快速失败服务端不可达时马上暴露问题读取超时负责给模型生成留足时间。并发越高读取超时要给得越宽因为排队时间的方差会变大。4.3 顺序错乱与结果对不上并发脚本里最常见的翻车点现象并发跑完结果文件里第10条文本对应的输出内容明显是对第25条文本的回复。检查代码逻辑没错但结果就是乱了。原因线程池方案里如果你用executor.map或者不按下标收集结果任务的返回顺序和提交顺序并不保证一致。asyncio方案里gather虽然保持顺序但如果你提前把部分结果写进了文件后续结果覆盖错位同样会乱。解决统一用「输入带ID、输出带ID」的方式。每条待处理文本在一开始就分配一个唯一ID请求消息里带上这个ID返回结果按ID写回对应槽位。整理结果时不做顺序假设只按ID合并。这是我踩过最大的坑两千条数据跑了几十分钟最后因为结果对应不上全部重跑。4.4 本地部署与API调用的取舍vllm部署时并发参数要换一套如果你不是用DeepSeek官方API而是通过vllm在本地部署DeepSeek模型批量请求的并发逻辑要调整。vllm服务端本身支持高并发请求排队它的吞吐量瓶颈在GPU显存和batch size而不是RPM限制。本地部署时客户端的并发数可以调到50甚至更高但要注意max_num_seqs参数vllm默认值是256超过这个值请求会在服务端排队客户端超时设置同样要放宽。本地部署还有一层参数在服务端启动命令里--max-model-len控制最大上下文长度--gpu-memory-utilization控制显存利用比例。批量任务如果文本长度参差不齐建议把max-model-len设得比最长样本大20%左右否则长文本直接给你截断。本地部署的好处是彻底避开限流和费用问题坏处是服务端的并发能力受硬件限制参数调优的复杂度上了一个台阶。5. 把批量请求推到生产级任务队列、断点续跑与成本控制5.1 用队列替代裸并发批量任务里tokens、优先级与失败补偿怎么设计裸并发方案做几百条数据可以做几万条就不够用了。生产级做法是把待处理文本丢进任务队列消费者从队列里取任务处理完写结果。队列的好处有三个一是你可以随时暂停和恢复不丢任务二是可以按优先级插队紧急文本插到队首三是失败补偿可以集中在队列层面做——某条任务重试耗尽后把它重新塞回队列尾部标记重试次数上限累计重试超限才放弃。实现上不必上重型消息队列用Python标准库queue.PriorityQueue就够。优先级设计成小数值优先普通任务优先级设10紧急任务设1这样插队逻辑天然成立。队列里存待处理文本的元组(priority, index, text, retry_count)消费者取任务时按优先级出队。5.2 断点续跑程序被中断后怎么不从头再来批量任务最怕的不是慢是跑了一半挂了然后从头开始。断点续跑的常规做法是给每条任务加状态标记结果落盘采用「先写临时文件完成后改后缀」的策略。每条任务处理完把结果追加到一个done.jsonl文件里启动时扫描done.jsonl已处理的样本ID跳过。import os import json DONE_FILE done.jsonl def load_done_ids(file_pathDONE_FILE): if not os.path.exists(file_path): return set() done_ids set() with open(file_path, r, encodingutf-8) as f: for line in f: record json.loads(line) done_ids.add(record[id]) return done_ids def append_done(record, file_pathDONE_FILE): with open(file_path, a, encodingutf-8) as f: f.write(json.dumps(record, ensure_asciiFalse) \n) # 在消费任务前检查 done_ids load_done_ids() for item in all_samples: if item[id] in done_ids: continue result process(item[text]) append_done({id: item[id], result: result})这个做法的核心逻辑是把「进度」固化到磁盘上而不是依赖内存变量。任何时刻进程崩溃、断网、手动停止重启后load_done_ids自动跳过已完成任务。done.jsonl是追加写不会因为覆盖而丢数据。我在处理3000条文本时就是靠这一招省掉了多次返工——更早期的方案是整体结果统一写一个文件写到一半挂掉整个文件损坏全部重跑。5.3 成本翻倍不是效率翻倍并发调用如何控制token消耗并发把时间压缩了10倍不代表成本会同样增长。DeepSeek按token计费成本取决于你发送了多少输入token和接收了多少输出token。并发改变的是请求节奏不改变单条请求的token消耗。唯一可能让成本爆炸的是重试——429限流后重试的请求会重复计费输入token超时后重发的请求也会重复算。你把重试逻辑写好等于直接控制了成本上限。另一个成本控制点是输入token压缩。批量任务里提示词模板越精简越好。比如你要让DeepSeek做关键词提取不要在每条请求里重复一大段背景说明尽量把背景信息压缩进一两条system消息里短时间内重复的内容不再反复写。几千条短文本批量调用时提示词精简20%总成本可能降30%以上。5.4 长任务分割与合并一条大请求拆成多条小请求反而更快遇到超长文本处理比如几万字的长文档总结不要指望一条请求解决。DeepSeek的上下文窗口虽然支持长文本但超长输入有两个代价单条请求耗时暴涨、max_tokens输出不够用。实践做法是把长文档按段落或固定长度切片每个切片独立生成摘要最后把分摘要合并成一条新请求让它基于分摘要生成总摘要。分割粒度放在500字到2000字之间比较合适太碎了调用次数增加太粗了摘要质量下降。6. 验证压测结果与调优一次批量任务从90分钟到9分钟的完整路径把上面的方案组合起来做一次真实压测。样本3050条中文短文本每条需要DeepSeek输出100到150字的摘要。第一次跑用的是同步循环耗时94分钟。改成ThreadPoolExecutor并发10线程后耗时12分钟。换asyncio加信号量控制、并发数调整到15后稳定在9分半左右。中间穿插了三次429限流指数退避都兜住了任务没有中断。验证效率提升有个简单方法日志里给每条任务记下开始时间和结束时间最后统计整个批次的p50和p95单条耗时。p95显著低于平均值加上网络延迟时说明并发起了作用如果p95反而升高说明并发打到了限流边界降并发数或者拉大重试间隔。最后一个习惯我每次跑大批量任务之前都会先跑20条小样本验证参数。20条样本3分钟内出结果能看出有没有限流、输出格式对不对、token预算够不够。不要拿全部数据直接上生产配置后悔药只有断点续跑这一种。希望这篇实战笔记帮你把DeepSeek的批量调用效率真正拉起来。本文还有配套的精品资源点击获取