引言

本文介绍了如何基于 OpenAI Python SDK 封装大语言模型调用，统一消息格式，轻松实现流式和非流式输出。通过 Message 类和 ChatOpenAI 封装，我们可以快速构建智能 Agent 系统，兼顾稳定性、兼容性和生产化需求，无需重复实现 HTTP 调用和流式解析。完整示例与代码请见项目仓库：Vero 。

为什么选择openai-python库

1. 功能完备，避免重复造轮子

   • OpenAI Python SDK 已经封装了大部分常用功能，包括：

     • ChatCompletion / Completion / Embedding / Moderation 等不同接口

     • 流式输出（streaming），适合逐步生成和工具调用场景

     • 请求参数管理（temperature、max_tokens、top_p 等）

   • 如果直接用 requests 从零实现，你需要自己处理这些接口差异，还要实现流式解析、JSON 校验、参数组合等，工作量非常大，而且容易出错。

2. 内置重试和异常处理

   • 在生产环境或高并发场景中，调用 API 常常会遇到：
     • 网络超时
     • 连接被拒绝
     • 临时性 5xx 错误
   • OpenAI Python SDK 内部已经处理了重试、超时和异常，封装后只需要再做一层统一异常包装即可。直接用 requests 实现，你需要自己写完整的重试机制、超时控制和错误分类，容易遗漏，增加调试难度。

3. OpenAI API 协议模型的兼容性

   • 现在很多开源或商业模型都支持 OpenAI API 协议（GPT4All、Qwen、MOSS、Azure OpenAI 等）。

     • 封装 OpenAI Python SDK 后，上层 Agent 逻辑几乎不变
     • 可以方便地切换底层模型或部署环境，实现“模型即插即用”

   • 如果直接用 requests 实现，每次切换模型或服务提供方，都可能需要修改底层请求逻辑，维护成本高。

4. 节省开发和维护成本

   • 封装官方库，意味着你可以：
   • 聚焦 Agent 逻辑（推理链、工具调用、记忆管理、规划）
   • 不必重复实现 HTTP 请求、身份验证、JSON 解析
   • 轻松获得官方更新和新功能支持
   • 从零实现 requests 调用，只能解决“能调用接口”这一点，功能、稳定性、兼容性和安全性都需要自己承担，代价远高。

5. 更易学习和生产化

   • 对于教学和工程实践来说：
   • 我们可以专注理解智能体设计和模式
   • 生产环境可以直接封装异常、日志和流式输出
   • 代码简洁，利于扩展 Agent 功能，如 ReAct、RAG、MCP、多智能体协作

openai-python的版本选择的是1.0.0,<2.0.0，从2.00版本引入了responses功能，而Qwen、DeepSeek等主流开源模型尚未原生支持OpenAI Responses API，所以我们限定低于这个版本。

配置类

基于目前最新的pydantic-settings(2.12.0)实现一个能读取本地.env环境变量配置的类，

from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    openai_api_key: str             # 从.env的OPENAI_API_KEY中读取
    openai_base_url: str = "https://api.openai.com/v1"  
    debug: bool = False
    timeout: int = 60
    model_name: str = "Qwen/Qwen3-32B"
    temperature: float = 0.7

    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        extra="ignore" # 避免未来加入字段导致报错
    )


settings = Settings()

这里可以配置OpenAI协议兼容的base_url和api_key，如果不想用官方的话也可以使用自己本地部署或模型供应商提供的配置。

比如国内的硅基流动就是一个不错的选择，它为大多数开源模型提供了API服务，按量计费，对于轻量的模型还可以免费使用(有速率限制)，比如Qwen/Qwen3-8B。

我们创建一个.env文件，并加入以下两个配置：

OPENAI_API_KEY=sk-xx  # 从  https://cloud.siliconflow.cn/me/account/ak 获取
OPENAI_API_BASE=https://api.siliconflow.cn/v1

Message类

在Agent系统重，消息是最核心的抽象之一，我们封装这个类的目的：

• 统一消息格式(避免到处传dict)
• 为Agent未来扩展留了接口(metadata里面可以添加tool_name/error/reasoning/token_count等)
• 解决LLM调用输入格式兼容问题(统一转换为{ "role": "...", "content": "..." })

