Coverage for haystack/components/generators/chat/openai_responses.py: 92%

1# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai> 

2# 

3# SPDX-License-Identifier: Apache-2.0 

4 

5import json 

6import os 

7from datetime import datetime 

8from typing import Any, ClassVar 

9 

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 

14 

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 ChatRole, 

20 ComponentInfo, 

21 FileContent, 

22 ImageContent, 

23 ReasoningContent, 

24 StreamingCallbackT, 

25 StreamingChunk, 

26 SyncStreamingCallbackT, 

27 TextContent, 

28 ToolCall, 

29 ToolCallDelta, 

30 select_streaming_callback, 

31) 

32from haystack.dataclasses.chat_message import _parse_openai_tool_call_arguments 

33from haystack.dataclasses.streaming_chunk import FinishReason, _invoke_streaming_callback 

34from haystack.tools import ( 

35 ToolsType, 

36 _check_duplicate_tool_names, 

37 deserialize_tools_or_toolset_inplace, 

38 flatten_tools_or_toolsets, 

39 serialize_tools_or_toolset, 

40) 

41from haystack.utils import Secret, deserialize_callable, serialize_callable 

42from haystack.utils.http_client import init_http_client 

43 

44logger = logging.getLogger(__name__) 

45 

46_INCOMPLETE_REASON_MAPPING: dict[str, FinishReason] = { 

47 "max_output_tokens": "length", 

48 "content_filter": "content_filter", 

49} 

50 

51 

52def _get_response_finish_reason(response: Response | ParsedResponse) -> FinishReason: 

53 """Return the normalized finish reason for a terminal OpenAI Responses API response.""" 

54 if ( 

55 response.status == "incomplete" 

56 and response.incomplete_details is not None 

57 and response.incomplete_details.reason is not None 

58 ): 

59 finish_reason = _INCOMPLETE_REASON_MAPPING.get(response.incomplete_details.reason) 

60 if finish_reason is not None: 

61 return finish_reason 

62 if any(output.type == "function_call" for output in response.output): 

63 return "tool_calls" 

64 return "stop" 

65 

66 

67@component 

68class OpenAIResponsesChatGenerator: 

69 """ 

70 Completes chats using OpenAI's Responses API. 

71 

72 It works with OpenAI's GPT and o-series models and supports streaming responses 

73 from OpenAI API. It uses [ChatMessage](https://docs.haystack.deepset.ai/docs/chatmessage) 

74 format in input and output. 

75 

76 You can customize how the text is generated by passing parameters to the 

77 OpenAI API. Use the `**generation_kwargs` argument when you initialize 

78 the component or when you run it. Any parameter that works with 

79 `openai.Responses.create` will work here too. 

80 

81 For details on OpenAI API parameters, see 

82 [OpenAI documentation](https://platform.openai.com/docs/api-reference/responses). 

83 

84 ### Usage example 

85 

86 ```python 

87 from haystack.components.generators.chat import OpenAIResponsesChatGenerator 

88 from haystack.dataclasses import ChatMessage 

89 

90 messages = [ChatMessage.from_user("What's Natural Language Processing?")] 

91 

92 client = OpenAIResponsesChatGenerator(generation_kwargs={"reasoning": {"effort": "low", "summary": "auto"}}) 

93 response = client.run(messages) 

94 print(response) 

95 ``` 

96 """ 

97 

98 SUPPORTED_MODELS: ClassVar[list[str]] = [ 

99 "gpt-5-mini", 

100 "gpt-5-nano", 

101 "gpt-5", 

102 "gpt-5.1", 

103 "gpt-5.2", 

104 "gpt-5.2-pro", 

105 "gpt-5.4", 

106 "gpt-5-pro", 

107 "gpt-4.1", 

108 "gpt-4.1-mini", 

109 "gpt-4.1-nano", 

110 "gpt-4o", 

111 "gpt-4o-mini", 

112 "o1", 

113 "o1-mini", 

114 "o1-pro", 

115 "o3", 

116 "o3-mini", 

117 "o3-pro", 

118 "o4-mini", 

119 ] 

120 """A non-exhaustive list of chat models supported by this component. 

121 See https://platform.openai.com/docs/models for the full list and snapshot IDs.""" 

122 

123 def __init__( 

124 self, 

125 *, 

126 api_key: Secret = Secret.from_env_var("OPENAI_API_KEY"), 

127 model: str = "gpt-5-mini", 

128 streaming_callback: StreamingCallbackT | None = None, 

129 api_base_url: str | None = None, 

130 organization: str | None = None, 

131 generation_kwargs: dict[str, Any] | None = None, 

132 timeout: float | None = None, 

133 max_retries: int | None = None, 

134 tools: ToolsType | list[dict] | None = None, 

135 tools_strict: bool = False, 

136 http_client_kwargs: dict[str, Any] | None = None, 

137 ) -> None: 

