Python 基础体系 · 第 100/112 篇。示例统一以 Python 3.14 为语言基线；第三方库使用与其兼容的现代稳定版本，版本敏感行为会单独说明。

Python OpenAI SDK：Responses、流式、结构化输出、工具和重试

OpenAI Python SDK 当前以 Responses API 作为主要的模型调用接口。它统一了文本输入、文本输出、结构化输出、函数工具以及部分内置工具的调用方式；SDK 同时提供同步客户端 OpenAI 和异步客户端 AsyncOpenAI。当前 SDK 的要求是 Python 3.10 或更高版本，因此 Python 3.14 在版本范围内。(github.com)

本文使用 Python 3.14 语法，示例基于当前 openai 和 Pydantic v2 接口。模型名称不写死为某个具体型号，而是通过环境变量传入：

python -m pip install -U openai pydantic
export OPENAI_API_KEY="你的 API Key"
export OPENAI_MODEL="支持 Responses API 的模型"

Windows PowerShell：

python -m pip install -U openai pydantic
$env:OPENAI_API_KEY = "你的 API Key"
$env:OPENAI_MODEL = "支持 Responses API 的模型"

生产环境应固定依赖版本，并在程序启动时记录实际使用的版本：

import openai

print(openai.__version__)

SDK 官方文档也建议通过运行时检查版本确认当前环境实际加载的包，而不是只看安装命令或锁文件。(github.com)

---

一、先建立正确的对象模型：请求、响应和输出项

一次 Responses API 调用可以抽象为：

$$\text{Response} = F(\text{model}, \text{instructions}, \text{input}, \text{tools}, \text{text})$$

其中：

• model 决定使用哪个模型；
• instructions 是面向整个任务的高层指令；
• input 是本次输入，可以是字符串，也可以是带角色和内容的消息项列表；
• tools 描述模型可以调用的工具；
• text 描述文本输出格式，例如普通文本或结构化输出；
• 返回值是一个 Response 对象，而不是简单字符串。

最小调用如下：

import os

from openai import OpenAI

client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])

response = client.responses.create(
    model=os.environ["OPENAI_MODEL"],
    instructions="你是一个严谨的 Python 助手。",
    input="解释 Python 中生成器和列表的区别。",
)

print(response.output_text)

response.output_text 是 SDK 提供的便捷文本属性。它适合“我只关心最终文本”的场景，但不能替代对完整 response.output 的检查，因为模型输出可能包含消息、函数调用、内置工具调用或其他输出项。Responses API 的输出是一个有类型的项目列表，而不是固定的单一消息结构。(github.com)

可以把输出理解为：

Response
└── output: list[ResponseOutputItem]
    ├── message
    │   └── content
    │       └── output_text
    ├── function_call
    ├── web_search_call
    └── ...

因此，下面两种读取方式含义不同：

# 便捷读取：只取合并后的文本
print(response.output_text)

# 精确读取：检查每一个输出项的类型
for item in response.output:
    print(item.type)

当程序需要判断“模型是回答了问题，还是要求调用工具”时，必须使用第二种方式。

---

二、Responses API 与 Chat Completions 的关系

Chat Completions 仍然存在，并且官方 SDK 继续支持它：

completion = client.chat.completions.create(
    model=os.environ["OPENAI_MODEL"],
    messages=[
        {"role": "developer", "content": "你是一个严谨的 Python 助手。"},
        {"role": "user", "content": "什么是上下文管理器？"},
    ],
)

print(completion.choices[0].message.content)

但两者的对象模型不同：

维度	Chat Completions	Responses
文本读取	choices[0].message.content	output_text
工具调用	位于 message/tool_calls	位于 response.output
输出结构	以 choice 为中心	以 output item 为中心
流式事件	completion chunk	类型化 Responses 事件
结构化输出	response_format 体系	text_format 或 text.format 体系

新代码通常应优先选择 Responses API，除非已有系统依赖 Chat Completions 的消息格式或兼容层。官方 Python SDK 将 Responses API 描述为主要调用接口，同时保留 Chat Completions 作为长期支持的既有接口。(github.com)

---

三、流式响应：传输层流式不等于语义层完成

3.1 流式的基本机制

流式响应是指服务端通过 Server-Sent Events，也就是 SSE，将生成过程中的事件逐步发送给客户端。客户端不必等待完整答案生成后才显示首个字符。(github.com)

