Coverage for haystack/components/generators/chat/openai_responses.py: 91%
360 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-21 13:53 +0000
1# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
2#
3# SPDX-License-Identifier: Apache-2.0
5import json
6import os
7from datetime import datetime
8from typing import Any, ClassVar
10from openai import AsyncOpenAI, AsyncStream, OpenAI, Stream
11from openai.lib._pydantic import to_strict_json_schema
12from openai.types.responses import ParsedResponse, Response, ResponseOutputRefusal, ResponseStreamEvent
13from pydantic import BaseModel
15from haystack import component, default_from_dict, default_to_dict, logging
16from haystack.components.generators.utils import _normalize_messages, _serialize_object
17from haystack.dataclasses import (
18 ChatMessage,
19 ComponentInfo,
20 FileContent,
21 ImageContent,
22 ReasoningContent,
23 StreamingCallbackT,
24 StreamingChunk,
25 SyncStreamingCallbackT,
26 TextContent,
27 ToolCall,
28 ToolCallDelta,
29 select_streaming_callback,
30)
31from haystack.dataclasses.streaming_chunk import _invoke_streaming_callback
32from haystack.tools import (
33 ToolsType,
34 _check_duplicate_tool_names,
35 deserialize_tools_or_toolset_inplace,
36 flatten_tools_or_toolsets,
37 serialize_tools_or_toolset,
38 warm_up_tools,
39)
40from haystack.utils import Secret, deserialize_callable, serialize_callable
41from haystack.utils.http_client import init_http_client
43logger = logging.getLogger(__name__)
46@component
47class OpenAIResponsesChatGenerator:
48 """
49 Completes chats using OpenAI's Responses API.
51 It works with the gpt-4 and o-series models and supports streaming responses
52 from OpenAI API. It uses [ChatMessage](https://docs.haystack.deepset.ai/docs/chatmessage)
53 format in input and output.
55 You can customize how the text is generated by passing parameters to the
56 OpenAI API. Use the `**generation_kwargs` argument when you initialize
57 the component or when you run it. Any parameter that works with
58 `openai.Responses.create` will work here too.
60 For details on OpenAI API parameters, see
61 [OpenAI documentation](https://platform.openai.com/docs/api-reference/responses).
63 ### Usage example
65 ```python
66 from haystack.components.generators.chat import OpenAIResponsesChatGenerator
67 from haystack.dataclasses import ChatMessage
69 messages = [ChatMessage.from_user("What's Natural Language Processing?")]
71 client = OpenAIResponsesChatGenerator(generation_kwargs={"reasoning": {"effort": "low", "summary": "auto"}})
72 response = client.run(messages)
73 print(response)
74 ```
75 """
77 SUPPORTED_MODELS: ClassVar[list[str]] = [
78 "gpt-5-mini",
79 "gpt-5-nano",
80 "gpt-5",
81 "gpt-5.1",
82 "gpt-5.2",
83 "gpt-5.2-pro",
84 "gpt-5.4",
85 "gpt-5-pro",
86 "gpt-4.1",
87 "gpt-4.1-mini",
88 "gpt-4.1-nano",
89 "gpt-4o",
90 "gpt-4o-mini",
91 "o1",
92 "o1-mini",
93 "o1-pro",
94 "o3",
95 "o3-mini",
96 "o3-pro",
97 "o4-mini",
98 ]
99 """A non-exhaustive list of chat models supported by this component.
100 See https://platform.openai.com/docs/models for the full list and snapshot IDs."""
102 def __init__(
103 self,
104 *,
105 api_key: Secret = Secret.from_env_var("OPENAI_API_KEY"),
106 model: str = "gpt-5-mini",
107 streaming_callback: StreamingCallbackT | None = None,
108 api_base_url: str | None = None,
109 organization: str | None = None,
110 generation_kwargs: dict[str, Any] | None = None,
111 timeout: float | None = None,
112 max_retries: int | None = None,
113 tools: ToolsType | list[dict] | None = None,
114 tools_strict: bool = False,
115 http_client_kwargs: dict[str, Any] | None = None,
116 ) -> None:
117 """
118 Creates an instance of OpenAIResponsesChatGenerator. Uses OpenAI's gpt-5-mini by default.
120 Before initializing the component, you can set the 'OPENAI_TIMEOUT' and 'OPENAI_MAX_RETRIES'
121 environment variables to override the `timeout` and `max_retries` parameters respectively
122 in the OpenAI client.
124 :param api_key: The OpenAI API key.
125 You can set it with an environment variable `OPENAI_API_KEY`, or pass with this parameter
126 during initialization.
127 :param model: The name of the model to use.
128 :param streaming_callback: A callback function that is called when a new token is received from the stream.
129 The callback function accepts [StreamingChunk](https://docs.haystack.deepset.ai/docs/data-classes#streamingchunk)
130 as an argument.
131 :param api_base_url: An optional base URL.
132 :param organization: Your organization ID, defaults to `None`. See
133 [production best practices](https://platform.openai.com/docs/guides/production-best-practices/setting-up-your-organization).
134 :param generation_kwargs: Other parameters to use for the model. These parameters are sent
135 directly to the OpenAI endpoint.
136 See OpenAI [documentation](https://platform.openai.com/docs/api-reference/responses) for
137 more details.
138 Some of the supported parameters:
139 - `temperature`: What sampling temperature to use. Higher values like 0.8 will make the output more random,
140 while lower values like 0.2 will make it more focused and deterministic.
141 - `top_p`: An alternative to sampling with temperature, called nucleus sampling, where the model
142 considers the results of the tokens with top_p probability mass. For example, 0.1 means only the tokens
143 comprising the top 10% probability mass are considered.
144 - `previous_response_id`: The ID of the previous response.
145 Use this to create multi-turn conversations.
146 - `text_format`: A Pydantic model that enforces the structure of the model's response.
147 If provided, the output will always be validated against this
148 format (unless the model returns a tool call).
149 For details, see the [OpenAI Structured Outputs documentation](https://platform.openai.com/docs/guides/structured-outputs).
150 - `text`: A JSON schema that enforces the structure of the model's response.
151 If provided, the output will always be validated against this
152 format (unless the model returns a tool call).
153 Notes:
154 - Both JSON Schema and Pydantic models are supported for latest models starting from GPT-4o.
155 - If both are provided, `text_format` takes precedence and json schema passed to `text` is ignored.
156 - Currently, this component doesn't support streaming for structured outputs.
157 - Older models only support basic version of structured outputs through `{"type": "json_object"}`.
158 For detailed information on JSON mode, see the [OpenAI Structured Outputs documentation](https://platform.openai.com/docs/guides/structured-outputs#json-mode).
159 - `reasoning`: A dictionary of parameters for reasoning. For example:
160 - `summary`: The summary of the reasoning.
161 - `effort`: The level of effort to put into the reasoning. Can be `low`, `medium` or `high`.
162 - `generate_summary`: Whether to generate a summary of the reasoning.
163 - `mode`: The reasoning mode. Can be `standard`, or `pro`. Supported since GPT-5.6.
164 Note: OpenAI does not return the reasoning tokens, but we can view summary if its enabled.
165 For details, see the [OpenAI Reasoning documentation](https://platform.openai.com/docs/guides/reasoning).
166 - `include`: Specify additional output data to include in the model response. Supported values are:
167 - web_search_call.action.sources: Include the sources of the web search tool call.
168 - code_interpreter_call.outputs: Includes the outputs of python code execution in code interpreter tool
169 call items.
170 - computer_call_output.output.image_url: Include image urls from the computer call output.
171 - file_search_call.results: Include the search results of the file search tool call.
172 - message.input_image.image_url: Include image urls from the input message.
173 - message.output_text.logprobs: Include logprobs with assistant messages.
174 - reasoning.encrypted_content: Includes an encrypted version of reasoning tokens in reasoning item
175 outputs. This enables reasoning items to be used in multi-turn conversations when using the
176 Responses API statelessly (like when the store parameter is set to false, or when an organization
177 is enrolled in the zero data retention program).
178 :param timeout:
179 Timeout for OpenAI client calls. If not set, it defaults to either the
180 `OPENAI_TIMEOUT` environment variable, or 30 seconds.
181 :param max_retries:
182 Maximum number of retries to contact OpenAI after an internal error.
183 If not set, it defaults to either the `OPENAI_MAX_RETRIES` environment variable, or set to 5.
184 :param tools:
185 The tools that the model can use to prepare calls. This parameter can accept either a
186 mixed list of Haystack `Tool` objects and Haystack `Toolset`. Or you can pass a dictionary of
187 OpenAI/MCP tool definitions.
188 Note: You cannot pass OpenAI/MCP tools and Haystack tools together.
189 For details on tool support, see [OpenAI documentation](https://platform.openai.com/docs/api-reference/responses/create#responses-create-tools).
190 :param tools_strict:
191 Whether to enable strict schema adherence for tool calls. If set to `False`, the model may not exactly
192 follow the schema provided in the `parameters` field of the tool definition. In Response API, tool calls
193 are strict by default.
194 :param http_client_kwargs:
195 A dictionary of keyword arguments to configure a custom `httpx.Client`or `httpx.AsyncClient`.
196 For more information, see the [HTTPX documentation](https://www.python-httpx.org/api/#client).
198 """
199 self.api_key = api_key
200 self.model = model
201 self.generation_kwargs = generation_kwargs or {}
202 self.streaming_callback = streaming_callback
203 self.api_base_url = api_base_url
204 self.organization = organization
205 self.timeout = timeout
206 self.max_retries = max_retries
207 self.tools = tools # Store tools as-is, whether it's a list or a Toolset
208 self.tools_strict = tools_strict
209 self.http_client_kwargs = http_client_kwargs
211 self.client: OpenAI | None = None
212 self.async_client: AsyncOpenAI | None = None
213 self._tools_warmed_up = False
215 def _client_kwargs(self) -> dict[str, Any]:
216 timeout = self.timeout if self.timeout is not None else float(os.environ.get("OPENAI_TIMEOUT", "30.0"))
217 max_retries = (
218 self.max_retries if self.max_retries is not None else int(os.environ.get("OPENAI_MAX_RETRIES", "5"))
219 )
220 resolved_api_key = self.api_key.resolve_value() if isinstance(self.api_key, Secret) else self.api_key
221 return {
222 "api_key": resolved_api_key,
223 "organization": self.organization,
224 "base_url": self.api_base_url,
225 "timeout": timeout,
226 "max_retries": max_retries,
227 }
229 def _warm_up_tools(self) -> None:
230 if not self._tools_warmed_up:
231 is_openai_tool = isinstance(self.tools, list) and bool(self.tools) and isinstance(self.tools[0], dict)
232 # We only warm up Haystack tools, not OpenAI/MCP tools
233 # The type ignore is needed because mypy cannot infer the type correctly
234 if not is_openai_tool:
235 warm_up_tools(self.tools) # type: ignore[arg-type]
236 self._tools_warmed_up = True
238 def warm_up(self) -> None:
239 """
240 Warm up the tools and initialize the synchronous OpenAI client.
241 """
242 self._warm_up_tools()
243 if self.client is None:
244 # openai>=3 annotates http_client as httpx2, but legacy httpx clients are supported at runtime.
245 # https://github.com/openai/openai-python/blob/main/httpx2.md
246 http_client = init_http_client(self.http_client_kwargs, async_client=False)
247 self.client = OpenAI(
248 http_client=http_client, # type: ignore[arg-type]
249 **self._client_kwargs(),
250 )
252 async def warm_up_async(self) -> None: # noqa: RUF029
253 """
254 Warm up the tools and initialize the asynchronous OpenAI client on the serving event loop.
255 """
256 self._warm_up_tools()
257 if self.async_client is None:
258 # openai>=3 annotates http_client as httpx2, but legacy httpx clients are supported at runtime.
259 # https://github.com/openai/openai-python/blob/main/httpx2.md
260 http_client = init_http_client(self.http_client_kwargs, async_client=True)
261 self.async_client = AsyncOpenAI(
262 http_client=http_client, # type: ignore[arg-type]
263 **self._client_kwargs(),
264 )
266 def close(self) -> None:
267 """
268 Releases the synchronous OpenAI client.
269 """
270 if self.client is not None:
271 self.client.close()
272 self.client = None
274 async def close_async(self) -> None:
275 """
276 Releases the asynchronous OpenAI client.
277 """
278 if self.async_client is not None:
279 await self.async_client.close()
280 self.async_client = None
282 def _get_telemetry_data(self) -> dict[str, Any]:
283 """
284 Data that is sent to Posthog for usage analytics.
285 """
286 return {"model": self.model}
288 def to_dict(self) -> dict[str, Any]:
289 """
290 Serialize this component to a dictionary.
292 :returns:
293 The serialized component as a dictionary.
294 """
295 callback_name = serialize_callable(self.streaming_callback) if self.streaming_callback else None
296 generation_kwargs = self.generation_kwargs.copy()
297 text_format = generation_kwargs.pop("text_format", None)
299 # If the response format is a Pydantic model, it's converted to openai's json schema format
300 # If it's already a json schema, it's left as is
301 if text_format and isinstance(text_format, type) and issubclass(text_format, BaseModel):
302 json_schema = {
303 "format": {
304 "type": "json_schema",
305 "name": text_format.__name__,
306 "strict": True,
307 "schema": to_strict_json_schema(text_format),
308 }
309 }
310 # json schema needs to be passed to text parameter instead of text_format
311 generation_kwargs["text"] = json_schema
313 # OpenAI/MCP tools are passed as list of dictionaries
314 serialized_tools: dict[str, Any] | list[dict[str, Any]] | None
315 if self.tools and isinstance(self.tools, list) and isinstance(self.tools[0], dict):
316 # mypy can't infer that self.tools is list[dict] here
317 serialized_tools = self.tools
318 else:
319 serialized_tools = serialize_tools_or_toolset(self.tools) # type: ignore[arg-type]
321 return default_to_dict(
322 self,
323 model=self.model,
324 streaming_callback=callback_name,
325 api_base_url=self.api_base_url,
326 organization=self.organization,
327 generation_kwargs=generation_kwargs,
328 api_key=self.api_key,
329 timeout=self.timeout,
330 max_retries=self.max_retries,
331 tools=serialized_tools,
332 tools_strict=self.tools_strict,
333 http_client_kwargs=self.http_client_kwargs,
334 )
336 @classmethod
337 def from_dict(cls, data: dict[str, Any]) -> "OpenAIResponsesChatGenerator":
338 """
339 Deserialize this component from a dictionary.
341 :param data: The dictionary representation of this component.
342 :returns:
343 The deserialized component instance.
344 """
345 # we only deserialize the tools if they are haystack tools
346 # because openai tools are not serialized in the same way
347 tools = data["init_parameters"].get("tools")
348 if tools and (
349 isinstance(tools, dict)
350 and tools.get("type") == "haystack.tools.toolset.Toolset"
351 or isinstance(tools, list)
352 and tools[0].get("type") == "haystack.tools.tool.Tool"
353 ):
354 deserialize_tools_or_toolset_inplace(data["init_parameters"], key="tools")
356 init_params = data.get("init_parameters", {})
357 serialized_callback_handler = init_params.get("streaming_callback")
359 if serialized_callback_handler:
360 data["init_parameters"]["streaming_callback"] = deserialize_callable(serialized_callback_handler)
361 return default_from_dict(cls, data)
363 @component.output_types(replies=list[ChatMessage])
364 def run(
365 self,
366 messages: list[ChatMessage] | str,
367 *,
368 streaming_callback: StreamingCallbackT | None = None,
369 generation_kwargs: dict[str, Any] | None = None,
370 tools: ToolsType | list[dict] | None = None,
371 tools_strict: bool | None = None,
372 ) -> dict[str, list[ChatMessage]]:
373 """
374 Invokes response generation based on the provided messages and generation parameters.
376 :param messages:
377 A list of ChatMessage instances representing the input messages.
378 :param streaming_callback:
379 A callback function that is called when a new token is received from the stream.
380 :param generation_kwargs:
381 Additional keyword arguments for text generation. These are merged per key with the
382 `generation_kwargs` passed at initialization: keys provided here take precedence, keys set
383 only at initialization are kept.
384 For details on OpenAI API parameters, see [OpenAI documentation](https://platform.openai.com/docs/api-reference/responses/create).
385 :param tools:
386 The tools that the model can use to prepare calls. If set, it will override the
387 `tools` parameter set during component initialization. This parameter can accept either a
388 mixed list of Haystack `Tool` objects and Haystack `Toolset`. Or you can pass a dictionary of
389 OpenAI/MCP tool definitions.
390 Note: You cannot pass OpenAI/MCP tools and Haystack tools together.
391 For details on tool support, see [OpenAI documentation](https://platform.openai.com/docs/api-reference/responses/create#responses-create-tools).
392 :param tools_strict:
393 Whether to enable strict schema adherence for tool calls. If set to `False`, the model may not exactly
394 follow the schema provided in the `parameters` field of the tool definition. In Response API, tool calls
395 are strict by default.
396 If set, it will override the `tools_strict` parameter set during component initialization.
398 :returns:
399 A dictionary with the following key:
400 - `replies`: A list containing the generated responses as ChatMessage instances.
401 """
402 self.warm_up()
404 messages = _normalize_messages(messages)
406 if len(messages) == 0:
407 return {"replies": []}
409 streaming_callback = select_streaming_callback(
410 init_callback=self.streaming_callback, runtime_callback=streaming_callback, requires_async=False
411 )
412 responses: Stream[ResponseStreamEvent] | Response
414 api_args = self._prepare_api_call(
415 messages=messages,
416 streaming_callback=streaming_callback,
417 generation_kwargs=generation_kwargs,
418 tools=tools,
419 tools_strict=tools_strict,
420 )
421 openai_endpoint = api_args.pop("openai_endpoint")
422 assert self.client is not None # mypy: client is built by warm_up above
423 openai_endpoint_method = getattr(self.client.responses, openai_endpoint)
424 responses = openai_endpoint_method(**api_args)
426 if streaming_callback is not None:
427 response_output = self._handle_stream_response(
428 responses, # type: ignore
429 streaming_callback,
430 )
431 else:
432 assert isinstance(responses, Response), "Unexpected response type for non-streaming request."
433 response_output = [_convert_response_to_chat_message(responses)]
434 return {"replies": response_output}
436 @component.output_types(replies=list[ChatMessage])
437 async def run_async(
438 self,
439 messages: list[ChatMessage] | str,
440 *,
441 streaming_callback: StreamingCallbackT | None = None,
442 generation_kwargs: dict[str, Any] | None = None,
443 tools: ToolsType | list[dict] | None = None,
444 tools_strict: bool | None = None,
445 ) -> dict[str, list[ChatMessage]]:
446 """
447 Asynchronously invokes response generation based on the provided messages and generation parameters.
449 This is the asynchronous version of the `run` method. It has the same parameters and return values
450 but can be used with `await` in async code.
452 :param messages:
453 A list of ChatMessage instances representing the input messages.
454 :param streaming_callback:
455 A callback function that is called when a new token is received from the stream. Async callbacks are
456 preferred; a sync callback is accepted but will run synchronously on the event loop and may block it.
457 :param generation_kwargs:
458 Additional keyword arguments for text generation. These are merged per key with the
459 `generation_kwargs` passed at initialization: keys provided here take precedence, keys set
460 only at initialization are kept.
461 For details on OpenAI API parameters, see [OpenAI documentation](https://platform.openai.com/docs/api-reference/responses/create).
462 :param tools:
463 A list of tools or a Toolset for which the model can prepare calls. If set, it will override the
464 `tools` parameter set during component initialization. This parameter can accept either a list of
465 mixed list of Haystack `Tool` objects and Haystack `Toolset`. Or you can pass a dictionary of
466 OpenAI/MCP tool definitions.
467 Note: You cannot pass OpenAI/MCP tools and Haystack tools together.
468 :param tools_strict:
469 Whether to enable strict schema adherence for tool calls. If set to `True`, the model will follow exactly
470 the schema provided in the `parameters` field of the tool definition, but this may increase latency.
471 If set, it will override the `tools_strict` parameter set during component initialization.
473 :returns:
474 A dictionary with the following key:
475 - `replies`: A list containing the generated responses as ChatMessage instances.
476 """
477 await self.warm_up_async()
479 messages = _normalize_messages(messages)
481 # validate and select the streaming callback
482 streaming_callback = select_streaming_callback(
483 init_callback=self.streaming_callback, runtime_callback=streaming_callback, requires_async=True
484 )
485 responses: AsyncStream[ResponseStreamEvent] | Response
487 if len(messages) == 0:
488 return {"replies": []}
490 api_args = self._prepare_api_call(
491 messages=messages,
492 streaming_callback=streaming_callback,
493 generation_kwargs=generation_kwargs,
494 tools=tools,
495 tools_strict=tools_strict,
496 )
498 openai_endpoint = api_args.pop("openai_endpoint")
499 assert self.async_client is not None # mypy: async_client is built by warm_up_async above
500 openai_endpoint_method = getattr(self.async_client.responses, openai_endpoint)
501 responses = await openai_endpoint_method(**api_args)
503 if streaming_callback is not None:
504 response_output = await self._handle_async_stream_response(
505 responses, # type: ignore
506 streaming_callback,
507 )
509 else:
510 assert isinstance(responses, Response), "Unexpected response type for non-streaming request."
511 response_output = [_convert_response_to_chat_message(responses)]
512 return {"replies": response_output}
514 def _prepare_api_call( # noqa: PLR0913
515 self,
516 *,
517 messages: list[ChatMessage],
518 streaming_callback: StreamingCallbackT | None = None,
519 generation_kwargs: dict[str, Any] | None = None,
520 tools: ToolsType | list[dict] | None = None,
521 tools_strict: bool | None = None,
522 ) -> dict[str, Any]:
523 # update generation kwargs by merging with the generation kwargs passed to the run method
524 generation_kwargs = {**self.generation_kwargs, **(generation_kwargs or {})}
525 generation_kwargs = self._resolve_flattened_kwargs(generation_kwargs)
527 # adapt ChatMessage(s) to the format expected by the OpenAI API
528 openai_formatted_messages: list[dict[str, Any]] = []
529 for message in messages:
530 openai_formatted_messages.extend(_convert_chat_message_to_responses_api_format(message))
532 tools = tools or self.tools
533 tools_strict = tools_strict if tools_strict is not None else self.tools_strict
535 openai_tools = {}
536 # Build tool definitions
537 if tools:
538 tool_definitions: list[Any] = []
539 if isinstance(tools, list) and isinstance(tools[0], dict):
540 # Predefined OpenAI/MCP-style tools
541 tool_definitions = tools
543 # Convert all tool objects to the correct OpenAI-compatible structure
544 else:
545 # mypy can't infer that tools is ToolsType here
546 flattened_tools = flatten_tools_or_toolsets(tools) # type: ignore[arg-type]
547 _check_duplicate_tool_names(flattened_tools)
548 for t in flattened_tools:
549 function_spec = {**t.tool_spec}
550 if not tools_strict:
551 function_spec["strict"] = False
552 # Copy the parameters schema before editing it. tool_spec exposes
553 # Tool.parameters by reference, so mutating it here would permanently alter
554 # the user's Tool (and any other generator that shares the same Tool instance).
555 function_spec["parameters"] = {**function_spec["parameters"], "additionalProperties": False}
556 tool_definitions.append({"type": "function", **function_spec})
558 openai_tools = {"tools": tool_definitions}
560 base_args = {"model": self.model, "input": openai_formatted_messages, **openai_tools, **generation_kwargs}
562 # if `text_format` is provided, we use the `parse` endpoint for response type parsing
563 if generation_kwargs.get("text_format"):
564 # if both `text_format` and `text` are provided, `text_format` takes precedence
565 # and json schema passed to `text` is ignored
566 return {**base_args, "stream": streaming_callback is not None, "openai_endpoint": "parse"}
567 # we pass a key `openai_endpoint` as a hint to the run method to use the create or parse endpoint
568 # this key will be removed before the API call is made
569 return {**base_args, "stream": streaming_callback is not None, "openai_endpoint": "create"}
571 def _resolve_flattened_kwargs(self, generation_kwargs: dict[str, Any]) -> dict[str, Any]:
572 generation_kwargs = generation_kwargs.copy()
574 # avoid mutating the caller's dict
575 reasoning_overrides = {}
576 reasoning_effort = generation_kwargs.pop("reasoning_effort", None)
577 if reasoning_effort is not None:
578 reasoning_overrides["effort"] = reasoning_effort
580 reasoning_summary = generation_kwargs.pop("reasoning_summary", None)
581 if reasoning_summary is not None:
582 reasoning_overrides["summary"] = reasoning_summary
584 reasoning_mode = generation_kwargs.pop("reasoning_mode", None)
585 if reasoning_mode is not None:
586 reasoning_overrides["mode"] = reasoning_mode
588 if reasoning_overrides:
589 generation_kwargs["reasoning"] = {**generation_kwargs.get("reasoning", {}), **reasoning_overrides}
591 include_reasoning_encrypted_content = generation_kwargs.pop("include_reasoning_encrypted_content", None)
592 if include_reasoning_encrypted_content is True:
593 include = generation_kwargs.get("include", [])
594 if "reasoning.encrypted_content" not in include:
595 generation_kwargs["include"] = [*include, "reasoning.encrypted_content"]
597 verbosity = generation_kwargs.pop("verbosity", None)
598 if verbosity is not None:
599 generation_kwargs["text"] = {**generation_kwargs.get("text", {}), "verbosity": verbosity}
601 return generation_kwargs
603 def _handle_stream_response(self, responses: Stream, callback: SyncStreamingCallbackT) -> list[ChatMessage]:
604 component_info = ComponentInfo.from_component(self)
605 chunks: list[StreamingChunk] = []
607 for openai_chunk in responses:
608 chunk_delta = _convert_response_chunk_to_streaming_chunk(
609 chunk=openai_chunk, previous_chunks=chunks, component_info=component_info
610 )
611 chunks.append(chunk_delta)
612 callback(chunk_delta)
613 chat_message = _convert_streaming_chunks_to_chat_message(chunks=chunks)
614 return [chat_message]
616 async def _handle_async_stream_response(
617 self, responses: AsyncStream, callback: StreamingCallbackT
618 ) -> list[ChatMessage]:
619 component_info = ComponentInfo.from_component(self)
620 chunks: list[StreamingChunk] = []
621 async for openai_chunk in responses:
622 chunk_delta = _convert_response_chunk_to_streaming_chunk(
623 chunk=openai_chunk, previous_chunks=chunks, component_info=component_info
624 )
625 chunks.append(chunk_delta)
626 await _invoke_streaming_callback(callback, chunk_delta)
627 chat_message = _convert_streaming_chunks_to_chat_message(chunks=chunks)
628 return [chat_message]
631def _convert_response_to_chat_message(responses: Response | ParsedResponse) -> ChatMessage:
632 """
633 Converts the non-streaming response from the OpenAI API to a ChatMessage.
635 :param responses: The responses returned by the OpenAI API.
636 :returns: The ChatMessage.
637 """
639 tool_calls = []
640 reasoning = None
641 logprobs: list[dict] = []
642 for output in responses.output:
643 if isinstance(output, ResponseOutputRefusal):
644 logger.warning("OpenAI returned a refusal output: {output}", output=output)
645 continue
647 if output.type == "message":
648 for content in output.content:
649 if hasattr(content, "logprobs") and content.logprobs is not None:
650 logprobs.append(_serialize_object(content.logprobs))
652 if output.type == "reasoning":
653 # openai doesn't return the reasoning tokens, but we can view summary if its enabled
654 # https://platform.openai.com/docs/guides/reasoning#reasoning-summaries
655 summaries = output.summary
656 extra = output.to_dict()
657 # we dont need the summary in the extra
658 extra.pop("summary")
659 if output.content:
660 logger.warning(
661 "OpenAI returned a non-empty 'content' field on a reasoning item ({_id}). "
662 "The content is preserved in ReasoningContent.extra['content'] but is NOT "
663 "reflected in ReasoningContent.reasoning_text.",
664 _id=output.id,
665 )
666 reasoning_text = "\n".join([summary.text for summary in summaries if summaries])
667 reasoning = ReasoningContent(reasoning_text=reasoning_text, extra=extra)
669 elif output.type == "function_call":
670 try:
671 arguments = json.loads(output.arguments)
672 tool_calls.append(
673 ToolCall(
674 id=output.id, tool_name=output.name, arguments=arguments, extra={"call_id": output.call_id}
675 )
676 )
677 except json.JSONDecodeError:
678 logger.warning(
679 "The LLM provider returned a malformed JSON string for tool call arguments. This tool call "
680 "will be skipped. To always generate a valid JSON, set `tools_strict` to `True`. "
681 "Tool call ID: {_id}, Tool name: {_name}, Arguments: {_arguments}",
682 _id=output.id,
683 _name=output.name,
684 _arguments=output.arguments,
685 )
686 arguments = {}
688 # we save the response as dict because it contains resp_id etc.
689 meta = responses.to_dict()
691 # remove output from meta because it contains toolcalls, reasoning, text etc.
692 meta.pop("output")
694 if logprobs:
695 meta["logprobs"] = logprobs
697 return ChatMessage.from_assistant(
698 text=responses.output_text if responses.output_text else None,
699 reasoning=reasoning,
700 tool_calls=tool_calls,
701 meta=meta,
702 )
705def _convert_response_chunk_to_streaming_chunk( # noqa: PLR0911
706 chunk: ResponseStreamEvent, previous_chunks: list[StreamingChunk], component_info: ComponentInfo | None = None
707) -> StreamingChunk:
708 """
709 Converts the streaming response chunk from the OpenAI Responses API to a StreamingChunk.
711 :param chunk: The chunk returned by the OpenAI Responses API.
712 :param previous_chunks: A list of previously received StreamingChunks.
713 :param component_info: An optional `ComponentInfo` object containing information about the component that
714 generated the chunk, such as the component name and type.
715 :returns:
716 A StreamingChunk object representing the content of the chunk from the OpenAI Responses API.
717 """
718 if chunk.type == "response.output_item.added":
719 # Responses API always returns reasoning chunks even if there is no summary
720 if chunk.item.type == "reasoning":
721 reasoning = ReasoningContent(reasoning_text="", extra=chunk.item.to_dict())
722 return StreamingChunk(
723 content="",
724 component_info=component_info,
725 index=chunk.output_index,
726 reasoning=reasoning,
727 start=True,
728 meta={"received_at": datetime.now().isoformat()},
729 )
731 # the function name is only streamed at the start and end of the function call
732 if chunk.item.type == "function_call":
733 tool_call = ToolCallDelta(
734 index=chunk.output_index, id=chunk.item.id, tool_name=chunk.item.name, extra=chunk.item.to_dict()
735 )
736 return StreamingChunk(
737 content="",
738 component_info=component_info,
739 index=chunk.output_index,
740 tool_calls=[tool_call],
741 start=True,
742 meta={"received_at": datetime.now().isoformat()},
743 )
745 elif chunk.type == "response.output_item.done":
746 # The done event carries the completed reasoning item, which includes encrypted_content
747 # when include=["reasoning.encrypted_content"] was requested. Without this handler the
748 # event falls through to the generic default and reasoning=None, so encrypted_content
749 # is never available for multi-turn conversations.
750 if chunk.item.type == "reasoning":
751 if chunk.item.content:
752 logger.warning(
753 "OpenAI returned a non-empty 'content' field on a reasoning item ({_id}). "
754 "This field is currently undocumented and was never observed in practice. "
755 "The content is preserved in ReasoningContent.extra['content'] but is NOT "
756 "reflected in ReasoningContent.reasoning_text. Please report this at "
757 "https://github.com/deepset-ai/haystack/issues so we can update the mapping.",
758 _id=chunk.item.id,
759 )
760 reasoning = ReasoningContent(reasoning_text="", extra=chunk.item.to_dict())
761 return StreamingChunk(
762 content="",
763 component_info=component_info,
764 index=chunk.output_index,
765 reasoning=reasoning,
766 meta={"received_at": datetime.now().isoformat()},
767 )
769 elif chunk.type == "response.completed":
770 # This means a full response is finished
771 # If there are tool_calls present in the final output we mark finish_reason as tool_calls otherwise it's
772 # marked as stop
773 return StreamingChunk(
774 content="",
775 component_info=component_info,
776 finish_reason="tool_calls" if any(o.type == "function_call" for o in chunk.response.output) else "stop",
777 meta={**chunk.to_dict(), "received_at": datetime.now().isoformat()},
778 )
780 elif chunk.type == "response.output_text.delta":
781 # Start is determined by checking if this is the first text delta event of a new output_index
782 # 1) Check if all previous chunks have different output_index
783 # 2) If any chunks do have the same output_index, check if they have content
784 # If none of them have content, this is the start of a new text output
785 start = all(c.index != chunk.output_index for c in previous_chunks) or all(
786 c.content == "" for c in previous_chunks if c.index == chunk.output_index
787 )
788 return StreamingChunk(
789 content=chunk.delta,
790 component_info=component_info,
791 index=chunk.output_index,
792 start=start,
793 meta={**chunk.to_dict(), "received_at": datetime.now().isoformat()},
794 )
796 elif chunk.type == "response.reasoning_summary_text.delta":
797 # We remove the delta from the extra because it is already in the reasoning_text
798 # Remaining information needs to be saved for chat message
799 extra = chunk.to_dict()
800 extra.pop("delta")
801 reasoning = ReasoningContent(reasoning_text=chunk.delta, extra=extra)
802 return StreamingChunk(
803 content="",
804 component_info=component_info,
805 index=chunk.output_index,
806 reasoning=reasoning,
807 meta={"received_at": datetime.now().isoformat()},
808 )
810 # the function arguments are streamed in parts
811 # function name is not passed in these chunks
812 elif chunk.type == "response.function_call_arguments.delta":
813 arguments = chunk.delta
814 extra = chunk.to_dict()
815 extra.pop("delta")
816 # in delta of tool calls there is no call_id so we use the item_id which is the function call id
817 tool_call = ToolCallDelta(index=chunk.output_index, id=chunk.item_id, arguments=arguments, extra=extra)
818 return StreamingChunk(
819 content="",
820 component_info=component_info,
821 index=chunk.output_index,
822 tool_calls=[tool_call],
823 meta={"received_at": datetime.now().isoformat()},
824 )
826 # we return rest of the chunk as is
827 return StreamingChunk(
828 content="",
829 component_info=component_info,
830 index=getattr(chunk, "output_index", None),
831 meta={**chunk.to_dict(), "received_at": datetime.now().isoformat()},
832 )
835def _convert_streaming_chunks_to_chat_message(chunks: list[StreamingChunk]) -> ChatMessage:
836 """
837 Connects the streaming chunks into a single ChatMessage.
839 :param chunks: The list of all `StreamingChunk` objects.
841 :returns: The ChatMessage.
842 """
844 # Get the full text by concatenating all text chunks
845 text = "".join([chunk.content for chunk in chunks])
846 logprobs = []
847 for chunk in chunks:
848 if chunk.meta.get("logprobs"):
849 logprobs.append(chunk.meta.get("logprobs"))
851 # Gather reasoning information if present
852 reasoning_id = None
853 reasoning_text = ""
854 for chunk in chunks:
855 if chunk.reasoning:
856 reasoning_text += chunk.reasoning.reasoning_text
857 if chunk.reasoning.extra.get("id"):
858 reasoning_id = chunk.reasoning.extra.get("id")
860 # Process tool calls if present in any chunk
861 tool_call_data: dict[str, dict[str, Any]] = {} # Track tool calls by id
862 for chunk in chunks:
863 if chunk.tool_calls:
864 for tool_call in chunk.tool_calls:
865 # here the tool_call.id is fc_id not call_id
866 assert tool_call.id is not None
867 # We use the tool call id to track the tool call across chunks
868 if tool_call.id not in tool_call_data:
869 tool_call_data[tool_call.id] = {"name": "", "arguments": ""}
871 if tool_call.arguments is not None:
872 tool_call_data[tool_call.id]["arguments"] += tool_call.arguments
874 # We capture the tool name from one of the chunks
875 if tool_call.tool_name is not None:
876 tool_call_data[tool_call.id]["name"] = tool_call.tool_name
878 # We capture the call_id from one of the chunks
879 if tool_call.extra and "call_id" in tool_call.extra:
880 tool_call_data[tool_call.id]["extra"] = {"call_id": tool_call.extra["call_id"]}
882 # Convert accumulated tool call data into ToolCall objects
883 tool_calls = []
884 sorted_keys = sorted(tool_call_data.keys())
885 for key in sorted_keys:
886 tool_call_dict = tool_call_data[key]
887 try:
888 arguments = json.loads(tool_call_dict.get("arguments", "{}")) if tool_call_dict.get("arguments") else {}
889 extra: dict[str, Any] = tool_call_dict.get("extra", {})
890 tool_calls.append(ToolCall(id=key, tool_name=tool_call_dict["name"], arguments=arguments, extra=extra))
891 except json.JSONDecodeError:
892 logger.warning(
893 "The LLM provider returned a malformed JSON string for tool call arguments. This tool call "
894 "will be skipped. To always generate a valid JSON, set `tools_strict` to `True`. "
895 "Tool call ID: {_id}, Tool name: {_name}, Arguments: {_arguments}",
896 _id=key,
897 _name=tool_call_dict["name"],
898 _arguments=tool_call_dict["arguments"],
899 )
901 # We dump the entire final response into meta to be consistent with non-streaming response
902 final_response = chunks[-1].meta.get("response") or {}
903 final_response.pop("output", None)
904 if logprobs:
905 final_response["logprobs"] = logprobs
907 # Add reasoning content if id is available
908 # Note: the API expects a reasoning id even if there is no reasoning text
909 # function calls without reasoning ids are not supported by the API
910 reasoning = None
911 if reasoning_id:
912 # Preserve all extra fields from streaming chunks (e.g. encrypted_content) while ensuring id and
913 # type are present
914 reasoning_extra = {}
915 for chunk in chunks:
916 if chunk.reasoning and chunk.reasoning.extra:
917 reasoning_extra.update(chunk.reasoning.extra)
918 # Ensure id and type are always set, but don't override if already present
919 reasoning_extra.setdefault("id", reasoning_id)
920 reasoning_extra.setdefault("type", "reasoning")
921 reasoning = ReasoningContent(reasoning_text=reasoning_text, extra=reasoning_extra)
923 return ChatMessage.from_assistant(
924 text=text or None, tool_calls=tool_calls, meta=final_response, reasoning=reasoning
925 )
928def _convert_chat_message_to_responses_api_format(message: ChatMessage) -> list[dict[str, Any]]:
929 """
930 Convert a ChatMessage to the dictionary format expected by OpenAI's Responses API.
932 :param message: The ChatMessage to convert to OpenAI's Responses API format.
933 :returns:
934 The ChatMessage in the format expected by OpenAI's Responses API.
936 :raises ValueError:
937 If the message format is invalid.
938 """
940 def convert_part(part: Any) -> dict[str, str | None]:
941 if isinstance(part, TextContent):
942 return {"type": "input_text", "text": part.text}
943 if isinstance(part, ImageContent):
944 return {
945 "type": "input_image",
946 # If no MIME type is provided, default to JPEG. OpenAI API appears to tolerate MIME type mismatches.
947 "image_url": f"data:{part.mime_type or 'image/jpeg'};base64,{part.base64_image}",
948 }
949 if isinstance(part, FileContent):
950 return {
951 "type": "input_file",
952 # Filename is optional but if not provided, OpenAI expects a file_id of a previous file upload.
953 # We use a dummy filename to avoid this issue.
954 "filename": part.filename or "filename",
955 "file_data": f"data:{part.mime_type or 'application/pdf'};base64,{part.base64_data}",
956 }
957 raise ValueError(f"Unsupported content type: {type(part)}")
959 text_contents = message.texts
960 tool_calls = message.tool_calls
961 tool_call_results = message.tool_call_results
962 images = message.images
963 reasonings = message.reasonings
964 files = message.files
966 if not any([text_contents, tool_calls, tool_call_results, images, reasonings, files]):
967 raise ValueError(
968 """A `ChatMessage` must contain at least one `TextContent`, `ToolCall`, `ToolCallResult`,
969 `ImageContent`, `FileContent`, or `ReasoningContent`."""
970 )
971 if len(tool_call_results) > 0 and len(message._content) > 1:
972 raise ValueError(
973 "For OpenAI compatibility, a `ChatMessage` with a `ToolCallResult` cannot contain any other content."
974 )
976 formatted_messages: list[dict[str, Any]] = []
977 openai_msg: dict[str, Any] = {"role": message._role.value}
978 if message._name is not None:
979 openai_msg["name"] = message._name
981 # user message
982 if message._role.value == "user":
983 content = [convert_part(part) for part in message._content]
984 openai_msg["content"] = content
985 return [openai_msg]
987 # tool message
988 if tool_call_results:
989 formatted_tool_results = []
990 for result in tool_call_results:
991 if result.origin.id is not None:
992 # Handle multimodal tool results (list of TextContent/ImageContent/FileContent)
993 if isinstance(result.result, list):
994 output_content = [convert_part(part) for part in result.result]
995 elif isinstance(result.result, str):
996 output_content = [{"type": "input_text", "text": result.result}]
997 else:
998 raise ValueError(f"Unsupported tool result: {result.result}")
999 tool_result = {
1000 "type": "function_call_output",
1001 "call_id": result.origin.extra.get("call_id") if result.origin.extra else "",
1002 "output": output_content,
1003 }
1004 formatted_tool_results.append(tool_result)
1005 formatted_messages.extend(formatted_tool_results)
1007 # Note: the API expects a reasoning id even if there is no reasoning text
1008 # function calls without reasoning ids are not supported by the API
1009 if reasonings:
1010 formatted_reasonings = []
1011 for reasoning in reasonings:
1012 # Streaming events (e.g. response.reasoning_summary_text.delta) store event-level
1013 # fields like item_id, output_index, summary_index, event_id, sequence_number into
1014 # reasoning.extra. Those are not valid reasoning input item fields and the API
1015 # rejects them with "Unknown parameter" when sent back in subsequent turns.
1016 # Valid fields per ResponseReasoningItem schema: id, type, summary (handled separately),
1017 # content, encrypted_content, status.
1018 _valid_reasoning_fields = {"id", "type", "encrypted_content", "status", "content"}
1019 filtered_extra = {k: v for k, v in reasoning.extra.items() if k in _valid_reasoning_fields}
1020 reasoning_item = {"summary": [], **filtered_extra}
1021 if reasoning.reasoning_text:
1022 reasoning_item["summary"] = [{"text": reasoning.reasoning_text, "type": "summary_text"}]
1023 formatted_reasonings.append(reasoning_item)
1024 formatted_messages.extend(formatted_reasonings)
1026 if tool_calls:
1027 formatted_tool_calls = []
1028 for tc in tool_calls:
1029 openai_tool_call = {
1030 "type": "function_call",
1031 # We disable ensure_ascii so special chars like emojis are not converted
1032 "name": tc.tool_name,
1033 "arguments": json.dumps(tc.arguments, ensure_ascii=False),
1034 "id": tc.id,
1035 "call_id": tc.extra.get("call_id") if tc.extra else "",
1036 }
1038 formatted_tool_calls.append(openai_tool_call)
1039 formatted_messages.extend(formatted_tool_calls)
1041 # system and assistant messages
1042 if text_contents:
1043 openai_msg["content"] = " ".join(text_contents)
1044 formatted_messages.append(openai_msg)
1046 return formatted_messages