138 """ 

139 Creates an instance of OpenAIResponsesChatGenerator. Uses OpenAI's gpt-5-mini by default. 

140 

141 Before initializing the component, you can set the 'OPENAI_TIMEOUT' and 'OPENAI_MAX_RETRIES' 

142 environment variables to override the `timeout` and `max_retries` parameters respectively 

143 in the OpenAI client. 

144 

145 :param api_key: The OpenAI API key. 

146 You can set it with an environment variable `OPENAI_API_KEY`, or pass with this parameter 

147 during initialization. 

148 :param model: The name of the model to use. 

149 :param streaming_callback: A callback function that is called when a new token is received from the stream. 

150 The callback function accepts [StreamingChunk](https://docs.haystack.deepset.ai/docs/data-classes#streamingchunk) 

151 as an argument. 

152 :param api_base_url: An optional base URL. 

153 :param organization: Your organization ID, defaults to `None`. See 

154 [production best practices](https://platform.openai.com/docs/guides/production-best-practices/setting-up-your-organization). 

155 :param generation_kwargs: Other parameters to use for the model. These parameters are sent 

156 directly to the OpenAI endpoint. 

157 See OpenAI [documentation](https://platform.openai.com/docs/api-reference/responses) for 

158 more details. 

159 Some of the supported parameters: 

160 - `temperature`: What sampling temperature to use. Higher values like 0.8 will make the output more random, 

161 while lower values like 0.2 will make it more focused and deterministic. 

162 - `top_p`: An alternative to sampling with temperature, called nucleus sampling, where the model 

163 considers the results of the tokens with top_p probability mass. For example, 0.1 means only the tokens 

164 comprising the top 10% probability mass are considered. 

165 - `previous_response_id`: The ID of the previous response. 

166 Use this to create multi-turn conversations. 

167 - `text_format`: A Pydantic model that enforces the structure of the model's response. 

168 If provided, the output will always be validated against this 

169 format (unless the model returns a tool call). 

170 For details, see the [OpenAI Structured Outputs documentation](https://platform.openai.com/docs/guides/structured-outputs). 

171 - `text`: A JSON schema that enforces the structure of the model's response. 

172 If provided, the output will always be validated against this 

173 format (unless the model returns a tool call). 

174 Notes: 

175 - Both JSON Schema and Pydantic models are supported for latest models starting from GPT-4o. 

176 - If both are provided, `text_format` takes precedence and json schema passed to `text` is ignored. 

177 - Currently, this component doesn't support streaming for structured outputs. 

178 - Older models only support basic version of structured outputs through `{"type": "json_object"}`. 

179 For detailed information on JSON mode, see the [OpenAI Structured Outputs documentation](https://platform.openai.com/docs/guides/structured-outputs#json-mode). 

180 - `reasoning`: A dictionary of parameters for reasoning. For example: 

181 - `summary`: The summary of the reasoning. 

182 - `effort`: The level of effort to put into the reasoning. Can be `low`, `medium` or `high`. 

183 - `generate_summary`: Whether to generate a summary of the reasoning. 

184 - `mode`: The reasoning mode. Can be `standard`, or `pro`. Supported since GPT-5.6. 

185 Note: OpenAI does not return the reasoning tokens, but we can view summary if its enabled. 

186 If a provider returns the raw reasoning as `reasoning_text` content instead, 

187 it is mapped to `ReasoningContent.reasoning_text` when no summary is returned. 

188 For details, see the [OpenAI Reasoning documentation](https://platform.openai.com/docs/guides/reasoning). 

189 - `include`: Specify additional output data to include in the model response. Supported values are: 

190 - web_search_call.action.sources: Include the sources of the web search tool call. 

191 - code_interpreter_call.outputs: Includes the outputs of python code execution in code interpreter tool 

192 call items. 

193 - computer_call_output.output.image_url: Include image urls from the computer call output. 

194 - file_search_call.results: Include the search results of the file search tool call. 

195 - message.input_image.image_url: Include image urls from the input message. 

196 - message.output_text.logprobs: Include logprobs with assistant messages. 

197 - reasoning.encrypted_content: Includes an encrypted version of reasoning tokens in reasoning item 

198 outputs. This enables reasoning items to be used in multi-turn conversations when using the 

199 Responses API statelessly (like when the store parameter is set to false, or when an organization 

200 is enrolled in the zero data retention program). 

201 :param timeout: 

202 Timeout for OpenAI client calls. If not set, it defaults to either the 

203 `OPENAI_TIMEOUT` environment variable, or 30 seconds. 

204 :param max_retries: 

205 Maximum number of retries to contact OpenAI after an internal error. 

206 If not set, it defaults to either the `OPENAI_MAX_RETRIES` environment variable, or set to 5. 

207 :param tools: 

208 The tools that the model can use to prepare calls. This parameter can accept either a 

209 mixed list of Haystack `Tool` objects and Haystack `Toolset`. Or you can pass a dictionary of 

210 OpenAI/MCP tool definitions. 

211 Note: You cannot pass OpenAI/MCP tools and Haystack tools together. 

212 For details on tool support, see [OpenAI documentation](https://platform.openai.com/docs/api-reference/responses/create#responses-create-tools). 

213 :param tools_strict: 

214 Whether to enable strict schema adherence for tool calls. If set to `False`, the model may not exactly 

215 follow the schema provided in the `parameters` field of the tool definition. In Response API, tool calls 

216 are strict by default. 

217 :param http_client_kwargs: 

218 A dictionary of keyword arguments to configure a custom `httpx.Client`or `httpx.AsyncClient`. 

219 For more information, see the [HTTPX documentation](https://www.python-httpx.org/api/#client). 

220 

221 """ 