最直接的写法是：

from openai import OpenAI

client = OpenAI()

stream = client.responses.create(
    model=os.environ["OPENAI_MODEL"],
    input="用一句话解释什么是幂等性。",
    stream=True,
)

for event in stream:
    print(event)

这里的 event 不是最终文本，而是一个流式事件。对于只想把文本增量显示到终端的程序，可以筛选包含文本输出的事件：

stream = client.responses.create(
    model=os.environ["OPENAI_MODEL"],
    input="写一个不超过三句的 Python 学习建议。",
    stream=True,
)

for event in stream:
    if event.type == "response.output_text.delta":
        print(event.delta, end="", flush=True)

print()

事件名称和字段属于 Responses 流式协议的一部分。工程代码不应假设所有事件都有 delta 属性，因为工具调用、响应开始、响应完成和错误事件的结构不同。

一个典型的数据流可以表示为：

sequenceDiagram
    participant App as Python 应用
    participant API as Responses API
    participant Model as 模型

    App->>API: responses.create(stream=True)
    API->>Model: 提交输入
    Model-->>API: 生成事件
    API-->>App: response.created
    API-->>App: response.output_text.delta
    API-->>App: response.output_text.delta
    API-->>App: response.completed
    App->>App: 拼接 delta 或读取最终结果

关键点是：delta 只表示“目前新增的一小段内容”，不能当作一个完整语义对象使用。

3.2 使用 responses.stream() 管理生命周期

SDK 还提供流式辅助器：

with client.responses.stream(
    model=os.environ["OPENAI_MODEL"],
    input="写一个三行以内的 Python 生成器示例。",
) as stream:
    for event in stream:
        if event.type == "response.output_text.delta":
            print(event.delta, end="", flush=True)

    final_response = stream.get_final_response()

print()
print("最终文本：", final_response.output_text)

上下文管理器的作用不是语法装饰，而是确保流对象在正常结束或异常退出时被关闭。若直接创建流却没有完整消费或关闭它，连接可能长期占用，尤其是在 Web 服务中会放大连接池压力。

3.3 异步流式

异步客户端使用 AsyncOpenAI，调用接口基本一致：

import asyncio
import os

from openai import AsyncOpenAI

client = AsyncOpenAI(api_key=os.environ["OPENAI_API_KEY"])


async def main() -> None:
    stream = await client.responses.create(
        model=os.environ["OPENAI_MODEL"],
        input="用两句话解释 asyncio 的事件循环。",
        stream=True,
    )

    async for event in stream:
        if event.type == "response.output_text.delta":
            print(event.delta, end="", flush=True)

    print()


asyncio.run(main())

同步客户端适合命令行程序、脚本和同步 Web 处理函数；异步客户端适合已经采用 asyncio 的服务。不要在异步服务中大量调用同步客户端，否则网络等待会阻塞事件循环。

3.4 流式输出的边界

流式只解决“尽早收到结果”的问题，不自动解决：

1. 输出是否符合业务格式；
2. 响应是否完整；
3. 工具调用参数是否可执行；
4. 网络中断后是否需要恢复；
5. 用户取消请求后如何停止生成。

例如，以下做法不安全：

buffer = ""

for event in stream:
    if hasattr(event, "delta"):
        buffer += event.delta

data = json.loads(buffer)

原因是：

• 事件不一定都是文本事件；
• 流可能在 JSON 完成前中断；
• delta 的边界不等于 JSON token 的边界；
• 模型可能返回拒答、截断或工具调用，而不是目标 JSON。

如果结果必须满足结构化约束，应优先使用 SDK 的结构化输出解析能力，并在最终响应完成后再使用解析结果。

---

四、结构化输出：让模型输出符合 Schema，但不要混淆“约束”和“业务校验”

4.1 什么是结构化输出

结构化输出要求模型生成符合指定 JSON Schema 的数据。例如，我们希望模型把一段文本提取成：

{
  "title": "字符串",
  "priority": 1,
  "tags": ["字符串"]
}

这与普通提示词：

请只输出 JSON，不要输出解释。

不是一回事。

普通提示词只是行为要求，模型仍可能输出 Markdown、额外说明、错误类型或缺失字段。结构化输出则将模型的文本生成与一个 schema 关联起来，SDK 可以进一步将返回内容解析为 Pydantic 模型。

