fix: preserve reasoning from compatible Chat Completions providers

docs-website/docs/pipeline-components/generators/openaichatgenerator.mdx

@@ -93,6 +93,8 @@ print(response["replies"][0].text)
 
 ### Streaming
 
+For OpenAI-compatible providers that return reasoning text, non-empty string `reasoning` or `reasoning_content` fields populate `ChatMessage.reasoning` and `StreamingChunk.reasoning`, separately from the answer text. `reasoning` takes precedence when both contain text; empty or non-string values are ignored. A provider delta containing both reasoning and text produces separate callback chunks. Reasoning tokens in usage metadata alone do not provide reasoning text.
+
 You can stream output as it’s generated. Pass a callback to `streaming_callback`. Use the built-in `print_streaming_chunk` to print text tokens and tool events (tool calls and tool results).
 
 ```python

docs-website/versioned_docs/version-3.3/pipeline-components/generators/openaichatgenerator.mdx

@@ -93,6 +93,8 @@ print(response["replies"][0].text)
 
 ### Streaming
 
+For OpenAI-compatible providers that return reasoning text, non-empty string `reasoning` or `reasoning_content` fields populate `ChatMessage.reasoning` and `StreamingChunk.reasoning`, separately from the answer text. `reasoning` takes precedence when both contain text; empty or non-string values are ignored. A provider delta containing both reasoning and text produces separate callback chunks. Reasoning tokens in usage metadata alone do not provide reasoning text.
+
 You can stream output as it’s generated. Pass a callback to `streaming_callback`. Use the built-in `print_streaming_chunk` to print text tokens and tool events (tool calls and tool results).
 
 ```python

haystack/components/generators/chat/openai.py

@@ -5,6 +5,7 @@
 import asyncio
 import json
 import os
+from dataclasses import replace
 from datetime import datetime
 from typing import Any, ClassVar
 
@@ -32,6 +33,7 @@
     ChatMessage,
     ComponentInfo,
     FinishReason,
+    ReasoningContent,
     StreamingCallbackT,
     StreamingChunk,
     SyncStreamingCallbackT,