222 self.api_key = api_key 

223 self.model = model 

224 self.generation_kwargs = generation_kwargs or {} 

225 self.streaming_callback = streaming_callback 

226 self.api_base_url = api_base_url 

227 self.organization = organization 

228 self.timeout = timeout 

229 self.max_retries = max_retries 

230 self.tools = tools # Store tools as-is, whether it's a list or a Toolset 

231 self.tools_strict = tools_strict 

232 self.http_client_kwargs = http_client_kwargs 

233 

234 self.client: OpenAI | None = None 

235 self.async_client: AsyncOpenAI | None = None 

236 

237 def _client_kwargs(self) -> dict[str, Any]: 

238 timeout = self.timeout if self.timeout is not None else float(os.environ.get("OPENAI_TIMEOUT", "30.0")) 

239 max_retries = ( 

240 self.max_retries if self.max_retries is not None else int(os.environ.get("OPENAI_MAX_RETRIES", "5")) 

241 ) 

242 resolved_api_key = self.api_key.resolve_value() if isinstance(self.api_key, Secret) else self.api_key 

243 return { 

244 "api_key": resolved_api_key, 

245 "organization": self.organization, 

246 "base_url": self.api_base_url, 

247 "timeout": timeout, 

248 "max_retries": max_retries, 

249 } 

250 

251 def warm_up(self) -> None: 

252 """ 

253 Initialize the synchronous OpenAI client. 

254 """ 

255 if self.client is None: 

256 # openai>=3 annotates http_client as httpx2, but legacy httpx clients are supported at runtime. 

257 # https://github.com/openai/openai-python/blob/main/httpx2.md 

258 http_client = init_http_client(self.http_client_kwargs, async_client=False) 

259 self.client = OpenAI( 

260 http_client=http_client, # type: ignore[arg-type] 

261 **self._client_kwargs(), 

262 ) 

263 

264 async def warm_up_async(self) -> None: # noqa: RUF029 

265 """ 

266 Initialize the asynchronous OpenAI client on the serving event loop. 

267 """ 

268 if self.async_client is None: 

269 # openai>=3 annotates http_client as httpx2, but legacy httpx clients are supported at runtime. 

270 # https://github.com/openai/openai-python/blob/main/httpx2.md 

271 http_client = init_http_client(self.http_client_kwargs, async_client=True) 

272 self.async_client = AsyncOpenAI( 

273 http_client=http_client, # type: ignore[arg-type] 

274 **self._client_kwargs(), 

275 ) 

276 

277 def close(self) -> None: 

278 """ 

279 Releases the synchronous OpenAI client. 

280 """ 

281 if self.client is not None: 

282 self.client.close() 

283 self.client = None 

284 

285 async def close_async(self) -> None: 

286 """ 

287 Releases the asynchronous OpenAI client. 

288 """ 

289 if self.async_client is not None: 

290 await self.async_client.close() 

291 self.async_client = None 

292 

293 def _get_telemetry_data(self) -> dict[str, Any]: 

294 """ 

295 Data that is sent to Posthog for usage analytics. 

296 """ 

297 return {"model": self.model} 

298 

299 def to_dict(self) -> dict[str, Any]: 

300 """ 

301 Serialize this component to a dictionary. 

302 

303 :returns: 

304 The serialized component as a dictionary. 

305 """ 

306 callback_name = serialize_callable(self.streaming_callback) if self.streaming_callback else None 

307 generation_kwargs = self.generation_kwargs.copy() 

308 text_format = generation_kwargs.pop("text_format", None) 

309 

310 # If the response format is a Pydantic model, it's converted to openai's json schema format 

311 # If it's already a json schema, it's left as is 

312 if text_format and isinstance(text_format, type) and issubclass(text_format, BaseModel): 

313 json_schema = { 

314 "format": { 

315 "type": "json_schema", 

316 "name": text_format.__name__, 

317 "strict": True, 

318 "schema": to_strict_json_schema(text_format), 

319 } 

320 } 

