大语言模型 API 进阶指南：DeepSeek 与 Qwen 的深度应用与封装实践

在昨天小宁已经教大家怎么去获取各个平台的API-kEY,然后也带大家了解了最简单的大模型的调用，那么今天就带大家进阶了解一些更加详细的功能。

在大语言模型的实际应用中，除了基础的问答功能，深度思考能力、推理过程的获取以及灵活的对话模式往往是提升应用体验的关键。本文将围绕 DeepSeek 和 Qwen 两款优秀的大语言模型，深入探讨其进阶功能的实现，包括深度思考机制、推理过程处理、多轮对话实现以及 API 封装技巧，帮助开发者更高效地集成大语言模型能力。

一、不支持reasoning_content的deepseek-chat：深度思考与输出形式

支持reasoning_content（是否自带原生的思考过程）的模型有 Qwen 系列模型、DeepSeek-R1 系列模型等，以下是具体介绍：

• Qwen 系列模型：如 Qwen3、QwQ（基于 Qwen2.5）等模型具有强大的推理能力，会先输出思考过程，再输出回答内容。可以通过enable_thinking参数开启思考模式，通过响应的reasoning_content字段获取思考过程，content字段获取回复内容。
• DeepSeek-R1 系列模型：包含 DeepSeek-R1 及其蒸馏版本如 DeepSeek-R1-Distill-Qwen-32B 等。该系列模型在后训练阶段大规模使用了强化学习技术，在数学、代码、自然语言推理等任务上表现出色，思考过程会通过reasoning_content字段返回，最终回复会通过content字段返回。
• 基于 vLLM 的相关模型：vLLM 支持 DeepSeek R1 系列、QwQ-32B 等推理模型，这些模型输出中包含reasoning_content字段，该字段包含导致最终结论的推理步骤，而content字段包含最终结论。

1.1 深度思考功能的启用

DeepSeek 的深度思考功能(回答的答案更加详细)并非默认开启，需要通过特定参数进行配置。在 API 调用中，通过extra_body参数传递"deep_think": True即可启用该功能，代码示例如下：

from openai import OpenAI
import os
from dotenv import load_dotenv

load_dotenv()

# 初始化DeepSeek客户端
client = OpenAI(
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com/v1"
)

# 启用深度思考功能调用API
response = client.chat.completions.create(
    model="deepseek-chat",
    messages=[
        {"role": "system", "content": "你是一个逻辑清晰的助手，擅长分步解决问题"},
        {"role": "user", "content": "如何用Python实现斐波那契数列？要求输出前10项"}
    ],
    extra_body={
        "deep_think": True  # 启用深度思考
    }
)

 启用深度思考后，模型会在生成最终答案前，先进行内部的逻辑推演，尤其适合需要复杂推理的问题，如算法实现、数学证明、逻辑分析等场景。

1.2 推理过程的获取

与部分模型不同，DeepSeek 的响应对象中没有自带reasoning_content属性，其推理过程需要通过对content内容的结构化要求来获取。具体做法是在系统提示中明确要求模型将 "思考过程" 和 "最终答案" 分开输出，再通过字符串处理提取推理过程。

非流式输出下的推理过程提取

# 非流式输出时获取推理过程
full_response = response.choices[0].message.content

# 按照预设格式分割内容（假设系统提示要求以"思考过程："和"最终答案："为标识）
parts = full_response.split("最终答案：")
if len(parts) >= 2:
    reasoning = parts[0].replace("思考过程：", "").strip()  # 提取思考过程
    answer = parts[1].strip()  # 提取最终答案
else:
    # 处理格式不符合预期的情况
    reasoning = "未获取到推理过程"
    answer = full_response

print("【DeepSeek思考过程】")
print(reasoning)
print("\n【DeepSeek最终答案】")
print(answer)

流式输出下的推理过程提取

流式输出时，内容会分块返回，需要通过状态判断区分 "思考过程" 和 "最终答案" 阶段：

