Responses API 调用

演示新一代 Responses API 的用法：以 input（消息列表）+ instructions（系统指令） 发起请求，并从响应中提取文本。真实调用需 OPENAI_API_KEY；本示例通过依赖注入， 可在测试中使用伪造客户端验证调用逻辑。 对应教程：第3卷《OpenAI API》第1章 Responses API。

文件路径: examples/volume-03-openai-api\responses_api.py

---

源代码

"""OpenAI Responses API 示例。

演示新一代 Responses API 的用法：以 ``input``（消息列表）+ ``instructions``（系统指令）
发起请求，并从响应中提取文本。真实调用需 ``OPENAI_API_KEY``；本示例通过依赖注入，
可在测试中使用伪造客户端验证调用逻辑。

对应教程：第3卷《OpenAI API》第1章 Responses API。
"""

from __future__ import annotations

from typing import Any, Protocol


class ResponsesClient(Protocol):
    """Responses API 客户端的最小协议。"""

    def responses(self) -> Any: ...

    # 真实用法：client.responses.create(model=..., instructions=..., input=...)


def build_input(messages: list[dict[str, str]]) -> list[dict[str, str]]:
    """将聊天风格的消息转换为 Responses API 的 ``input`` 格式。

    Args:
        messages: ``[{"role": "user", "content": "..."}]`` 形式的消息列表。

    Returns:
        可直接作为 ``input`` 传入的列表（此处格式兼容，原样返回并过滤空内容）。

    Raises:
        ValueError: 消息缺少 ``role`` 或 ``content`` 时。
    """
    result: list[dict[str, str]] = []
    for msg in messages:
        role = msg.get("role")
        content = msg.get("content")
        if not role or not content:
            raise ValueError(f"消息缺少 role 或 content: {msg!r}")
        result.append({"role": role, "content": content})
    return result


def extract_text(response: Any) -> str:
    """从 Responses API 响应中提取文本。

    优先使用 ``output_text``，其次尝试 ``output[0].content[0].text``。

    Args:
        response: Responses API 响应对象。

    Returns:
        提取到的文本；若无法提取返回空字符串。
    """
    output_text = getattr(response, "output_text", None)
    if isinstance(output_text, str) and output_text:
        return output_text
    output = getattr(response, "output", None)
    if isinstance(output, list) and output:
        first = output[0]
        content = getattr(first, "content", None)
        if isinstance(content, list) and content:
            return getattr(content[0], "text", "") or ""
    return ""


def call_responses_api(
    client: ResponsesClient,
    user_input: str,
    instructions: str = "你是一个友好的助手。",
    model: str = "gpt-4o-mini",
) -> str:
    """调用 Responses API 并返回回复文本。

    Args:
        client: 符合 ``ResponsesClient`` 协议的客户端。
        user_input: 用户输入文本。
        instructions: 系统指令。
        model: 模型名称。

    Returns:
        模型回复文本。
    """
    response = client.responses.create(  # type: ignore[union-attr]
        model=model,
        instructions=instructions,
        input=[{"role": "user", "content": user_input}],
    )
    return extract_text(response)


def main() -> None:
    """演示 Responses API 调用（需真实 API Key）。"""
    import os

    api_key = os.environ.get("OPENAI_API_KEY")
    if not api_key:
        print("未设置 OPENAI_API_KEY，跳过真实调用。")
        print("示例调用方式：")
        print("  from openai import OpenAI")
        print("  client = OpenAI()")
        print("  reply = call_responses_api(client, '你好')")
        return

    from openai import OpenAI

    client = OpenAI(api_key=api_key)
    reply = call_responses_api(client, "用一句话解释什么是 AI Agent")
    print(f"回复: {reply}")


if __name__ == "__main__":
    main()