4.2 使用 Pydantic v2 定义输出模型

from pydantic import BaseModel, Field


class Ticket(BaseModel):
    title: str = Field(description="问题标题")
    priority: int = Field(ge=1, le=5, description="优先级，1 最紧急，5 最不紧急")
    tags: list[str] = Field(description="问题标签")

这里存在两层约束：

$$\text{模型可接受输出} \subseteq \text{JSON Schema 允许的结构}$$

但业务真正需要的是：

$$\text{可接受业务结果} = \text{Schema 合法} \cap \text{业务规则成立}$$

例如，priority=1 满足 1 <= priority <= 5，但它是否真的应该被判定为最高优先级，仍然取决于业务语义。Pydantic 可以检查类型和声明式约束，却无法自动证明模型理解了业务事实。

4.3 使用 responses.parse()

import os

from openai import OpenAI
from pydantic import BaseModel, Field


class Ticket(BaseModel):
    title: str = Field(description="问题标题")
    priority: int = Field(ge=1, le=5)
    tags: list[str]


client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])

response = client.responses.parse(
    model=os.environ["OPENAI_MODEL"],
    instructions="从用户描述中提取工单信息。",
    input="支付页面在提交订单时返回错误，影响所有用户。",
    text_format=Ticket,
)

for output in response.output:
    if output.type != "message":
        continue

    for content in output.content:
        if content.type != "output_text":
            continue

        if content.parsed is None:
            raise RuntimeError("模型没有得到可解析的结构化输出")

        ticket: Ticket = content.parsed
        print(ticket)
        print(ticket.model_dump())

官方示例使用的也是 client.responses.parse(..., text_format=SomePydanticModel)，然后从 output_text 内容项的 parsed 属性取得 Pydantic 实例。(github.com)

content.parsed 和 content.text 的区别是：

• content.text 是模型产生的原始文本；
• content.parsed 是 SDK 根据 Pydantic 模型解析后的对象；
• parsed is None 表示当前输出项没有成功得到解析对象。

不要把 response.output_text 当成 Pydantic 实例：

# 错误理解
ticket = response.output_text
print(ticket.priority)  # 字符串没有 priority 属性

正确做法是从对应的 output_text 内容项取得 parsed。

4.4 手动解析适合哪些情况

如果不能使用 responses.parse()，也可以让接口返回 JSON 文本，再手动调用 Pydantic v2：

from pydantic import ValidationError

try:
    ticket = Ticket.model_validate_json(response.output_text)
except ValidationError as exc:
    print("结构校验失败：")
    print(exc)
    raise

手动解析的优点是流程透明，适合调试和兼容旧接口；缺点是需要自行处理：

• 输出不是合法 JSON；
• JSON 外层包有 Markdown；
• 字段缺失；
• 类型不匹配；
• 模型拒答；
• 输出被截断。

responses.parse() 主要减少的是解析样板代码，并不意味着业务系统可以跳过异常处理。

4.5 结构化输出的真实边界

结构化输出保证的是“符合指定输出格式的概率和接口约束”，并不保证：

• 事实正确；
• 推理正确；
• 字段值来自真实数据库；
• 没有提示词注入；
• 没有敏感信息；
• 业务规则全部满足。

一个反例是：

class RefundDecision(BaseModel):
    approved: bool
    amount: float

即使返回值是：

{"approved": true, "amount": 999999.0}

它可能完全符合 Schema，但如果订单实际金额只有 10 元，仍然是错误业务结果。因此，结构化输出之后还应执行确定性的业务校验：

def validate_refund(decision: RefundDecision, order_amount: float) -> None:
    if decision.amount < 0:
        raise ValueError("退款金额不能为负数")

    if decision.amount > order_amount:
        raise ValueError("退款金额不能超过订单金额")

Pydantic 的类型校验和业务服务的状态校验不是同一层。

---

五、结构化输出与流式输出同时使用

当前 SDK 提供了流式辅助器与 Pydantic 输出格式结合的用法：

from pydantic import BaseModel


class MathResult(BaseModel):
    expression: str
    result: int


with client.responses.stream(
    model=os.environ["OPENAI_MODEL"],
    input="计算 8 * 7，并返回表达式和结果。",
    text_format=MathResult,
) as stream:
    for event in stream:
        if event.type == "response.output_text.delta":
            print(event.delta, end="", flush=True)

    final_response = stream.get_final_response()