from typing import Dict, Any, Self
from datetime import datetime
from pydantic import BaseModel, Field
from enum import Enum

class Message(BaseModel):
    class Role(str, Enum):
        system = "system"
        user = "user"
        assistant = "assistant"

    content: str
    role: Role
    timestamp: datetime = Field(default_factory=datetime.now)
    metadata: Dict[str, Any] = Field(default_factory=dict)

    @classmethod
    def new(cls, role: Role, content: str, **kwargs) -> Self:
        return cls(role=role, content=content, **kwargs)

    @classmethod
    def user(cls, content: str, **kw):
        return cls(role=Message.Role.user, content=content, **kw)

    @classmethod
    def system(cls, content: str, **kw):
        return cls(role=Message.Role.system, content=content, **kw)

    @classmethod
    def assistant(cls, content: str, **kw):
        return cls(role=Message.Role.assistant, content=content, **kw)

    def to_dict(self) -> Dict[str, Any]:
        return {"role": self.role, "content": self.content}
    
    def __str__(self):
        return f"[{self.role.value}] {self.content}"

我们可以通过来快速实例化不同角色的消息实例：

Message.user("你好")
Message.assistant("收到")
Message.system("你是助手")

封装LLM调用

from typing import Optional, Iterator, List, Union
from openai import OpenAI

from .message import Message
from vero.config import settings
from vero.core.exceptions import LLMCallError, LLMConfigError


class ChatOpenAI:
    """
    OpenAI 聊天模型封装类，用于与基于聊天的 LLM 进行交互。

    属性:
        model_name: 要使用的 LLM 模型名称。
        temperature: 生成文本的采样温度，值越高文本越随机。
        max_tokens: 响应的最大 token 数。
        timeout: 请求超时时间（秒）。
        api_key: OpenAI API 密钥。
        base_url: OpenAI API 基础 URL。
    """

    def __init__(
        self,
        model_name: Optional[str] = None,
        api_key: Optional[str] = None,
        base_url: Optional[str] = None,
        temperature: float = 0.7,
        max_tokens: Optional[int] = None,
        timeout: Optional[int] = None,
        **kwargs,
    ) -> None:
        """
        初始化 LLM 封装类。如果未提供参数，则使用默认配置。

        参数:
            model_name: 模型名称，如果为空则使用配置文件中的默认值。
            api_key: OpenAI API 密钥，如果为空则使用配置文件中的默认值。
            base_url: OpenAI API 基础 URL，如果为空则使用配置文件中的默认值。
            temperature: 文本生成的随机性温度。
            max_tokens: 响应的最大 token 数。
            timeout: 请求超时时间。
            kwargs: 其他可选参数。

        异常:
            LLMConfigError: 当缺少必要配置（api_key、base_url 或 model_name）时抛出。
        """
        self.model_name = model_name or settings.model_name
        self.temperature = temperature
        self.max_tokens = max_tokens
        self.timeout = timeout or getattr(settings, "timeout", None)
        self.kwargs = kwargs

        self.api_key = api_key or settings.openai_api_key
        self.base_url = base_url or settings.openai_base_url

        if not all([self.api_key, self.base_url, self.model_name]):
            raise LLMConfigError(
                "LLM 客户端缺少 api_key、base_url 或 model_name 配置"
            )

        self._client = self._create_client()

    def _create_client(self) -> OpenAI:
        """
        创建并返回 OpenAI 客户端实例。

        返回:
            OpenAI: 已配置的 OpenAI 客户端对象。
        """
        return OpenAI(
            api_key=self.api_key, base_url=self.base_url, timeout=self.timeout
        )

    def generate(
        self,
        messages: List[Union[Message, dict]],
        stream: bool = False,
        temperature: Optional[float] = None,
        **kwargs,
    ) -> Union[str, Iterator[str]]:
        """
        生成 LLM 响应，支持流式输出和非流式完整输出。

        参数:
            messages: 消息列表，可以是 Message 对象或字典，表示对话历史。
            stream: 是否以流式方式返回结果，True 时返回 generator。
            temperature: 可选参数，覆盖默认温度。
            kwargs: 其他生成参数，例如 max_tokens。

        返回:
            str: 当 stream=False 时，返回完整的响应字符串。
            Iterator[str]: 当 stream=True 时，返回流式文本片段的迭代器。

        异常:
            LLMCallError: LLM API 调用失败时抛出。
        """
        # 将 Message 对象转换为字典，以兼容 OpenAI API
        messages_dict = [
            msg.to_dict() if isinstance(msg, Message) else msg for msg in messages
        ]

        try:
            response = self._client.chat.completions.create(
                model=self.model_name,
                messages=messages_dict,
                temperature=temperature or self.temperature,
                max_tokens=kwargs.get("max_tokens", self.max_tokens),
                stream=stream,
                **{k: v for k, v in kwargs.items() if k not in {"temperature", "max_tokens"}},
            )

            if stream:
                # 使用生成器帮助函数返回流式结果
                # 这样可以在函数内部处理响应，但外部仍可以按需迭代输出
                def _stream_generator():
                    for chunk in response:
                        content = chunk.choices[0].delta.content or ""
                        if content:
                            yield content
                return _stream_generator()
            else:
                # 非流式调用直接返回完整响应文本
                return response.choices[0].message.content

        except Exception as e:
            raise LLMCallError(f"LLM 调用失败: {str(e)}") from e