321 # json schema needs to be passed to text parameter instead of text_format 

322 generation_kwargs["text"] = json_schema 

323 

324 # OpenAI/MCP tools are passed as list of dictionaries 

325 serialized_tools: dict[str, Any] | list[dict[str, Any]] | None 

326 if self.tools and isinstance(self.tools, list) and isinstance(self.tools[0], dict): 

327 # mypy can't infer that self.tools is list[dict] here 

328 serialized_tools = self.tools 

329 else: 

330 serialized_tools = serialize_tools_or_toolset(self.tools) # type: ignore[arg-type] 

331 

332 return default_to_dict( 

333 self, 

334 model=self.model, 

335 streaming_callback=callback_name, 

336 api_base_url=self.api_base_url, 

337 organization=self.organization, 

338 generation_kwargs=generation_kwargs, 

339 api_key=self.api_key, 

340 timeout=self.timeout, 

341 max_retries=self.max_retries, 

342 tools=serialized_tools, 

343 tools_strict=self.tools_strict, 

344 http_client_kwargs=self.http_client_kwargs, 

345 ) 

346 

347 @classmethod 

348 def from_dict(cls, data: dict[str, Any]) -> "OpenAIResponsesChatGenerator": 

349 """ 

350 Deserialize this component from a dictionary. 

351 

352 :param data: The dictionary representation of this component. 

353 :returns: 

354 The deserialized component instance. 

355 """ 

356 # we only deserialize the tools if they are haystack tools 

357 # because openai tools are not serialized in the same way 

358 tools = data.get("init_parameters", {}).get("tools") 

359 if tools and ( 

360 isinstance(tools, dict) 

361 and "type" in tools 

362 and "data" in tools 

363 or isinstance(tools, list) 

364 and isinstance(tools[0], dict) 

365 and "type" in tools[0] 

366 and "data" in tools[0] 

367 ): 

368 deserialize_tools_or_toolset_inplace(data["init_parameters"], key="tools") 

369 

370 init_params = data.get("init_parameters", {}) 

371 serialized_callback_handler = init_params.get("streaming_callback") 

372 

373 if serialized_callback_handler: 

374 data["init_parameters"]["streaming_callback"] = deserialize_callable(serialized_callback_handler) 

375 return default_from_dict(cls, data) 

376 

377 @component.output_types(replies=list[ChatMessage]) 

378 def run( 

379 self, 

380 messages: list[ChatMessage] | str, 

381 *, 

382 streaming_callback: StreamingCallbackT | None = None, 

383 generation_kwargs: dict[str, Any] | None = None, 

384 tools: ToolsType | list[dict] | None = None, 

385 tools_strict: bool | None = None, 

386 ) -> dict[str, list[ChatMessage]]: 

387 """ 

388 Invokes response generation based on the provided messages and generation parameters. 

389 

390 :param messages: 

391 A list of ChatMessage instances representing the input messages. 

392 :param streaming_callback: 

393 A callback function that is called when a new token is received from the stream. 

394 :param generation_kwargs: 

395 Additional keyword arguments for text generation. These are merged per key with the 

396 `generation_kwargs` passed at initialization: keys provided here take precedence, keys set 

397 only at initialization are kept. 

398 For details on OpenAI API parameters, see [OpenAI documentation](https://platform.openai.com/docs/api-reference/responses/create). 

399 :param tools: 

400 The tools that the model can use to prepare calls. If set, it will override the 

401 `tools` parameter set during component initialization. This parameter can accept either a 

402 mixed list of Haystack `Tool` objects and Haystack `Toolset`. Or you can pass a dictionary of 

403 OpenAI/MCP tool definitions. 

404 Note: You cannot pass OpenAI/MCP tools and Haystack tools together. 

405 For details on tool support, see [OpenAI documentation](https://platform.openai.com/docs/api-reference/responses/create#responses-create-tools). 

406 :param tools_strict: 

407 Whether to enable strict schema adherence for tool calls. If set to `False`, the model may not exactly 

408 follow the schema provided in the `parameters` field of the tool definition. In Response API, tool calls 

409 are strict by default. 

410 If set, it will override the `tools_strict` parameter set during component initialization. 

411 

412 :returns: 

413 A dictionary with the following key: 

414 - `replies`: A list containing the generated responses as ChatMessage instances. 

415 """ 

416 self.warm_up() 

417 

418 messages = _normalize_messages(messages) 

419 

420 if len(messages) == 0: 

421 return {"replies": []} 

422 

423 streaming_callback = select_streaming_callback( 

424 init_callback=self.streaming_callback, runtime_callback=streaming_callback, requires_async=False 

425 ) 

426 responses: Stream[ResponseStreamEvent] | Response 

427 

428 api_args = self._prepare_api_call( 

429 messages=messages, 

430 streaming_callback=streaming_callback, 

431 generation_kwargs=generation_kwargs, 

432 tools=tools, 

433 tools_strict=tools_strict, 

434 ) 

