Responses API 高级实战：structured outputs / 流式 SSE / Batch / prompt caching

Responses API（input 字段，不是 messages）是 OpenAI 当前主推接口——大部分开发者只用到基础 chat。本文覆盖四个生产级特性，让 Responses API 从『能调通』升级到『生产级高可用、低成本』。

1. JSON Schema 严格模式（structured outputs）

让模型 100% 按 schema 输出，零解析失败。

from pydantic import BaseModel
from openai import OpenAI

class OrderStatus(BaseModel):
    order_id: str
    status: str  # 'pending' / 'shipped' / 'delivered'
    eta: str     # ISO date
    tracking_url: str | None = None

client = OpenAI()
response = client.responses.create(
    model="gpt-5.6-terra",
    input=[{"role": "user", "content": "查询订单 12345 状态"}],
    text={
        "format": {
            "type": "json_schema",
            "name": "order_status",        # 必须有 name 字段
            "strict": True,                # 严格模式：100% 按 schema
            "schema": OrderStatus.model_json_schema(),
        }
    },
)

# 直接解析，不用 try/except
order = OrderStatus.model_validate_json(response.output_text)
print(order.order_id, order.status)

关键约束：

• name 必填——只放 schema 不放 name 会报错。
• strict: True 让模型输出严格符合 schema（没有额外字段、没有缺字段、没有类型错误）。
• schema 必须 JSON Schema 2020-12 兼容——Pydantic / Zod 生成的 schema 通常 OK。
• 代价：latency 略增（模型要做 schema 校验），但生产环境『不出错』价值远大于几十 ms。

2. 流式 SSE 解析

实时 UI + 中途 function calling。

import sseclient
import json

def stream_response(prompt):
    response = requests.post(
        "https://api.openai.com/v1/responses",
        headers={"Authorization": f"Bearer {OPENAI_API_KEY}"},
        json={
            "model": "gpt-5.6-terra",
            "input": [{"role": "user", "content": prompt}],
            "stream": True,
        },
        stream=True,
    )
    client = sseclient.SSEClient(response.iter_content(chunk_size=1024))

    text_buffer = ""
    for event in client.events():
        data = json.loads(event.data)

        if data["type"] == "response.output_text.delta":
            # 文本片段
            text_buffer += data["delta"]
            yield ("text", data["delta"])

        elif data["type"] == "response.function_call_arguments.delta":
            # function 参数流式
            yield ("function_arg_delta", data["delta"])

        elif data["type"] == "response.function_call_arguments.done":
            # function 参数收齐，调用 function
            args = json.loads(data["arguments"])
            result = my_function(**args)
            # 把结果回传（需要保持 session / conversation_id）
            yield ("function_done", result)

注意点：

• 文本输出和 function 调用是交织的——不能假设『先输出文本再调 function』。
• 中途 function 调用：监听 function_call_arguments.done 事件，调用 function，把结果通过 conversation.item.create 回传（要保持 conversation 状态）。
• 客户端要维护完整 state（之前的 message 历史 + 当前 token 累积）才能正确拼接。

3. Prompt caching（三层缓存）

重复 prompt 降本 80%+。

# === implicit caching（自动）===
# 当 prompt 超过 1k tokens 且前缀相同时，OpenAI 自动命中缓存
# 缓存 TTL：5-10 分钟
response = client.responses.create(
    model="gpt-5.6-terra",
    input=[
        {"role": "system", "content": LONG_SYSTEM_PROMPT},  # > 1k token
        {"role": "user", "content": "用户问题"},
    ],
)
# 第一次调用：cache miss，按原价
# 5 分钟内再调用相同 system prompt：cache hit，按 cache 价（约 input 的 1/4）

# === explicit caching（自定义 key）===
response = client.responses.create(
    model="gpt-5.6-terra",
    input=[...],
    prompt_cache_key="user_12345",  # 自定义 cache key，按 user / tenant 隔离
)

# === 控制 cache mode ===
response = client.responses.create(
    model="gpt-5.6-terra",
    input=[...],
    prompt_cache_options={
        "mode": "explicit",  # 'implicit' / 'explicit'
    },
)

何时有效：

• ✅ 长 system prompt（>1k token）+ 多次调用
• ✅ 多轮对话——历史对话累积在 prompt 前缀
• ✅ 批量相似任务——共享同一段大 prompt

何时失效：

• ❌ 每次调用 prompt 完全不一样
• ❌ prompt < 1k tokens
• ❌ 调用间隔超过 5-10 分钟（cache 过期）

配额限制：每请求 ≤4 个 write、≤50 个 breakpoints、每 key ≤15 次/分钟。

4. Batch API（离线降本 50%）

不需要实时的批量任务用 Batch API。

# === 创建 batch 任务 ===
batch = client.batches.create(
    input_file_id="file-abc123",  # 上传的 JSONL 请求文件
    endpoint="/v1/responses",
    completion_window="24h",       # 最长 24h 返回
    metadata={"description": "doc-summarize-batch-2026-08-14"},
)

# === 查询 batch 状态 ===
status = client.batches.retrieve(batch.id)
print(f"Status: {status.status}, completed: {status.request_counts.completed}/{status.request_counts.total}")

# === 下载结果（完成后）===
if status.status == "completed":
    output_file = client.files.content(status.output_file_id)
    for line in output_file.text.split("\n"):
        result = json.loads(line)
        print(result["custom_id"], result["response"]["body"])

