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, replay_reasoning_content: bool = False, **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._replay_reasoning_content = replay_reasoning_content
        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, replay_reasoning_content=self._replay_reasoning_content),
            "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)
            failed = False
            try:
                async for chunk in _translate_chunks(stream):
                    yield chunk
            except BaseException:
                failed = True
                raise
            finally:
                close = getattr(stream, "close", None) or getattr(stream, "aclose", None)
                if close is not None:
                    try:
                        await close()
                    except Exception:
                        # A cleanup failure must not replace the provider error or cancellation.
                        if not failed:
                            raise

        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, replay_reasoning_content=self._replay_reasoning_content),
        "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)
        failed = False
        try:
            async for chunk in _translate_chunks(stream):
                yield chunk
        except BaseException:
            failed = True
            raise
        finally:
            close = getattr(stream, "close", None) or getattr(stream, "aclose", None)
            if close is not None:
                try:
                    await close()
                except Exception:
                    # A cleanup failure must not replace the provider error or cancellation.
                    if not failed:
                        raise

    return _gen()

create_openai_client

Python

create_openai_client(api_key: str, base_url: str) -> Any

Create an OpenAI-compatible client for an explicitly configured provider endpoint.

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

def create_openai_client(api_key: str, base_url: str) -> Any:
    r"""Create an OpenAI-compatible client for an explicitly configured provider endpoint."""
    import openai

    return openai.AsyncOpenAI(api_key=api_key, base_url=base_url.rstrip("/"))

create_openai_backend

Python

create_openai_backend(*, client: Any, model: str, base_url: str | None, thinking_enabled: bool | None = None, thinking_budget: int | None = None, force_non_thinking: bool = False) -> OpenAIBackend

Build a backend with the configured provider’s thinking protocol.

DeepSeek uses the OpenAI-compatible thinking object and requires reasoning_content replay across tool-result turns. Qwen-compatible endpoints retain their enable_thinking / thinking_budget contract.

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

def create_openai_backend(
    *,
    client: Any,
    model: str,
    base_url: str | None,
    thinking_enabled: bool | None = None,
    thinking_budget: int | None = None,
    force_non_thinking: bool = False,
) -> OpenAIBackend:
    r"""Build a backend with the configured provider's thinking protocol.

    DeepSeek uses the OpenAI-compatible ``thinking`` object and requires
    ``reasoning_content`` replay across tool-result turns. Qwen-compatible
    endpoints retain their ``enable_thinking`` / ``thinking_budget`` contract.
    """
    deepseek = _uses_deepseek_thinking_protocol(base_url)
    extra_body: dict[str, Any] = {}
    if force_non_thinking:
        extra_body = {"thinking": {"type": "disabled"}} if deepseek else {"enable_thinking": False}
    elif deepseek:
        if thinking_enabled is not None:
            extra_body["thinking"] = {"type": "enabled" if thinking_enabled else "disabled"}
    else:
        if thinking_enabled is not None:
            extra_body["enable_thinking"] = thinking_enabled
        if thinking_budget is not None:
            extra_body["thinking_budget"] = thinking_budget
    defaults = {"extra_body": extra_body} if extra_body else {}
    return OpenAIBackend(
        client=client,
        model=model,
        replay_reasoning_content=deepseek,
        **defaults,
    )