这个类通过OpenAI封装了所有与模型交互的细节，并且generate同时兼容流式和非流式输出，这里碰到了一个坑就是不能直接像下面这样写：

if stream:
    for chunk in response:
        content = chunk.choices[0].delta.content or ""
        if content:
            yield content
else:
    # 非流式调用直接返回完整响应文本
    return response.choices[0].message.content

因为这样直接在一个函数中出现yield关键词的话，会把它转换为一个生成器，所以这里的return还是会返回生成器。感兴趣的可以试试。

使用示例

from vero.core import ChatOpenAI, Message
# 默认从.env中加载配置，前提是你创建好.env并配置好OPENAI_API_KEY和OPENAI_BASE_URL
llm = ChatOpenAI()
messages = [Message.user("你是谁")]
response = llm.generate(messages, stream=False)
print("Full response:")
print(response)

我是通义千问，阿里巴巴集团旗下的通义实验室自主研发的超大规模语言模型。我能够回答问题、创作文字，比如写故事、写公文、写邮件、写剧本、逻辑推理、编程等等，还能表达观点，玩游戏等。我支持多种语言，包括但不限于中文、英文、德语、法语、西班牙语等。我的目标是成为人类最值得信赖的AI助手，以我的知识和能力为用户提供帮助。

我们可以传入Message的列表，也可以传入对应的字典对象。同时测试一下流式输出(需要实际运行才能看出)：

from vero.core import ChatOpenAI, Message
# 默认从.env中加载配置，前提是你创建好.env并配置好OPENAI_API_KEY和OPENAI_BASE_URL
llm = ChatOpenAI()

messages = [{"role": "user", "content": "讲一个非常好笑的笑话，要有反转"}]
for chunk in llm.generate(messages, stream=True):
    print(chunk, end="") 

当然可以！下面这个笑话有反转效果，希望你会喜欢：

---

有一天，小明去面试。

面试官问：“如果你进入一个房间，发现里面有三个火炉，一个着火，一个冒烟，一个熄灭，你先扑灭哪个？”

小明想了想，认真地说：“我先扑灭着火的那个。”

面试官接着问：“为什么？”

小明回答：“因为如果我不扑灭着火的，它可能会烧到冒烟的。”

面试官点点头：“很好，那接下来呢？”

小明说：“然后我扑灭冒烟的。”

面试官：“为什么？”

小明：“因为冒烟的如果不处理，迟早也会着火。”

面试官赞许地说：“看来你很有逻辑思维。那最后呢？”

小明：“最后我会把熄灭的那个也扑灭了。”

面试官一愣：“它不是已经熄了吗？”

小明：“对啊，我已经把它编号为‘第四个火炉’了。”

---

希望这个反转能让你笑一笑！😄 如果你还想要更多类似的笑话，随时告诉我！