1. 什么是 OpenAI？ #

OpenAI Python库是官方提供的 Python 工具包，用于与 OpenAI 的各种 AI 服务进行交互。通过这个库，你可以轻松地在 Python 程序中使用 ChatGPT、图像生成、语音识别等强大的 AI 功能。

2. 调用Completions #

• 本节完成从安装依赖到发起一次聊天补全（Chat Completions）请求的完整流程。
• 项目使用官方 OpenAI Python SDK，并将 base_url 指向 DeepSeek 兼容接口，因此同一套代码可无缝切换 OpenAI 或兼容 OpenAI 协议的其他模型服务。

本节你将学到：

• 使用 uv 安装并管理 openai 依赖
• 通过 OpenAI 客户端配置 API Key 与 Base URL（支持环境变量覆盖）
• 调用 client.chat.completions.create() 发送用户消息并获取模型回复

调用时序概览：

2.1 安装 #

uv add openai

2.2 main.py #

# 导入操作系统相关的os模块
import os

# 从openai模块中导入OpenAI类
from openai import OpenAI

# 创建OpenAI客户端对象，并设置API密钥和基础URL（支持从环境变量读取，否则使用默认值）
client = OpenAI(
    api_key=os.getenv("OPENAI_API_KEY", "sk-6dac9405d2c74a8f9a5b921b80ebb0f2"),
    base_url=os.getenv("OPENAI_BASE_URL", "https://api.deepseek.com/v1"),
)

# 使用OpenAI客户端创建一次聊天补全请求，指定模型和用户消息内容
result = client.chat.completions.create(
    model=os.getenv("OPENAI_MODEL", "deepseek-v4-pro"),
    messages=[{"role": "user", "content": "你是谁?"}],
)

# 打印聊天补全结果的第一条回复的消息内容
print(result.choices[0].message.content)

2.3 ChatCompletion #

2.3.1 ChatCompletion #

# 调用ChatCompletion对象
ChatCompletion(
    # 指定唯一标识符
    id='ae232a31-d7f9-44fc-ab3c-c8dd8b33e0a2',
    # 指定choices参数，为一个列表，包含Choice对象
    choices=[
        # 创建一个Choice对象
        Choice(
            # 生成结束的原因是‘stop’
            finish_reason='stop',
            # 当前choice的索引为0
            index=0,
            # 没有提供logprobs
            logprobs=None,
            # message为ChatCompletionMessage对象
            message=ChatCompletionMessage(
                # 回答内容是“The answer is 2.”
                content='The answer is 2.',
                # 没有拒答内容
                refusal=None,
                # 角色是‘assistant’
                role='assistant',
                # 没有注释
                annotations=None,
                # 没有音频内容
                audio=None,
                # 没有函数调用
                function_call=None,
                # 没有工具调用
                tool_calls=None,
                # 推理过程的详细内容
                reasoning_content='We are asked: "1+1=?" This is a simple arithmetic question. The answer is 2. But we need to provide the response as the AI assistant. The user likely expects just "2". But we should output it in a proper format. The instruction: "You are an AI assistant. Help the user with their query." So I\'ll answer concisely.'
            )
        )
    ],
    # 创建时间的时间戳
    created=1781092901,
    # 模型名称
    model='deepseek-v4-pro',
    # 对象类型
    object='chat.completion',
    # 没有内容监管信息
    moderation=None,
    # 没有指定服务等级
    service_tier=None,
    # 系统指纹
    system_fingerprint='fp_9954b31ca7_prod0820_fp8_kvcache_20260402',
    # usage参数，描述本次调用的token用量
    usage=CompletionUsage(
        # 完成回答消耗的token数
        completion_tokens=85,
        # 提示语占用的token数
        prompt_tokens=8,
        # 总token数
        total_tokens=93,
        # 完成token的详细信息
        completion_tokens_details=CompletionTokensDetails(
            # 被接受的预测token数量
            accepted_prediction_tokens=None,
            # 音频相关token数量
            audio_tokens=None,
            # 推理过程所用的token数
            reasoning_tokens=78,
            # 被拒绝的预测token数量
            rejected_prediction_tokens=None
        ),
        # 提示token的详细信息
        prompt_tokens_details=PromptTokensDetails(
            # 音频相关token
            audio_tokens=None,
            # 缓存的token数量
            cached_tokens=0
        ),
        # 提示缓存命中token数量
        prompt_cache_hit_tokens=0,
        # 提示缓存未命中token数量
        prompt_cache_miss_tokens=8
    )
)

2.3.2 ChatCompletion #

2.3.3 Choice #

2.3.4 finish_reason #

2.3.5 ChatCompletionMessage #

