# SPDX-FileCopyrightText: 2022-present deepset GmbH # # SPDX-License-Identifier: Apache-2.0 import json import os from datetime import datetime from typing import Any, ClassVar from openai import AsyncOpenAI, AsyncStream, OpenAI, Stream from openai.lib._pydantic import to_strict_json_schema from openai.types.responses import ParsedResponse, Response, ResponseOutputRefusal, ResponseStreamEvent from pydantic import BaseModel from haystack import component, default_from_dict, default_to_dict, logging from haystack.components.generators.utils import _normalize_messages, _serialize_object from haystack.dataclasses import ( ChatMessage, ComponentInfo, FileContent, ImageContent, ReasoningContent, StreamingCallbackT, StreamingChunk, SyncStreamingCallbackT, TextContent, ToolCall, ToolCallDelta, select_streaming_callback, ) from haystack.dataclasses.streaming_chunk import _invoke_streaming_callback from haystack.tools import ( ToolsType, _check_duplicate_tool_names, deserialize_tools_or_toolset_inplace, flatten_tools_or_toolsets, serialize_tools_or_toolset, warm_up_tools, ) from haystack.utils import Secret, deserialize_callable, serialize_callable from haystack.utils.http_client import init_http_client logger = logging.getLogger(__name__) @component class OpenAIResponsesChatGenerator: """ Completes chats using OpenAI's Responses API. It works with the gpt-4 and o-series models and supports streaming responses from OpenAI API. It uses [ChatMessage](https://docs.haystack.deepset.ai/docs/chatmessage) format in input and output. You can customize how the text is generated by passing parameters to the OpenAI API. Use the `**generation_kwargs` argument when you initialize the component or when you run it. Any parameter that works with `openai.Responses.create` will work here too. For details on OpenAI API parameters, see [OpenAI documentation](https://platform.openai.com/docs/api-reference/responses). ### Usage example ```python from haystack.components.generators.chat import OpenAIResponsesChatGenerator from haystack.dataclasses import ChatMessage messages = [ChatMessage.from_user("What's Natural Language Processing?")] client = OpenAIResponsesChatGenerator(generation_kwargs={"reasoning": {"effort": "low", "summary": "auto"}}) response = client.run(messages) print(response) ``` """ SUPPORTED_MODELS: ClassVar[list[str]] = [ "gpt-5-mini", "gpt-5-nano", "gpt-5", "gpt-5.1", "gpt-5.2", "gpt-5.2-pro", "gpt-5.4", "gpt-5-pro", "gpt-4.1", "gpt-4.1-mini", "gpt-4.1-nano", "gpt-4o", "gpt-4o-mini", "o1", "o1-mini", "o1-pro", "o3", "o3-mini", "o3-pro", "o4-mini", ] """A non-exhaustive list of chat models supported by this component. See https://platform.openai.com/docs/models for the full list and snapshot IDs.""" def __init__( self, *, api_key: Secret = Secret.from_env_var("OPENAI_API_KEY"), model: str = "gpt-5-mini", streaming_callback: StreamingCallbackT | None = None, api_base_url: str | None = None, organization: str | None = None, generation_kwargs: dict[str, Any] | None = None, timeout: float | None = None, max_retries: int | None = None, tools: ToolsType | list[dict] | None = None, tools_strict: bool = False, http_client_kwargs: dict[str, Any] | None = None, ) -> None: """ Creates an instance of OpenAIResponsesChatGenerator. Uses OpenAI's gpt-5-mini by default. Before initializing the component, you can set the 'OPENAI_TIMEOUT' and 'OPENAI_MAX_RETRIES' environment variables to override the `timeout` and `max_retries` parameters respectively in the OpenAI client. :param api_key: The OpenAI API key. You can set it with an environment variable `OPENAI_API_KEY`, or pass with this parameter during initialization. :param model: The name of the model to use. :param streaming_callback: A callback function that is called when a new token is received from the stream. The callback function accepts [StreamingChunk](https://docs.haystack.deepset.ai/docs/data-classes#streamingchunk) as an argument. :param api_base_url: An optional base URL. :param organization: Your organization ID, defaults to `None`. See [production best practices](https://platform.openai.com/docs/guides/production-best-practices/setting-up-your-organization). :param generation_kwargs: Other parameters to use for the model. These parameters are sent directly to the OpenAI endpoint. See OpenAI [documentation](https://platform.openai.com/docs/api-reference/responses) for more details. Some of the supported parameters: - `temperature`: What sampling temperature to use. Higher values like 0.8 will make the output more random, while lower values like 0.2 will make it more focused and deterministic. - `top_p`: An alternative to sampling with temperature, called nucleus sampling, where the model considers the results of the tokens with top_p probability mass. For example, 0.1 means only the tokens comprising the top 10% probability mass are considered. - `previous_response_id`: The ID of the previous response. Use this to create multi-turn conversations. - `text_format`: A Pydantic model that enforces the structure of the model's response. If provided, the output will always be validated against this format (unless the model returns a tool call). For details, see the [OpenAI Structured Outputs documentation](https://platform.openai.com/docs/guides/structured-outputs). - `text`: A JSON schema that enforces the structure of the model's response. If provided, the output will always be validated against this format (unless the model returns a tool call). Notes: - Both JSON Schema and Pydantic models are supported for latest models starting from GPT-4o. - If both are provided, `text_format` takes precedence and json schema passed to `text` is ignored. - Currently, this component doesn't support streaming for structured outputs. - Older models only support basic version of structured outputs through `{"type": "json_object"}`. For detailed information on JSON mode, see the [OpenAI Structured Outputs documentation](https://platform.openai.com/docs/guides/structured-outputs#json-mode). - `reasoning`: A dictionary of parameters for reasoning. For example: - `summary`: The summary of the reasoning. - `effort`: The level of effort to put into the reasoning. Can be `low`, `medium` or `high`. - `generate_summary`: Whether to generate a summary of the reasoning. Note: OpenAI does not return the reasoning tokens, but we can view summary if its enabled. For details, see the [OpenAI Reasoning documentation](https://platform.openai.com/docs/guides/reasoning). :param timeout: Timeout for OpenAI client calls. If not set, it defaults to either the `OPENAI_TIMEOUT` environment variable, or 30 seconds. :param max_retries: Maximum number of retries to contact OpenAI after an internal error. If not set, it defaults to either the `OPENAI_MAX_RETRIES` environment variable, or set to 5. :param tools: The tools that the model can use to prepare calls. This parameter can accept either a mixed list of Haystack `Tool` objects and Haystack `Toolset`. Or you can pass a dictionary of OpenAI/MCP tool definitions. Note: You cannot pass OpenAI/MCP tools and Haystack tools together. For details on tool support, see [OpenAI documentation](https://platform.openai.com/docs/api-reference/responses/create#responses-create-tools). :param tools_strict: Whether to enable strict schema adherence for tool calls. If set to `False`, the model may not exactly follow the schema provided in the `parameters` field of the tool definition. In Response API, tool calls are strict by default. :param http_client_kwargs: A dictionary of keyword arguments to configure a custom `httpx.Client`or `httpx.AsyncClient`. For more information, see the [HTTPX documentation](https://www.python-httpx.org/api/#client). """ self.api_key = api_key self.model = model self.generation_kwargs = generation_kwargs or {} self.streaming_callback = streaming_callback self.api_base_url = api_base_url self.organization = organization self.timeout = timeout self.max_retries = max_retries self.tools = tools # Store tools as-is, whether it's a list or a Toolset self.tools_strict = tools_strict self.http_client_kwargs = http_client_kwargs self.client: OpenAI | None = None self.async_client: AsyncOpenAI | None = None self._tools_warmed_up = False def _client_kwargs(self) -> dict[str, Any]: timeout = self.timeout if self.timeout is not None else float(os.environ.get("OPENAI_TIMEOUT", "30.0")) max_retries = ( self.max_retries if self.max_retries is not None else int(os.environ.get("OPENAI_MAX_RETRIES", "5")) ) resolved_api_key = self.api_key.resolve_value() if isinstance(self.api_key, Secret) else self.api_key return { "api_key": resolved_api_key, "organization": self.organization, "base_url": self.api_base_url, "timeout": timeout, "max_retries": max_retries, } def _warm_up_tools(self) -> None: if not self._tools_warmed_up: is_openai_tool = isinstance(self.tools, list) and isinstance(self.tools[0], dict) # We only warm up Haystack tools, not OpenAI/MCP tools # The type ignore is needed because mypy cannot infer the type correctly if not is_openai_tool: warm_up_tools(self.tools) # type: ignore[arg-type] self._tools_warmed_up = True def warm_up(self) -> None: """ Warm up the tools and initialize the synchronous OpenAI client. """ self._warm_up_tools() if self.client is None: self.client = OpenAI( http_client=init_http_client(self.http_client_kwargs, async_client=False), **self._client_kwargs() ) async def warm_up_async(self) -> None: # noqa: RUF029 """ Warm up the tools and initialize the asynchronous OpenAI client on the serving event loop. """ self._warm_up_tools() if self.async_client is None: self.async_client = AsyncOpenAI( http_client=init_http_client(self.http_client_kwargs, async_client=True), **self._client_kwargs() ) def close(self) -> None: """ Releases the synchronous OpenAI client. """ if self.client is not None: self.client.close() self.client = None async def close_async(self) -> None: """ Releases the asynchronous OpenAI client. """ if self.async_client is not None: await self.async_client.close() self.async_client = None def _get_telemetry_data(self) -> dict[str, Any]: """ Data that is sent to Posthog for usage analytics. """ return {"model": self.model} def to_dict(self) -> dict[str, Any]: """ Serialize this component to a dictionary. :returns: The serialized component as a dictionary. """ callback_name = serialize_callable(self.streaming_callback) if self.streaming_callback else None generation_kwargs = self.generation_kwargs.copy() text_format = generation_kwargs.pop("text_format", None) # If the response format is a Pydantic model, it's converted to openai's json schema format # If it's already a json schema, it's left as is if text_format and isinstance(text_format, type) and issubclass(text_format, BaseModel): json_schema = { "format": { "type": "json_schema", "name": text_format.__name__, "strict": True, "schema": to_strict_json_schema(text_format), } } # json schema needs to be passed to text parameter instead of text_format generation_kwargs["text"] = json_schema # OpenAI/MCP tools are passed as list of dictionaries serialized_tools: dict[str, Any] | list[dict[str, Any]] | None if self.tools and isinstance(self.tools, list) and isinstance(self.tools[0], dict): # mypy can't infer that self.tools is list[dict] here serialized_tools = self.tools else: serialized_tools = serialize_tools_or_toolset(self.tools) # type: ignore[arg-type] return default_to_dict( self, model=self.model, streaming_callback=callback_name, api_base_url=self.api_base_url, organization=self.organization, generation_kwargs=generation_kwargs, api_key=self.api_key, timeout=self.timeout, max_retries=self.max_retries, tools=serialized_tools, tools_strict=self.tools_strict, http_client_kwargs=self.http_client_kwargs, ) @classmethod def from_dict(cls, data: dict[str, Any]) -> "OpenAIResponsesChatGenerator": """ Deserialize this component from a dictionary. :param data: The dictionary representation of this component. :returns: The deserialized component instance. """ # we only deserialize the tools if they are haystack tools # because openai tools are not serialized in the same way tools = data["init_parameters"].get("tools") if tools and ( isinstance(tools, dict) and tools.get("type") == "haystack.tools.toolset.Toolset" or isinstance(tools, list) and tools[0].get("type") == "haystack.tools.tool.Tool" ): deserialize_tools_or_toolset_inplace(data["init_parameters"], key="tools") init_params = data.get("init_parameters", {}) serialized_callback_handler = init_params.get("streaming_callback") if serialized_callback_handler: data["init_parameters"]["streaming_callback"] = deserialize_callable(serialized_callback_handler) return default_from_dict(cls, data) @component.output_types(replies=list[ChatMessage]) def run( self, messages: list[ChatMessage] | str, *, streaming_callback: StreamingCallbackT | None = None, generation_kwargs: dict[str, Any] | None = None, tools: ToolsType | list[dict] | None = None, tools_strict: bool | None = None, ) -> dict[str, list[ChatMessage]]: """ Invokes response generation based on the provided messages and generation parameters. :param messages: A list of ChatMessage instances representing the input messages. :param streaming_callback: A callback function that is called when a new token is received from the stream. :param generation_kwargs: Additional keyword arguments for text generation. These parameters will override the parameters passed during component initialization. For details on OpenAI API parameters, see [OpenAI documentation](https://platform.openai.com/docs/api-reference/responses/create). :param tools: The tools that the model can use to prepare calls. If set, it will override the `tools` parameter set during component initialization. This parameter can accept either a mixed list of Haystack `Tool` objects and Haystack `Toolset`. Or you can pass a dictionary of OpenAI/MCP tool definitions. Note: You cannot pass OpenAI/MCP tools and Haystack tools together. For details on tool support, see [OpenAI documentation](https://platform.openai.com/docs/api-reference/responses/create#responses-create-tools). :param tools_strict: Whether to enable strict schema adherence for tool calls. If set to `False`, the model may not exactly follow the schema provided in the `parameters` field of the tool definition. In Response API, tool calls are strict by default. If set, it will override the `tools_strict` parameter set during component initialization. :returns: A dictionary with the following key: - `replies`: A list containing the generated responses as ChatMessage instances. """ self.warm_up() messages = _normalize_messages(messages) if len(messages) == 0: return {"replies": []} streaming_callback = select_streaming_callback( init_callback=self.streaming_callback, runtime_callback=streaming_callback, requires_async=False ) responses: Stream[ResponseStreamEvent] | Response api_args = self._prepare_api_call( messages=messages, streaming_callback=streaming_callback, generation_kwargs=generation_kwargs, tools=tools, tools_strict=tools_strict, ) openai_endpoint = api_args.pop("openai_endpoint") assert self.client is not None # mypy: client is built by warm_up above openai_endpoint_method = getattr(self.client.responses, openai_endpoint) responses = openai_endpoint_method(**api_args) if streaming_callback is not None: response_output = self._handle_stream_response( responses, # type: ignore streaming_callback, ) else: assert isinstance(responses, Response), "Unexpected response type for non-streaming request." response_output = [_convert_response_to_chat_message(responses)] return {"replies": response_output} @component.output_types(replies=list[ChatMessage]) async def run_async( self, messages: list[ChatMessage] | str, *, streaming_callback: StreamingCallbackT | None = None, generation_kwargs: dict[str, Any] | None = None, tools: ToolsType | list[dict] | None = None, tools_strict: bool | None = None, ) -> dict[str, list[ChatMessage]]: """ Asynchronously invokes response generation based on the provided messages and generation parameters. This is the asynchronous version of the `run` method. It has the same parameters and return values but can be used with `await` in async code. :param messages: A list of ChatMessage instances representing the input messages. :param streaming_callback: A callback function that is called when a new token is received from the stream. Async callbacks are preferred; a sync callback is accepted but will run synchronously on the event loop and may block it. :param generation_kwargs: Additional keyword arguments for text generation. These parameters will override the parameters passed during component initialization. For details on OpenAI API parameters, see [OpenAI documentation](https://platform.openai.com/docs/api-reference/responses/create). :param tools: A list of tools or a Toolset for which the model can prepare calls. If set, it will override the `tools` parameter set during component initialization. This parameter can accept either a list of mixed list of Haystack `Tool` objects and Haystack `Toolset`. Or you can pass a dictionary of OpenAI/MCP tool definitions. Note: You cannot pass OpenAI/MCP tools and Haystack tools together. :param tools_strict: Whether to enable strict schema adherence for tool calls. If set to `True`, the model will follow exactly the schema provided in the `parameters` field of the tool definition, but this may increase latency. If set, it will override the `tools_strict` parameter set during component initialization. :returns: A dictionary with the following key: - `replies`: A list containing the generated responses as ChatMessage instances. """ await self.warm_up_async() messages = _normalize_messages(messages) # validate and select the streaming callback streaming_callback = select_streaming_callback( init_callback=self.streaming_callback, runtime_callback=streaming_callback, requires_async=True ) responses: AsyncStream[ResponseStreamEvent] | Response if len(messages) == 0: return {"replies": []} api_args = self._prepare_api_call( messages=messages, streaming_callback=streaming_callback, generation_kwargs=generation_kwargs, tools=tools, tools_strict=tools_strict, ) openai_endpoint = api_args.pop("openai_endpoint") assert self.async_client is not None # mypy: async_client is built by warm_up_async above openai_endpoint_method = getattr(self.async_client.responses, openai_endpoint) responses = await openai_endpoint_method(**api_args) if streaming_callback is not None: response_output = await self._handle_async_stream_response( responses, # type: ignore streaming_callback, ) else: assert isinstance(responses, Response), "Unexpected response type for non-streaming request." response_output = [_convert_response_to_chat_message(responses)] return {"replies": response_output} def _prepare_api_call( # noqa: PLR0913 self, *, messages: list[ChatMessage], streaming_callback: StreamingCallbackT | None = None, generation_kwargs: dict[str, Any] | None = None, tools: ToolsType | list[dict] | None = None, tools_strict: bool | None = None, ) -> dict[str, Any]: # update generation kwargs by merging with the generation kwargs passed to the run method generation_kwargs = {**self.generation_kwargs, **(generation_kwargs or {})} generation_kwargs = self._resolve_flattened_kwargs(generation_kwargs) # adapt ChatMessage(s) to the format expected by the OpenAI API openai_formatted_messages: list[dict[str, Any]] = [] for message in messages: openai_formatted_messages.extend(_convert_chat_message_to_responses_api_format(message)) tools = tools or self.tools tools_strict = tools_strict if tools_strict is not None else self.tools_strict openai_tools = {} # Build tool definitions if tools: tool_definitions: list[Any] = [] if isinstance(tools, list) and isinstance(tools[0], dict): # Predefined OpenAI/MCP-style tools tool_definitions = tools # Convert all tool objects to the correct OpenAI-compatible structure else: # mypy can't infer that tools is ToolsType here flattened_tools = flatten_tools_or_toolsets(tools) # type: ignore[arg-type] _check_duplicate_tool_names(flattened_tools) for t in flattened_tools: function_spec = {**t.tool_spec} if not tools_strict: function_spec["strict"] = False function_spec["parameters"]["additionalProperties"] = False tool_definitions.append({"type": "function", **function_spec}) openai_tools = {"tools": tool_definitions} base_args = {"model": self.model, "input": openai_formatted_messages, **openai_tools, **generation_kwargs} # if `text_format` is provided, we use the `parse` endpoint for response type parsing if generation_kwargs.get("text_format"): # if both `text_format` and `text` are provided, `text_format` takes precedence # and json schema passed to `text` is ignored return {**base_args, "stream": streaming_callback is not None, "openai_endpoint": "parse"} # we pass a key `openai_endpoint` as a hint to the run method to use the create or parse endpoint # this key will be removed before the API call is made return {**base_args, "stream": streaming_callback is not None, "openai_endpoint": "create"} def _resolve_flattened_kwargs(self, generation_kwargs: dict[str, Any]) -> dict[str, Any]: generation_kwargs = generation_kwargs.copy() reasoning_effort = generation_kwargs.pop("reasoning_effort", None) if reasoning_effort is not None: reasoning = generation_kwargs.setdefault("reasoning", {}) reasoning["effort"] = reasoning_effort reasoning_summary = generation_kwargs.pop("reasoning_summary", None) if reasoning_summary is not None: reasoning = generation_kwargs.setdefault("reasoning", {}) reasoning["summary"] = reasoning_summary verbosity = generation_kwargs.pop("verbosity", None) if verbosity is not None: text = generation_kwargs.setdefault("text", {}) text["verbosity"] = verbosity return generation_kwargs def _handle_stream_response(self, responses: Stream, callback: SyncStreamingCallbackT) -> list[ChatMessage]: component_info = ComponentInfo.from_component(self) chunks: list[StreamingChunk] = [] for openai_chunk in responses: chunk_delta = _convert_response_chunk_to_streaming_chunk( chunk=openai_chunk, previous_chunks=chunks, component_info=component_info ) chunks.append(chunk_delta) callback(chunk_delta) chat_message = _convert_streaming_chunks_to_chat_message(chunks=chunks) return [chat_message] async def _handle_async_stream_response( self, responses: AsyncStream, callback: StreamingCallbackT ) -> list[ChatMessage]: component_info = ComponentInfo.from_component(self) chunks: list[StreamingChunk] = [] async for openai_chunk in responses: chunk_delta = _convert_response_chunk_to_streaming_chunk( chunk=openai_chunk, previous_chunks=chunks, component_info=component_info ) chunks.append(chunk_delta) await _invoke_streaming_callback(callback, chunk_delta) chat_message = _convert_streaming_chunks_to_chat_message(chunks=chunks) return [chat_message] def _convert_response_to_chat_message(responses: Response | ParsedResponse) -> ChatMessage: """ Converts the non-streaming response from the OpenAI API to a ChatMessage. :param responses: The responses returned by the OpenAI API. :returns: The ChatMessage. """ tool_calls = [] reasoning = None logprobs: list[dict] = [] for output in responses.output: if isinstance(output, ResponseOutputRefusal): logger.warning("OpenAI returned a refusal output: {output}", output=output) continue if output.type == "message": for content in output.content: if hasattr(content, "logprobs") and content.logprobs is not None: logprobs.append(_serialize_object(content.logprobs)) if output.type == "reasoning": # openai doesn't return the reasoning tokens, but we can view summary if its enabled # https://platform.openai.com/docs/guides/reasoning#reasoning-summaries summaries = output.summary extra = output.to_dict() # we dont need the summary in the extra extra.pop("summary") if output.content: logger.warning( "OpenAI returned a non-empty 'content' field on a reasoning item ({_id}). " "The content is preserved in ReasoningContent.extra['content'] but is NOT " "reflected in ReasoningContent.reasoning_text.", _id=output.id, ) reasoning_text = "\n".join([summary.text for summary in summaries if summaries]) reasoning = ReasoningContent(reasoning_text=reasoning_text, extra=extra) elif output.type == "function_call": try: arguments = json.loads(output.arguments) tool_calls.append( ToolCall( id=output.id, tool_name=output.name, arguments=arguments, extra={"call_id": output.call_id} ) ) except json.JSONDecodeError: logger.warning( "The LLM provider returned a malformed JSON string for tool call arguments. This tool call " "will be skipped. To always generate a valid JSON, set `tools_strict` to `True`. " "Tool call ID: {_id}, Tool name: {_name}, Arguments: {_arguments}", _id=output.id, _name=output.name, _arguments=output.arguments, ) arguments = {} # we save the response as dict because it contains resp_id etc. meta = responses.to_dict() # remove output from meta because it contains toolcalls, reasoning, text etc. meta.pop("output") if logprobs: meta["logprobs"] = logprobs return ChatMessage.from_assistant( text=responses.output_text if responses.output_text else None, reasoning=reasoning, tool_calls=tool_calls, meta=meta, ) def _convert_response_chunk_to_streaming_chunk( # noqa: PLR0911 chunk: ResponseStreamEvent, previous_chunks: list[StreamingChunk], component_info: ComponentInfo | None = None ) -> StreamingChunk: """ Converts the streaming response chunk from the OpenAI Responses API to a StreamingChunk. :param chunk: The chunk returned by the OpenAI Responses API. :param previous_chunks: A list of previously received StreamingChunks. :param component_info: An optional `ComponentInfo` object containing information about the component that generated the chunk, such as the component name and type. :returns: A StreamingChunk object representing the content of the chunk from the OpenAI Responses API. """ if chunk.type == "response.output_item.added": # Responses API always returns reasoning chunks even if there is no summary if chunk.item.type == "reasoning": reasoning = ReasoningContent(reasoning_text="", extra=chunk.item.to_dict()) return StreamingChunk( content="", component_info=component_info, index=chunk.output_index, reasoning=reasoning, start=True, meta={"received_at": datetime.now().isoformat()}, ) # the function name is only streamed at the start and end of the function call if chunk.item.type == "function_call": tool_call = ToolCallDelta( index=chunk.output_index, id=chunk.item.id, tool_name=chunk.item.name, extra=chunk.item.to_dict() ) return StreamingChunk( content="", component_info=component_info, index=chunk.output_index, tool_calls=[tool_call], start=True, meta={"received_at": datetime.now().isoformat()}, ) elif chunk.type == "response.output_item.done": # The done event carries the completed reasoning item, which includes encrypted_content # when include=["reasoning.encrypted_content"] was requested. Without this handler the # event falls through to the generic default and reasoning=None, so encrypted_content # is never available for multi-turn conversations. if chunk.item.type == "reasoning": if chunk.item.content: logger.warning( "OpenAI returned a non-empty 'content' field on a reasoning item ({_id}). " "This field is currently undocumented and was never observed in practice. " "The content is preserved in ReasoningContent.extra['content'] but is NOT " "reflected in ReasoningContent.reasoning_text. Please report this at " "https://github.com/deepset-ai/haystack/issues so we can update the mapping.", _id=chunk.item.id, ) reasoning = ReasoningContent(reasoning_text="", extra=chunk.item.to_dict()) return StreamingChunk( content="", component_info=component_info, index=chunk.output_index, reasoning=reasoning, meta={"received_at": datetime.now().isoformat()}, ) elif chunk.type == "response.completed": # This means a full response is finished # If there are tool_calls present in the final output we mark finish_reason as tool_calls otherwise it's # marked as stop return StreamingChunk( content="", component_info=component_info, finish_reason="tool_calls" if any(o.type == "function_call" for o in chunk.response.output) else "stop", meta={**chunk.to_dict(), "received_at": datetime.now().isoformat()}, ) elif chunk.type == "response.output_text.delta": # Start is determined by checking if this is the first text delta event of a new output_index # 1) Check if all previous chunks have different output_index # 2) If any chunks do have the same output_index, check if they have content # If none of them have content, this is the start of a new text output start = all(c.index != chunk.output_index for c in previous_chunks) or all( c.content == "" for c in previous_chunks if c.index == chunk.output_index ) return StreamingChunk( content=chunk.delta, component_info=component_info, index=chunk.output_index, start=start, meta={**chunk.to_dict(), "received_at": datetime.now().isoformat()}, ) elif chunk.type == "response.reasoning_summary_text.delta": # We remove the delta from the extra because it is already in the reasoning_text # Remaining information needs to be saved for chat message extra = chunk.to_dict() extra.pop("delta") reasoning = ReasoningContent(reasoning_text=chunk.delta, extra=extra) return StreamingChunk( content="", component_info=component_info, index=chunk.output_index, reasoning=reasoning, meta={"received_at": datetime.now().isoformat()}, ) # the function arguments are streamed in parts # function name is not passed in these chunks elif chunk.type == "response.function_call_arguments.delta": arguments = chunk.delta extra = chunk.to_dict() extra.pop("delta") # in delta of tool calls there is no call_id so we use the item_id which is the function call id tool_call = ToolCallDelta(index=chunk.output_index, id=chunk.item_id, arguments=arguments, extra=extra) return StreamingChunk( content="", component_info=component_info, index=chunk.output_index, tool_calls=[tool_call], meta={"received_at": datetime.now().isoformat()}, ) # we return rest of the chunk as is return StreamingChunk( content="", component_info=component_info, index=getattr(chunk, "output_index", None), meta={**chunk.to_dict(), "received_at": datetime.now().isoformat()}, ) def _convert_streaming_chunks_to_chat_message(chunks: list[StreamingChunk]) -> ChatMessage: """ Connects the streaming chunks into a single ChatMessage. :param chunks: The list of all `StreamingChunk` objects. :returns: The ChatMessage. """ # Get the full text by concatenating all text chunks text = "".join([chunk.content for chunk in chunks]) logprobs = [] for chunk in chunks: if chunk.meta.get("logprobs"): logprobs.append(chunk.meta.get("logprobs")) # Gather reasoning information if present reasoning_id = None reasoning_text = "" for chunk in chunks: if chunk.reasoning: reasoning_text += chunk.reasoning.reasoning_text if chunk.reasoning.extra.get("id"): reasoning_id = chunk.reasoning.extra.get("id") # Process tool calls if present in any chunk tool_call_data: dict[str, dict[str, Any]] = {} # Track tool calls by id for chunk in chunks: if chunk.tool_calls: for tool_call in chunk.tool_calls: # here the tool_call.id is fc_id not call_id assert tool_call.id is not None # We use the tool call id to track the tool call across chunks if tool_call.id not in tool_call_data: tool_call_data[tool_call.id] = {"name": "", "arguments": ""} if tool_call.arguments is not None: tool_call_data[tool_call.id]["arguments"] += tool_call.arguments # We capture the tool name from one of the chunks if tool_call.tool_name is not None: tool_call_data[tool_call.id]["name"] = tool_call.tool_name # We capture the call_id from one of the chunks if tool_call.extra and "call_id" in tool_call.extra: tool_call_data[tool_call.id]["extra"] = {"call_id": tool_call.extra["call_id"]} # Convert accumulated tool call data into ToolCall objects tool_calls = [] sorted_keys = sorted(tool_call_data.keys()) for key in sorted_keys: tool_call_dict = tool_call_data[key] try: arguments = json.loads(tool_call_dict.get("arguments", "{}")) if tool_call_dict.get("arguments") else {} extra: dict[str, Any] = tool_call_dict.get("extra", {}) tool_calls.append(ToolCall(id=key, tool_name=tool_call_dict["name"], arguments=arguments, extra=extra)) except json.JSONDecodeError: logger.warning( "The LLM provider returned a malformed JSON string for tool call arguments. This tool call " "will be skipped. To always generate a valid JSON, set `tools_strict` to `True`. " "Tool call ID: {_id}, Tool name: {_name}, Arguments: {_arguments}", _id=key, _name=tool_call_dict["name"], _arguments=tool_call_dict["arguments"], ) # We dump the entire final response into meta to be consistent with non-streaming response final_response = chunks[-1].meta.get("response") or {} final_response.pop("output", None) if logprobs: final_response["logprobs"] = logprobs # Add reasoning content if id is available # Note: the API expects a reasoning id even if there is no reasoning text # function calls without reasoning ids are not supported by the API reasoning = None if reasoning_id: # Preserve all extra fields from streaming chunks (e.g. encrypted_content) while ensuring id and # type are present reasoning_extra = {} for chunk in chunks: if chunk.reasoning and chunk.reasoning.extra: reasoning_extra.update(chunk.reasoning.extra) # Ensure id and type are always set, but don't override if already present reasoning_extra.setdefault("id", reasoning_id) reasoning_extra.setdefault("type", "reasoning") reasoning = ReasoningContent(reasoning_text=reasoning_text, extra=reasoning_extra) return ChatMessage.from_assistant( text=text or None, tool_calls=tool_calls, meta=final_response, reasoning=reasoning ) def _convert_chat_message_to_responses_api_format(message: ChatMessage) -> list[dict[str, Any]]: """ Convert a ChatMessage to the dictionary format expected by OpenAI's Responses API. :param message: The ChatMessage to convert to OpenAI's Responses API format. :returns: The ChatMessage in the format expected by OpenAI's Responses API. :raises ValueError: If the message format is invalid. """ def convert_part(part: Any) -> dict[str, str | None]: if isinstance(part, TextContent): return {"type": "input_text", "text": part.text} if isinstance(part, ImageContent): return { "type": "input_image", # If no MIME type is provided, default to JPEG. OpenAI API appears to tolerate MIME type mismatches. "image_url": f"data:{part.mime_type or 'image/jpeg'};base64,{part.base64_image}", } if isinstance(part, FileContent): return { "type": "input_file", # Filename is optional but if not provided, OpenAI expects a file_id of a previous file upload. # We use a dummy filename to avoid this issue. "filename": part.filename or "filename", "file_data": f"data:{part.mime_type or 'application/pdf'};base64,{part.base64_data}", } raise ValueError(f"Unsupported content type: {type(part)}") text_contents = message.texts tool_calls = message.tool_calls tool_call_results = message.tool_call_results images = message.images reasonings = message.reasonings files = message.files if not any([text_contents, tool_calls, tool_call_results, images, reasonings, files]): raise ValueError( """A `ChatMessage` must contain at least one `TextContent`, `ToolCall`, `ToolCallResult`, `ImageContent`, `FileContent`, or `ReasoningContent`.""" ) if len(tool_call_results) > 0 and len(message._content) > 1: raise ValueError( "For OpenAI compatibility, a `ChatMessage` with a `ToolCallResult` cannot contain any other content." ) formatted_messages: list[dict[str, Any]] = [] openai_msg: dict[str, Any] = {"role": message._role.value} if message._name is not None: openai_msg["name"] = message._name # user message if message._role.value == "user": content = [convert_part(part) for part in message._content] openai_msg["content"] = content return [openai_msg] # tool message if tool_call_results: formatted_tool_results = [] for result in tool_call_results: if result.origin.id is not None: # Handle multimodal tool results (list of TextContent/ImageContent/FileContent) if isinstance(result.result, list): output_content = [convert_part(part) for part in result.result] elif isinstance(result.result, str): output_content = [{"type": "input_text", "text": result.result}] else: raise ValueError(f"Unsupported tool result: {result.result}") tool_result = { "type": "function_call_output", "call_id": result.origin.extra.get("call_id") if result.origin.extra else "", "output": output_content, } formatted_tool_results.append(tool_result) formatted_messages.extend(formatted_tool_results) # Note: the API expects a reasoning id even if there is no reasoning text # function calls without reasoning ids are not supported by the API if reasonings: formatted_reasonings = [] for reasoning in reasonings: # Streaming events (e.g. response.reasoning_summary_text.delta) store event-level # fields like item_id, output_index, summary_index, event_id, sequence_number into # reasoning.extra. Those are not valid reasoning input item fields and the API # rejects them with "Unknown parameter" when sent back in subsequent turns. # Valid fields per ResponseReasoningItem schema: id, type, summary (handled separately), # content, encrypted_content, status. _valid_reasoning_fields = {"id", "type", "encrypted_content", "status", "content"} filtered_extra = {k: v for k, v in reasoning.extra.items() if k in _valid_reasoning_fields} reasoning_item = {"summary": [], **filtered_extra} if reasoning.reasoning_text: reasoning_item["summary"] = [{"text": reasoning.reasoning_text, "type": "summary_text"}] formatted_reasonings.append(reasoning_item) formatted_messages.extend(formatted_reasonings) if tool_calls: formatted_tool_calls = [] for tc in tool_calls: openai_tool_call = { "type": "function_call", # We disable ensure_ascii so special chars like emojis are not converted "name": tc.tool_name, "arguments": json.dumps(tc.arguments, ensure_ascii=False), "id": tc.id, "call_id": tc.extra.get("call_id") if tc.extra else "", } formatted_tool_calls.append(openai_tool_call) formatted_messages.extend(formatted_tool_calls) # system and assistant messages if text_contents: openai_msg["content"] = " ".join(text_contents) formatted_messages.append(openai_msg) return formatted_messages