435 openai_endpoint = api_args.pop("openai_endpoint") 

436 assert self.client is not None # mypy: client is built by warm_up above 

437 openai_endpoint_method = getattr(self.client.responses, openai_endpoint) 

438 responses = openai_endpoint_method(**api_args) 

439 

440 if streaming_callback is not None: 

441 response_output = self._handle_stream_response( 

442 responses, # type: ignore 

443 streaming_callback, 

444 ) 

445 else: 

446 assert isinstance(responses, Response), "Unexpected response type for non-streaming request." 

447 response_output = [_convert_response_to_chat_message(responses)] 

448 return {"replies": response_output} 

449 

450 @component.output_types(replies=list[ChatMessage]) 

451 async def run_async( 

452 self, 

453 messages: list[ChatMessage] | str, 

454 *, 

455 streaming_callback: StreamingCallbackT | None = None, 

456 generation_kwargs: dict[str, Any] | None = None, 

457 tools: ToolsType | list[dict] | None = None, 

458 tools_strict: bool | None = None, 

459 ) -> dict[str, list[ChatMessage]]: 

460 """ 

461 Asynchronously invokes response generation based on the provided messages and generation parameters. 

462 

463 This is the asynchronous version of the `run` method. It has the same parameters and return values 

464 but can be used with `await` in async code. 

465 

466 :param messages: 

467 A list of ChatMessage instances representing the input messages. 

468 :param streaming_callback: 

469 A callback function that is called when a new token is received from the stream. Async callbacks are 

470 preferred; a sync callback is accepted but will run synchronously on the event loop and may block it. 

471 :param generation_kwargs: 

472 Additional keyword arguments for text generation. These are merged per key with the 

473 `generation_kwargs` passed at initialization: keys provided here take precedence, keys set 

474 only at initialization are kept. 

475 For details on OpenAI API parameters, see [OpenAI documentation](https://platform.openai.com/docs/api-reference/responses/create). 

476 :param tools: 

477 A list of tools or a Toolset for which the model can prepare calls. If set, it will override the 

478 `tools` parameter set during component initialization. This parameter can accept either a list of 

479 mixed list of Haystack `Tool` objects and Haystack `Toolset`. Or you can pass a dictionary of 

480 OpenAI/MCP tool definitions. 

481 Note: You cannot pass OpenAI/MCP tools and Haystack tools together. 

482 :param tools_strict: 

483 Whether to enable strict schema adherence for tool calls. If set to `True`, the model will follow exactly 

484 the schema provided in the `parameters` field of the tool definition, but this may increase latency. 

485 If set, it will override the `tools_strict` parameter set during component initialization. 

486 

487 :returns: 

488 A dictionary with the following key: 

489 - `replies`: A list containing the generated responses as ChatMessage instances. 

490 """ 

491 await self.warm_up_async() 

492 

493 messages = _normalize_messages(messages) 

494 

495 # validate and select the streaming callback 

496 streaming_callback = select_streaming_callback( 

497 init_callback=self.streaming_callback, runtime_callback=streaming_callback, requires_async=True 

498 ) 

499 responses: AsyncStream[ResponseStreamEvent] | Response 

500 

501 if len(messages) == 0: 

502 return {"replies": []} 

503 

504 api_args = self._prepare_api_call( 

505 messages=messages, 

506 streaming_callback=streaming_callback, 

507 generation_kwargs=generation_kwargs, 

508 tools=tools, 

509 tools_strict=tools_strict, 

510 ) 

511 

512 openai_endpoint = api_args.pop("openai_endpoint") 

513 assert self.async_client is not None # mypy: async_client is built by warm_up_async above 

514 openai_endpoint_method = getattr(self.async_client.responses, openai_endpoint) 

515 responses = await openai_endpoint_method(**api_args) 

516 

517 if streaming_callback is not None: 

518 response_output = await self._handle_async_stream_response( 

519 responses, # type: ignore 

520 streaming_callback, 

521 ) 

522 

523 else: 

524 assert isinstance(responses, Response), "Unexpected response type for non-streaming request." 

525 response_output = [_convert_response_to_chat_message(responses)] 

526 return {"replies": response_output} 

527 

528 def _prepare_api_call( # noqa: PLR0913 

529 self, 

530 *, 

531 messages: list[ChatMessage], 

532 streaming_callback: StreamingCallbackT | None = None, 

533 generation_kwargs: dict[str, Any] | None = None, 

534 tools: ToolsType | list[dict] | None = None, 

535 tools_strict: bool | None = None, 

536 ) -> dict[str, Any]: 

537 # update generation kwargs by merging with the generation kwargs passed to the run method 

538 generation_kwargs = {**self.generation_kwargs, **(generation_kwargs or {})} 

539 generation_kwargs = self._resolve_flattened_kwargs(generation_kwargs) 