2.3.6 CompletionUsage #

2.3.7 CompletionTokensDetails #

2.3.8 PromptTokensDetails #

2.3.9 字段层级关系 #

ChatCompletion
├── id
├── created
├── model
├── object
├── moderation
├── service_tier
├── system_fingerprint
├── choices[]
│   └── Choice
│       ├── index
│       ├── finish_reason
│       ├── logprobs
│       └── message
│           ├── role
│           ├── content          ← 最常用：最终回复
│           ├── refusal
│           ├── annotations
│           ├── audio
│           ├── function_call
│           ├── tool_calls
│           └── reasoning_content ← 推理模型扩展
└── usage
    ├── prompt_tokens
    ├── completion_tokens
    ├── total_tokens
    ├── prompt_cache_hit_tokens
    ├── prompt_cache_miss_tokens
    ├── completion_tokens_details
    │   ├── reasoning_tokens
    │   ├── accepted_prediction_tokens
    │   ├── rejected_prediction_tokens
    │   └── audio_tokens
    └── prompt_tokens_details
        ├── cached_tokens
        └── audio_tokens

3. 实现Completions #

• 本节实现 client.chat.completions.create()
• 通过分层设计（Client → Resource → Completions）理解 OpenAI SDK 的核心结构：HTTP 传输、资源路由、响应模型校验各司其职。

本节你将学到：

• 用 httpx 封装 _request / post，统一处理 URL 拼接与请求头构建
• 用 Resource 模式组织 API：OpenAI.chat → Chat.completions → Completions.create
• 用 Pydantic 模型（ChatCompletion）对 API 响应做结构化校验与类型安全访问

模块结构：

openai_lite/
├── __init__.py          # 对外导出 OpenAI
├── _constants.py        # 默认配置、URL/Header 构建
├── client.py            # HTTP 客户端核心
├── resources/chat/      # Chat Completions 资源层
│   ├── __init__.py      # Chat 类
│   └── completions.py   # create() 方法
└── types/               # 响应数据模型
    ├── __init__.py
    └── chat.py          # ChatCompletion / Choice / Message

调用时序概览：

3.1 init.py #

openai_lite/init.py

from .client import OpenAI
__all__ = [
    "OpenAI"
]

3.2 _constants.py #

openai_lite/_constants.py

# 默认的OpenAI API基地址
DEFAULT_BASE_URL = "https://api.openai.com/v1"
# 默认超时时间为60秒
DEFAULT_TIMEOUT = 60.0
# 聊天补全接口的路径
CHAT_COMPLETIONS_PATH = "/chat/completions"
# 构建HTTP请求头的函数，传入API密钥，返回字典
def build_headers(api_key: str) -> dict[str, str]:
    # 返回一个包含授权和内容类型的请求头字典
    return {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json",
    }


# 构建完整的请求URL的函数，传入基地址和路径
def build_url(base_url: str, path: str) -> str:
    # 如果路径没以斜杠开头，则补上斜杠
    if not path.startswith("/"):
        path = f"/{path}"
    # 返回拼接后的完整URL
    return f"{base_url}{path}"

3.3 client.py #

openai_lite/client.py

# 导入httpx库，用于发起HTTP请求
import httpx
# 从._constants模块中导入默认基础URL、默认超时时间、请求头和URL构建方法
from ._constants import DEFAULT_BASE_URL,DEFAULT_TIMEOUT,build_headers,build_url
# 从resources目录的chat模块导入Chat类
from .resources.chat import Chat

# 定义OpenAI类
class OpenAI:
    # 初始化方法，设置API key、基础URL、超时时间
    def __init__(
        self,
        *,
        api_key: str,
        base_url: str = DEFAULT_BASE_URL,
        timeout: float = DEFAULT_TIMEOUT,
    ):
        # 保存API Key
        self.api_key = api_key
        # 移除基础URL右侧的斜杠并保存
        self.base_url = base_url.rstrip("/")
        # 保存超时时间
        self.timeout = timeout
        # 创建Chat类实例并保存
        self.chat = Chat(self)
    # 内部请求方法
    def _request(self, method: str, path: str, *, json: dict | None = None):
        # 使用httpx发起HTTP请求
        response = httpx.request(
            method,
            # 构造完整的请求URL
            build_url(self.base_url, path),
            # 构造请求头
            headers=build_headers(self.api_key),
            # 请求体（JSON格式）
            json=json,
            # 请求超时时间
            timeout=self.timeout,
        )
        # 返回HTTP响应
        return response    
    # POST请求方法
    def post(self, path: str, *, json: dict | None = None):
        # 调用_request方法发起POST请求
        return self._request("POST", path, json=json)

