Python 连接 DeepSeek API，OpenAI 两种对话方式基础实现

上一节-Python 连接 DeepSeek API，OpenAI的基础操作

文章目录

• 三、进阶功能 1：实现多轮上下文对话
• • 1、技术原理：传递对话历史
  • 2、完整代码示例
  • 3、代码关键点解析
• 四、进阶功能 2：实现流式输出
• • 1、技术原理
  • 2、完整代码示例
  • 3、代码关键点解析

三、进阶功能 1：实现多轮上下文对话

上述基础示例仅适用于单轮问答场景，它的上下文是独立的 —— 模型无法记住上一次用户输入的信息；而在实际应用中，用户更需要的是连续多轮对话的交互效果，这就需要程序在每次请求时，都将完整的历史对话消息列表提交给 API。

1、技术原理：传递对话历史

DeepSeek API 本身是无状态的，它不会保存任何历史对话数据 —— 要实现连续多轮对话的上下文记忆，必须在每次请求时，将完整的历史对话消息列表（包含之前的所有用户输入和模型回复），按顺序通过messages参数提交给 API 服务端。

实现多轮对话的核心逻辑，是在客户端本地维护一个完整的对话消息列表：

1. 列表的第一个元素，通常是设定模型身份的system角色消息。
2. 每一轮用户输入新的提问后，将这个提问内容以user角色的形式，追加到消息列表的末尾。
3. 将这个包含完整历史对话的消息列表，作为messages参数的值，发送给 API 服务端，模型会基于这个完整的上下文信息生成回复。
4. 收到模型的有效回复后，将模型的回复内容以assistant角色的形式，追加到消息列表的末尾，用于下一轮对话的上下文构建。
5. 重复步骤 2 到 4 的操作，直到用户主动结束对话。

需要特别注意的是，必须保证messages数组的消息顺序和对话的实际发生顺序完全一致，且其中的role角色设置必须正确，这样模型才能正确理解完整的对话上下文。

2、完整代码示例

下面的代码将实现一个支持多轮上下文对话的机器人，它会在控制台等待用户的输入请求，连续进行对话，直到用户输入exit或quit命令主动退出程序：

importosfromdotenvimportload_dotenvfromopenaiimportOpenAI,APIError,APIConnectionError,RateLimitError load_dotenv()# 初始化客户端client=OpenAI(api_key=os.getenv("DEEPSEEK_API_KEY"),base_url=os.getenv("DEEPSEEK_BASE_URL","https://api.deepseek.com"))# 初始化消息列表，带上system角色的上下文设定messages=[{"role":"system","content":"你是一个知识渊博的AI助手，名叫「小迪」。在回答用户问题时，要尽量简洁、通俗易懂，必要时提供具体的例子或代码片段。请使用中文回答，并保持语气友好、专业。"}]print("=== DeepSeek 多轮对话机器人 ===")print("你可以持续输入内容进行对话，输入 `exit` 或 `quit` 来结束对话。")# 开启循环，持续等待用户输入并发送请求whileTrue:# 1. 接收用户的输入内容user_input=input("\n你：")ifuser_input.lower()in['exit','quit']:print("对话结束，再见！")break# 2. 将用户的新输入，追加到历史消息列表中messages.append({"role":"user","content":user_input})try:# 3. 携带完整的历史消息，发送请求response=client.chat.completions.create(model="deepseek-chat",messages=messages,# 提交完整的对话历史max_tokens=1024,temperature=0.7,stream=False)# 4. 解析模型的回复内容ai_reply=response.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(f"AI：抱歉，请求处理失败，错误码：{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模型支持的上限为 8192），API 会返回错误。在实际应用中，需要根据使用的模型上限，对历史消息列表进行管理（例如截断早期的历史消息、压缩上下文、摘要总结等），避免超出模型的上下文长度限制。

四、进阶功能 2：实现流式输出

在默认情况下，API 会在模型完整生成所有回复内容后，一次性返回完整的响应结果 —— 如果回复内容较长，或者网络延迟较高，用户可能会等待较长时间，看不到任何实时反馈，这会严重影响交互体验。为了优化这种场景下的用户体验，DeepSeek API 提供了流式输出的能力，让回复内容可以逐块实时返回，实现打字机效果的实时输出体验。

1、技术原理

流式输出是基于 HTTP 的长连接机制实现的：客户端在请求参数中设置stream=True，此时 API 服务端会与客户端保持长连接，将模型生成的回复内容分成多个小块（Chunk），以文本流的形式实时返回给客户端；客户端则通过迭代的方式，实时读取这些内容块并依次打印，最终在控制台形成类似人类打字的输出效果。

使用流式输出时，API 的响应格式与非流式输出完全不同 —— 响应结果是一个可迭代的响应对象，每个迭代元素都对应一个包含实际内容的小块，需要额外的逻辑来逐个读取、解析和处理这些内容块。需要注意的是，开启流式输出后，响应中的usage字段将默认不返回 —— 如果需要获取本次请求的令牌使用统计信息，需要在请求中添加stream_options参数，明确要求返回该字段。