# 流式输出时获取推理过程
thinking_phase = True  # 标记是否处于思考过程阶段
full_content = ""
print("【DeepSeek流式思考过程】")

for chunk in response:
    content = chunk.choices[0].delta.content
    if content:
        full_content += content
        # 当检测到"最终答案："时，切换到答案阶段
        if "最终答案：" in full_content:
            thinking_phase = False
            continue
        # 仅在思考阶段输出内容
        if thinking_phase:
            print(content, end="", flush=True)

# 提取最终答案
answer_start = full_content.find("最终答案：") + len("最终答案：")
answer = full_content[answer_start:].strip()
print("\n\n【DeepSeek流式最终答案】")
print(answer)

 通过这种方式，即使 DeepSeek 没有原生的推理过程属性，也能稳定获取其思考逻辑，便于开发者理解模型的决策过程，或用于教育、调试等场景。

1.3 输出形式的灵活控制

DeepSeek 支持通过系统提示和参数控制输出形式，例如：

• 要求输出代码时指定编程语言和格式（如 "用 Python 实现，代码用``` 包裹 "）；
• 要求推理过程分步骤（如 "请分 3 步分析问题"）；
• 控制输出长度（如 "简洁回答，不超过 200 字"）。

示例：通过系统提示控制输出形式

messages=[
    {"role": "system", "content": "请先分析问题解决思路（分点说明），再给出答案。分析部分不超过3点，答案部分用Markdown代码块包裹Python代码。"},
    {"role": "user", "content": "写一个函数，计算1到n的累加和"}
]

1.4 完整代码案例

案例一：

from openai import OpenAI
import os
from dotenv import load_dotenv

load_dotenv()

# 初始化DeepSeek客户端
client = OpenAI(
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com/v1"
)

# 启用深度思考功能调用API
response = client.chat.completions.create(
    model="deepseek-chat",
    # messages=[
    #     {"role": "system", "content": "你是一个逻辑清晰的助手，擅长分步解决问题,需要以两点来回答：1、思考过程2、最终答案"},
    #     {"role": "user", "content": "如何用Python实现斐波那契数列？要求输出前10项"}
    # ],
    messages=[
        {"role": "system", "content": "请先分析问题解决思路（分点说明），再给出答案。分析部分不超过3点，答案部分用Markdown代码块包裹Python代码。"},
        {"role": "user", "content": "写一个函数，计算1到n的累加和"}
    ],
    stream=True,
    # stream=False,# 非流式输出
    extra_body={
        "deep_think": True  # 启用深度思考
    }
)

# 1、非流式输出时获取推理过程
# full_response = response.choices[0].message.content
#
# # 按照预设格式分割内容（假设系统提示要求以"思考过程："和"最终答案："为标识）
# parts = full_response.split("最终答案：")
# if len(parts) >= 2:
#     reasoning = parts[0].replace("思考过程：", "").strip()  # 提取思考过程
#     answer = parts[1].strip()  # 提取最终答案
# else:
#     # 处理格式不符合预期的情况
#     reasoning = "未获取到推理过程"
#     answer = full_response
#
# print("【DeepSeek思考过程】")
# print(reasoning)
# print("\n【DeepSeek最终答案】")
# print(answer)

# 2、 流式输出时获取推理过程
thinking_phase = True  # 标记是否处于思考过程阶段
full_content = ""
print("【DeepSeek流式思考过程】")

for chunk in response:
    content = chunk.choices[0].delta.content
    if content:
        full_content += content
        # 当检测到"最终答案："时，切换到答案阶段
        if "最终答案：" in full_content:
            thinking_phase = False
            continue
        # 仅在思考阶段输出内容
        if thinking_phase:
            print(content, end="", flush=True)

# 提取最终答案
answer_start = full_content.find("最终答案：") + len("最终答案：")
answer = full_content[answer_start:].strip()
print("\n\n【DeepSeek流式最终答案】")
print(answer)

案例二：