@@ -63,6 +65,11 @@ class OpenAIChatGenerator:
     from OpenAI API. It uses [ChatMessage](https://docs.haystack.deepset.ai/docs/chatmessage)
     format in input and output.
 
+    For OpenAI-compatible providers, non-empty string `reasoning` or `reasoning_content` response fields
+    populate `ChatMessage.reasoning` and `StreamingChunk.reasoning`, separately from text. `reasoning` takes
+    precedence; empty or non-string values are ignored. A delta containing both reasoning and text produces
+    separate streaming chunks.
+
     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
@@ -566,28 +573,32 @@ def _prepare_api_call(  # noqa: PLR0913
     def _handle_stream_response(self, chat_completion: Stream, callback: SyncStreamingCallbackT) -> list[ChatMessage]:
         component_info = ComponentInfo.from_component(self)
         chunks: list[StreamingChunk] = []
+        block_indices: dict[str | int, int] = {}
         for chunk in chat_completion:
             assert len(chunk.choices) <= 1, "Streaming responses should have at most one choice."
-            chunk_delta = _convert_chat_completion_chunk_to_streaming_chunk(
-                chunk=chunk, previous_chunks=chunks, component_info=component_info
+            chunk_deltas = _convert_chat_completion_chunk_to_streaming_chunks(
+                chunk=chunk, block_indices=block_indices, previous_chunks=chunks, component_info=component_info
             )
-            chunks.append(chunk_delta)
-            callback(chunk_delta)
+            for chunk_delta in chunk_deltas:
+                chunks.append(chunk_delta)
+                callback(chunk_delta)
         return [_convert_streaming_chunks_to_chat_message(chunks=chunks)]
 
     async def _handle_async_stream_response(
         self, chat_completion: AsyncStream, callback: StreamingCallbackT
     ) -> list[ChatMessage]:
         component_info = ComponentInfo.from_component(self)
         chunks: list[StreamingChunk] = []
+        block_indices: dict[str | int, int] = {}
         try:
             async for chunk in chat_completion:
                 assert len(chunk.choices) <= 1, "Streaming responses should have at most one choice."
-                chunk_delta = _convert_chat_completion_chunk_to_streaming_chunk(
-                    chunk=chunk, previous_chunks=chunks, component_info=component_info
+                chunk_deltas = _convert_chat_completion_chunk_to_streaming_chunks(
+                    chunk=chunk, block_indices=block_indices, previous_chunks=chunks, component_info=component_info
                 )
-                chunks.append(chunk_delta)
-                await _invoke_streaming_callback(callback, chunk_delta)
+                for chunk_delta in chunk_deltas:
+                    chunks.append(chunk_delta)
+                    await _invoke_streaming_callback(callback, chunk_delta)
 
         except asyncio.CancelledError:
             await asyncio.shield(chat_completion.close())
@@ -648,6 +659,16 @@ def _check_finish_reason(meta: dict[str, Any]) -> None:
         )
 
 
+def _get_reasoning_text(message: BaseModel | None) -> str | None:
+    # Compatible providers expose these non-standard fields as SDK model extras.
+    extra = (message.model_extra or {}) if message is not None else {}
+    for field in ("reasoning", "reasoning_content"):
+        value = extra.get(field)
+        if isinstance(value, str) and value:
+            return value
+    return None
+
+
 def _convert_chat_completion_to_chat_message(
     completion: ChatCompletion | ParsedChatCompletion, choice: Choice
 ) -> ChatMessage:
@@ -690,22 +711,26 @@ def _convert_chat_completion_to_chat_message(
     if logprobs:
         meta["logprobs"] = logprobs
 
-    return ChatMessage.from_assistant(text=text, tool_calls=tool_calls, meta=meta)
+    return ChatMessage.from_assistant(
+        text=text, tool_calls=tool_calls, meta=meta, reasoning=_get_reasoning_text(message)
+    )
 
 
-def _convert_chat_completion_chunk_to_streaming_chunk(
-    chunk: ChatCompletionChunk, previous_chunks: list[StreamingChunk], component_info: ComponentInfo | None = None
-) -> StreamingChunk:
+def _convert_chat_completion_chunk_to_streaming_chunks(
+    chunk: ChatCompletionChunk,
+    block_indices: dict[str | int, int],
+    previous_chunks: list[StreamingChunk],
+    component_info: ComponentInfo | None = None,
+) -> list[StreamingChunk]:
     """
-    Converts the streaming response chunk from the OpenAI API to a StreamingChunk.
+    Convert an SDK delta, splitting reasoning and other content into separate blocks.
 
     :param chunk: The chunk returned by the OpenAI 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 API.
+    :param block_indices: Per-response mapping of text, reasoning, and provider tool indices to content block indices.
+        Updated as new blocks arrive.
+    :param previous_chunks: Previously emitted chunks, used to preserve the non-reasoning start markers.
+    :param component_info: Information about the component that generated the chunk.
+    :returns: StreamingChunks representing the delta, with completion metadata on the last chunk.
     """
     finish_reason_mapping: dict[str, FinishReason] = {
         "stop": "stop",
@@ -714,93 +739,94 @@ def _convert_chat_completion_chunk_to_streaming_chunk(
         "tool_calls": "tool_calls",
         "function_call": "tool_calls",
     }
-    # On very first chunk so len(previous_chunks) == 0, the Choices field only provides role info (e.g. "assistant")
-    # Choices is empty if include_usage is set to True where the usage information is returned.
-    if len(chunk.choices) == 0:
-        return StreamingChunk(
-            content="",
-            component_info=component_info,
-            # Index is None since it's only set to an int when a content block is present
-            index=None,
-            finish_reason=None,
-            meta={
-                "model": chunk.model,
-                "received_at": datetime.now().isoformat(),
-                "usage": _serialize_object(chunk.usage),
-            },
-        )
+    meta = {"model": chunk.model, "received_at": datetime.now().isoformat(), "usage": _serialize_object(chunk.usage)}
+    if not chunk.choices:
+        return [StreamingChunk(content="", component_info=component_info, meta=meta)]
 
     choice: ChunkChoice = chunk.choices[0]
+    delta = choice.delta
+    meta.update(
+        index=choice.index,
+        tool_calls=delta.tool_calls if delta and delta.tool_calls else None,
+        finish_reason=choice.finish_reason,
+    )
+    finish_reason = finish_reason_mapping.get(choice.finish_reason) if choice.finish_reason else None
+    reasoning_text = _get_reasoning_text(delta)
+    has_reasoning = reasoning_text is not None or "reasoning" in block_indices
+
+    result = []
+    if reasoning_text is not None:
+        start = "reasoning" not in block_indices
+        index = block_indices.setdefault("reasoning", max(block_indices.values(), default=-1) + 1)
+        result.append(
+            StreamingChunk(
+                content="",
+                reasoning=ReasoningContent(reasoning_text=reasoning_text),
+                index=index,
+                start=start,
+                component_info=component_info,
+                meta={**meta, "tool_calls": None},
+            )
+        )
+
+    if delta and delta.content and (has_reasoning or not delta.tool_calls):
+        start = "text" not in block_indices if has_reasoning else len(previous_chunks) == 1
+        text_index = (
+            None
+            if not has_reasoning and delta.role is not None
+            else block_indices.setdefault("text", max(block_indices.values(), default=-1) + 1 if has_reasoning else 0)
+        )
+        text_meta = {**meta, "tool_calls": None}
+        if choice.logprobs:
+            text_meta["logprobs"] = _serialize_object(choice.logprobs)
+        result.append(
+            StreamingChunk(
+                content=delta.content, index=text_index, start=start, component_info=component_info, meta=text_meta
+            )
+        )
 
-    # create a list of ToolCallDelta objects from the tool calls
-    if choice.delta and choice.delta.tool_calls:
+    if delta and delta.tool_calls:
         tool_calls_deltas = []
-        for tool_call in choice.delta.tool_calls:
+        for tool_call in delta.tool_calls:
+            block_indices.setdefault(
+                tool_call.index, max(block_indices.values(), default=-1) + 1 if has_reasoning else tool_call.index
+            )
             function = tool_call.function
             tool_calls_deltas.append(
                 ToolCallDelta(
+                    # Reconstruction groups and sorts tools by this provider index, not by arrival order.
                     index=tool_call.index,
                     id=tool_call.id,
                     tool_name=function.name if function else None,
                     arguments=function.arguments if function and function.arguments else None,
                 )
             )
-        return StreamingChunk(
-            content=choice.delta.content or "",
-            component_info=component_info,
-            # We adopt the first tool_calls_deltas.index as the overall index of the chunk.
-            index=tool_calls_deltas[0].index,
-            tool_calls=tool_calls_deltas,
-            start=tool_calls_deltas[0].tool_name is not None,
-            finish_reason=finish_reason_mapping.get(choice.finish_reason) if choice.finish_reason else None,
-            meta={
-                "model": chunk.model,
-                "index": choice.index,
-                "tool_calls": choice.delta.tool_calls,
-                "finish_reason": choice.finish_reason,
-                "received_at": datetime.now().isoformat(),
-                "usage": _serialize_object(chunk.usage),
-            },
+        result.append(
+            StreamingChunk(
+                content="" if has_reasoning else delta.content or "",
+                index=block_indices[tool_calls_deltas[0].index],
+                tool_calls=tool_calls_deltas,
+                start=tool_calls_deltas[0].tool_name is not None,
+                component_info=component_info,
+                meta=meta,
+            )
         )
 
-    # On very first chunk the choice field only provides role info (e.g. "assistant") so we set index to None
-    # We set all chunks missing the content field to index of None. E.g. can happen if chunk only contains finish
-    # reason.
-    if choice.delta and (choice.delta.content is None or choice.delta.role is not None):
-        resolved_index = None
-    else:
-        # We set the index to be 0 since if text content is being streamed then no tool calls are being streamed
-        # NOTE: We may need to revisit this if OpenAI allows planning/thinking content before tool calls like
-        #       Anthropic Claude
-        resolved_index = 0
-
-    # Initialize meta dictionary
-    meta = {
-        "model": chunk.model,
-        "index": choice.index,
-        "tool_calls": choice.delta.tool_calls if choice.delta and choice.delta.tool_calls else None,
-        "finish_reason": choice.finish_reason,
-        "received_at": datetime.now().isoformat(),
-        "usage": _serialize_object(chunk.usage),
-    }
-
-    # check if logprobs are present
-    # logprobs are returned only for text content
-    logprobs = _serialize_object(choice.logprobs) if choice.logprobs else None
-    if logprobs:
-        meta["logprobs"] = logprobs
+    if not result:
+        if choice.logprobs:
+            meta["logprobs"] = _serialize_object(choice.logprobs)
+        result.append(
+            StreamingChunk(
+                content="",
+                component_info=component_info,
+                meta=meta,
+                # Preserve legacy empty/null delta indices and start markers when reasoning is absent.
+                index=None if has_reasoning or (delta and (delta.content is None or delta.role is not None)) else 0,
+                start=not has_reasoning and len(previous_chunks) == 1,
+            )
+        )
 
-    content = ""
-    if choice.delta and choice.delta.content:
-        content = choice.delta.content
-
-    return StreamingChunk(
-        content=content,
-        component_info=component_info,
-        index=resolved_index,
-        # The first chunk is always a start message chunk that only contains role information, so if we reach here
-        # and previous_chunks is length 1 then this is the start of text content.
-        start=len(previous_chunks) == 1,
-        finish_reason=finish_reason_mapping.get(choice.finish_reason) if choice.finish_reason else None,
-        meta=meta,
-    )
+    # A single SDK delta can contain reasoning, text, and a finish reason. Signal completion only after all its content.
+    return [
+        replace(converted, meta={**converted.meta, "finish_reason": None, "usage": None}) for converted in result[:-1]
+    ] + [replace(result[-1], finish_reason=finish_reason)]

releasenotes/notes/openai-chat-reasoning-7e25f15ac5ff3d32.yaml

@@ -0,0 +1,9 @@
+---
+fixes:
+  - |
+    ``OpenAIChatGenerator`` now preserves non-empty string ``reasoning`` and
+    ``reasoning_content`` fields returned by OpenAI-compatible Chat Completions
+    providers in ``ChatMessage.reasoning`` and ``StreamingChunk.reasoning``.
+    Previously these fields were discarded. Reasoning stays separate from answer
+    text, including when both arrive in the same streaming delta. The
+    ``reasoning`` field takes precedence; empty or non-string values are ignored.