2、完整代码示例

执行流程：发起请求 → 迭代接收 chunk → 实时输出 → 拼接完整文本 → 取最后 chunk 元数据 (token、finish_reason) → 业务处理 → 组装记录 → 保存 json

下面是一个支持流式输出的多轮对话代码示例，它将实现打字机效果的实时回复输出：

importosimportjsonfromopenaiimportOpenAI client=OpenAI(api_key=os.environ["DEEPSEEK_API_KEY"],base_url="https://api.deepseek.com")defstream_chat_demo(messages:list):# -------------------------- 步骤1：发起流式请求，拿到迭代器stream --------------------------stream=client.chat.completions.create(model="deepseek-chat",messages=messages,temperature=0.7,max_tokens=1024,stream=True,# 开启流式stream_options={"include_usage":True}# 必须开启，否则拿不到usage)full_ai_text=""# 用来拼接完整回答文本last_chunk=None# 保存循环的最后一个chunk，用于拿usage、request_idprint("\n🤖 AI回复：",end="",flush=True)# -------------------------- 步骤2：循环迭代每一个chunk碎片 --------------------------forchunkinstream:# 每一轮覆盖，循环结束后last_chunk就是网络收到的最后一包数据last_chunk=chunk# delta 是增量内容，只有新生成的一小段文字delta=chunk.choices[0].deltaifdelta.content:# 实时打印，实现打字机效果print(delta.content,end="",flush=True)# 把增量拼接到完整字符串full_ai_text+=delta.contentprint("\n")# 流式结束换行# -------------------------- 步骤3：循环结束！解析【最后一个chunk】的元数据 --------------------------# ⚠️ 重点：usage / request_id / finish_reason 全部在 last_chunkrequest_id=last_chunk.idmodel_name=last_chunk.model finish_reason=last_chunk.choices[0].finish_reason# token消耗，只存在最后一块chunkprompt_tokens=last_chunk.usage.prompt_tokens completion_tokens=last_chunk.usage.completion_tokens total_tokens=last_chunk.usage.total_tokens# -------------------------- 步骤4：业务处理原始文本 --------------------------raw_reply=full_ai_text processed_reply=raw_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_file="stream_chat_history.jsonl"withopen(save_file,"a",encoding="utf-8")asf:f.write(json.dumps(chat_record,ensure_ascii=False)+"\n")print(f"✅ 对话记录已保存，本次总token：{total_tokens}")# 返回处理完的文本 + 完整记录，外部可以继续使用returnprocessed_reply,chat_recordif__name__=="__main__":# 测试上下文msg_list=[{"role":"system","content":"你是简洁的Python技术助手"},{"role":"user","content":"简单讲下生成器yield的原理"}]resp_text,record=stream_chat_demo(msg_list)print("\n===返回给程序使用的处理后文本===")print(resp_text)

将这段代码保存为deepseek_streaming_chat.py文件，然后在虚拟环境中执行该脚本，即可看到实时输出的对话效果。

3、代码关键点解析

与之前的非流式代码相比，这个流式输出的代码有以下三个关键调整点：

• 开启流式输出：在请求参数中设置stream=True，此时 API 会返回一个可迭代的响应对象，而不是完整的响应结果。同时，为了在流式输出结束后获取本次请求的令牌使用统计信息，需要额外设置stream_options={"include_usage": True}—— 如果不添加这个配置，响应的usage字段将返回空值，无法统计令牌使用情况。
• 迭代读取内容块：通过for chunk in stream循环，逐个读取响应流中的内容块。每个内容块中，都包含了模型本次返回的增量文本内容。这些增量内容需要直接打印到控制台，实现实时输出的效果。
• 记录完整回复：由于每个内容块中只包含增量内容，我们需要在客户端侧，将所有内容块的增量内容拼接成完整的回复文本 —— 否则，无法将完整的assistant角色回复，加入到下一轮请求的上下文消息列表中。

需要特别注意的是，在打印每个内容块的增量内容时，必须将print()方法的end参数设置为空字符串（end=""），避免每个内容块打印后自动换行，破坏完整回复的排版效果；同时，需要将flush参数设置为True，强制 Python 的标准输出流，立即将内容输出到控制台，而不是暂存到缓冲区中 —— 这是保证打字机效果实时性的关键前提。

下一节-Python 连接 DeepSeek API，OpenAI 对话方式总结

---

免责声明：本文档中描述的 API 调用方法及相关代码示例，均基于 DeepSeek 官方文档和公开技术社区的讨论内容。在实际使用过程中，因 API 版本差异、服务端接口升级、账户权限配置等导致的程序异常或数据风险，本文作者及 DeepSeek 官方不承担任何责任。请务必先在测试环境中完成充分验证后，再将逻辑部署到生产环境中。

本文档的内容会随着 DeepSeek API 的版本迭代和功能升级而更新，最新版本请以官方文档的内容为准。

（注：文档部分内容可能由 AI 生成）