3.4 init.py #

openai_lite/resources/chat/init.py

# 从当前文件夹导入Completions类
from .completions import Completions

# 定义一个Chat类
class Chat:
    # 定义构造函数，接收一个client参数
    def __init__(self, client):
        # 创建Completions实例并赋值给self.completions属性
        self.completions = Completions(client)

3.5 completions.py #

openai_lite/resources/chat/completions.py

# 从项目的_constants模块导入CHAT_COMPLETIONS_PATH常量
from ..._constants import CHAT_COMPLETIONS_PATH
# 从openai.types.chat模块导入ChatCompletion类
from openai.types.chat import ChatCompletion
# 定义Completions类
class Completions():
    # 初始化方法，接收client对象
    def __init__(self, client):
        # 从client对象中获取post方法并赋值给实例变量_post
        self._post = client.post
    # 定义create方法，用于创建聊天补全
    def create(
        self,
        *,
        model,
        messages,
        **kwargs,
    ):
        # 构造请求负载，包括模型名、消息列表及其它可选参数
        payload = {"model": model, "messages": messages, **kwargs}
        # 发送POST请求到CHAT_COMPLETIONS_PATH，并传递payload作为json参数
        response = self._post(CHAT_COMPLETIONS_PATH, json=payload)
        # 对响应的json数据进行模型验证，并返回ChatCompletion对象
        return ChatCompletion.model_validate(response.json())

3.6 init.py #

openai_lite/types/init.py

from openai.types.chat import (
    ChatCompletion,
)

__all__ = [
    "ChatCompletion"
]

3.7 chat.py #

openai_lite/types/chat.py

# 导入Pydantic的BaseModel用于数据模型的定义
from pydantic import BaseModel
# 定义消息（Message）类，继承自BaseModel
class Message(BaseModel):
    # role字段，表示消息的角色（如user、assistant）
    role: str
    # content字段，表示消息内容，可以为None
    content: str | None = None

# 定义选择（Choice）类，继承自BaseModel
class Choice(BaseModel):
    # index字段，表示选择的索引
    index: int
    # message字段，表示对应的Message对象
    message: Message
    # finish_reason字段，表示完成的原因，可以为None
    finish_reason: str | None = None

# 定义聊天完成（ChatCompletion）类，继承自BaseModel
class ChatCompletion(BaseModel):
    # id字段，表示聊天完成对象的唯一标识符
    id: str
    # model字段，表示使用的模型
    model: str
    # choices字段，表示Choice对象的列表
    choices: list[Choice]

3.8 main.py #

main.py

# 导入操作系统相关的os模块
import os

# 从openai_lite模块中导入OpenAI类
+from openai_lite import OpenAI

# 创建OpenAI客户端对象，并设置API密钥和基础URL（支持从环境变量读取，否则使用默认值）
client = OpenAI(
    api_key=os.getenv("OPENAI_API_KEY", "sk-6dac9405d2c74a8f9a5b921b80ebb0f2"),
    base_url=os.getenv("OPENAI_BASE_URL", "https://api.deepseek.com/v1"),
)

# 使用OpenAI客户端创建一次聊天补全请求，指定模型和用户消息内容
result = client.chat.completions.create(
    model=os.getenv("OPENAI_MODEL", "deepseek-v4-pro"),
    messages=[{"role": "user", "content": "你是谁?"}],
)

# 打印聊天补全结果的第一条回复的消息内容
print(result.choices[0].message.content)

4. 流式响应 #

• 本节在 openai_lite 基础上扩展流式输出能力：设置 stream=True 后，模型回复不再一次性返回，而是逐块（chunk）推送，实现打字机式实时显示。
• 服务端采用 SSE（Server-Sent Events） 协议，每行以 data: 前缀携带 JSON 片段，最终以 data: [DONE] 标记流结束。
• Completions.create() 根据 stream 参数分流：非流式走 post() 返回完整 ChatCompletion；流式走 post_stream() 返回可迭代的 ChatCompletionChunk 生成器。

本节你将学到：

• 用 httpx.stream 建立长连接，通过 response.iter_lines() 逐行读取 SSE 数据
• 解析 SSE 行：识别 data: 前缀、跳过无效行、遇 [DONE] 终止
• 定义增量数据模型（ChoiceDelta → ChunkChoice → ChatCompletionChunk），逐块提取 delta.content 并实时打印

SSE 数据格式示例：

data: {"id":"...","choices":[{"index":0,"delta":{"content":"你"}}]}

data: {"id":"...","choices":[{"index":0,"delta":{"content":"好"}}]}

data: [DONE]

调用时序概览：

4.1 main.py #

main.py