540 

541 # adapt ChatMessage(s) to the format expected by the OpenAI API 

542 openai_formatted_messages: list[dict[str, Any]] = [] 

543 for message in messages: 

544 openai_formatted_messages.extend(_convert_chat_message_to_responses_api_format(message)) 

545 

546 tools = tools if tools is not None else self.tools 

547 tools_strict = tools_strict if tools_strict is not None else self.tools_strict 

548 

549 openai_tools = {} 

550 # Build tool definitions 

551 if tools: 

552 tool_definitions: list[Any] = [] 

553 if isinstance(tools, list) and isinstance(tools[0], dict): 

554 # Predefined OpenAI/MCP-style tools 

555 tool_definitions = tools 

556 

557 # Convert all tool objects to the correct OpenAI-compatible structure 

558 else: 

559 # mypy can't infer that tools is ToolsType here 

560 flattened_tools = flatten_tools_or_toolsets(tools) # type: ignore[arg-type] 

561 _check_duplicate_tool_names(flattened_tools) 

562 for t in flattened_tools: 

563 function_spec = {**t.tool_spec} 

564 if not tools_strict: 

565 function_spec["strict"] = False 

566 # Copy the parameters schema before editing it. tool_spec exposes 

567 # Tool.parameters by reference, so mutating it here would permanently alter 

568 # the user's Tool (and any other generator that shares the same Tool instance). 

569 function_spec["parameters"] = {**function_spec["parameters"], "additionalProperties": False} 

570 tool_definitions.append({"type": "function", **function_spec}) 

571 

572 openai_tools = {"tools": tool_definitions} 

573 

574 base_args = {"model": self.model, "input": openai_formatted_messages, **openai_tools, **generation_kwargs} 

575 

576 # if `text_format` is provided, we use the `parse` endpoint for response type parsing 

577 if generation_kwargs.get("text_format"): 

578 # if both `text_format` and `text` are provided, `text_format` takes precedence 

579 # and json schema passed to `text` is ignored 

580 return {**base_args, "stream": streaming_callback is not None, "openai_endpoint": "parse"} 

581 # we pass a key `openai_endpoint` as a hint to the run method to use the create or parse endpoint 

582 # this key will be removed before the API call is made 

583 return {**base_args, "stream": streaming_callback is not None, "openai_endpoint": "create"} 

584 

585 def _resolve_flattened_kwargs(self, generation_kwargs: dict[str, Any]) -> dict[str, Any]: 

586 generation_kwargs = generation_kwargs.copy() 

587 

588 # avoid mutating the caller's dict 

589 reasoning_overrides = {} 

590 reasoning_effort = generation_kwargs.pop("reasoning_effort", None) 

591 if reasoning_effort is not None: 

592 reasoning_overrides["effort"] = reasoning_effort 

593 

594 reasoning_summary = generation_kwargs.pop("reasoning_summary", None) 

595 if reasoning_summary is not None: 

596 reasoning_overrides["summary"] = reasoning_summary 

597 

598 reasoning_mode = generation_kwargs.pop("reasoning_mode", None) 

599 if reasoning_mode is not None: 

600 reasoning_overrides["mode"] = reasoning_mode 

601 

602 if reasoning_overrides: 

603 generation_kwargs["reasoning"] = {**generation_kwargs.get("reasoning", {}), **reasoning_overrides} 

604 

605 include_reasoning_encrypted_content = generation_kwargs.pop("include_reasoning_encrypted_content", None) 

606 if include_reasoning_encrypted_content is True: 

607 include = generation_kwargs.get("include", []) 

608 if "reasoning.encrypted_content" not in include: 

609 generation_kwargs["include"] = [*include, "reasoning.encrypted_content"] 

610 

611 verbosity = generation_kwargs.pop("verbosity", None) 

612 if verbosity is not None: 

613 generation_kwargs["text"] = {**generation_kwargs.get("text", {}), "verbosity": verbosity} 

614 

615 return generation_kwargs 

616 

617 def _handle_stream_response(self, responses: Stream, callback: SyncStreamingCallbackT) -> list[ChatMessage]: 

618 component_info = ComponentInfo.from_component(self) 

619 chunks: list[StreamingChunk] = [] 

620 

621 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 callback(chunk_delta) 

627 chat_message = _convert_streaming_chunks_to_chat_message(chunks=chunks) 

628 return [chat_message] 

629 

630 async def _handle_async_stream_response( 

631 self, responses: AsyncStream, callback: StreamingCallbackT 

632 ) -> list[ChatMessage]: 

633 component_info = ComponentInfo.from_component(self) 

634 chunks: list[StreamingChunk] = [] 

635 async for openai_chunk in responses: 

636 chunk_delta = _convert_response_chunk_to_streaming_chunk( 

637 chunk=openai_chunk, previous_chunks=chunks, component_info=component_info 

638 ) 

