Python连接DeepSeek API和OpenAI对话方式基础实现_Python

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

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

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

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

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

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

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

2、完整代码示例

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

import os
from dotenv import load_dotenv
from openai import openai, 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` 来结束对话。")

# 开启循环，持续等待用户输入并发送请求
while true:
  # 1. 接收用户的输入内容
  user_input = input("\n你：")
  if user_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.content
    print(f"\nai：{ai_reply}")

    # 5. 将模型的回复，作为assistant角色消息，追加到历史消息列表中
    messages.append({"role": "assistant", "content": ai_reply})

  except apiconnectionerror as e:
    print("ai：抱歉，我当前无法连接到服务，请检查你的网络连接或稍后再试。")
    print(f"[错误详情] 连接失败：{str(e)}")
    # 移除用户的提问消息，避免污染下一次请求的上下文环境
    messages.pop()
  except ratelimiterror as e:
    print("ai：抱歉，当前我的访问压力过大或账户额度不足，请稍后再试。")
    print(f"[错误详情] 访问受限：{str(e)}")
    messages.pop()
  except apierror as e:
    print(f"ai：抱歉，请求处理失败，错误码：{e.status_code}")
    print(f"[错误详情] {str(e)}")
    messages.pop()
  except exception as e:
    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

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

import os
import json
from openai import openai

client = openai(
    api_key=os.environ["deepseek_api_key"],
    base_url="https://api.deepseek.com"
)

def stream_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_id

    print("\n🤖 ai回复：", end="", flush=true)

    # -------------------------- 步骤2：循环迭代每一个chunk碎片 --------------------------
    for chunk in stream:
        # 每一轮覆盖，循环结束后last_chunk就是网络收到的最后一包数据
        last_chunk = chunk

        # delta 是增量内容，只有新生成的一小段文字
        delta = chunk.choices[0].delta

        if delta.content:
            # 实时打印，实现打字机效果
            print(delta.content, end="", flush=true)
            # 把增量拼接到完整字符串
            full_ai_text += delta.content

    print("\n") # 流式结束换行

    # -------------------------- 步骤3：循环结束！解析【最后一个chunk】的元数据 --------------------------
    # ⚠️ 重点：usage / request_id / finish_reason 全部在 last_chunk
    request_id = last_chunk.id
    model_name = last_chunk.model
    finish_reason = last_chunk.choices[0].finish_reason

    # token消耗，只存在最后一块chunk
    prompt_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"
    with open(save_file, "a", encoding="utf-8") as f:
        f.write(json.dumps(chat_record, ensure_ascii=false) + "\n")

    print(f"✅ 对话记录已保存，本次总token：{total_tokens}")

    # 返回处理完的文本 + 完整记录，外部可以继续使用
    return processed_reply, chat_record


if __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对话方式基础实现的文章就介绍到这了,更多相关python连接deepseek和openai对话内容请搜索代码网以前的文章或继续浏览下面的相关文章希望大家以后多多支持代码网！