零基础入门 DeepSeek API 流式传输：兼容 OpenAI SDK 调用、核心工作原理解析、流事件处理与文本提取、Token 用量统计、首个 Token 时间性能对比、自定义流处理器与多轮流式

全文精简速览复习专用 本文是面向编程小白的 DeepSeek 流式 API 全攻略核心内容可浓缩为 7 点DeepSeek API 完全兼容 OpenAI SDK仅需修改base_url为https://api.deepseek.com即可快速初始化客户端支持 dotenv 管理密钥。流式传输核心是streamTrue参数可让 AI 生成内容边输出边展示无需等待全部生成完成大幅优化用户感知体验。流式响应为统一的ChatCompletionChunk对象文本内容存于choices[0].delta.content需判断非空后逐段拼接或实时打印。Token 统计有两种方案非流式二次调用精准统计、流式循环中近似累计部分 API 版本可从结束 chunk 直接读取 usage 数据。流式传输可将首个 Token 响应时间TTFT从数秒压缩到 1 秒内但总生成时长与非流式基本一致并不加快模型本身生成速度。可通过封装自定义流处理器类统一管理文本输出、事件回调与结果缓存提升代码复用性还可搭配 ANSI 颜色优化终端显示。维护会话历史列表 流式输出可快速实现带上下文记忆、实时打字机效果的多轮聊天机器人支持输入 quit 退出。 一、环境准备DeepSeek 客户端初始化DeepSeek API 100% 兼容 OpenAI 的 SDK 格式不用学习新的调用语法只需修改接口地址即可快速上手。 知识点汇总表对比项Claude APIDeepSeek API兼容 OpenAI依赖安装pip install anthropicpip install openai python-dotenv客户端类Anthropic()OpenAI()核心配置仅需配置 api_key需同时配置 api_key base_url密钥管理支持环境变量支持环境变量推荐 dotenv 文件管理 完整代码样例# 导入依赖dotenv读取环境变量OpenAI为官方SDKos用于系统交互 from dotenv import load_dotenv from openai import OpenAI import os 加载项目根目录.env文件中的配置避免密钥硬编码泄露 load_dotenv() 初始化DeepSeek客户端完全复用OpenAI SDK语法 client OpenAI( api_keyos.getenv(DEEPSEEK_API_KEY), # 从环境变量读取API密钥 base_urlhttps://api.deepseek.com # DeepSeek官方接口地址 ) 代码逐行解释依赖导入python-dotenv是密钥管理工具可将 API 密钥写在.env文件中避免直接写在代码里造成泄露OpenAI是官方 SDKDeepSeek 完全兼容其接口规范。加载环境变量load_dotenv()会自动读取当前目录下.env文件的配置例如文件内写入DEEPSEEK_API_KEYsk-你的密钥即可。客户端初始化和原生 OpenAI 调用的唯一区别是base_url参数将地址指向 DeepSeek 官方接口后续所有调用语法和 OpenAI 完全一致。小白提示本地测试时如果没有配置环境变量也可以直接填写密钥字符串api_keysk-xxx但生产环境强烈不推荐。⚡ 二、基础调用非流式 vs 流式传输流式传输是本文的核心概念。普通非流式调用必须等 AI 生成完全部内容才会返回结果而流式传输可以生成一段、返回一段实现类似打字机的实时效果。 知识点汇总表对比项非流式调用流式调用核心参数无特殊参数streamTrue返回类型完整响应对象包含全部文本Stream 生成器对象逐段返回文本片段展示时机全部生成完成后一次性展示生成一段展示一段实时输出适用场景短文本、后台批量调用、需完整结果再处理前端对话界面、长文本生成、优化用户体验用户感知等待时间长易产生卡顿感首字响应快感知上更流畅 代码样例 1非流式调用基础版from dotenv import load_dotenv from openai import OpenAI import os load_dotenv() client OpenAI( api_keyos.getenv(DEEPSEEK_API_KEY), base_urlhttps://api.deepseek.com ) 发起非流式请求语法和OpenAI完全一致 response client.chat.completions.create( messages[{role: user, content: 请写一篇关于亚马逊金刚鹦鹉和黏土食的短文}], modeldeepseek-chat, max_tokens1000, # 限制最大生成token数 temperature0, # 温度为0输出更确定、更严谨 ) print(我们已收到回复) print() 直接读取完整响应文本 print(response.choices[0].message.content) 代码解释非流式是最基础的调用方式发起请求后程序会阻塞等待直到 AI 生成完全部内容才返回完整的响应对象。文本固定存储在response.choices[0].message.content中直接读取即可。缺点是生成长文本时用户要等待数秒才能看到第一个字体验较差。 代码样例 2流式调用最简版from dotenv import load_dotenv from openai import OpenAI import os load_dotenv() client OpenAI( api_keyos.getenv(DEEPSEEK_API_KEY), base_urlhttps://api.deepseek.com ) 仅需添加streamTrue即可开启流式传输 r_stream client.chat.completions.create( modeldeepseek-chat, max_tokens1000, temperature0, streamTrue, messages[{role: user, content: 请写一篇关于亚马逊金刚鹦鹉和黏土食的短文}] ) print(我们已收到回复) print() for chunk in r_stream: print(chunk) collected [] # 用于收集所有文本片段 for chunk in r_stream: # 判断当前片段是否包含文本内容首尾chunk可能为空仅含元数据 if chunk.choices[0].delta.content is not None: content chunk.choices[0].delta.content print(repr(content)) # 打印原始片段带引号方便观察分段 collected.append(content) 拼接所有片段得到完整文本 print(\nFull response:, .join(collected)) 代码解释开启流式请求参数中添加streamTrue后返回的不再是完整响应对象而是一个可迭代的Stream生成器。遍历数据块通过for循环逐个读取chunk数据块每个 chunk 对应 AI 生成的一小段内容。内容提取流式片段的文本存储在chunk.choices[0].delta.content中首尾 chunk 通常不含文本内容因此必须加is not None判断避免报错。结果拼接将所有非空文本片段存入列表最后用.join()拼接即可得到和非流式一致的完整响应内容。 三、进阶处理流式文本实时输出流式传输最常用的场景是实时打字机效果通过调整打印参数即可在终端实现边生成边显示的效果。 知识点汇总表对比项Claude 流式事件DeepSeek 流式事件事件类型5 种以上开始、内容块、结束等仅 1 种ChatCompletionChunk文本字段event.delta.text需匹配事件类型chunk.choices[0].delta.content结束标记MessageStopEvent事件finish_reason不为 None上手难度高事件类型多逻辑复杂低结构简单易理解 代码样例终端打字机实时效果stream client.chat.completions.create( messages[{role: user, content: 大语言模型是怎么工作的}], modeldeepseek-chat, max_tokens1000, temperature0, streamTrue, ) 逐段实时打印不换行强制刷新缓冲区 for chunk in stream: if chunk.choices[0].delta.content is not None: # end 取消print默认换行flushTrue 强制立刻输出不等待缓冲区 print(chunk.choices[0].delta.content, flushTrue, end) 代码解释endPython 的print函数默认会在结尾添加换行符使用end可以让所有文本在同一行连续输出模拟打字机的连续显示效果。flushTruePython 输出默认会积攒到缓冲区满了才打印到终端添加该参数可以让每一个文本片段生成后立刻显示真正实现 边生成边输出 的实时体验。小白提示运行后会看到文字逐片段蹦出和主流 AI 对话产品的显示效果完全一致。 四、用量统计Token 计数的两种实现方案Token 是大模型的计费单位流式场景下无法直接从响应对象读取用量有两种主流的统计方案。 知识点汇总表方案实现方式优点缺点方案一非流式二次查询流式输出后再发起一次非流式请求获取 usage计数 100% 精准额外产生一次 API 调用增加费用方案二流式循环累计遍历 chunk 时逐段计数结束时尝试读取 chunk 的 usage无额外调用零成本手动计数为近似值仅部分 API 版本返回结束 chunk 的 usage 代码样例 1非流式精准统计from dotenv import load_dotenv from openai import OpenAI import os load_dotenv() client OpenAI( api_keyos.getenv(DEEPSEEK_API_KEY), base_urlhttps://api.deepseek.com ) 发起非流式请求限制生成长度主要用于获取精准token统计 usage_response client.chat.completions.create( messages[ {role: user, content: 用中文回答 How do large language models work? 简要的用中文回答这个英文问题} ], modeldeepseek-v4-flash, max_tokens1000, temperature0, max_completion_tokens1, # 仅生成1个token最小成本获取输入token统计 ) print(\n) print(fInput tokens: {usage_response.usage.prompt_tokens}) print(fOutput tokens: {usage_response.usage.completion_tokens}) print(fTotal tokens: {usage_response.usage.total_tokens}) 代码解释max_completion_tokens1限制 AI 只生成 1 个输出 token以最低成本拿到输入 token 的精准计数。usage对象包含三个核心数据prompt_tokens是用户输入的 token 数completion_tokens是 AI 生成的 token 数total_tokens为两者之和是计费的核心依据。适用场景需要精准核算费用、做账单统计的生产环境。 Token 类型大白话解释 1. Input tokens输入令牌数—— 你发的消息打印的是usage.prompt_tokens大白话你发给 AI 的那句提问被模型拆解后总共占了多少个 Token。类比相当于你寄快递时包裹的重量不含回信。 2. Output tokens输出令牌数—— AI 回的答案打印的是usage.completion_tokens大白话AI 生成的回答一共消耗了多少个 Token。类比相当于你收到回信时回信包裹的重量。 3. Total tokens总令牌数—— 本次对话总计打印的是usage.total_tokens大白话上面两个加起来的总和Input Output。类比相当于本次聊天总的流量消耗。 代码样例 2流式近似统计 结束 chunk 读取from dotenv import load_dotenv from openai import OpenAI import os load_dotenv() client OpenAI( api_keyos.getenv(DEEPSEEK_API_KEY), base_urlhttps://api.deepseek.com ) stream client.chat.completions.create( modeldeepseek-chat, max_tokens1000, messages[ {role: user, content: 用中文回答 How do large language models work? 简要的用中文回答这个英文问题} ], streamTrue ) output_tokens 0 for chunk in stream: if chunk.choices[0].delta.content: print(chunk.choices[0].delta.content, end, flushTrue) output_tokens 1 # 近似数 # 判断是否为结束chunk部分API版本会在此返回完整usage if chunk.choices[0].finish_reason is not None: if hasattr(chunk, usage) and chunk.usage: print(f\nUsage: {chunk.usage}) 代码解释近似计数每个带文本的 chunk 大致对应 1 个 token循环中累加计数可得到输出 token 的近似值适合做进度预估。结束判断当finish_reason不为None时代表生成结束通常值为stop表示正常结束。读取 usage部分 DeepSeek API 版本会在最后一个 chunk 中附带完整的 usage 数据用hasattr判断属性是否存在存在则直接读取精准且无额外成本。 缓存机制DeepSeek 特有的省钱妙招下面这一坨看起来复杂的字段其实是 DeepSeek 的上下文缓存Prompt Caching技术。简单说就是如果你短时间内问相同或类似的问题AI 不用重新计算直接调取缓存价格会便宜很多通常打 1 折。字段数值大白话解释prompt_cache_hit_tokens0命中了缓存的 Token 数量。这里是 0意味着你的提问是全新的没有命中缓存所以这次没有享受到折扣。prompt_cache_miss_tokens23未命中缓存的 Token 数量。这里也是 23意味着你输入的 23 个 Token 全部需要重新计算所以按原价收费。cached_tokens在prompt_tokens_details里0这个和上面的hit是一个意思就是命中了 0 个。简单数学prompt_tokens (23) prompt_cache_hit_tokens (0) prompt_cache_miss_tokens (23)。❓ 那些 None 和看不懂的细节字段数值解释completion_tokens_detailsNone这个是用来记录输出回答里有没有特殊音频或视频 Token 的。你这里是纯文字所以是空None不用管。audio_tokensNone同上你没有输入音频所以为空。️ 安全判断两层防护️ 第一层hasattr(chunk, usage)字面意思检查这个chunk数据包有没有一个叫做usage的属性。通俗类比就像拆快递时先看看包裹上有没有贴发票清单。如果没有这个标签你就别去撕它否则会撕坏包装程序报错。✅ 第二层and chunk.usage字面意思如果确实有这个属性再进一步检查这个属性里面的内容是不是存在不为空、不为None、不为 0。通俗类比就算贴了发票清单标签你还要看一眼清单上是不是真的写了字有数据。如果只是个空白的标签那也没用不需要打印。⚡ 五、性能对比首个 Token 时间TTFT流式传输最大的价值是优化首个 Token 时间TTFT这是衡量 AI 对话体验的核心指标。 知识点汇总表指标非流式调用流式调用首个 Token 时间TTFT~3-5 秒与完整响应时间一致~0.5-1 秒完整响应总时长~3-5 秒~3-5 秒生成 token 总数相同相同用户感知速度慢需等待全部生成快立刻能看到内容 核心概念TTFTTime To First Token从发起请求到收到第一个生成文本片段的时间。流式传输并不会加快模型的总生成速度但能让用户更快看到第一句话大幅降低等待的焦虑感感知上 AI 反应更快。 代码样例 1非流式 TTFT 测试import time start_time time.time() # 记录请求开始时间 response client.chat.completions.create( max_tokens500, messages[{role: user, content: 写一篇长文介绍美国独立战争的历史}], temperature0, modeldeepseek-chat, ) response_time time.time() - start_time # 计算总耗时 print(f首个token时间: {response_time:.3f} 秒) print(f完整响应总时间: {response_time:.3f} 秒) print(f生成总token数: {response.usage.completion_tokens}) print(response.choices[0].message.content[:200] ...) # 仅打印前200字 代码解释非流式模式下第一个 token 和最后一个 token 会同时返回因此首个 token 时间等于完整响应总时间。生成长文本时用户需要等待数秒才能看到任何内容体验较差。 代码样例 2流式 TTFT 测试def measure_streaming_ttft(): start_time time.time() stream client.chat.completions.create( max_tokens500, messages[{role: user, content: 写一篇长文介绍美国独立战争的历史}], temperature0, modeldeepseek-chat, streamTrue ) have_received_first_token False ttft 0 output_tokens 0 for chunk in stream: if chunk.choices[0].delta.content is not None: # 第一次收到文本时记录TTFT if not have_received_first_token: ttft time.time() - start_time have_received_first_token True print(chunk.choices[0].delta.content, flushTrue, end) output_tokens 1 total_time time.time() - start_time print(f\n\n首个token时间: {ttft:.3f} 秒, flushTrue) print(f完整响应总时间: {total_time:.3f} 秒, flushTrue) print(f生成总token数近似: {output_tokens}, flushTrue) measure_streaming_ttft() 代码解释用布尔变量have_received_first_token标记是否收到第一个文本片段。第一次进入内容判断时计算从请求开始到当前的时间差即为 TTFT。最终可观察到流式的 TTFT 通常不到 1 秒但总生成时长和非流式几乎一致。流式的本质是 边做边上菜而不是 做饭更快。 六、进阶封装自定义流处理器当项目中频繁使用流式调用时重复编写 chunk 判断、文本拼接的代码会很冗余可以封装成通用的流处理器类。 知识点汇总表功能说明统一事件处理封装 chunk 解析、内容提取、结束判断主代码更简洁自定义回调可给文本加颜色、触发日志、调用业务函数结果自动缓存自动拼接完整响应无需手动维护列表高可复用性一次封装所有流式调用场景均可复用 代码样例自定义 StreamHandler 类class StreamHandler: 自定义流处理器封装流式响应全流程处理逻辑 def __init__(self): self.full_content [] # 缓存完整响应文本 def on_content(self, text): 处理每一段文本绿色打印 存入缓存 print(f\033[92m{text}\033[0m, end, flushTrue) self.full_content.append(text) def on_complete(self): 流式传输结束时的回调处理 print(\n\n✅ 生成完成) return .join(self.full_content) 使用示例 handler StreamHandler() stream client.chat.completions.create( messages[{role: user, content: 大语言模型是怎么工作的}], modeldeepseek-chat, max_tokens1000, temperature0, streamTrue, ) for chunk in stream: if chunk.choices[0].delta.content is not None: handler.on_content(chunk.choices[0].delta.content) full_response handler.on_complete() print(f\n完整响应长度: {len(full_response)} 字符) 代码解释类封装思想将所有流式处理逻辑封装进StreamHandler类外部调用只需创建实例、循环传入 chunk 即可代码更整洁。事件回调on_content、on_complete属于回调函数对应流式过程中的不同阶段可在函数内自定义业务逻辑比如存入数据库、触发消息通知。ANSI 颜色代码\033[92m等是终端 ANSI 转义序列可控制文字颜色和样式丰富终端显示效果。 七、实战项目多轮流式聊天机器人结合前面所有知识点我们可以实现一个带上下文记忆、实时打字机效果的终端聊天机器人。 功能汇总表功能点实现方式多轮上下文记忆用列表维护会话历史每次请求携带全部历史消息实时打字机效果流式传输 end flushTrue角色视觉区分ANSI 颜色代码用户蓝色、AI 绿色退出机制输入 quit 关键词跳出循环会话持久化每轮结束后将 AI 完整回复追加到会话列表 完整代码样例from openai import OpenAI import os # 初始化 DeepSeek 客户端 client OpenAI( api_keyos.getenv(DEEPSEEK_API_KEY), base_urlhttps://api.deepseek.com ) # ANSI 颜色代码 BLUE \033[94m GREEN \033[92m RESET \033[0m def chat_with_deepseek(): print(欢迎使用 DeepSeek 聊天机器人) print(输入 quit 退出聊天。\n) conversation[] while True: user_inputinput(f{BLUE}You:{RESET}) if user_input.lower()quit: print(再见) break conversation.append({role:user,content:user_input}) print(f{GREEN}DeepSeek:{RESET},end,flushTrue) streamclient.chat.completions.create( modeldeepseek-chat, max_tokens1000, messagesconversation, streamTrue ) assistant_response for chunk in stream: if chunk.choices[0].delta.content is not None: contentchunk.choices[0].delta.content print(f{GREEN}{content}{RESET},end,flushTrue) assistant_responsecontent print() #完整响应后换行 conversation.append({role:assistant,content:assistant_response}) if __name____main__: chat_with_deepseek() 代码逐部分解释会话管理conversation列表是多轮对话的核心按顺序存储所有用户提问和 AI 回复每次请求都将完整列表传给 APIAI 即可记住之前的对话内容。输入循环while True构建无限循环持续等待用户输入直到输入quit触发break跳出循环。流式输出复用前文的流式打印逻辑配合 ANSI 颜色代码实现蓝色用户、绿色 AI 的视觉区分。 quit 判断机制详解用户输入.lower()转换后是否等于quit结果quitquit✅ 是退出程序Quitquit✅ 是退出程序QUITquit✅ 是退出程序qUiTquit✅ 是退出程序exitexit❌ 否当成聊天内容发给 AI 八、总览DeepSeek vs Claude API 核心差异 差异汇总表对比维度Claude (Anthropic)DeepSeek (OpenAI 兼容)SDK 安装命令pip install anthropicpip install openai客户端初始化Anthropic(api_key...)OpenAI(api_key..., base_urlhttps://api.deepseek.com)非流式调用方法client.messages.create(...)client.chat.completions.create(...)流式开启方式streamTruestreamTrue非流式文本路径response.content[0].textresponse.choices[0].message.content流式文本路径event.delta.text需匹配事件类型chunk.choices[0].delta.content主流模型名称claude-3-haiku-20240307等deepseek-chat、deepseek-reasoner系统提示设置独立的system参数放入 messages 列表role 设为 system流式事件复杂度高5 种以上事件类型低仅一种 chunk 结构上手更快 流式传输核心概念详解delta 是什么delta是流式传输里最核心的概念简单一句话delta就是增加的那一小部分不是全部。 先打个比方你让 AI 写一句话比如今天天气真好。如果不开流式streamFalseAI 会一次性把整个箱子扔给你。但你开了流式streamTrueAI 就像发微信消息一样一个字一个字地往外蹦其实是按片段发。这段代码的作用就是把这些蹦出来的字一个一个接住最后拼成完整的一句话。 核心代码示例collected [] for chunk in stream: if chunk.choices[0].delta.content is not None: content chunk.choices[0].delta.content print(repr(content)) collected.append(content) print(\nFull response:, .join(collected)) 逐行大白话解释collected []意思拿一个空盒子列表放这里。准备一会儿把收到的字都丢进这个盒子里。for chunk in stream:意思stream就是 AI 给你发来的一串快递包裹流。for chunk in意思就是快递员每送来一个小包裹chunk我就打开看一次。直到所有包裹送完为止。if chunk.choices[0].delta.content is not None:意思打开包裹后先看一眼里面有没有字。因为有的包裹只是说我还在打字哦里面是空的None。这句代码就是过滤掉空包裹只处理有内容的。content chunk.choices[0].delta.content意思确认包裹里有字后把里面的那一小段文字拿出来给它起个名字叫content内容。print(repr(content))意思把拿到的这一小段字打印在屏幕上。这里的repr()有点讲究它会带上引号打印比如今、天。这样你就能清楚看到它收到了什么连空格、换行都能看见方便调试。collected.append(content)意思把这小段字放进刚才准备的空盒子collected 列表里。print(\nFull response:, .join(collected))意思等所有包裹都收完了循环结束把盒子里的小碎片全部拼接起来。.join(collected)意思是用空字符串当做胶水把列表里的字一个个粘起来。比如[今,天,好]变成今天好。最前面的\n是换行让结果另起一行显示更美观。 概念速查表概念大白话解释chunk快递员送来的一次包裹delta这个包裹里装的新增的碎片不是全部拼好的content具体的那几个字比如今、天 本文已全部整理完毕所有代码均放入规范的代码块中内容按章节归类清晰希望能帮助大家快速上手 DeepSeek 流式 API