# 导入操作系统相关的os模块
import os

# 从openai_lite模块中导入OpenAI类
from openai_lite import OpenAI

# 创建OpenAI客户端对象，并设置API密钥和基础URL（支持从环境变量读取，否则使用默认值）
client = OpenAI(
    api_key=os.getenv("OPENAI_API_KEY", "sk-6dac9405d2c74a8f9a5b921b80ebb0f2"),
    base_url=os.getenv("OPENAI_BASE_URL", "https://api.deepseek.com/v1"),
)

# 使用OpenAI客户端创建一次聊天补全请求，指定模型和用户消息内容
+stream  = client.chat.completions.create(
    model=os.getenv("OPENAI_MODEL", "deepseek-v4-pro"),
    messages=[{"role": "user", "content": "你是谁?"}],
+   stream=True,
)

# 遍历流式返回的每一个chunk
+for chunk in stream:
   # 如果chunk中没有choices则跳过本次循环
+  if not chunk.choices:
+      continue
   # 获取第一个choice中的delta的content内容
+  content = chunk.choices[0].delta.content
   # 如果content有内容，则打印，不换行，立即输出
+  if content:
+      print(content, end="", flush=True)

4.2 client.py #

openai_lite/client.py

# 导入httpx库，用于发起HTTP请求
import httpx
# 导入json库，用于解析JSON数据
+import json
# 从._constants模块中导入默认基础URL、默认超时时间、请求头和URL构建方法
from ._constants import DEFAULT_BASE_URL,DEFAULT_TIMEOUT,build_headers,build_url
# 从resources目录的chat模块导入Chat类
from .resources.chat import Chat
# 从.types.chat模块导入ChatCompletionChunk类
+from .types.chat import ChatCompletionChunk
# 定义SSE前缀为"data: "
+_SSE_PREFIX = "data: "
# 定义解析SSE行的函数
+def _parse_sse_line(line):
    # 如果行为空或不是以_SSE_PREFIX开头
+   if not line or not line.startswith(_SSE_PREFIX):
        # 返回None
+       return None
    # 去除前缀后获取数据，并去除首尾空白符
+   data = line[len(_SSE_PREFIX) :].strip()
    # 如果数据等于"[DONE]"，返回空字符串，否则返回数据本身
+   return "" if data == "[DONE]" else data
# 定义OpenAI类
class OpenAI:
    # 初始化方法，设置API key、基础URL、超时时间
    def __init__(
        self,
        *,
        api_key: str,
        base_url: str = DEFAULT_BASE_URL,
        timeout: float = DEFAULT_TIMEOUT,
    ):
        # 保存API Key
        self.api_key = api_key
        # 移除基础URL右侧的斜杠并保存
        self.base_url = base_url.rstrip("/")
        # 保存超时时间
        self.timeout = timeout
        # 创建Chat类实例并保存
        self.chat = Chat(self)
    # 内部请求方法
    def _request(self, method: str, path: str, *, json: dict | None = None):
        # 使用httpx发起HTTP请求
        response = httpx.request(
            method,
            # 构造完整的请求URL
            build_url(self.base_url, path),
            # 构造请求头
            headers=build_headers(self.api_key),
            # 请求体（JSON格式）
            json=json,
            # 请求超时时间
            timeout=self.timeout,
        )
        # 返回HTTP响应
        return response    
    # POST请求方法
    def post(self, path: str, *, json: dict | None = None):
        # 调用_request方法发起POST请求
        return self._request("POST", path, json=json)
    # 定义post_stream方法，用于以流式方式发送POST请求
+   def post_stream(self, path, *, body=None):
        # 定义内部生成器函数iter_chunks，用于逐块处理响应数据
+       def iter_chunks():
            # 使用httpx.stream方法以流式发送POST请求
+           with httpx.stream(
+               "POST",  # 请求方法为POST
+               build_url(self.base_url, path),  # 构造完整的请求URL
+               headers=build_headers(self.api_key),  # 设置请求头部，包括API KEY等信息
+               json=body,  # 以JSON格式发送请求体
+               timeout=self.timeout,  # 超时时间配置
+           ) as response:  # 使用with自动关闭响应资源
                # 迭代响应体的每一行数据
+               for line in response.iter_lines():
                    # 解析每一行的SSE（Server Sent Event）数据
+                   data = _parse_sse_line(line)
                    # 如果解析结果为None，说明该行不是以指定前缀开头，跳过
+                   if data is None:
+                       continue
                    # 如果解析结果为空字符串，说明是流的结束标记，退出循环
+                   if data == "":
+                       break
                    # 将解析出的json字符串转对象，并验证其格式，作为块数据产出