# 导入OpenAI模块，用于调用OpenAI的API
from openai import OpenAI
import os
from dotenv import load_dotenv

# 加载环境变量：不直接暴露秘钥，防止秘钥泄露
load_dotenv()

# 1、秘钥，指定基础url地址
# 使用指定的API密钥和基础URL创建OpenAI客户端实例
client = OpenAI(
    api_key=os.getenv("deepseek_API_KER"),  # 获取环境变量中的API密钥
    base_url="https://api.deepseek.com/v1",  # 指定基础URL
)

# 2、指定模型名称，messages的会话内容，流式输出，是否深度思考
# 调用OpenAI的聊天完成接口
response = client.chat.completions.create(
    model="deepseek-chat",  # 指定模型名称
    messages=[
        {"role": "system", "content": "你是AI助手，请分两部分回答：1.思考过程；2.最终答案"},  # 系统回答问题的要求
        {"role": "user", "content": "帮我实现九九乘法表，给出python源码"},  # 用户的问题
    ],
    # stream=False,  # 设置非流式响应，回答的结果一次性输出
    stream=True,  # 设置流式响应，回答的结果会分段输出
    extra_body={
        "deep_think": True  # 通过 extra_body 传递 DeepSeek 特有参数
    } # 启用深度思考功能
)

# 注意：deepseek返回的对象中没有reasoning_content这个思考过程属性，但是我们可以要求设置"content"对答案提要求，然后获取到思考过程


# 非流式输出
# full_response = response.choices[0].message.content
# parts = full_response.split("最终答案：")
# reasoning = parts[0].replace("思考过程：", "").strip()  # 提取并清理思考过程
# answer = parts[1].strip() if len(parts) > 1 else full_response
#
# print("思考过程：")
# print(reasoning)
#
# print("\n最终答案：")
# print(answer)

# 流式输出
thinking_phase = True
full_content = ""

print("【思考过程】")
for chunk in response:
    content = chunk.choices[0].delta.content
    if content:
        full_content += content

        # 判断是否进入“最终答案”阶段
        if "最终答案：" in full_content:
            thinking_phase = False
            continue

        if thinking_phase:
            print(content, end="", flush=True)

输出结果：
 

二、支持reasoning_content的deepseek-r1：深度思考与输出形式

2.1 深度思考功能的启用

Qwen 的深度思考功能同样需要通过参数启用，与 DeepSeek 类似，可通过extra_body传递"deep_think": True参数，让模型进行更充分的逻辑推演：

import os
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()

# 创建 DashScope API 客户端，使用兼容 OpenAI 接口的模式
client = OpenAI(
    # 若没有配置环境变量，请用你的 API Key 替换下一行：
    api_key=os.getenv("QWEN_API_KEY"),  # 获取 API Key 的方式：https://help.aliyun.com/zh/model-studio/developer-reference/get-api-key
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)

# 调用模型生成回复
reasoning = client.chat.completions.create(
    model="deepseek-r1",  # 可替换为其他所需模型
    messages=[
        {'role': 'user', 'content': '9.9和9.11谁大'}
    ],
     stream=False, # 非流式输出
     # stream=True, # 流式输出
     # 启用深度思考功能  回答的问题会更加详细
    extra_body={
            "deep_think": True  # 通过 extra_body 传递 DeepSeek 特有参数
        }
)

启用后，Qwen 会对问题进行多步骤分析，尤其适合数学运算、逻辑推理等需要分步验证的场景。

2.2 推理过程的直接获取

Qwen 的显著优势是其响应对象中自带reasoning_content属性，专门用于存储推理过程，无需像 DeepSeek 那样通过字符串分割提取，极大简化了开发流程。

非流式输出下的推理过程获取

# 非流式输出时直接获取推理过程和最终答案
reasoning = response.choices[0].message.reasoning_content  # 原生推理过程属性
answer = response.choices[0].message.content  # 最终答案

print("【Qwen思考过程】")
print(reasoning)
print("\n【Qwen最终答案】")
print(answer)

