# -*- coding: utf-8 -*- """The OpenAI Responses API chat model implementation.""" from collections import OrderedDict from datetime import datetime from typing import Literal, Any, AsyncGenerator, List, TYPE_CHECKING, Type from pydantic import BaseModel, Field from ..._utils._common import _generate_id from .._base import ChatModelBase, _TOOL_CHOICE_LITERAL_MODES from .._model_response import ChatResponse from .._model_usage import ChatUsage from ...credential import OpenAICredential from ...formatter import FormatterBase, OpenAIResponseFormatter from ...message import Msg, ThinkingBlock, ToolCallBlock, TextBlock from ...tool import ToolChoice if TYPE_CHECKING: from openai.types.responses import Response from openai.types.responses import ResponseStreamEvent from openai import AsyncStream else: Response = Any ResponseStreamEvent = Any AsyncStream = Any # kwargs accepted by Chat Completions but NOT by the Responses API. _RESPONSES_UNSUPPORTED_KWARGS = frozenset({"modalities", "audio"}) class OpenAIResponseModel(ChatModelBase): """The OpenAI Responses API chat model. Compared with the Chat Completions API, the Responses API provides first-class streaming events for reasoning / thinking, text output and function-call arguments, which makes it a natural fit for models that expose chain-of-thought reasoning (e.g. ``o3``, ``o4-mini``). """ class Parameters(BaseModel): """The parameters for the OpenAI Response API model.""" max_tokens: int | None = Field( default=None, title="Max Tokens", description="The maximum number of output tokens.", gt=0, ) thinking_enable: bool = Field( default=False, title="Thinking", description=( "Whether to enable reasoning for reasoning models " "(e.g. o3, o4-mini, gpt-5.5). Use reasoning_effort to " "control the depth of reasoning." ), ) reasoning_effort: ( Literal["none", "minimal", "low", "medium", "high", "xhigh"] | None ) = Field( default=None, title="Reasoning Effort", description=( "Controls the depth of reasoning for reasoning models " "(e.g. o3, o4-mini, gpt-5.5). Supported values are " "model-dependent and may include: none, minimal, low, " "medium, high, xhigh." ), ) temperature: float | None = Field( default=None, title="Temperature", description="The temperature for the LLM output.", ge=0, le=2, ) type: Literal["openai_response"] = "openai_response" """The type of the chat model.""" def __init__( self, credential: OpenAICredential, model: str, parameters: "OpenAIResponseModel.Parameters | None" = None, stream: bool = True, max_retries: int = 3, retry_delay: float = 1.0, context_size: int = 200000, formatter: FormatterBase | None = None, client_kwargs: dict[str, Any] | None = None, ) -> None: """Initialize the OpenAI Responses API chat model. Args: credential (`OpenAICredential`): The OpenAI credential used to authenticate API calls. model (`str`): The OpenAI model name, e.g. ``o3`` or ``o4-mini``. parameters (`OpenAIResponseModel.Parameters | None`, defaults \ to `None`): The OpenAI Responses API parameters. When ``None``, the default parameters will be used. stream (`bool`, defaults to `True`): Whether to enable streaming output. max_retries (`int`, defaults to `3`): The maximum number of retries for the OpenAI Responses API. retry_delay (`float`, defaults to `1.0`): Seconds to sleep between retry attempts. context_size (`int`, defaults to `200000`): The model context size used for context compression. formatter (`FormatterBase | None`, defaults to `None`): The formatter that converts ``Msg`` objects to the format required by the OpenAI Responses API. When ``None``, an ``OpenAIResponseFormatter`` instance will be used. client_kwargs (`dict[str, Any] | None`, defaults to `None`): Extra keyword arguments forwarded to ``openai.AsyncClient`` (e.g. ``timeout``, ``default_headers``, ``http_client``). """ super().__init__( credential=credential, model=model, parameters=parameters or self.Parameters(), stream=stream, max_retries=max_retries, retry_delay=retry_delay, context_size=context_size, ) self.formatter = formatter or OpenAIResponseFormatter() self.client_kwargs = client_kwargs or {} @classmethod def _get_retryable_exceptions(cls) -> tuple[Type[Exception], ...]: import openai return ( openai.APIConnectionError, openai.APITimeoutError, openai.RateLimitError, openai.InternalServerError, ) async def _call_api( self, model_name: str, messages: list[Msg], tools: list[dict] | None = None, tool_choice: ToolChoice | None = None, **generate_kwargs: Any, ) -> ChatResponse | AsyncGenerator[ChatResponse, None]: """Call the OpenAI Responses API. Args: model_name (`str`): The model name to use for this call. messages (`list`): A list of input items for the Responses API. tools (`list[dict]`, default `None`): The tools JSON schemas. tool_choice (`ToolChoice | None`, optional): Controls which (if any) tool is called by the model. **generate_kwargs (`Any`): Extra keyword arguments forwarded to the API. Returns: `ChatResponse | AsyncGenerator[ChatResponse, None]`: A ``ChatResponse`` when streaming is disabled, or an async generator of ``ChatResponse`` objects when streaming is enabled. """ import openai client = openai.AsyncClient( **{ "api_key": self.credential.api_key.get_secret_value(), "organization": self.credential.organization, "base_url": self.credential.base_url, **self.client_kwargs, }, ) formatted_messages = await self.formatter.format(messages) api_kwargs: dict[str, Any] = { "model": model_name, "input": formatted_messages, "stream": self.stream, } if self.parameters.max_tokens is not None: api_kwargs["max_output_tokens"] = self.parameters.max_tokens if self.parameters.temperature is not None: api_kwargs["temperature"] = self.parameters.temperature if ( self.parameters.thinking_enable and self.parameters.reasoning_effort ): api_kwargs["reasoning"] = { "effort": self.parameters.reasoning_effort, } # The Responses API does not yet support audio output # (modalities / audio params). Strip them so callers that # mistakenly pass Chat-Completions-style audio kwargs don't # trigger a TypeError. # https://developers.openai.com/api/docs/guides/migrate-to-responses api_kwargs.update( { k: v for k, v in generate_kwargs.items() if k not in _RESPONSES_UNSUPPORTED_KWARGS }, ) fmt_tools, fmt_tool_choice = self._format_tools(tools, tool_choice) if fmt_tools is not None: api_kwargs["tools"] = fmt_tools if fmt_tool_choice is not None: api_kwargs["tool_choice"] = fmt_tool_choice start_datetime = datetime.now() response = await client.responses.create(**api_kwargs) if self.stream: return self._parse_stream_response(start_datetime, response) return self._parse_completion_response(start_datetime, response) async def _parse_stream_response( self, start_datetime: datetime, response: "AsyncStream[ResponseStreamEvent]", ) -> AsyncGenerator[ChatResponse, None]: """Parse the OpenAI Responses API streaming response. Each event yields only the *delta* content produced by that event so that the base ``ChatModelBase.__call__`` can accumulate the full response. Because the Responses API only returns the upstream response id on the final ``response.completed`` event, we assign a locally-generated ``response_id`` up-front so every delta chunk carries a stable id. Args: start_datetime (`datetime`): The start datetime of the response generation. response (`AsyncStream[ResponseStreamEvent]`): The OpenAI Responses API async stream object. Yields: `ChatResponse`: Incremental ``ChatResponse`` objects with ``is_last=False``. The base class ``__call__`` accumulates them and emits the final ``is_last=True`` chunk. """ usage: ChatUsage | None = None response_id: str = _generate_id() text_id: str = _generate_id() thinking_id: str = _generate_id() # Mapping from Responses API item id (fc_xxx) to (call_id, name) # so subsequent argument deltas can be routed to the right tool # call block. tool_call_mapping: dict = OrderedDict() async for event in response: event_type = event.type delta_res = ChatResponse( content=[], is_last=False, id=response_id, ) if event_type == "response.reasoning_summary_text.delta": # Reasoning summary text is NOT emitted by all models. # As of 2026-05, o1 and o4-mini do not stream reasoning # summary deltas. This handler exists for forward # compatibility with models that do expose it. delta_res.append_thinking( event.delta, block_id=thinking_id, ) elif event_type == "response.output_text.delta": delta_res.append_text(event.delta, block_id=text_id) elif event_type == "response.output_item.added": item = event.item if getattr(item, "type", None) == "function_call": # item.call_id (call_xxx) is used as ToolCallBlock.id # so the formatter can echo it back as # function_call.call_id / function_call_output.call_id. # Only record the mapping here — do NOT emit an # empty-input delta so downstream consumers don't see # a leading no-op chunk. The block is created on the # first argument delta below. tool_call_mapping[item.id] = ( item.call_id, getattr(item, "name", "") or "unknown", ) elif event_type == "response.function_call_arguments.delta": item_id = event.item_id if item_id in tool_call_mapping: call_id, name = tool_call_mapping[item_id] delta_res.append_tool_call( block_id=call_id, name=name, input=event.delta or "", ) elif event_type == "response.completed": resp = event.response if resp.usage: u = resp.usage details = getattr(u, "input_tokens_details", None) usage = ChatUsage( input_tokens=u.input_tokens, output_tokens=u.output_tokens, time=(datetime.now() - start_datetime).total_seconds(), cache_input_tokens=getattr( details, "cached_tokens", 0, ) if details else 0, ) # Attach reasoning item id metadata from the completed # response so the formatter can echo it back in # multi-turn history. The Responses API requires every # function_call item to be accompanied by its preceding # reasoning item (see the function-calling guide); the # reasoning item may have an empty summary when the model # does not expose it (e.g. o1/o4-mini as of 2026-05). for output_item in getattr(resp, "output", []): if getattr(output_item, "type", None) == "reasoning": reasoning_item_id = getattr(output_item, "id", None) if reasoning_item_id: delta_res.append_thinking( thinking="", block_id=thinking_id, reasoning_item_id=reasoning_item_id, ) if delta_res.content or usage: delta_res.usage = usage yield delta_res def _parse_completion_response( self, start_datetime: datetime, response: "Response", ) -> ChatResponse: """Parse the OpenAI Responses API non-streaming response. Args: start_datetime (`datetime`): The start datetime of the response generation. response (`Response`): The OpenAI Responses API response object. Returns: `ChatResponse`: A single ``ChatResponse`` with ``is_last=True``. """ content_blocks: List[TextBlock | ToolCallBlock | ThinkingBlock] = [] for item in response.output: item_type = getattr(item, "type", None) if item_type == "reasoning": reasoning_item_id = getattr(item, "id", None) combined_summary = " ".join( getattr(s, "text", "") for s in getattr(item, "summary", []) if getattr(s, "text", "") ) if combined_summary: content_blocks.append( ThinkingBlock( type="thinking", thinking=combined_summary, reasoning_item_id=reasoning_item_id, ), ) elif item_type == "message": for part in getattr(item, "content", []): if getattr(part, "type", None) == "output_text": content_blocks.append( TextBlock(type="text", text=part.text), ) elif item_type == "function_call": content_blocks.append( ToolCallBlock( id=item.call_id, name=item.name, input=getattr(item, "arguments", "") or "{}", ), ) usage = None if response.usage: u = response.usage details = getattr(u, "input_tokens_details", None) usage = ChatUsage( input_tokens=u.input_tokens, output_tokens=u.output_tokens, time=(datetime.now() - start_datetime).total_seconds(), cache_input_tokens=getattr( details, "cached_tokens", 0, ) if details else 0, ) resp_kwargs: dict[str, Any] = { "content": content_blocks, "is_last": True, "usage": usage, } response_id = getattr(response, "id", None) if response_id: resp_kwargs["id"] = response_id return ChatResponse(**resp_kwargs) def _format_tools( self, tools: list[dict] | None, tool_choice: ToolChoice | None, ) -> tuple[list[dict] | None, str | dict | None]: """Validate and format tools and tool_choice for the Responses API. The full ``tools`` list is always sent unchanged to maximise prompt cache hits. When ``tool_choice.tools`` restricts the callable subset, the ``allowed_tools`` tool_choice format is used instead of filtering the schemas list. Args: tools (`list[dict] | None`, optional): The raw tool schemas. tool_choice (`ToolChoice | None`, optional): The tool choice configuration. Returns: `tuple[list[dict] | None, str | dict | None]`: A tuple of ``(formatted_tools, formatted_tool_choice)`` ready for the Responses API. """ if tool_choice and tools: self._validate_tool_choice(tool_choice, tools) fmt_tools = None if tools: fmt_tools = [ {"type": "function", **tool["function"]} for tool in tools ] if not tool_choice: return fmt_tools, None mode = tool_choice.mode if mode not in _TOOL_CHOICE_LITERAL_MODES: return fmt_tools, {"type": "function", "name": mode} if tool_choice.tools: return fmt_tools, { "type": "allowed_tools", "mode": mode, "tools": [ {"type": "function", "name": name} for name in tool_choice.tools ], } return fmt_tools, mode