+                   yield ChatCompletionChunk.model_validate(json.loads(data))
        # 返回迭代器对象，实现流式数据处理
+       return iter_chunks()

4.3 completions.py #

openai_lite/resources/chat/completions.py

# 从项目的_constants模块导入CHAT_COMPLETIONS_PATH常量
from ..._constants import CHAT_COMPLETIONS_PATH
# 从openai.types.chat模块导入ChatCompletion类
from openai.types.chat import ChatCompletion
# 定义Completions类
class Completions():
    # 初始化方法，接收client对象
    def __init__(self, client):
        # 从client对象中获取post方法并赋值给实例变量_post
        self._post = client.post
        # 从client对象中获取post_stream方法并赋值给实例变量_post_stream
+       self._post_stream = client.post_stream
    # 定义create方法，用于创建聊天补全
    def create(
+       self,#self表示当前实例
+       *, # 表示可选参数
+       model,#模型名称
+       stream=False,#是否流式返回，默认不流式返回
+       messages,#消息列表，必选参数
+       **kwargs,#其他可选参数，如temperature、max_tokens等
    ):
        # 构造请求负载，包括模型名、消息列表及其它可选参数
        payload = {"model": model, "messages": messages, **kwargs}
        # 如果stream为真，说明需要以流式方式获取聊天补全
+       if stream:
            # 调用_post_stream方法，向CHAT_COMPLETIONS_PATH发送包含stream=True的payload，实现流式响应
+           return self._post_stream(CHAT_COMPLETIONS_PATH, body={**payload, "stream": True})
        # 发送POST请求到CHAT_COMPLETIONS_PATH，并传递payload作为json参数
        response = self._post(CHAT_COMPLETIONS_PATH, json=payload)
        # 对响应的json数据进行模型验证，并返回ChatCompletion对象
        return ChatCompletion.model_validate(response.json())

4.4 chat.py #

openai_lite/types/chat.py

# 导入Pydantic的BaseModel用于数据模型的定义
from pydantic import BaseModel

# 定义消息（Message）类，继承自BaseModel
class Message(BaseModel):
    # role字段，表示消息的角色（如user、assistant）
    role: str
    # content字段，表示消息内容，可以为None
    content: str | None = None

# 定义选择（Choice）类，继承自BaseModel
class Choice(BaseModel):
    # index字段，表示选择的索引
    index: int
    # message字段，表示对应的Message对象
    message: Message
    # finish_reason字段，表示完成的原因，可以为None
    finish_reason: str | None = None

# 定义聊天完成（ChatCompletion）类，继承自BaseModel
class ChatCompletion(BaseModel):
    # id字段，表示聊天完成对象的唯一标识符
    id: str
    # model字段，表示使用的模型
    model: str
    # choices字段，表示Choice对象的列表
    choices: list[Choice]

# 定义ChoiceDelta类，继承自BaseModel，用于描述消息变更（delta）的结构
+class ChoiceDelta(BaseModel):
    # role字段，表示当前delta的角色信息，可以为None
+   role: str | None = None
    # content字段，表示当前delta的内容，可以为None
+   content: str | None = None

# 定义ChunkChoice类，继承自BaseModel，用于描述stream时每个选择的内容
+class ChunkChoice(BaseModel):
    # index字段，表示当前选择的索引
+   index: int
    # delta字段，类型为ChoiceDelta，表示此选择的增量内容
+   delta: ChoiceDelta
    # finish_reason字段，表示结束的原因，可以为None
+   finish_reason: str | None = None

# 定义ChatCompletionChunk类，继承自BaseModel，表示拆分的聊天完成片段
+class ChatCompletionChunk(BaseModel):
    # id字段，表示本次请求的唯一ID，可以为None
+   id: str | None = None
    # model字段，表示使用的模型名称，可以为None
+   model: str | None = None
    # choices字段，包含ChunkChoice对象的列表，默认是空列表
+   choices: list[ChunkChoice] = []  

5. 推理模式 #

• 本节在流式输出的基础上，适配 DeepSeek 推理模型（如 deepseek-v4-pro）的 双通道输出：delta.reasoning_content 承载思考过程，delta.content 承载最终答案。
• 推理阶段与回答阶段分先后到达，同一 chunk 通常只携带其中一个字段，客户端需用 phase 变量追踪当前阶段，并在切换时打印对应标题。
• 在 ChoiceDelta 模型中补充 reasoning_content 字段，确保 Pydantic 校验时不会丢弃该扩展字段。

本节你将学到：

• 通过 getattr(delta, "reasoning_content", None) 安全读取推理内容
• 用 if reasoning / elif content 区分两个阶段，避免 content 为 None 时误输出
• 用 phase 状态机控制标题只打印一次，实现 【思考过程】 → 【回答】 的分段展示