流式输出下的推理过程获取

流式输出时，Qwen 的推理过程和最终答案会通过不同的字段分块返回，可直接通过reasoning_content字段提取：

# 流式输出时获取推理过程
print("【Qwen流式思考过程】")
for chunk in response:
    reasoning_chunk = chunk.choices[0].delta.reasoning_content
    if reasoning_chunk:
        print(reasoning_chunk, end="", flush=True)

# 重新调用以获取最终答案（实际应用中可在同一循环中处理）
print("\n\n【Qwen流式最终答案】")
for chunk in response:
    answer_chunk = chunk.choices[0].delta.content
    if answer_chunk:
        print(answer_chunk, end="", flush=True)

2.3 输出形式的控制

Qwen 支持通过系统提示和模型参数灵活控制输出形式，例如指定答案格式（列表、表格、代码块）、限制长度、调整语气等。示例如下：

# 控制Qwen输出为Markdown表格形式
messages=[
    {"role": "system", "content": "请用Markdown表格对比下列两种动物的特征：猫和狗。表格需包含'特征'、'猫'、'狗'三列。"},
    {"role": "user", "content": "对比猫和狗的生活习性"}
]

通过简单的提示词设计，Qwen 即可输出结构化的内容，降低后续数据处理的复杂度。

2.4 完整代码案例

代码：

import os
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()

# 创建 DashScope API 客户端，使用兼容 OpenAI 接口的模式
client = OpenAI(
    # 若没有配置环境变量，请用你的 API Key 替换下一行：
    api_key=os.getenv("QWEN_API_KEY"),  # 获取 API Key 的方式：https://help.aliyun.com/zh/model-studio/developer-reference/get-api-key
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)

# 调用模型生成回复
response = client.chat.completions.create(
    model="deepseek-r1",  # 可替换为其他所需模型
    messages=[
        {'role': 'user', 'content': '9.9和9.11谁大'}
    ],
# 控制Qwen输出为Markdown表格形式
    # messages=[
    #     {"role": "system", "content": "请用Markdown表格对比下列两种动物的特征：猫和狗。表格需包含'特征'、'猫'、'狗'三列。"},
    #     {"role": "user", "content": "对比猫和狗的生活习性"}
    # ],
     # stream=False, # 非流式输出
     stream=True, # 流式输出
     # 启用深度思考功能  回答的问题会更加详细
    extra_body={
            "deep_think": True  # 通过 extra_body 传递 DeepSeek 特有参数
        }
)

# 注意：Qwen返回的内容对象中自带推理过程和最终答案

# 非流式输出时直接获取推理过程和最终答案
# reasoning = response.choices[0].message.reasoning_content  # 原生推理过程属性
# answer = response.choices[0].message.content  # 最终答案
#
# print("【Qwen思考过程】")
# print(reasoning)
# print("\n【Qwen最终答案】")
# print(answer)

# 流式输出时获取推理过程
print("【Qwen流式思考过程】")
for chunk in response:
    reasoning_chunk = chunk.choices[0].delta.reasoning_content
    if reasoning_chunk:
        print(reasoning_chunk, end="", flush=True)

# 重新调用以获取最终答案（实际应用中可在同一循环中处理）
print("\n\n【Qwen流式最终答案】")
for chunk in response:
    answer_chunk = chunk.choices[0].delta.content
    if answer_chunk:
        print(answer_chunk, end="", flush=True)

运行结果：

三、多轮对话的实现

多轮对话是大语言模型应用的核心场景之一，其核心在于维护对话上下文，让模型能够理解历史交互信息。无论是 DeepSeek 还是 Qwen，多轮对话的实现逻辑类似：通过一个列表存储历史消息，每次调用 API 时将该列表传入，实现上下文的延续。

3.1 多轮对话的核心逻辑

多轮对话的实现需要以下关键步骤：