print()

流式阶段可以用于展示增量文本；最终阶段再取得完整响应并检查解析结果。官方 SDK 示例也展示了 responses.stream(..., text_format=MathResponse) 和 get_final_response() 的组合。(github.com)

需要注意，结构化 JSON 在生成中间状态时通常是不完整的：

{"expression":"8 * 7","res

所以不要在每个增量事件到达时都运行：

MathResult.model_validate_json(partial_text)

这会把一个尚未完成的 JSON 当成最终数据，导致大量无意义的解析异常。正确的生命周期是：

1. 接收并展示增量文本；
2. 等待终止事件；
3. 获取最终响应；
4. 检查状态；
5. 读取 parsed 或执行 model_validate_json()；
6. 通过业务校验后再提交后续操作。

流式结构化解析的具体行为与 SDK 版本相关，升级 SDK 时应运行自己的截断、拒答和工具调用测试，而不要依赖内部模块。

---

六、工具调用：模型只能提出调用请求，不能直接执行 Python 函数

6.1 工具调用的角色划分

函数工具调用至少包含三个角色：

用户
  ↓
模型：决定是否调用工具，并生成参数
  ↓
应用：校验参数、执行函数
  ↓
模型：读取工具结果，生成最终回答

模型不能直接访问你的 Python 进程。它只能输出一个结构化的函数调用请求，例如：

{
  "type": "function_call",
  "name": "get_weather",
  "call_id": "call_123",
  "arguments": "{\"city\":\"杭州\"}"
}

应用程序必须：

1. 找到 function_call；
2. 根据名称选择本地函数；
3. 解析 JSON 参数；
4. 校验参数；
5. 执行函数；
6. 将结果作为 function_call_output 发回模型。

6.2 定义一个工具

import json
import os
from typing import Any

from openai import OpenAI


def get_weather(city: str) -> dict[str, Any]:
    """
    示例函数。真实系统中这里应调用天气服务，
    而不是相信模型提供的天气事实。
    """
    return {
        "city": city,
        "temperature_c": 26,
        "condition": "晴",
    }


tools = [
    {
        "type": "function",
        "name": "get_weather",
        "description": "查询指定城市的当前天气。",
        "parameters": {
            "type": "object",
            "properties": {
                "city": {
                    "type": "string",
                    "description": "城市名称，例如杭州",
                }
            },
            "required": ["city"],
            "additionalProperties": False,
        },
        "strict": True,
    }
]

这里的 JSON Schema 是模型生成参数时使用的接口契约。strict=True 可以使参数结构更加严格，但它不能代替本地校验，也不能保证城市一定存在。

6.3 完整工具循环

import json
import os
from typing import Any

from openai import OpenAI
from pydantic import BaseModel, ValidationError


class WeatherArgs(BaseModel):
    city: str


def get_weather(city: str) -> dict[str, Any]:
    return {
        "city": city,
        "temperature_c": 26,
        "condition": "晴",
    }


tool_definitions = [
    {
        "type": "function",
        "name": "get_weather",
        "description": "查询指定城市的当前天气。",
        "parameters": WeatherArgs.model_json_schema(),
        "strict": True,
    }
]

tool_registry = {
    "get_weather": get_weather,
}

client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])

input_items: list[Any] = [
    {
        "role": "user",
        "content": "杭州现在天气怎么样？",
    }
]

for step in range(5):
    response = client.responses.create(
        model=os.environ["OPENAI_MODEL"],
        instructions="如果需要当前天气，必须调用 get_weather。",
        input=input_items,
        tools=tool_definitions,
    )

    # 必须保留模型产生的输出项，后续工具结果要与 call_id 对应。
    input_items.extend(response.output)

    function_calls = [
        item
        for item in response.output
        if item.type == "function_call"
    ]

    if not function_calls:
        print(response.output_text)
        break

    for call in function_calls:
        try:
            raw_args = json.loads(call.arguments)
            args = WeatherArgs.model_validate(raw_args)
        except (json.JSONDecodeError, ValidationError) as exc:
            tool_output = {
                "error": "工具参数无效",
                "detail": str(exc),
            }
        else:
            function = tool_registry.get(call.name)

            if function is None:
                tool_output = {
                    "error": f"未知工具：{call.name}",
                }
            else:
                try:
                    result = function(**args.model_dump())
                except Exception as exc:
                    tool_output = {
                        "error": "工具执行失败",
                        "detail": str(exc),
                    }
                else:
                    tool_output = result

        input_items.append(
            {
                "type": "function_call_output",
                "call_id": call.call_id,
                "output": json.dumps(
                    tool_output,
                    ensure_ascii=False,
                ),
            }
        )
