openai

feishu.agent.adapters.openai

OpenAIBackend

OpenAI Chat Completions API 的 feishu.agent.llm.LlmBackend 实现。

将与厂商无关的 feishu.agent.llm.Message、feishu.agent.llm.ToolSpec 翻译为 Chat Completions 请求格式（系统提示词作为 role="system" 的消息），并把其流式分片归一化为 [feishu.agent.llm.StreamChunk][]。openai SDK 仅在未注入 client 时按需懒加载，核心模块本身不依赖 该 SDK。本适配器仅实现已确认的 Chat Completions 流式协议。

参数：

名称	类型	描述	默认
client	Any	已构造的 openai.AsyncOpenAI 客户端。为 None 时自动创建一个。	None
model	str	模型名称，例如 gpt-4o。	必需
**defaults	Any	透传给 OpenAI API 的默认参数，例如 temperature。	{}

示例：

>>> backend = OpenAIBackend(model="gpt-4o", temperature=0.7)

源代码位于： feishu/agent/adapters/openai.py

class OpenAIBackend:
    r"""
    OpenAI Chat Completions API 的 [feishu.agent.llm.LlmBackend][] 实现。

    将与厂商无关的 [feishu.agent.llm.Message][]、[feishu.agent.llm.ToolSpec][] 翻译为 Chat Completions
    请求格式（系统提示词作为 `role="system"` 的消息），并把其流式分片归一化为
    [feishu.agent.llm.StreamChunk][]。`openai` SDK 仅在未注入 `client` 时按需懒加载，核心模块本身不依赖
    该 SDK。本适配器仅实现已确认的 Chat Completions 流式协议。

    Args:
        client: 已构造的 `openai.AsyncOpenAI` 客户端。为 `None` 时自动创建一个。
        model: 模型名称，例如 `gpt-4o`。
        **defaults: 透传给 OpenAI API 的默认参数，例如 `temperature`。

    Examples:
        >>> backend = OpenAIBackend(model="gpt-4o", temperature=0.7)  # doctest:+SKIP
    """

    def __init__(self, client: Any = None, *, model: str, **defaults: Any):
        if client is None:
            import openai  # imported lazily; core never imports the SDK

            client = openai.AsyncOpenAI()
        self._client = client
        self._model = model
        self._defaults = defaults

    def stream(
        self,
        *,
        messages: Sequence[Message],
        tools: Sequence[ToolSpec] = (),
        system: str | None = None,
        **kwargs: Any,
    ) -> AsyncIterator[StreamChunk]:
        r"""
        调用 OpenAI Chat Completions API 流式生成一轮响应。

        Args:
            messages: 截至当前轮次的对话历史。
            tools: 本轮可供模型调用的工具声明。
            system: 系统提示词，将作为首条 `role="system"` 消息加入请求。
            **kwargs: 覆盖默认参数（如 `model`）或追加的额外参数。

        Returns:
            逐个产出归一化 [feishu.agent.llm.StreamChunk][] 的异步迭代器。

        飞书文档:
            [OpenAI Chat Completions API](https://platform.openai.com/docs/api-reference/chat)

        Examples:
            >>> async for chunk in backend.stream(messages=history):  # doctest:+SKIP
            ...     print(chunk)
            TextDelta(text='你好')
            MessageStop(stop_reason=<StopReason.END_TURN: 'end_turn'>, usage={'prompt_tokens': 8})
        """
        params: dict[str, Any] = {
            "model": kwargs.pop("model", self._model),
            "messages": _to_openai_messages(messages, system),
            "stream": True,
            "stream_options": {"include_usage": True},
            **self._defaults,
            **kwargs,
        }
        if tools:
            params["tools"] = _to_openai_tools(tools)

        async def _gen() -> AsyncIterator[StreamChunk]:
            stream = await self._client.chat.completions.create(**params)
            async for chunk in _translate_chunks(stream):
                yield chunk

        return _gen()

stream

Python

stream(*, messages: Sequence[Message], tools: Sequence[ToolSpec] = (), system: str | None = None, **kwargs: Any) -> AsyncIterator[StreamChunk]

调用 OpenAI Chat Completions API 流式生成一轮响应。

参数：

名称	类型	描述	默认
messages	Sequence[Message]	截至当前轮次的对话历史。	必需
tools	Sequence[ToolSpec]	本轮可供模型调用的工具声明。	()
system	str | None	系统提示词，将作为首条 role="system" 消息加入请求。	None
**kwargs	Any	覆盖默认参数（如 model）或追加的额外参数。	{}

返回：

类型	描述
AsyncIterator[StreamChunk]	逐个产出归一化 [feishu.agent.llm.StreamChunk][] 的异步迭代器。

飞书文档

OpenAI Chat Completions API

示例：

>>> async for chunk in backend.stream(messages=history):
...     print(chunk)
TextDelta(text='你好')
MessageStop(stop_reason=<StopReason.END_TURN: 'end_turn'>, usage={'prompt_tokens': 8})

源代码位于： feishu/agent/adapters/openai.py

def stream(
    self,
    *,
    messages: Sequence[Message],
    tools: Sequence[ToolSpec] = (),
    system: str | None = None,
    **kwargs: Any,
) -> AsyncIterator[StreamChunk]:
    r"""
    调用 OpenAI Chat Completions API 流式生成一轮响应。

    Args:
        messages: 截至当前轮次的对话历史。
        tools: 本轮可供模型调用的工具声明。
        system: 系统提示词，将作为首条 `role="system"` 消息加入请求。
        **kwargs: 覆盖默认参数（如 `model`）或追加的额外参数。

    Returns:
        逐个产出归一化 [feishu.agent.llm.StreamChunk][] 的异步迭代器。

    飞书文档:
        [OpenAI Chat Completions API](https://platform.openai.com/docs/api-reference/chat)

    Examples:
        >>> async for chunk in backend.stream(messages=history):  # doctest:+SKIP
        ...     print(chunk)
        TextDelta(text='你好')
        MessageStop(stop_reason=<StopReason.END_TURN: 'end_turn'>, usage={'prompt_tokens': 8})
    """
    params: dict[str, Any] = {
        "model": kwargs.pop("model", self._model),
        "messages": _to_openai_messages(messages, system),
        "stream": True,
        "stream_options": {"include_usage": True},
        **self._defaults,
        **kwargs,
    }
    if tools:
        params["tools"] = _to_openai_tools(tools)

    async def _gen() -> AsyncIterator[StreamChunk]:
        stream = await self._client.chat.completions.create(**params)
        async for chunk in _translate_chunks(stream):
            yield chunk

    return _gen()