639 chunks.append(chunk_delta) 

640 await _invoke_streaming_callback(callback, chunk_delta) 

641 chat_message = _convert_streaming_chunks_to_chat_message(chunks=chunks) 

642 return [chat_message] 

643 

644 

645def _convert_response_to_chat_message(responses: Response | ParsedResponse) -> ChatMessage: 

646 """ 

647 Converts the non-streaming response from the OpenAI API to a ChatMessage. 

648 

649 :param responses: The responses returned by the OpenAI API. 

650 :returns: The ChatMessage. 

651 """ 

652 

653 tool_calls = [] 

654 reasoning = None 

655 logprobs: list[dict] = [] 

656 for output in responses.output: 

657 if isinstance(output, ResponseOutputRefusal): 

658 logger.warning("OpenAI returned a refusal output: {output}", output=output) 

659 continue 

660 

661 if output.type == "message": 

662 for content in output.content: 

663 if hasattr(content, "logprobs") and content.logprobs is not None: 

664 logprobs.append(_serialize_object(content.logprobs)) 

665 

666 if output.type == "reasoning": 

667 extra = output.to_dict() 

668 # we dont need the summary in the extra 

669 extra.pop("summary") 

670 # OpenAI returns reasoning summaries, other providers may return raw reasoning as `reasoning_text` content 

671 reasoning_text = "\n".join(summary.text for summary in output.summary) 

672 if not reasoning_text: 

673 reasoning_text = "\n".join(part.text for part in output.content or [] if part.type == "reasoning_text") 

674 reasoning = ReasoningContent(reasoning_text=reasoning_text, extra=extra) 

675 

676 elif output.type == "function_call": 

677 try: 

678 arguments = _parse_openai_tool_call_arguments(output.arguments) 

679 tool_calls.append( 

680 ToolCall( 

681 id=output.id, tool_name=output.name, arguments=arguments, extra={"call_id": output.call_id} 

682 ) 

683 ) 

684 except json.JSONDecodeError: 

685 logger.warning( 

686 "The LLM provider returned a malformed JSON string for tool call arguments. This tool call " 

687 "will be skipped. To always generate a valid JSON, set `tools_strict` to `True`. " 

688 "Tool call ID: {_id}, Tool name: {_name}, Arguments: {_arguments}", 

689 _id=output.id, 

690 _name=output.name, 

691 _arguments=output.arguments, 

692 ) 

693 arguments = {} 

694 

695 # we save the response as dict because it contains resp_id etc. 

696 meta = responses.to_dict() 

697 

698 # remove output from meta because it contains toolcalls, reasoning, text etc. 

699 meta.pop("output") 

700 meta["finish_reason"] = _get_response_finish_reason(responses) 

701 

702 if logprobs: 

703 meta["logprobs"] = logprobs 

704 

705 return ChatMessage.from_assistant( 

706 text=responses.output_text if responses.output_text else None, 

707 reasoning=reasoning, 

708 tool_calls=tool_calls, 

709 meta=meta, 

710 ) 

711 

712 

713def _convert_response_chunk_to_streaming_chunk( # noqa: PLR0911 

714 chunk: ResponseStreamEvent, previous_chunks: list[StreamingChunk], component_info: ComponentInfo | None = None 

715) -> StreamingChunk: 

716 """ 

717 Converts the streaming response chunk from the OpenAI Responses API to a StreamingChunk. 

718 

719 :param chunk: The chunk returned by the OpenAI Responses API. 

720 :param previous_chunks: A list of previously received StreamingChunks. 

721 :param component_info: An optional `ComponentInfo` object containing information about the component that 

722 generated the chunk, such as the component name and type. 

723 :returns: 

724 A StreamingChunk object representing the content of the chunk from the OpenAI Responses API. 

725 """ 

726 if chunk.type == "response.output_item.added": 

727 # Responses API always returns reasoning chunks even if there is no summary 

728 if chunk.item.type == "reasoning": 

729 reasoning = ReasoningContent(reasoning_text="", extra=chunk.item.to_dict()) 

730 return StreamingChunk( 

731 content="", 

732 component_info=component_info, 

733 index=chunk.output_index, 

734 reasoning=reasoning, 

735 start=True, 

736 meta={"received_at": datetime.now().isoformat()}, 

737 ) 

738 

739 # the function name is only streamed at the start and end of the function call 

740 if chunk.item.type == "function_call": 

741 tool_call = ToolCallDelta( 

742 index=chunk.output_index, id=chunk.item.id, tool_name=chunk.item.name, extra=chunk.item.to_dict() 

743 ) 

744 return StreamingChunk( 

745 content="", 

746 component_info=component_info, 

747 index=chunk.output_index, 

748 tool_calls=[tool_call], 

749 start=True, 

750 meta={"received_at": datetime.now().isoformat()}, 

751 ) 