else:
    raise RuntimeError("工具调用超过最大步数")

这段循环的状态变化如下：

S0: input_items = 用户问题
  ↓
S1: 模型返回 function_call
  ↓
S2: 应用解析 arguments
  ↓
S3: 应用执行 get_weather
  ↓
S4: 应用追加 function_call_output
  ↓
S5: 再次请求模型
  ↓
S6: 模型返回最终 message

其中 call_id 是关联工具调用和工具结果的关键。如果工具结果使用了错误的 call_id，模型无法可靠地知道这个结果对应哪一次调用。

6.4 为什么不能直接执行任意工具名

以下代码存在严重风险：

function = globals()[call.name]
result = function(**json.loads(call.arguments))

模型输出的函数名是外部输入，globals() 会把本地命名空间暴露为可调用集合。正确做法是使用显式白名单：

tool_registry = {
    "get_weather": get_weather,
    "search_order": search_order,
}

工具参数也必须经过 Pydantic 或其他确定性校验。特别是文件路径、SQL 条件、URL、shell 参数和支付金额，不能仅依赖 JSON Schema。

6.5 工具调用的副作用与重试

工具可能执行真实副作用：

• 创建订单；
• 发送邮件；
• 删除文件；
• 扣款；
• 修改数据库；
• 发布消息。

此时必须区分两个动作：

模型请求被重试
≠
工具应该被重复执行

SDK 会对某些网络错误和 HTTP 状态自动重试，但如果请求已经在服务端执行成功，只是客户端没有收到响应，客户端可能再次提交同一请求。官方 SDK 默认会对连接错误、408、409、429 和 500 以上的错误重试两次。(github.com)

因此，副作用工具必须设计业务幂等键。例如：

def create_order(user_id: str, request_id: str, items: list[str]) -> dict:
    """
    request_id 作为数据库唯一键。
    已处理过的 request_id 直接返回原结果。
    """
    ...

幂等性的条件是：

$$f(f(x, k), k) = f(x, k)$$

其中：

• $x$ 是业务输入；
• $k$ 是同一次业务操作的唯一键；
• $f$ 是带副作用的操作。

如果两次使用同一个 $k$，第二次执行不应产生新的订单或新的扣款。

---

七、重试：网络可靠性机制，不是模型质量修复机制

7.1 SDK 自动重试什么

当前 SDK 默认重试次数为 2。默认可重试的情况包括：

• 网络连接错误；
• 请求超时；
• HTTP 408；
• HTTP 409；
• HTTP 429；
• HTTP 500 及以上服务端错误。

可以在客户端级别配置：

from openai import OpenAI

client = OpenAI(
    api_key=os.environ["OPENAI_API_KEY"],
    max_retries=4,
)

也可以只对某一次调用覆盖：

response = client.with_options(
    max_retries=0,
).responses.create(
    model=os.environ["OPENAI_MODEL"],
    input="执行一次不应重复提交的请求。",
)

SDK 使用短暂的指数退避。指数退避的基本形式可以写成：

$$t_n = \min(t_{\max}, t_0 \cdot 2^n) + j$$

其中：

• $t_0$ 是初始等待时间；
• $n$ 是已经发生的重试次数；
• $t_{\max}$ 是等待时间上限；
• $j$ 是随机抖动；
• Retry-After 响应头存在且合理时，客户端会优先参考它。

自动重试的目标是应对暂时性故障，而不是修复以下问题：

• 400 参数错误；
• 401 认证失败；
• 403 权限错误；
• Schema 不支持；
• 工具函数抛出业务异常；
• 模型产生了错误事实；
• 提示词本身不完整。

7.2 异常分类

SDK 异常都继承自 openai.APIError。常见分类如下：(github.com)

import openai

try:
    response = client.responses.create(
        model=os.environ["OPENAI_MODEL"],
        input="你好",
    )