适用场景：

• ✅ 批量文档摘要（几千篇新闻）
• ✅ 批量数据标注（10 万条数据）
• ✅ 批量翻译（不要求实时）
• ✅ 批量评测（生成测试用例）
• ✅ 夜间离线任务

硬限制：

• ❌ 不支持流式
• ❌ 不支持 web search / file search
• ❌ 不支持 Realtime API
• ✅ 仅支持 responses.create / chat.completions.create 等同步接口

成本对比：Batch API 价格 -50%，但任务调度周期长（最多 24h）。

5. 成本优化组合拳

三件套上齐，月成本从 $100K 降到 $30K 不是梦。

# === 三件套组合示例：批量文档摘要 ===

# Step 1：上传 JSONL 请求（每个 request 一个）
requests_jsonl = []
for doc_id, doc_text in documents:
    requests_jsonl.append({
        "custom_id": f"doc-{doc_id}",
        "method": "POST",
        "url": "/v1/responses",
        "body": {
            "model": "gpt-5.6-luna",  # 最便宜的模型做摘要
            "input": [
                {"role": "system", "content": LONG_SUMMARIZE_PROMPT},  # 共享 prompt 走 cache
                {"role": "user", "content": doc_text},
            ],
            "text": {
                "format": {
                    "type": "json_schema",
                    "name": "summary",
                    "strict": True,
                    "schema": {
                        "type": "object",
                        "properties": {
                            "summary": {"type": "string"},
                            "key_points": {"type": "array", "items": {"type": "string"}},
                        },
                        "required": ["summary", "key_points"],
                        "additionalProperties": False,
                    },
                }
            },
        },
    })

# Step 2：上传 + 创建
file = client.files.create(file=("\n".join(json.dumps(r) for r in requests_jsonl)).encode(), purpose="batch")
batch = client.batches.create(input_file_id=file.id, endpoint="/v1/responses", completion_window="24h")

成本节省估算（10 万文档摘要场景）：

• 不用三件套：$100K/月
• 只用 Luna 模型：$20K/月
• • JSON Schema（避免重解析）：$20K/月（节省的解析失败重试费用）
• • Batch API：$10K/月（-50%）
• • Prompt caching：$8K/月（共享 prompt 减半）

总节省 92%。

常见问题

1. structured outputs 一定要用 strict 模式吗？

强烈建议。strict 模式（{'strict': True}）让模型 100% 按 JSON Schema 输出，没有额外字段、没有缺字段、没有类型错误。非 strict 模式（默认）模型可能输出『近似符合』的 JSON，客户端解析时偶尔失败。strict 模式的代价是 latency 略增（schema 校验要做），但生产环境用 strict 模式几乎不出错。

2. 流式输出中途遇到 function calling 怎么处理？

三步：(1) 监听 response.output_text.delta 事件累积文本片段；(2) 监听 response.function_call_arguments.delta 事件流式接收参数；(3) 当收到 response.function_call_arguments.done 事件时，调用 function 并把结果通过 conversation.item.create 回传。注意点：function 调用和文本生成是交织的，不能假设『先输出文本再调用 function』。

3. Batch API 什么时候用？

判断标准：(1) 是否要求实时（要求 → 用 Realtime API 或常规 Responses）；(2) 是否能等 24 小时（能等 → 用 Batch API 降本 50%）。典型场景：批量文档摘要、批量数据标注、批量翻译、批量评测、夜间离线任务。硬限制：no streaming、不支持 web search / file search。如果任务中途要调工具或查外部数据，Batch API 不能用。

4. prompt caching 在什么场景有效？

三个场景：(1) 长 system prompt（>1k tokens）+ 多次调用——每次都重用同一段 prompt，命中缓存；(2) 多轮对话——历史对话累积在 prompt 前缀，命中缓存；(3) 批量相似任务——所有调用共享同一段大 prompt。失效场景：每次调用 prompt 完全不一样 / prompt < 1k tokens / 调用间隔太长（缓存过期，默认 5-10 分钟）。

5. Batch API + prompt caching 能叠加吗？

能，但有 trade-off。Batch API 已经 -50% 成本，叠加 prompt caching 还能再省（cache 命中部分按 cache 价，约 input 的 1/4）。但 Batch API 任务调度周期长（最多 24h），cache TTL 可能过期。建议：长期高频调用走 prompt caching 实时调用；低频 / 大批量走 Batch API。两者结合的 ROI 需要按具体场景算。

6. 监控指标

生产环境必看的几个数：

# === Cache hit rate ===
# 从 response.usage 里取 cached_tokens
cached = response.usage.input_tokens_details.cached_tokens
total_input = response.usage.input_tokens
cache_hit_rate = cached / total_input if total_input > 0 else 0
# 目标 ≥ 50%

# === Batch 完成率 ===
batch = client.batches.retrieve(batch_id)
print(f"completed: {batch.request_counts.completed}/{batch.request_counts.total}")
# 目标 = 100%

# === Function call 触发率 ===
# 从 log 里统计 function_call 事件 / total response

下一步

• 想了解 Responses API 入门？读 《OpenAI API 入门：第一个 GPT-5.6 调用详解》。
• 想了解 function calling？读 《OpenAI API 函数调用实战：Responses API 工具使用完全指南》。
• 想了解错误处理？读 《OpenAI API 错误处理与重试：401/429/5xx 实战模式》。