752 

753 elif chunk.type == "response.output_item.done": 

754 # The done event carries the completed reasoning item, which includes encrypted_content 

755 # when include=["reasoning.encrypted_content"] was requested. Without this handler the 

756 # event falls through to the generic default and reasoning=None, so encrypted_content 

757 # is never available for multi-turn conversations. 

758 if chunk.item.type == "reasoning": 

759 # The completed item repeats the reasoning text already streamed as deltas, so we only keep its fields 

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 ) 

768 

769 elif chunk.type == "response.completed" or chunk.type == "response.incomplete": 

770 # This means a full response is finished. 

771 return StreamingChunk( 

772 content="", 

773 component_info=component_info, 

774 finish_reason=_get_response_finish_reason(chunk.response), 

775 meta={**chunk.to_dict(), "received_at": datetime.now().isoformat()}, 

776 ) 

777 

778 elif chunk.type == "response.output_text.delta": 

779 # Start is determined by checking if this is the first text delta event of a new output_index 

780 # 1) Check if all previous chunks have different output_index 

781 # 2) If any chunks do have the same output_index, check if they have content 

782 # If none of them have content, this is the start of a new text output 

783 start = all(c.index != chunk.output_index for c in previous_chunks) or all( 

784 c.content == "" for c in previous_chunks if c.index == chunk.output_index 

785 ) 

786 return StreamingChunk( 

787 content=chunk.delta, 

788 component_info=component_info, 

789 index=chunk.output_index, 

790 start=start, 

791 meta={**chunk.to_dict(), "received_at": datetime.now().isoformat()}, 

792 ) 

793 

794 # Unlike non-streaming, where raw reasoning text is only a fallback when there is no summary, summary and raw 

795 # reasoning deltas are both streamed as they arrive and end up concatenated in the final reasoning_text. 

796 # In practice this rarely matters, because providers usually send only one kind. 

797 elif chunk.type in ("response.reasoning_summary_text.delta", "response.reasoning_text.delta"): 

798 # We remove the delta from the extra because it is already in the reasoning_text 

799 # Remaining information needs to be saved for chat message 

800 extra = chunk.to_dict() 

801 extra.pop("delta") 

802 reasoning = ReasoningContent(reasoning_text=chunk.delta, extra=extra) 

803 return StreamingChunk( 

804 content="", 

805 component_info=component_info, 

806 index=chunk.output_index, 

807 reasoning=reasoning, 

808 meta={"received_at": datetime.now().isoformat()}, 

809 ) 

810 

811 # the function arguments are streamed in parts 

812 # function name is not passed in these chunks 

813 elif chunk.type == "response.function_call_arguments.delta": 

814 arguments = chunk.delta 

815 extra = chunk.to_dict() 

816 extra.pop("delta") 

817 # in delta of tool calls there is no call_id so we use the item_id which is the function call id 

818 tool_call = ToolCallDelta(index=chunk.output_index, id=chunk.item_id, arguments=arguments, extra=extra) 

819 return StreamingChunk( 

820 content="", 

821 component_info=component_info, 

822 index=chunk.output_index, 

823 tool_calls=[tool_call], 

824 meta={"received_at": datetime.now().isoformat()}, 

825 ) 

826 

827 # we return rest of the chunk as is 

828 return StreamingChunk( 

829 content="", 

830 component_info=component_info, 

831 index=getattr(chunk, "output_index", None), 

832 meta={**chunk.to_dict(), "received_at": datetime.now().isoformat()}, 

833 ) 

834 

835 

836def _convert_streaming_chunks_to_chat_message(chunks: list[StreamingChunk]) -> ChatMessage: 

837 """ 

838 Connects the streaming chunks into a single ChatMessage. 

839 

840 :param chunks: The list of all `StreamingChunk` objects. 

841 

842 :returns: The ChatMessage. 

843 """ 

844 

845 # Get the full text by concatenating all text chunks 

846 text = "".join([chunk.content for chunk in chunks]) 

847 logprobs = [] 

848 for chunk in chunks: 

849 if chunk.meta.get("logprobs"): 

850 logprobs.append(chunk.meta.get("logprobs")) 

851 

852 # Gather reasoning information if present 

853 reasoning_id = None 

854 reasoning_text = "" 

855 for chunk in chunks: 

856 if chunk.reasoning: 

857 reasoning_text += chunk.reasoning.reasoning_text 

858 if chunk.reasoning.extra.get("id"): 

859 reasoning_id = chunk.reasoning.extra.get("id") 

860 

861 # Process tool calls if present in any chunk 

862 tool_call_data: dict[str, dict[str, Any]] = {} # Track tool calls by id 

863 for chunk in chunks: 

864 if chunk.tool_calls: 

865 for tool_call in chunk.tool_calls: 

866 # here the tool_call.id is fc_id not call_id 

867 assert tool_call.id is not None