1. 初始化一个消息列表（messages），用于存储历史对话；
2. 每次用户输入后，将用户消息添加到列表；
3. 调用 API 时，将消息列表作为参数传入，模型会基于全部历史消息生成回复；
4. 将模型的回复添加到消息列表，完成一轮对话；
5. 可选：限制消息列表长度，避免上下文过长导致性能下降。

3.2  qwen-plus多轮对话示例

代码：

# 封装之后以后可以直接导入项目直接使用，不需要每次使用的时候都单独写
import os
from openai import OpenAI
from dotenv import load_dotenv


load_dotenv()

def get_response(messages):
    client = OpenAI(
        # 若没有配置环境变量，请用阿里云百炼API Key将下行替换为：api_key="sk-xxx",
        api_key=os.getenv("Qwen_API_KER"),
        base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    )
    response = client.chat.completions.create(
        model="qwen-plus",
        messages=messages
    )
    return response

# 定义一个初始的聊天消息列表
new_messages = [
    {
        "role": "system",
        "content": """你是一名智能咖啡店助手，请根据用户的喜好推荐一杯专属咖啡。
        你需要逐步询问以下参数：
        1. 咖啡种类（如美式、拿铁、卡布奇诺等）
        2. 浓度偏好（淡、标准、浓）
        3. 饮品温度（热饮、冷饮）
        4. 糖分偏好（无糖、少糖、标准糖）
        5. 是否添加额外配料（如奶泡、焦糖、香草糖浆）

        每次只问一个问题，直到所有信息齐全后，输出完成推荐。
        """
    }
]
assistant_output = "欢迎来到智能咖啡店！请问您喜欢什么类型的咖啡？例如：美式、拿铁或卡布奇诺？"
print(f"【助手】{assistant_output}")

while "您的专属咖啡已准备就绪" not in assistant_output:
    user_input = input("【顾客】")
    # 将顾客输入的内容添加到new_messages消息列表中
    new_messages.append({"role": "user", "content": user_input})
    # 获取ai回答的结果
    assistant_output = get_response(new_messages).choices[0].message.content
    # 将大模型的回复信息追加到messages列表中
    new_messages.append({"role": "assistant", "content": assistant_output})
    print(f"【助手】{assistant_output}")
    print("\n")

 运行结果：

3.3 DeepSeek 多轮对话示例

def deepseek_multi_turn_demo():
    client = OpenAI(
        api_key=os.getenv("DEEPSEEK_API_KEY"),
        base_url="https://api.deepseek.com/v1"
    )
    
    # 初始化对话上下文
    messages = [
        {"role": "system", "content": "你是一个旅行助手，会根据用户的问题逐步推荐旅行方案。请分步骤回答，先分析需求，再给建议。"}
    ]
    
    print("DeepSeek多轮对话演示（输入'退出'结束）")
    while True:
        user_input = input("\n用户：")
        if user_input == "退出":
            break
        
        # 添加用户输入到上下文
        messages.append({"role": "user", "content": user_input})
        
        # 调用API，传入完整上下文
        response = client.chat.completions.create(
            model="deepseek-chat",
            messages=messages,
            extra_body={"deep_think": True}
        )
        
        # 获取并输出回复
        assistant_reply = response.choices[0].message.content
        print("DeepSeek助手：", assistant_reply)
        
        # 将回复添加到上下文
        messages.append({"role": "assistant", "content": assistant_reply})
        
        # 限制上下文长度（保留最近10条消息）
        if len(messages) > 10:
            messages = messages[-10:]

# 运行示例
deepseek_multi_turn_demo()

3.4 多轮对话的优化技巧

• 上下文长度控制：通过限制消息列表长度（如保留最近 10-20 条），避免因上下文过长导致的 API 调用延迟和费用增加。
• 历史消息精简：对过长的历史消息进行摘要处理，保留关键信息，减少冗余内容。
• 角色区分清晰：严格使用"system"、"user"、"assistant"三种角色，确保模型正确理解消息来源。
• 状态重置：提供 "清空对话" 功能，通过重置消息列表实现上下文的重新开始。