delta 双字段示例：

# 思考阶段
data: {"choices":[{"delta":{"reasoning_content":"先比较整数部分...","content":null}}]}

# 回答阶段
data: {"choices":[{"delta":{"reasoning_content":null,"content":"9.11 更大"}}]}

调用时序概览：

5.1 main.py #

main.py

# 导入操作系统相关的os模块
import os

# 从openai_lite模块中导入OpenAI类
from openai_lite import OpenAI

# 创建OpenAI客户端对象，并设置API密钥和基础URL（支持从环境变量读取，否则使用默认值）
client = OpenAI(
    api_key=os.getenv("OPENAI_API_KEY", "sk-6dac9405d2c74a8f9a5b921b80ebb0f2"),
    base_url=os.getenv("OPENAI_BASE_URL", "https://api.deepseek.com/v1"),
)

# 使用OpenAI客户端创建一次聊天补全请求，指定模型和用户消息内容
stream  = client.chat.completions.create(
    model=os.getenv("OPENAI_MODEL", "deepseek-v4-pro"),
+   messages=[{"role": "user", "content": "9.11 和 9.8 哪个更大？请简要说明。"}],
    stream=True,
)
+phase = None  # "reasoning" | "content"
for chunk in stream:
+   delta = getattr(chunk.choices[0], 'delta', None) if getattr(chunk, 'choices', None) else None
+   if not delta:
+       continue
+   reasoning = getattr(delta, 'reasoning_content', None)
+   content = getattr(delta, 'content', None)
+   if reasoning:
+       if phase != "reasoning":
+           print("【思考过程】", end="", flush=True)
+           phase = "reasoning"
+       print(reasoning, end="", flush=True)
+   elif content:
+       if phase != "content":
+           if phase is not None:
+               print()
+           print("【回答】", end="", flush=True)
+           phase = "content"
+       print(content, end="", flush=True)
+print()

5.2 chat.py #

openai_lite/types/chat.py

# 导入Pydantic的BaseModel用于数据模型的定义
from pydantic import BaseModel

# 定义消息（Message）类，继承自BaseModel
class Message(BaseModel):
    # role字段，表示消息的角色（如user、assistant）
    role: str
    # content字段，表示消息内容，可以为None
    content: str | None = None

# 定义选择（Choice）类，继承自BaseModel
class Choice(BaseModel):
    # index字段，表示选择的索引
    index: int
    # message字段，表示对应的Message对象
    message: Message
    # finish_reason字段，表示完成的原因，可以为None
    finish_reason: str | None = None

# 定义聊天完成（ChatCompletion）类，继承自BaseModel
class ChatCompletion(BaseModel):
    # id字段，表示聊天完成对象的唯一标识符
    id: str
    # model字段，表示使用的模型
    model: str
    # choices字段，表示Choice对象的列表
    choices: list[Choice]

# 定义ChoiceDelta类，继承自BaseModel，用于描述消息变更（delta）的结构
class ChoiceDelta(BaseModel):
    # role字段，表示当前delta的角色信息，可以为None
    role: str | None = None
    # content字段，表示当前delta的内容，可以为None
    content: str | None = None
    # reasoning_content字段，表示推理相关内容，可以为None
+   reasoning_content: str | None = None

# 定义ChunkChoice类，继承自BaseModel，用于描述stream时每个选择的内容
class ChunkChoice(BaseModel):
    # index字段，表示当前选择的索引
    index: int
    # delta字段，类型为ChoiceDelta，表示此选择的增量内容
    delta: ChoiceDelta
    # finish_reason字段，表示结束的原因，可以为None
    finish_reason: str | None = None

# 定义ChatCompletionChunk类，继承自BaseModel，表示拆分的聊天完成片段
class ChatCompletionChunk(BaseModel):
    # id字段，表示本次请求的唯一ID，可以为None
    id: str | None = None
    # model字段，表示使用的模型名称，可以为None
    model: str | None = None
    # choices字段，包含ChunkChoice对象的列表，默认是空列表
    choices: list[ChunkChoice] = []  

6. 工具调用 #

• 本节实现 Function Calling：向模型注册工具 schema，由模型决定何时调用、传什么参数，客户端本地执行后将结果回传，模型再生成最终自然语言回复。
• 完整流程需要 两次 API 请求：第一次模型返回 tool_calls；本地执行工具并追加 role: "tool" 消息；第二次模型整合工具结果输出答案。
• 在数据模型中补充 Function、ToolCall，扩展 Message 的 tool_calls / tool_call_id 字段，使响应解析与多轮消息拼接具备类型支持。