except openai.APIConnectionError as exc:
    print("网络连接失败")
    print("底层异常：", exc.__cause__)
except openai.APITimeoutError:
    print("请求超时")
except openai.RateLimitError:
    print("触发限流")
except openai.APIStatusError as exc:
    print("HTTP 状态错误：", exc.status_code)
    print("请求 ID：", exc.request_id)
except openai.APIError:
    print("其他 OpenAI SDK 错误")

不要把所有异常都捕获后返回“模型调用失败”：

try:
    ...
except Exception:
    return "失败"

这样会丢失认证失败、参数错误、限流、服务端错误和本地工具错误之间的差异，导致诊断和恢复策略全部失效。

7.3 超时不是简单的失败

超时可能发生在不同阶段：

连接阶段超时
发送请求超时
等待首个响应超时
读取流中间超时
读取完整响应超时

SDK 支持通过 timeout 配置请求超时，也可以使用 HTTP 客户端提供的更细粒度超时对象。当前 README 说明默认请求超时为 10 分钟，并且超时默认会参与重试。(github.com)

脚本可以使用简单配置：

client = OpenAI(
    api_key=os.environ["OPENAI_API_KEY"],
    timeout=30.0,
)

交互式应用通常不应让用户等待十分钟。服务端则需要根据请求类型区分：

• 普通短文本：较短连接和读取超时；
• 长输出：较长读取超时；
• 流式响应：客户端取消时及时关闭流；
• 工具调用：为每个工具设置独立超时。

---

八、重试与流式响应的特殊关系

非流式请求失败时，SDK 可以在请求层重新发送；流式请求已经向客户端输出部分内容后，如果连接中断，就不能简单地把新流拼接到旧流后面：

第一次流：
    “订单已经创建，订单号是 ...”
连接中断

第二次重试：
    “订单已经创建，订单号是 ...”

如果直接拼接，用户可能看到重复文本。更严重的是，如果模型在中断前已经触发了工具，重试还可能再次触发工具。

因此，流式系统通常需要区分：

1. 尚未向用户展示任何数据：可以考虑重新请求；
2. 已经展示部分文本：应报告流中断，或使用应用层恢复协议；
3. 已经执行副作用工具：必须使用幂等键；
4. 无法确认服务端是否执行成功：进入待确认状态，而不是盲目重试。

流式响应的恢复不是单纯的 HTTP 重试问题，而是状态机问题。

---

九、请求追踪：记录 _request_id，但不要记录敏感内容

SDK 返回的对象提供 _request_id，它来自服务端的 x-request-id 响应头，可用于定位失败请求。失败的状态异常可以从异常对象读取 request_id。(github.com)

建议至少记录：

import logging

logger = logging.getLogger(__name__)

response = client.responses.create(
    model=os.environ["OPENAI_MODEL"],
    input="你好",
)

logger.info(
    "OpenAI response completed request_id=%s",
    response._request_id,
)

异常路径：

try:
    response = client.responses.create(
        model=os.environ["OPENAI_MODEL"],
        input="你好",
    )
except openai.APIStatusError as exc:
    logger.exception(
        "OpenAI request failed status=%s request_id=%s",
        exc.status_code,
        exc.request_id,
    )
    raise

不要默认记录：

• API Key；
• 完整用户输入；
• 完整模型输出；
• 身份证号、手机号、订单信息；
• 工具参数中的密码或访问令牌。

请求 ID 用于定位请求；它不是业务订单号，也不是可以暴露给最终用户的诊断替代品。

---

十、一个可复用的同步封装

将调用、结构化解析和异常分类集中起来，可以减少业务代码对 SDK 细节的依赖：

import os
from typing import TypeVar

import openai
from openai import OpenAI
from pydantic import BaseModel


T = TypeVar("T", bound=BaseModel)


