OpenAI 客户端 SDK 与 Responses API

from openai import OpenAI

# 默认值：从环境变量 OPENAI_API_KEY 读取密钥；超时 10 分钟；自动重试 2 次
client = OpenAI(timeout=60.0, max_retries=3)

response = client.responses.create(
    model="gpt-6-astra",
    instructions="你是一个资深后端工程师，回答要简洁。",  # 相当于系统提示词
    input="用一句话解释什么是幂等性。",
)
print(response.output_text)  # SDK 提供的便捷属性：把所有输出文本拼接起来
print(response.id)  # resp_...，可用于 previous_response_id
print(response.usage)  # token 用量

import asyncio

from openai import AsyncOpenAI

client = AsyncOpenAI()


async def main() -> None:
    response = await client.responses.create(model="gpt-6-astra", input="你好")
    print(response.output_text)


asyncio.run(main())

from openai import OpenAI

client = OpenAI()

# Chat Completions（上一代接口，仍然可用）
completion = client.chat.completions.create(
    model="gpt-6-astra",
    messages=[{"role": "user", "content": "写一句关于独角兽的睡前故事。"}],
)
print(completion.choices[0].message.content)

# Responses（新项目推荐使用）
response = client.responses.create(
    model="gpt-6-astra",
    input="写一句关于独角兽的睡前故事。",
)
print(response.output_text)

{
  "id": "resp_68af4030592c81938ec0a5fbab4a3e9f05438e46b5f69a3b",
  "object": "response",
  "model": "gpt-5.5",
  "output": [
    { "id": "rs_...", "type": "reasoning", "content": [], "summary": [] },
    {
      "id": "msg_...",
      "type": "message",
      "status": "completed",
      "role": "assistant",
      "content": [{ "type": "output_text", "annotations": [], "text": "Under a quilt of moonlight, ..." }]
    }
  ]
}

from openai import OpenAI

client = OpenAI()
response = client.responses.create(model="gpt-6-astra", input="1+1 等于几？")

for item in response.output:
    if item.type == "reasoning":
        print("[推理 item]", item.id)
    elif item.type == "message":
        for part in item.content:
            if part.type == "output_text":
                print("[文本]", part.text)
            elif part.type == "refusal":
                print("[拒绝]", part.refusal)

from typing import Any

from openai import OpenAI

client = OpenAI()
MODEL = "gpt-6-astra"

# 方式 1：手动管理历史。配合 store=False 时，OpenAI 不保存这次 response
history: list[Any] = [{"role": "user", "content": "讲个笑话"}]
r1 = client.responses.create(model=MODEL, input=history, store=False)
history += r1.output  # 包括加密的 reasoning item，必须原样保留
history.append({"role": "user", "content": "再来一个"})
r2 = client.responses.create(model=MODEL, input=history, store=False)

# 方式 2：previous_response_id 链式接续
r1 = client.responses.create(model=MODEL, input="讲个笑话")
r2 = client.responses.create(
    model=MODEL,
    previous_response_id=r1.id,
    input=[{"role": "user", "content": "解释一下笑点"}],
)

# 方式 3：Conversations API，得到一个持久的会话 ID
conv = client.conversations.create()
r3 = client.responses.create(
    model=MODEL,
    conversation=conv.id,
    input=[{"role": "user", "content": "躲避球的 5 个 D 是什么？"}],
)
print(r2.output_text, r3.output_text)

from openai import OpenAI

client = OpenAI()

# 1) Web 搜索
r = client.responses.create(
    model="gpt-6-astra",
    tools=[{"type": "web_search"}],
    input="今天有什么积极的科技新闻？",
)
print(r.output_text)

# 2) 远程 MCP 服务器：这里使用 OpenAI 官方的文档 MCP，只读，不需要鉴权
r = client.responses.create(
    model="gpt-6-astra",
    tools=[
        {
            "type": "mcp",
            "server_label": "openai_docs",
            "server_description": "Search and read the public OpenAI documentation.",
            "server_url": "https://developers.openai.com/mcp",
            "require_approval": "never",
        }
    ],
    input="在 OpenAI 文档中搜索 Responses API streaming，并返回相关链接。",
)
print(r.output_text)

from pydantic import BaseModel

from openai import OpenAI

client = OpenAI()


class CalendarEvent(BaseModel):
    name: str
    date: str
    participants: list[str]


response = client.responses.parse(
    model="gpt-6-astra",
    input=[
        {"role": "system", "content": "抽取事件信息。"},
        {"role": "user", "content": "Alice 和 Bob 周五去参加科学展。"},
    ],
    text_format=CalendarEvent,  # SDK 自动把 Pydantic 模型转成 JSON Schema
)
event = response.output_parsed  # 已经校验过的 CalendarEvent 实例（可能为 None）
print(event)

from openai import OpenAI

client = OpenAI()

stream = client.responses.create(
    model="gpt-6-astra",
    input="把 'double bubble bath' 快速说十遍。",
    stream=True,
)
for event in stream:
    if event.type == "response.output_text.delta":
        print(event.delta, end="", flush=True)
    elif event.type == "response.completed":
        print("\n[完成]", event.response.usage)

import openai
from openai import OpenAI

client = OpenAI()

try:
    response = client.responses.create(model="gpt-6-astra", input="hello")
    print(response._request_id)  # 例如 req_123；以 _ 开头，但这是一个公开属性
except openai.APIConnectionError as e:
    print("连不上服务器：", e.__cause__)
except openai.RateLimitError:
    print("429：触发限流，需要退避重试")
except openai.APIStatusError as e:
    print("其他非 2xx 状态码：", e.status_code, e.request_id)