四、大语言模型 API 的封装实践

直接调用 API 虽然可行，但在实际开发中，未封装的代码往往存在密钥管理混乱、异常处理缺失、复用性低等问题。通过封装，可以将重复逻辑抽象为工具类，提高代码的可维护性和扩展性。下面我们将基于 DeepSeek 和 Qwen 的特性，实现一个通用的大语言模型 API 客户端。

4.1 封装的核心目标

一个完善的 API 封装应实现以下目标：

• 统一的接口调用方式，屏蔽不同模型的差异；
• 安全的密钥管理，支持环境变量或配置文件；
• 完善的异常处理，包括网络错误、API 限制等；
• 支持单轮对话、多轮对话、流式输出三种模式；
• 内置日志系统，便于调试和问题追踪。

代码：

import os
from openai import OpenAI, APIError, Timeout
import logging
from dotenv import load_dotenv
from typing import Dict, Generator

load_dotenv()

class DeepSeekAPI:

    # 初始化函数
    def __init__(self, api_key: str = None, base_url: str = "https://api.deepseek.com/v1",
                 model: str = "deepseek-chat", log_level: str = "INFO"):
        """
        初始化 DeepSeek API 客户端。

        Args:
            api_key (str): 可选的 API 密钥。如果未提供，则从环境变量 `DEEPSEEK_API_KEY` 获取。
            base_url (str): API 的基础 URL，默认为 DeepSeek 的官方地址。
            model (str): 使用的模型名称，默认为 "deepseek-chat"。
            log_level (str): 日志级别，默认为 "INFO"。可选：DEBUG/INFO/WARNING/ERROR。
        Raises:
            ValueError: 如果未提供 API 密钥且环境变量中也未找到，则抛出异常。
        """
        # 优先使用传入的 API Key，否则尝试从环境变量获取
        self.api_key = api_key or os.getenv("DEEPSEEK_API_KEY")
        if not self.api_key:
            raise ValueError("API 密钥未提供且未在环境变量中找到")

        # 初始化 OpenAI 客户端
        self.client = OpenAI(
            api_key=self.api_key,
            base_url=base_url
        )

        # 设置默认模型
        self.model = model

        # 初始化消息列表用于多轮对话
        self.messages = []

        # 配置日志系统
        logging.basicConfig(level=getattr(logging, log_level.upper(), logging.INFO))
        self.logger = logging.getLogger(__name__)


    def _process_response(self, completion) -> Dict:
        """
        处理 API 响应，提取思考过程和最终答案。

        Args:
            completion: API 返回的响应对象。

        Returns:
            dict: 包含 'reasoning' 和 'answer' 的字典。
        """
        reasoning_content = None
        answer_content = None

        # 检查响应对象中是否存在 reasoning_content 属性，前面讲过QWEN有这个属性，deepseek没有这个属性
        if hasattr(completion.choices[0].message, 'reasoning_content'):
            reasoning_content = completion.choices[0].message.reasoning_content
        if hasattr(completion.choices[0].message, 'content'):
            answer_content = completion.choices[0].message.content

        return {
            'reasoning': reasoning_content,
            'answer': answer_content
        }

    def chat(self, message: str) -> Dict:
        """
        单轮对话模式。

        Args:
            message (str): 用户输入的消息。

        Returns:
            dict: 包含 'reasoning' 和 'answer' 的字典。
        """
        try:
            completion = self.client.chat.completions.create(
                model=self.model,
                messages=[{"role": "user", "content": message}]
            )
            return self._process_response(completion)
        except (APIError, Timeout) as e:
            self.logger.error(f"API 调用失败: {type(e).__name__}")
            return {'reasoning': None, 'answer': "抱歉，我现在无法正常回复，请稍后再试。"}
        except Exception as e:
            self.logger.error(f"未知错误: {type(e).__name__}")
            return {'reasoning': None, 'answer': "抱歉，我现在无法正常回复，请稍后再试。"}

    def chat_with_context(self, message: str, max_context_length: int = 10) -> Dict:
        """
        多轮对话模式。

        Args:
            message (str): 用户输入的消息。
            max_context_length (int): 最大上下文长度，用于限制历史对话条数。

        Returns:
            dict: 包含 'reasoning' 和 'answer' 的字典。
        """
        try:
            self.messages.append({"role": "user", "content": message})
            self.messages = self.messages[-max_context_length:]

            completion = self.client.chat.completions.create(
                model=self.model,
                messages=self.messages
            )

            response = self._process_response(completion)

            self.messages.append({"role": "assistant", "content": response['answer']})
            return response
        except (APIError, Timeout) as e:
            self.logger.error(f"API 调用失败: {type(e).__name__}")
            return {'reasoning': None, 'answer': "抱歉，我现在无法正常回复，请稍后再试。"}
        except Exception as e:
            self.logger.error(f"未知错误: {type(e).__name__}")
            return {'reasoning': None, 'answer': "抱歉，我现在无法正常回复，请稍后再试。"}

    def stream_chat(self, message: str) -> Generator[Dict, None, None]:
        """
        流式对话模式，逐步返回模型生成内容。

        Args:
            message (str): 用户输入的消息。

        Yields:
            dict: 包含 'type' 和 'content' 的字典。
                - type: 'reasoning', 'answer', 'start_answer', 'error'
                - content: 对应的内容字符串
        """
        try:
            completion = self.client.chat.completions.create(
                model=self.model,
                messages=[{"role": "user", "content": message}],
                stream=True
            )

            reasoning_content = ""
            answer_content = ""
            is_answering = False

            for chunk in completion:
                if not chunk.choices:
                    continue

                delta = chunk.choices[0].delta

                if hasattr(delta, 'reasoning_content') and delta.reasoning_content:
                    reasoning_content += delta.reasoning_content
                    yield {'type': 'reasoning', 'content': delta.reasoning_content}

                elif hasattr(delta, 'content') and delta.content:
                    if not is_answering:
                        is_answering = True
                        if self.model == "deepseek-reasoner":
                            yield {'type': 'start_answer'}

                    answer_content += delta.content
                    yield {'type': 'answer', 'content': delta.content}
        except (APIError, Timeout) as e:
            self.logger.error(f"API 调用失败: {type(e).__name__}")
            yield {'type': 'error', 'content': "抱歉，我现在无法正常回复，请稍后再试。"}
        except Exception as e:
            self.logger.error(f"未知错误: {type(e).__name__}")
            yield {'type': 'error', 'content': "抱歉，我现在无法正常回复，请稍后再试。"}

    def clear_context(self) -> None:
        """
        清空当前对话上下文。
        """
        self.messages = []


# 示例运行入口
if __name__ == "__main__":
    api = DeepSeekAPI(api_key=os.getenv("DEEPSEEK_API_KEY"), model="deepseek-chat")

    print("\n============== 普通对话模式 ===============")
    response = api.chat("9.9和9.11谁大？")
    print("思考过程：")
    print(response['reasoning'])
    print("最终答案：")
    print(response['answer'])

    print("\n============== 多轮对话模式 ==============")
    print("输入 '退出' 结束对话")
    while True:
        user_input = input("\n用户: ")
        if user_input.strip().lower() == '退出':
            break
        response = api.chat_with_context(user_input)
        print("\nAI 思考过程:", response['reasoning'])
        print("AI 回答:", response['answer'])

    print("\n============== 流式对话模式 ==============")
    for chunk in api.stream_chat("9.9和9.11谁大？"):
        if chunk['type'] == 'reasoning':
            print(chunk['content'], end='', flush=True)
        elif chunk['type'] == 'start_answer':
            print("\n=== 完整回复 ===\n")
        elif chunk['type'] == 'answer':
            print(chunk['content'], end='', flush=True)
        elif chunk['type'] == 'error':
            print("\n错误：", chunk['content'])

运行结果：