class LLMClient:
    def __init__(self) -> None:
        self.client = OpenAI(
            api_key=os.environ["OPENAI_API_KEY"],
            timeout=60.0,
            max_retries=2,
        )
        self.model = os.environ["OPENAI_MODEL"]

    def parse(self, prompt: str, schema: type[T]) -> T:
        try:
            response = self.client.responses.parse(
                model=self.model,
                input=prompt,
                text_format=schema,
            )
        except openai.RateLimitError:
            raise RuntimeError("模型服务限流，请稍后重试")
        except openai.APIConnectionError as exc:
            raise RuntimeError("模型服务网络不可用") from exc
        except openai.APITimeoutError as exc:
            raise RuntimeError("模型服务响应超时") from exc
        except openai.APIStatusError as exc:
            raise RuntimeError(
                f"模型服务返回 HTTP {exc.status_code}"
            ) from exc

        for output in response.output:
            if output.type != "message":
                continue

            for content in output.content:
                if content.type == "output_text":
                    if content.parsed is None:
                        raise RuntimeError("结构化输出解析失败")
                    return content.parsed

        raise RuntimeError("响应中没有可用的结构化文本")

使用：

from pydantic import BaseModel


class Summary(BaseModel):
    title: str
    points: list[str]


llm = LLMClient()

summary = llm.parse(
    "总结：Python 的类型注解不会在默认情况下执行运行时类型检查。",
    Summary,
)

print(summary.title)
print(summary.points)

这个封装没有把所有问题“吞掉”，而是把 SDK 的异常映射为应用层异常，同时保留原始异常作为 __cause__。上层可以根据应用需要决定返回 HTTP 429、HTTP 503，还是进入消息队列重试。

---

十一、常见误解与诊断路径

误解一：output_text 永远存在

错误假设：

print(response.output_text)

在工具调用场景中，模型可能只返回 function_call，此时应用应该先执行工具并将结果发回模型。诊断时先检查：

for item in response.output:
    print(item.type)

误解二：Pydantic 校验通过就代表答案正确

Pydantic 只能说明数据满足声明的结构和字段约束。例如：

class User(BaseModel):
    age: int

age=-100 仍然是合法整数。若业务要求年龄必须非负，需要明确声明：

class User(BaseModel):
    age: int = Field(ge=0, le=150)

即使如此，模型是否正确识别了用户年龄，仍需要业务语义验证。

误解三：工具 Schema 可以替代权限控制

Schema 只描述参数形状：

{"user_id": "u-123"}

它不能证明当前用户有权读取 u-123 的数据。权限检查必须在工具函数内部执行，并且使用服务端可信身份，而不是模型传入的用户 ID 作为唯一凭据。

误解四：重试次数越多越可靠

重试会增加恢复暂时性故障的机会，但也会增加：

$$\text{总延迟} = \sum \text{请求耗时} + \sum \text{退避时间}$$

对副作用操作而言，重试次数增加还可能增加重复执行风险。因此，重试必须与超时、幂等性、限流和业务状态机一起设计。

误解五：流式输出可以随时恢复

如果没有记录已经发送给客户端的内容、当前响应状态、工具调用状态和请求 ID，单纯重新建立连接无法判断哪些内容已经产生。流式恢复需要应用层协议，而不是只增加一个 try/except。

---

十二、如何选择调用方式

可以按照输出和交互要求选择：

只要最终文本
    → responses.create()

需要逐字显示
    → responses.create(stream=True)
    或 responses.stream()

需要 Pydantic 对象
    → responses.parse(..., text_format=Schema)

需要工具调用
    → responses.create(..., tools=[...])
    并实现 function_call → function_call_output 循环

既要工具又要结构化最终结果
    → 工具循环使用 responses.parse 或 create
    → 最终 message 再进行结构化解析

请求可能重复且包含副作用
    → SDK 重试 + 工具幂等键 + 业务状态检查

最重要的边界是：Responses API 负责模型交互协议，SDK 负责 Python 对象映射、流式事件和部分重试；工具的权限、幂等、事务和业务正确性仍然属于应用程序。理解这条边界后，Responses、流式、结构化输出、工具和重试才能组合成可诊断、可恢复的系统，而不是一组孤立的调用示例。

---

系列导航与关联阅读

• 系列入口：Python 完整学习路线：从语言模型、并发到 Web、数据、AI 与生产交付
• 上一篇：Jupyter 工程化：Kernel、Notebook 状态、复现、参数化和安全
• 下一篇：Python Agent 框架选型：原生循环、Agents SDK、LangGraph 和 PydanticAI
• 延伸：Pydantic v2：模型、校验器、序列化、Settings 和性能边界

官方资料

• PyTorch Documentation
• OpenAI Python SDK

本文依据 Python 官方文档、相关 PEP 与生态项目官方文档重新梳理；正文、示例与工程清单由 WR BLOG 编写。