上一节-Python 连接 DeepSeek APIOpenAI的基础操作文章目录三、进阶功能 1实现多轮上下文对话1、技术原理传递对话历史2、完整代码示例3、代码关键点解析四、进阶功能 2实现流式输出1、技术原理2、完整代码示例3、代码关键点解析三、进阶功能 1实现多轮上下文对话上述基础示例仅适用于单轮问答场景它的上下文是独立的 —— 模型无法记住上一次用户输入的信息而在实际应用中用户更需要的是连续多轮对话的交互效果这就需要程序在每次请求时都将完整的历史对话消息列表提交给 API。1、技术原理传递对话历史DeepSeek API 本身是无状态的它不会保存任何历史对话数据 —— 要实现连续多轮对话的上下文记忆必须在每次请求时将完整的历史对话消息列表包含之前的所有用户输入和模型回复按顺序通过messages参数提交给 API 服务端。实现多轮对话的核心逻辑是在客户端本地维护一个完整的对话消息列表列表的第一个元素通常是设定模型身份的system角色消息。每一轮用户输入新的提问后将这个提问内容以user角色的形式追加到消息列表的末尾。将这个包含完整历史对话的消息列表作为messages参数的值发送给 API 服务端模型会基于这个完整的上下文信息生成回复。收到模型的有效回复后将模型的回复内容以assistant角色的形式追加到消息列表的末尾用于下一轮对话的上下文构建。重复步骤 2 到 4 的操作直到用户主动结束对话。需要特别注意的是必须保证messages数组的消息顺序和对话的实际发生顺序完全一致且其中的role角色设置必须正确这样模型才能正确理解完整的对话上下文。2、完整代码示例下面的代码将实现一个支持多轮上下文对话的机器人它会在控制台等待用户的输入请求连续进行对话直到用户输入exit或quit命令主动退出程序importosfromdotenvimportload_dotenvfromopenaiimportOpenAI,APIError,APIConnectionError,RateLimitError load_dotenv()# 初始化客户端clientOpenAI(api_keyos.getenv(DEEPSEEK_API_KEY),base_urlos.getenv(DEEPSEEK_BASE_URL,https://api.deepseek.com))# 初始化消息列表带上system角色的上下文设定messages[{role:system,content:你是一个知识渊博的AI助手名叫「小迪」。在回答用户问题时要尽量简洁、通俗易懂必要时提供具体的例子或代码片段。请使用中文回答并保持语气友好、专业。}]print( DeepSeek 多轮对话机器人 )print(你可以持续输入内容进行对话输入 exit 或 quit 来结束对话。)# 开启循环持续等待用户输入并发送请求whileTrue:# 1. 接收用户的输入内容user_inputinput(\n你)ifuser_input.lower()in[exit,quit]:print(对话结束再见)break# 2. 将用户的新输入追加到历史消息列表中messages.append({role:user,content:user_input})try:# 3. 携带完整的历史消息发送请求responseclient.chat.completions.create(modeldeepseek-chat,messagesmessages,# 提交完整的对话历史max_tokens1024,temperature0.7,streamFalse)# 4. 解析模型的回复内容ai_replyresponse.choices[0].message.contentprint(f\nAI{ai_reply})# 5. 将模型的回复作为assistant角色消息追加到历史消息列表中messages.append({role:assistant,content:ai_reply})exceptAPIConnectionErrorase:print(AI抱歉我当前无法连接到服务请检查你的网络连接或稍后再试。)print(f[错误详情] 连接失败{str(e)})# 移除用户的提问消息避免污染下一次请求的上下文环境messages.pop()exceptRateLimitErrorase:print(AI抱歉当前我的访问压力过大或账户额度不足请稍后再试。)print(f[错误详情] 访问受限{str(e)})messages.pop()exceptAPIErrorase:print(fAI抱歉请求处理失败错误码{e.status_code})print(f[错误详情]{str(e)})messages.pop()exceptExceptionase:print(AI抱歉当前我无法处理你的请求请稍后再试。)print(f[错误详情] 未知异常{str(e)})messages.pop()将这段代码保存为deepseek_multi_turn_chat.py文件然后在虚拟环境中执行该脚本即可启动一个持续对话会话。3、代码关键点解析与基础单轮对话的代码相比这个多轮对话的代码有以下三个关键调整点消息列表的持久化维护使用一个全局的messages列表在内存中持续维护完整的对话历史包含用户的所有提问和模型的所有回复。在对话开始前需要在列表中加入system角色的消息设定对话的整体背景和约束。提交完整上下文在每次调用 API 时将这个完整的messages列表作为参数值提交给服务端 —— 这是让模型理解上下文、实现连续对话效果的核心前提。异常回滚机制如果某次 API 请求触发了异常例如网络连接失败、服务端返回错误为了保证上下文的逻辑一致性需要将用户刚刚输入的最后一条消息从messages列表中移除 —— 否则这条消息会残留在历史列表中被带入下一轮正常请求的上下文中导致重复处理的问题。需要注意的是受限于模型的上下文长度上限这个历史消息列表不能无限增长 —— 如果对话的轮次过多导致总令牌数超过了模型的上下文上限例如deepseek-chat模型支持的上限为 8192API 会返回错误。在实际应用中需要根据使用的模型上限对历史消息列表进行管理例如截断早期的历史消息、压缩上下文、摘要总结等避免超出模型的上下文长度限制。四、进阶功能 2实现流式输出在默认情况下API 会在模型完整生成所有回复内容后一次性返回完整的响应结果 —— 如果回复内容较长或者网络延迟较高用户可能会等待较长时间看不到任何实时反馈这会严重影响交互体验。为了优化这种场景下的用户体验DeepSeek API 提供了流式输出的能力让回复内容可以逐块实时返回实现打字机效果的实时输出体验。1、技术原理流式输出是基于 HTTP 的长连接机制实现的客户端在请求参数中设置streamTrue此时 API 服务端会与客户端保持长连接将模型生成的回复内容分成多个小块Chunk以文本流的形式实时返回给客户端客户端则通过迭代的方式实时读取这些内容块并依次打印最终在控制台形成类似人类打字的输出效果。使用流式输出时API 的响应格式与非流式输出完全不同 —— 响应结果是一个可迭代的响应对象每个迭代元素都对应一个包含实际内容的小块需要额外的逻辑来逐个读取、解析和处理这些内容块。需要注意的是开启流式输出后响应中的usage字段将默认不返回 —— 如果需要获取本次请求的令牌使用统计信息需要在请求中添加stream_options参数明确要求返回该字段。2、完整代码示例执行流程发起请求 → 迭代接收 chunk → 实时输出 → 拼接完整文本 → 取最后 chunk 元数据 (token、finish_reason) → 业务处理 → 组装记录 → 保存 json下面是一个支持流式输出的多轮对话代码示例它将实现打字机效果的实时回复输出importosimportjsonfromopenaiimportOpenAI clientOpenAI(api_keyos.environ[DEEPSEEK_API_KEY],base_urlhttps://api.deepseek.com)defstream_chat_demo(messages:list):# -------------------------- 步骤1发起流式请求拿到迭代器stream --------------------------streamclient.chat.completions.create(modeldeepseek-chat,messagesmessages,temperature0.7,max_tokens1024,streamTrue,# 开启流式stream_options{include_usage:True}# 必须开启否则拿不到usage)full_ai_text# 用来拼接完整回答文本last_chunkNone# 保存循环的最后一个chunk用于拿usage、request_idprint(\n AI回复,end,flushTrue)# -------------------------- 步骤2循环迭代每一个chunk碎片 --------------------------forchunkinstream:# 每一轮覆盖循环结束后last_chunk就是网络收到的最后一包数据last_chunkchunk# delta 是增量内容只有新生成的一小段文字deltachunk.choices[0].deltaifdelta.content:# 实时打印实现打字机效果print(delta.content,end,flushTrue)# 把增量拼接到完整字符串full_ai_textdelta.contentprint(\n)# 流式结束换行# -------------------------- 步骤3循环结束解析【最后一个chunk】的元数据 --------------------------# ⚠️ 重点usage / request_id / finish_reason 全部在 last_chunkrequest_idlast_chunk.idmodel_namelast_chunk.model finish_reasonlast_chunk.choices[0].finish_reason# token消耗只存在最后一块chunkprompt_tokenslast_chunk.usage.prompt_tokens completion_tokenslast_chunk.usage.completion_tokens total_tokenslast_chunk.usage.total_tokens# -------------------------- 步骤4业务处理原始文本 --------------------------raw_replyfull_ai_text processed_replyraw_reply.strip()# 简单清洗去除首尾空白换行# 如果需要模型输出JSON可以在这里做json.loads捕获异常# try:# processed_reply json.loads(raw_reply)# except json.JSONDecodeError:# pass# -------------------------- 步骤5组装可序列化的对话记录用于保存 --------------------------chat_record{request_id:request_id,model:model_name,finish_reason:finish_reason,input_messages:messages,ai_raw_reply:raw_reply,ai_processed_reply:processed_reply,usage:{prompt_tokens:prompt_tokens,completion_tokens:completion_tokens,total_tokens:total_tokens}}# -------------------------- 步骤6保存到 jsonl 文件一行一条记录 --------------------------save_filestream_chat_history.jsonlwithopen(save_file,a,encodingutf-8)asf:f.write(json.dumps(chat_record,ensure_asciiFalse)\n)print(f✅ 对话记录已保存本次总token{total_tokens})# 返回处理完的文本 完整记录外部可以继续使用returnprocessed_reply,chat_recordif__name____main__:# 测试上下文msg_list[{role:system,content:你是简洁的Python技术助手},{role:user,content:简单讲下生成器yield的原理}]resp_text,recordstream_chat_demo(msg_list)print(\n返回给程序使用的处理后文本)print(resp_text)将这段代码保存为deepseek_streaming_chat.py文件然后在虚拟环境中执行该脚本即可看到实时输出的对话效果。3、代码关键点解析与之前的非流式代码相比这个流式输出的代码有以下三个关键调整点开启流式输出在请求参数中设置streamTrue此时 API 会返回一个可迭代的响应对象而不是完整的响应结果。同时为了在流式输出结束后获取本次请求的令牌使用统计信息需要额外设置stream_options{include_usage: True}—— 如果不添加这个配置响应的usage字段将返回空值无法统计令牌使用情况。迭代读取内容块通过for chunk in stream循环逐个读取响应流中的内容块。每个内容块中都包含了模型本次返回的增量文本内容。这些增量内容需要直接打印到控制台实现实时输出的效果。记录完整回复由于每个内容块中只包含增量内容我们需要在客户端侧将所有内容块的增量内容拼接成完整的回复文本 —— 否则无法将完整的assistant角色回复加入到下一轮请求的上下文消息列表中。需要特别注意的是在打印每个内容块的增量内容时必须将print()方法的end参数设置为空字符串end避免每个内容块打印后自动换行破坏完整回复的排版效果同时需要将flush参数设置为True强制 Python 的标准输出流立即将内容输出到控制台而不是暂存到缓冲区中 —— 这是保证打字机效果实时性的关键前提。下一节-Python 连接 DeepSeek APIOpenAI 对话方式总结免责声明本文档中描述的 API 调用方法及相关代码示例均基于 DeepSeek 官方文档和公开技术社区的讨论内容。在实际使用过程中因 API 版本差异、服务端接口升级、账户权限配置等导致的程序异常或数据风险本文作者及 DeepSeek 官方不承担任何责任。请务必先在测试环境中完成充分验证后再将逻辑部署到生产环境中。本文档的内容会随着 DeepSeek API 的版本迭代和功能升级而更新最新版本请以官方文档的内容为准。注文档部分内容可能由 AI 生成 Python 连接 DeepSeek API，OpenAI 两种对话方式基础实现
上一节-Python 连接 DeepSeek APIOpenAI的基础操作文章目录三、进阶功能 1实现多轮上下文对话1、技术原理传递对话历史2、完整代码示例3、代码关键点解析四、进阶功能 2实现流式输出1、技术原理2、完整代码示例3、代码关键点解析三、进阶功能 1实现多轮上下文对话上述基础示例仅适用于单轮问答场景它的上下文是独立的 —— 模型无法记住上一次用户输入的信息而在实际应用中用户更需要的是连续多轮对话的交互效果这就需要程序在每次请求时都将完整的历史对话消息列表提交给 API。1、技术原理传递对话历史DeepSeek API 本身是无状态的它不会保存任何历史对话数据 —— 要实现连续多轮对话的上下文记忆必须在每次请求时将完整的历史对话消息列表包含之前的所有用户输入和模型回复按顺序通过messages参数提交给 API 服务端。实现多轮对话的核心逻辑是在客户端本地维护一个完整的对话消息列表列表的第一个元素通常是设定模型身份的system角色消息。每一轮用户输入新的提问后将这个提问内容以user角色的形式追加到消息列表的末尾。将这个包含完整历史对话的消息列表作为messages参数的值发送给 API 服务端模型会基于这个完整的上下文信息生成回复。收到模型的有效回复后将模型的回复内容以assistant角色的形式追加到消息列表的末尾用于下一轮对话的上下文构建。重复步骤 2 到 4 的操作直到用户主动结束对话。需要特别注意的是必须保证messages数组的消息顺序和对话的实际发生顺序完全一致且其中的role角色设置必须正确这样模型才能正确理解完整的对话上下文。2、完整代码示例下面的代码将实现一个支持多轮上下文对话的机器人它会在控制台等待用户的输入请求连续进行对话直到用户输入exit或quit命令主动退出程序importosfromdotenvimportload_dotenvfromopenaiimportOpenAI,APIError,APIConnectionError,RateLimitError load_dotenv()# 初始化客户端clientOpenAI(api_keyos.getenv(DEEPSEEK_API_KEY),base_urlos.getenv(DEEPSEEK_BASE_URL,https://api.deepseek.com))# 初始化消息列表带上system角色的上下文设定messages[{role:system,content:你是一个知识渊博的AI助手名叫「小迪」。在回答用户问题时要尽量简洁、通俗易懂必要时提供具体的例子或代码片段。请使用中文回答并保持语气友好、专业。}]print( DeepSeek 多轮对话机器人 )print(你可以持续输入内容进行对话输入 exit 或 quit 来结束对话。)# 开启循环持续等待用户输入并发送请求whileTrue:# 1. 接收用户的输入内容user_inputinput(\n你)ifuser_input.lower()in[exit,quit]:print(对话结束再见)break# 2. 将用户的新输入追加到历史消息列表中messages.append({role:user,content:user_input})try:# 3. 携带完整的历史消息发送请求responseclient.chat.completions.create(modeldeepseek-chat,messagesmessages,# 提交完整的对话历史max_tokens1024,temperature0.7,streamFalse)# 4. 解析模型的回复内容ai_replyresponse.choices[0].message.contentprint(f\nAI{ai_reply})# 5. 将模型的回复作为assistant角色消息追加到历史消息列表中messages.append({role:assistant,content:ai_reply})exceptAPIConnectionErrorase:print(AI抱歉我当前无法连接到服务请检查你的网络连接或稍后再试。)print(f[错误详情] 连接失败{str(e)})# 移除用户的提问消息避免污染下一次请求的上下文环境messages.pop()exceptRateLimitErrorase:print(AI抱歉当前我的访问压力过大或账户额度不足请稍后再试。)print(f[错误详情] 访问受限{str(e)})messages.pop()exceptAPIErrorase:print(fAI抱歉请求处理失败错误码{e.status_code})print(f[错误详情]{str(e)})messages.pop()exceptExceptionase:print(AI抱歉当前我无法处理你的请求请稍后再试。)print(f[错误详情] 未知异常{str(e)})messages.pop()将这段代码保存为deepseek_multi_turn_chat.py文件然后在虚拟环境中执行该脚本即可启动一个持续对话会话。3、代码关键点解析与基础单轮对话的代码相比这个多轮对话的代码有以下三个关键调整点消息列表的持久化维护使用一个全局的messages列表在内存中持续维护完整的对话历史包含用户的所有提问和模型的所有回复。在对话开始前需要在列表中加入system角色的消息设定对话的整体背景和约束。提交完整上下文在每次调用 API 时将这个完整的messages列表作为参数值提交给服务端 —— 这是让模型理解上下文、实现连续对话效果的核心前提。异常回滚机制如果某次 API 请求触发了异常例如网络连接失败、服务端返回错误为了保证上下文的逻辑一致性需要将用户刚刚输入的最后一条消息从messages列表中移除 —— 否则这条消息会残留在历史列表中被带入下一轮正常请求的上下文中导致重复处理的问题。需要注意的是受限于模型的上下文长度上限这个历史消息列表不能无限增长 —— 如果对话的轮次过多导致总令牌数超过了模型的上下文上限例如deepseek-chat模型支持的上限为 8192API 会返回错误。在实际应用中需要根据使用的模型上限对历史消息列表进行管理例如截断早期的历史消息、压缩上下文、摘要总结等避免超出模型的上下文长度限制。四、进阶功能 2实现流式输出在默认情况下API 会在模型完整生成所有回复内容后一次性返回完整的响应结果 —— 如果回复内容较长或者网络延迟较高用户可能会等待较长时间看不到任何实时反馈这会严重影响交互体验。为了优化这种场景下的用户体验DeepSeek API 提供了流式输出的能力让回复内容可以逐块实时返回实现打字机效果的实时输出体验。1、技术原理流式输出是基于 HTTP 的长连接机制实现的客户端在请求参数中设置streamTrue此时 API 服务端会与客户端保持长连接将模型生成的回复内容分成多个小块Chunk以文本流的形式实时返回给客户端客户端则通过迭代的方式实时读取这些内容块并依次打印最终在控制台形成类似人类打字的输出效果。使用流式输出时API 的响应格式与非流式输出完全不同 —— 响应结果是一个可迭代的响应对象每个迭代元素都对应一个包含实际内容的小块需要额外的逻辑来逐个读取、解析和处理这些内容块。需要注意的是开启流式输出后响应中的usage字段将默认不返回 —— 如果需要获取本次请求的令牌使用统计信息需要在请求中添加stream_options参数明确要求返回该字段。2、完整代码示例执行流程发起请求 → 迭代接收 chunk → 实时输出 → 拼接完整文本 → 取最后 chunk 元数据 (token、finish_reason) → 业务处理 → 组装记录 → 保存 json下面是一个支持流式输出的多轮对话代码示例它将实现打字机效果的实时回复输出importosimportjsonfromopenaiimportOpenAI clientOpenAI(api_keyos.environ[DEEPSEEK_API_KEY],base_urlhttps://api.deepseek.com)defstream_chat_demo(messages:list):# -------------------------- 步骤1发起流式请求拿到迭代器stream --------------------------streamclient.chat.completions.create(modeldeepseek-chat,messagesmessages,temperature0.7,max_tokens1024,streamTrue,# 开启流式stream_options{include_usage:True}# 必须开启否则拿不到usage)full_ai_text# 用来拼接完整回答文本last_chunkNone# 保存循环的最后一个chunk用于拿usage、request_idprint(\n AI回复,end,flushTrue)# -------------------------- 步骤2循环迭代每一个chunk碎片 --------------------------forchunkinstream:# 每一轮覆盖循环结束后last_chunk就是网络收到的最后一包数据last_chunkchunk# delta 是增量内容只有新生成的一小段文字deltachunk.choices[0].deltaifdelta.content:# 实时打印实现打字机效果print(delta.content,end,flushTrue)# 把增量拼接到完整字符串full_ai_textdelta.contentprint(\n)# 流式结束换行# -------------------------- 步骤3循环结束解析【最后一个chunk】的元数据 --------------------------# ⚠️ 重点usage / request_id / finish_reason 全部在 last_chunkrequest_idlast_chunk.idmodel_namelast_chunk.model finish_reasonlast_chunk.choices[0].finish_reason# token消耗只存在最后一块chunkprompt_tokenslast_chunk.usage.prompt_tokens completion_tokenslast_chunk.usage.completion_tokens total_tokenslast_chunk.usage.total_tokens# -------------------------- 步骤4业务处理原始文本 --------------------------raw_replyfull_ai_text processed_replyraw_reply.strip()# 简单清洗去除首尾空白换行# 如果需要模型输出JSON可以在这里做json.loads捕获异常# try:# processed_reply json.loads(raw_reply)# except json.JSONDecodeError:# pass# -------------------------- 步骤5组装可序列化的对话记录用于保存 --------------------------chat_record{request_id:request_id,model:model_name,finish_reason:finish_reason,input_messages:messages,ai_raw_reply:raw_reply,ai_processed_reply:processed_reply,usage:{prompt_tokens:prompt_tokens,completion_tokens:completion_tokens,total_tokens:total_tokens}}# -------------------------- 步骤6保存到 jsonl 文件一行一条记录 --------------------------save_filestream_chat_history.jsonlwithopen(save_file,a,encodingutf-8)asf:f.write(json.dumps(chat_record,ensure_asciiFalse)\n)print(f✅ 对话记录已保存本次总token{total_tokens})# 返回处理完的文本 完整记录外部可以继续使用returnprocessed_reply,chat_recordif__name____main__:# 测试上下文msg_list[{role:system,content:你是简洁的Python技术助手},{role:user,content:简单讲下生成器yield的原理}]resp_text,recordstream_chat_demo(msg_list)print(\n返回给程序使用的处理后文本)print(resp_text)将这段代码保存为deepseek_streaming_chat.py文件然后在虚拟环境中执行该脚本即可看到实时输出的对话效果。3、代码关键点解析与之前的非流式代码相比这个流式输出的代码有以下三个关键调整点开启流式输出在请求参数中设置streamTrue此时 API 会返回一个可迭代的响应对象而不是完整的响应结果。同时为了在流式输出结束后获取本次请求的令牌使用统计信息需要额外设置stream_options{include_usage: True}—— 如果不添加这个配置响应的usage字段将返回空值无法统计令牌使用情况。迭代读取内容块通过for chunk in stream循环逐个读取响应流中的内容块。每个内容块中都包含了模型本次返回的增量文本内容。这些增量内容需要直接打印到控制台实现实时输出的效果。记录完整回复由于每个内容块中只包含增量内容我们需要在客户端侧将所有内容块的增量内容拼接成完整的回复文本 —— 否则无法将完整的assistant角色回复加入到下一轮请求的上下文消息列表中。需要特别注意的是在打印每个内容块的增量内容时必须将print()方法的end参数设置为空字符串end避免每个内容块打印后自动换行破坏完整回复的排版效果同时需要将flush参数设置为True强制 Python 的标准输出流立即将内容输出到控制台而不是暂存到缓冲区中 —— 这是保证打字机效果实时性的关键前提。下一节-Python 连接 DeepSeek APIOpenAI 对话方式总结免责声明本文档中描述的 API 调用方法及相关代码示例均基于 DeepSeek 官方文档和公开技术社区的讨论内容。在实际使用过程中因 API 版本差异、服务端接口升级、账户权限配置等导致的程序异常或数据风险本文作者及 DeepSeek 官方不承担任何责任。请务必先在测试环境中完成充分验证后再将逻辑部署到生产环境中。本文档的内容会随着 DeepSeek API 的版本迭代和功能升级而更新最新版本请以官方文档的内容为准。注文档部分内容可能由 AI 生成