本节你将学到：

• 用 JSON Schema 定义 tools 列表，描述函数名、描述和参数结构
• 解析 assistant_message.tool_calls，通过 TOOL_HANDLERS 映射到本地实现
• 构造完整消息链：user → assistant（含 tool_calls）→ tool（含 tool_call_id）→ 再次请求

消息链示例：

[
  {"role": "user", "content": "北京今天天气怎么样？"},
  {"role": "assistant", "tool_calls": [{"id": "call_xxx", "type": "function", "function": {"name": "get_weather", "arguments": "{\"city\": \"北京\"}"}}]},
  {"role": "tool", "tool_call_id": "call_xxx", "content": "北京：晴，25°C，微风"}
]

调用时序概览：

6.1 main.py #

main.py

# 导入json模块，用于处理JSON数据
+import json
# 导入os模块，用于与操作系统交互
import os
# 导入sys模块，用于访问系统特定的参数和函数
+import sys
# 从openai模块导入OpenAI类
from openai_lite import OpenAI

# 定义TOOLS工具列表
+TOOLS = [
+   {
        # 工具类型为函数
+       "type": "function",
        # 工具的详细信息
+       "function": {
            # 函数名称为get_weather
+           "name": "get_weather",
            # 函数功能描述
+           "description": "获取指定城市的当前天气",
            # 函数接受的参数定义
+           "parameters": {
                # 参数类型是对象
+               "type": "object",
                # 对象的属性定义
+               "properties": {
                    # 属性city，表示城市名称
+                   "city": {
                        # city的数据类型为字符串
+                       "type": "string",
                        # city的描述
+                       "description": "城市名称，如北京、上海",
+                   },
+               },
                # city字段为必需字段
+               "required": ["city"],
+           },
+       },
+   }
+]

# 定义获取天气的函数，输入城市名称，返回天气信息字符串
+def get_weather(city: str) -> str:
    # 返回格式化后的天气信息
+   return f"{city}：晴，25°C，微风"

# 定义工具处理器映射，将工具名映射到对应的实现函数
+TOOL_HANDLERS = {"get_weather": get_weather}

# 创建OpenAI客户端对象，设置API密钥和基础URL，支持从环境变量读取，否则使用默认值
client = OpenAI(
    api_key=os.getenv("OPENAI_API_KEY", "sk-6dac9405d2c74a8f9a5b921b80ebb0f2"),
    base_url=os.getenv("OPENAI_BASE_URL", "https://api.deepseek.com/v1"),
)

# 构造对话消息列表，初始只有用户的提问
+messages = [
    # 用户的提问内容
+   {"role": "user", "content": "北京今天天气怎么样？"},
+]
# 打印用户的提问内容到控制台
+print(f"【用户】{messages[0]['content']}")

# 向OpenAI API发起第一次聊天请求，模型会决定是否调用函数工具
+response = client.chat.completions.create(
    model=os.getenv("OPENAI_MODEL", "deepseek-v4-pro"),
+   messages=messages,
+   tools=TOOLS,
)
# 获取助手生成的回复消息
+assistant_message = response.choices[0].message

# 如果助手消息中tool_calls字段存在，说明模型请求调用工具
+if assistant_message.tool_calls:
    # 将助手消息转换为字典并添加到消息列表
+   messages.append(assistant_message.model_dump())
    # 遍历所有要调用的工具
+   for tool_call in assistant_message.tool_calls:
        # 获取函数名称
+       func_name = tool_call.function.name
        # 获取函数调用的参数，解析为Python字典
+       func_args = json.loads(tool_call.function.arguments)
        # 打印调用的工具名和参数
+       print(f"【工具调用】{func_name}({func_args})")

        # 根据函数名从TOOL_HANDLERS中获取实际处理函数
+       handler = TOOL_HANDLERS[func_name]
        # 调用实现函数并获得结果
+       result = handler(**func_args)
        # 打印工具调用结果
+       print(f"【工具结果】{result}")

        # 将工具调用结果以工具角色的消息追加到消息队列
+       messages.append(
+           {
                # 角色为tool
+               "role": "tool",
                # 工具调用唯一id
+               "tool_call_id": tool_call.id,
                # 工具执行的结果
+               "content": result,
+           }
+       )

    # 第二次请求OpenAI API，根据工具结果生成最终的助手回复
+   response = client.chat.completions.create(
+       model=os.getenv("OPENAI_MODEL", "deepseek-v4-pro"),
+       messages=messages,
+       tools=TOOLS,
+   )
    # 打印助手的最终回复内容
+   print(f"【助手】{response.choices[0].message.content}")
# 如果助手没有请求调用工具，直接输出助手回复内容
+else:
+   print(f"【助手】{assistant_message.content}")

6.2 completions.py #

openai_lite/resources/chat/completions.py

# 从项目的_constants模块导入CHAT_COMPLETIONS_PATH常量
from ..._constants import CHAT_COMPLETIONS_PATH
# 从openai.types.chat模块导入ChatCompletion类
from openai.types.chat import ChatCompletion
# 定义Completions类
class Completions():
    # 初始化方法，接收client对象
    def __init__(self, client):
        # 从client对象中获取post方法并赋值给实例变量_post
        self._post = client.post
        # 从client对象中获取post_stream方法并赋值给实例变量_post_stream
        self._post_stream = client.post_stream
    # 定义create方法，用于创建聊天补全
    def create(
        self,#self表示当前实例
        *, # 表示可选参数
        model,#模型名称
        stream=False,#是否流式返回，默认不流式返回
        messages,#消息列表，必选参数
+       tools,#工具列表，必选参数
        **kwargs,#其他可选参数，如temperature、max_tokens等
    ):
        # 构造请求负载，包括模型名、消息列表及其它可选参数
+       payload = {"model": model, "messages": messages, "tools": tools, **kwargs}
        # 如果stream为真，说明需要以流式方式获取聊天补全
        if stream:
            # 调用_post_stream方法，向CHAT_COMPLETIONS_PATH发送包含stream=True的payload，实现流式响应
            return self._post_stream(CHAT_COMPLETIONS_PATH, body={**payload, "stream": True})
        # 发送POST请求到CHAT_COMPLETIONS_PATH，并传递payload作为json参数
        response = self._post(CHAT_COMPLETIONS_PATH, json=payload)
        # 对响应的json数据进行模型验证，并返回ChatCompletion对象
        return ChatCompletion.model_validate(response.json())

6.3 chat.py #

openai_lite/types/chat.py

# 导入Pydantic的BaseModel用于数据模型的定义
from pydantic import BaseModel

# 定义函数（Function）类，继承自BaseModel
+class Function(BaseModel):
    # name字段，表示函数名称
+   name: str
    # arguments字段，表示函数参数
+   arguments: str
# 定义工具调用（ToolCall）类，继承自BaseModel
+class ToolCall(BaseModel):
    # id字段，表示工具调用的唯一标识符
+   id: str
    # type字段，表示工具类型
+   type: str = "function"
    # function字段，表示工具函数
+   function: Function
# 定义消息（Message）类，继承自BaseModel
class Message(BaseModel):
    # role字段，表示消息的角色（如user、assistant）
    role: str
    # content字段，表示消息内容，可以为None
    content: str | None = None
    # tool_calls字段，表示工具调用列表，可以为None
+   tool_calls: list[ToolCall] | None = None
    # tool_call_id字段，表示工具调用的唯一标识符
+   tool_call_id: str | None = None

# 定义选择（Choice）类，继承自BaseModel
class Choice(BaseModel):
    # index字段，表示选择的索引
    index: int
    # message字段，表示对应的Message对象
    message: Message
    # finish_reason字段，表示完成的原因，可以为None
    finish_reason: str | None = None

# 定义聊天完成（ChatCompletion）类，继承自BaseModel
class ChatCompletion(BaseModel):
    # id字段，表示聊天完成对象的唯一标识符
    id: str
    # model字段，表示使用的模型
    model: str
    # choices字段，表示Choice对象的列表
    choices: list[Choice]

# 定义ChoiceDelta类，继承自BaseModel，用于描述消息变更（delta）的结构
class ChoiceDelta(BaseModel):
    # role字段，表示当前delta的角色信息，可以为None
    role: str | None = None
    # content字段，表示当前delta的内容，可以为None
    content: str | None = None
    # reasoning_content字段，表示推理相关内容，可以为None
    reasoning_content: str | None = None

# 定义ChunkChoice类，继承自BaseModel，用于描述stream时每个选择的内容
class ChunkChoice(BaseModel):
    # index字段，表示当前选择的索引
    index: int
    # delta字段，类型为ChoiceDelta，表示此选择的增量内容
    delta: ChoiceDelta
    # finish_reason字段，表示结束的原因，可以为None
    finish_reason: str | None = None

# 定义ChatCompletionChunk类，继承自BaseModel，表示拆分的聊天完成片段
class ChatCompletionChunk(BaseModel):
    # id字段，表示本次请求的唯一ID，可以为None
    id: str | None = None
    # model字段，表示使用的模型名称，可以为None
    model: str | None = None
    # choices字段，包含ChunkChoice对象的列表，默认是空列表
    choices: list[ChunkChoice] = []