fix: index streamed thinking_blocks so the OpenAI SDK .stream() accumulator accepts them (#46611)

* fix: index streamed thinking_blocks so the OpenAI SDK .stream() accumulator accepts them

* fix: satisfy the type discipline and basedpyright gates for the thinking block index

* chore: sync the OpenAPI snapshot and schema.d.ts with the thinking block index field

---------

Co-authored-by: devin-ai-integration[bot] <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
Kerry Lu 2026-10-14 22:21:45 -07:00 • committed by GitHub
parent 9444f53300
commit 0d741a9816
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
11 changed files with 230 additions and 18 deletions

View file

@ -2479,7 +2479,22 @@ def sanitize_messages_for_tool_calling(
def _replayable_thinking_blocks(
thinking_blocks: list[ChatCompletionThinkingBlock | ChatCompletionRedactedThinkingBlock],
) -> list[ChatCompletionThinkingBlock | ChatCompletionRedactedThinkingBlock]:
return [_with_native_redacted_data(block) for block in thinking_blocks if not is_unsignable_thinking_block(block)]
return [
_without_stream_index(_with_native_redacted_data(block))
for block in thinking_blocks
if not is_unsignable_thinking_block(block)
]
def _without_stream_index(
block: ChatCompletionThinkingBlock | ChatCompletionRedactedThinkingBlock,
) -> ChatCompletionThinkingBlock | ChatCompletionRedactedThinkingBlock:
if "index" not in block:
return block
return cast( # cast-ok: dropping one key keeps the block's TypedDict shape
ChatCompletionThinkingBlock | ChatCompletionRedactedThinkingBlock,
{key: value for key, value in block.items() if key != "index"},
)
def _with_native_redacted_data(

View file

@ -74,6 +74,8 @@ if TYPE_CHECKING:
from litellm.litellm_core_utils.streaming_handler import CustomStreamWrapper
from litellm.llms.base_llm.chat.transformation import BaseConfig
_THINKING_BLOCK_TYPES: Final = frozenset({"thinking", "redacted_thinking"})
def _loads_stream_chunk(payload: str) -> dict[str, object]:
return json.loads(payload)
@ -685,6 +687,7 @@ class ModelResponseIterator:
self._response = response
self.content_blocks: list[ContentBlockDelta] = []
self.tool_index = -1
self.thinking_block_index = -1
self.json_mode = json_mode
self.speed = speed
self._cumulative_usage: Mapping[str, object] = MappingProxyType({})
@ -832,13 +835,18 @@ class ModelResponseIterator:
ChatCompletionThinkingBlock(
type="thinking",
thinking=thinking_content,
index=self.thinking_block_index,
)
]
provider_specific_fields["thinking_blocks"] = thinking_blocks
signature: Final = content_block["delta"].get("signature")
if isinstance(signature, str) and signature:
thinking_blocks = [ChatCompletionThinkingBlock(type="thinking", thinking="", signature=signature)]
thinking_blocks = [
ChatCompletionThinkingBlock(
type="thinking", thinking="", signature=signature, index=self.thinking_block_index
)
]
provider_specific_fields["thinking_blocks"] = thinking_blocks
if reasoning_content is None:
reasoning_content = ""
@ -869,6 +877,7 @@ class ModelResponseIterator:
ChatCompletionRedactedThinkingBlock(
type="redacted_thinking",
data=content_block_start["content_block"]["data"],
index=self.thinking_block_index,
)
]
provider_specific_fields["thinking_blocks"] = thinking_blocks
@ -970,6 +979,8 @@ class ModelResponseIterator:
self.content_blocks = [] # reset content blocks when new block starts
# Track current content block type for filtering deltas
self.current_content_block_type = content_block_start["content_block"]["type"]
if self.current_content_block_type in _THINKING_BLOCK_TYPES:
self.thinking_block_index += 1
if content_block_start["content_block"]["type"] == "text":
text = self._release_answer_close() + content_block_start["content_block"]["text"]
elif (

View file

@ -483,6 +483,8 @@ class AWSEventStreamDecoder:
self.json_mode = json_mode
self._current_tool_name: str | None = None
self._thinking_ran = False
self._thinking_block_index = -1
self._thinking_content_block_index: int | None = None
self._provider_reasoning_tokens: int | None = None
def check_empty_tool_call_args(self) -> bool:
@ -514,8 +516,14 @@ class AWSEventStreamDecoder:
return reasoning_content_block["text"]
return None
def _thinking_block_index_for(self, content_block_index: int) -> int:
if self._thinking_content_block_index != content_block_index:
self._thinking_content_block_index = content_block_index
self._thinking_block_index += 1
return self._thinking_block_index
def translate_thinking_blocks(
self, thinking_block: BedrockConverseReasoningContentBlockDelta
self, thinking_block: BedrockConverseReasoningContentBlockDelta, content_block_index: int = 0
) -> list[ChatCompletionThinkingBlock | ChatCompletionRedactedThinkingBlock] | None:
"""
Translate the thinking blocks to a string
@ -523,17 +531,20 @@ class AWSEventStreamDecoder:
thinking_blocks_list: Final[list[ChatCompletionThinkingBlock | ChatCompletionRedactedThinkingBlock]] = []
_thinking_block: ChatCompletionThinkingBlock | ChatCompletionRedactedThinkingBlock | None = None
index: Final = self._thinking_block_index_for(content_block_index)
if "text" in thinking_block:
_thinking_block = ChatCompletionThinkingBlock(type="thinking")
_thinking_block = ChatCompletionThinkingBlock(type="thinking", index=index)
_thinking_block["thinking"] = thinking_block["text"]
elif "signature" in thinking_block:
_thinking_block = ChatCompletionThinkingBlock(type="thinking")
_thinking_block = ChatCompletionThinkingBlock(type="thinking", index=index)
_thinking_block["signature"] = thinking_block["signature"]
_thinking_block["thinking"] = "" # consistent with anthropic response
elif "redactedContent" in thinking_block:
_thinking_block = ChatCompletionRedactedThinkingBlock(
type="redacted_thinking", data=native_redacted_thinking_data(thinking_block["redactedContent"])
type="redacted_thinking",
data=native_redacted_thinking_data(thinking_block["redactedContent"]),
index=index,
)
if _thinking_block is not None:
thinking_blocks_list.append(_thinking_block)
@ -553,6 +564,7 @@ class AWSEventStreamDecoder:
def _handle_converse_start_event(
self,
start_obj: ContentBlockStartEvent,
content_block_index: int = 0,
) -> tuple[
ChatCompletionToolCallChunk | None,
dict,
@ -589,7 +601,7 @@ class AWSEventStreamDecoder:
elif (
"reasoningContent" in start_obj and start_obj["reasoningContent"] is not None
): # redacted thinking can be in start object
thinking_blocks = self.translate_thinking_blocks(start_obj["reasoningContent"])
thinking_blocks = self.translate_thinking_blocks(start_obj["reasoningContent"], content_block_index)
provider_specific_fields = {
"reasoningContent": start_obj["reasoningContent"],
}
@ -636,7 +648,7 @@ class AWSEventStreamDecoder:
"reasoningContent": delta_obj["reasoningContent"],
}
reasoning_content = self.extract_reasoning_content_str(delta_obj["reasoningContent"])
thinking_blocks = self.translate_thinking_blocks(delta_obj["reasoningContent"])
thinking_blocks = self.translate_thinking_blocks(delta_obj["reasoningContent"], index)
if thinking_blocks and len(thinking_blocks) > 0 and reasoning_content is None:
reasoning_content = "" # set to non-empty string to ensure consistency with Anthropic
elif "citationsContent" in delta_obj:
@ -698,7 +710,7 @@ class AWSEventStreamDecoder:
tool_use,
provider_specific_fields,
thinking_blocks,
) = self._handle_converse_start_event(start_obj)
) = self._handle_converse_start_event(start_obj, content_block_index)
elif "delta" in chunk_data:
delta_obj: Final = ContentBlockDeltaEvent(**chunk_data["delta"])
(

View file

@ -823,7 +823,11 @@ class DatabricksChatResponseIterator(BaseModelResponseIterator):
if block_reasoning_content is not None
else DatabricksConfig.extract_top_level_reasoning_content(choice["delta"])
)
choice["delta"]["thinking_blocks"] = thinking_blocks
choice["delta"]["thinking_blocks"] = (
[{**block, "index": position} for position, block in enumerate(thinking_blocks)]
if thinking_blocks
else thinking_blocks
)
translated_choices.append(choice)
service_tier: Final = chunk.get("service_tier")
if isinstance(service_tier, str) and service_tier:

View file

@ -46344,6 +46344,10 @@
"title": "Data",
"type": "string"
},
"index": {
"title": "Index",
"type": "integer"
},
"type": {
"const": "redacted_thinking",
"title": "Type",
@ -46429,6 +46433,10 @@
],
"title": "Cache Control"
},
"index": {
"title": "Index",
"type": "integer"
},
"signature": {
"anyOf": [
{

View file

@ -688,12 +688,14 @@ class ChatCompletionThinkingBlock(TypedDict, total=False):
thinking: str
signature: str | None
cache_control: dict | ChatCompletionCachedContent | None
index: ReadOnly[int]
class ChatCompletionRedactedThinkingBlock(TypedDict, total=False):
type: Required[Literal["redacted_thinking"]]
data: str
cache_control: dict | ChatCompletionCachedContent | None
index: ReadOnly[int]
class ChatCompletionReasoningSummaryTextBlock(TypedDict, total=False):

View file

@ -312,6 +312,31 @@ def test_convert_to_ollama_image_downloads_an_http_image_url_to_bare_base64(monk
@pytest.mark.asyncio
def test_anthropic_messages_pt_drops_the_stream_index_from_replayed_thinking_blocks():
"""Streamed thinking blocks carry the OpenAI SDK list-position `index`. Anthropic's native thinking block schema
rejects unknown fields, so the field must not reach the request when a client replays the streamed message"""
messages: Final = [
{"role": "user", "content": "Capital of France?"},
{
"role": "assistant",
"content": "Paris",
"thinking_blocks": [
{"type": "thinking", "thinking": "Step 1.", "signature": "sig-a", "index": 0},
{"type": "redacted_thinking", "data": "EmwKAhgB", "index": 1},
],
},
{"role": "user", "content": "And Spain?"},
]
replayed: Final = anthropic_messages_pt(messages=messages, model="claude-haiku-4-5", llm_provider="anthropic")
assert replayed[1]["content"] == [
{"type": "thinking", "thinking": "Step 1.", "signature": "sig-a"},
{"type": "redacted_thinking", "data": "EmwKAhgB"},
{"type": "text", "text": "Paris"},
]
async def test_anthropic_bedrock_thinking_blocks_with_none_content():
"""
Test the specific function that processes thinking_blocks when content is None

View file

@ -8,6 +8,8 @@ from unittest.mock import AsyncMock, MagicMock, patch
import httpx
import pytest
import respx
from openai.lib.streaming.chat import ChatCompletionStreamState
from openai.types.chat import ChatCompletionChunk
import litellm
from litellm._uuid import uuid
@ -309,10 +311,10 @@ def test_streaming_thinking_blocks_are_replayable_after_signature_delta():
block for chunk in parsed_chunks for block in (getattr(chunk.choices[0].delta, "thinking_blocks", None) or [])
)
expected_delta_blocks = (
{"type": "thinking", "thinking": "Step 1. "},
{"type": "thinking", "thinking": "Step 2."},
{"type": "thinking", "thinking": "Step 1. ", "index": 0},
{"type": "thinking", "thinking": "Step 2.", "index": 0},
)
expected_signature_block = {"type": "thinking", "thinking": "", "signature": "sig-final"}
expected_signature_block = {"type": "thinking", "thinking": "", "signature": "sig-final", "index": 0}
assert reasoning_content == "Step 1. Step 2."
assert thinking_blocks == (*expected_delta_blocks, expected_signature_block)
@ -422,8 +424,8 @@ def test_streaming_unsigned_thinking_deltas_keep_reasoning_content():
assert reasoning_content == "Step 1. Step 2."
assert thinking_blocks == (
{"type": "thinking", "thinking": "Step 1. "},
{"type": "thinking", "thinking": "Step 2."},
{"type": "thinking", "thinking": "Step 1. ", "index": 0},
{"type": "thinking", "thinking": "Step 2.", "index": 0},
)
@ -457,11 +459,74 @@ def test_streaming_truncated_thinking_deltas_keep_reasoning_content():
assert reasoning_content == "Step 1. Step 2."
assert thinking_blocks == (
{"type": "thinking", "thinking": "Step 1. "},
{"type": "thinking", "thinking": "Step 2."},
{"type": "thinking", "thinking": "Step 1. ", "index": 0},
{"type": "thinking", "thinking": "Step 2.", "index": 0},
)
def test_streamed_thinking_blocks_accumulate_in_the_openai_sdk_and_replay_without_index():
"""`client.chat.completions.stream()` feeds every delta through the OpenAI SDK's `ChatCompletionStreamState`,
which requires each dict inside a list to carry a list-position `index`. Unindexed `thinking_blocks` crashed it
with `RuntimeError: Expected list delta entry to have an \`index\` key` on the second thinking chunk"""
iterator: Final = ModelResponseIterator(streaming_response=MagicMock(), sync_stream=True, json_mode=False)
events: Final = (
{
"type": "message_start",
"message": {
"id": "msg_two_blocks",
"type": "message",
"role": "assistant",
"model": "claude-haiku-4-5",
"content": [],
"stop_reason": None,
"usage": {"input_tokens": 20, "output_tokens": 1},
},
},
{"type": "content_block_start", "index": 0, "content_block": {"type": "thinking", "thinking": ""}},
{"type": "content_block_delta", "index": 0, "delta": {"type": "thinking_delta", "thinking": "Step 1. "}},
{"type": "content_block_delta", "index": 0, "delta": {"type": "thinking_delta", "thinking": "Step 2."}},
{"type": "content_block_delta", "index": 0, "delta": {"type": "signature_delta", "signature": "sig-a"}},
{"type": "content_block_stop", "index": 0},
{"type": "content_block_start", "index": 1, "content_block": {"type": "text", "text": ""}},
{"type": "content_block_delta", "index": 1, "delta": {"type": "text_delta", "text": "Paris"}},
{"type": "content_block_stop", "index": 1},
{"type": "content_block_start", "index": 2, "content_block": {"type": "thinking", "thinking": ""}},
{"type": "content_block_delta", "index": 2, "delta": {"type": "thinking_delta", "thinking": "Step 3."}},
{"type": "content_block_delta", "index": 2, "delta": {"type": "signature_delta", "signature": "sig-b"}},
{"type": "content_block_stop", "index": 2},
{"type": "message_delta", "delta": {"stop_reason": "end_turn", "stop_sequence": None}, "usage": {"output_tokens": 9}},
{"type": "message_stop"},
)
state: Final = ChatCompletionStreamState()
for event in events:
state.handle_chunk(ChatCompletionChunk.model_validate(iterator.chunk_parser(chunk=event).model_dump()))
message: Final = state.get_final_completion().choices[0].message.model_dump()
assert message["content"] == "Paris"
assert message["thinking_blocks"] == [
{"type": "thinking", "thinking": "Step 1. Step 2.", "signature": "sig-a", "index": 0},
{"type": "thinking", "thinking": "Step 3.", "signature": "sig-b", "index": 1},
]
next_turn: Final = {
"role": "assistant",
"content": message["content"],
"thinking_blocks": message["thinking_blocks"],
}
replayed: Final = anthropic_messages_pt(
messages=[{"role": "user", "content": "Capital of France?"}, next_turn, {"role": "user", "content": "And Spain?"}],
model="claude-haiku-4-5",
llm_provider="anthropic",
)
assert replayed[1]["content"] == [
{"type": "thinking", "thinking": "Step 1. Step 2.", "signature": "sig-a"},
{"type": "thinking", "thinking": "Step 3.", "signature": "sig-b"},
{"type": "text", "text": "Paris"},
]
def test_handle_json_mode_chunk_response_format_tool():
model_response_iterator = ModelResponseIterator(streaming_response=MagicMock(), sync_stream=True, json_mode=True)
response_format_tool = ChatCompletionToolCallChunk(

View file

@ -51,7 +51,33 @@ def test_transform_thinking_blocks_unwraps_converse_redacted_content_to_claude_n
{"redactedContent": base64.b64encode(native_data.encode()).decode()}
)
assert thinking_blocks == [{"type": "redacted_thinking", "data": native_data}]
assert thinking_blocks == [{"type": "redacted_thinking", "data": native_data, "index": 0}]
def test_converse_reasoning_deltas_index_thinking_blocks_by_reasoning_block_for_the_openai_sdk():
"""The OpenAI SDK's `.stream()` accumulator needs a list-position `index` on every streamed `thinking_blocks`
entry, stable across one reasoning block's fragments and sequential across blocks even when a text block sits
between them (Converse `contentBlockIndex` 0 and 2 become thinking indexes 0 and 1)"""
decoder: Final = AWSEventStreamDecoder(model="test")
chunks: Final = (
{"delta": {"reasoningContent": {"text": "Step 1. "}}, "contentBlockIndex": 0},
{"delta": {"reasoningContent": {"text": "Step 2."}}, "contentBlockIndex": 0},
{"delta": {"reasoningContent": {"signature": "sig-a"}}, "contentBlockIndex": 0},
{"delta": {"text": "Paris"}, "contentBlockIndex": 1},
{"delta": {"reasoningContent": {"text": "Step 3."}}, "contentBlockIndex": 2},
{"delta": {"reasoningContent": {"signature": "sig-b"}}, "contentBlockIndex": 2},
)
deltas: Final = tuple(decoder.chunk_parser(chunk).choices[0].delta for chunk in chunks)
streamed: Final = tuple(block for delta in deltas for block in (getattr(delta, "thinking_blocks", None) or []))
assert streamed == (
{"type": "thinking", "thinking": "Step 1. ", "index": 0},
{"type": "thinking", "thinking": "Step 2.", "index": 0},
{"type": "thinking", "thinking": "", "signature": "sig-a", "index": 0},
{"type": "thinking", "thinking": "Step 3.", "index": 1},
{"type": "thinking", "thinking": "", "signature": "sig-b", "index": 1},
)
def test_transform_tool_calls_index():

View file

@ -7,6 +7,8 @@ from unittest.mock import MagicMock, patch
import httpx
import pytest
import respx
from openai.lib.streaming.chat import ChatCompletionStreamState
from openai.types.chat import ChatCompletionChunk
from pydantic import TypeAdapter
from fastapi.testclient import TestClient
from pydantic import BaseModel, ConfigDict
@ -333,6 +335,44 @@ def test_transform_choices_with_citations():
}
def _databricks_reasoning_chunk(chunk_id: str, summaries: list[dict[str, str]], text: str | None = None) -> dict:
content: Final = [{"type": "reasoning", "summary": summaries}] + ([{"type": "text", "text": text}] if text else [])
return {
"id": chunk_id,
"object": "chat.completion.chunk",
"created": 0,
"model": "databricks-qwen",
"choices": [{"delta": {"role": "assistant", "content": content}, "index": 0, "finish_reason": None}],
}
def test_streamed_reasoning_summaries_accumulate_in_the_openai_sdk_stream_state():
"""`client.chat.completions.stream()` runs every delta through the OpenAI SDK's `ChatCompletionStreamState`,
which raises `RuntimeError: Expected list delta entry to have an \`index\` key` on the second chunk unless each
streamed `thinking_blocks` entry carries its list position"""
iterator: Final = DatabricksChatResponseIterator(None, sync_stream=True)
chunks: Final = (
_databricks_reasoning_chunk("c1", [{"type": "summary_text", "text": "Step 1. "}]),
_databricks_reasoning_chunk("c1", [{"type": "summary_text", "text": "Step 2.", "signature": "sig-a"}]),
_databricks_reasoning_chunk(
"c1",
[{"type": "summary_text", "text": "", "signature": ""}, {"type": "summary_text", "text": "Step 3."}],
text="391",
),
)
state: Final = ChatCompletionStreamState()
for chunk in chunks:
state.handle_chunk(ChatCompletionChunk.model_validate(iterator.chunk_parser(chunk).model_dump()))
message: Final = state.get_final_completion().choices[0].message.model_dump()
assert message["content"] == "391"
assert message["thinking_blocks"] == [
{"type": "thinking", "thinking": "Step 1. Step 2.", "signature": "sig-a", "index": 0},
{"type": "thinking", "thinking": "Step 3.", "signature": "", "index": 1},
]
def test_chunk_parser_with_citation():
iterator = DatabricksChatResponseIterator(None, sync_stream=True)
chunk = {

View file

@ -28762,6 +28762,8 @@ export interface components {
} | components["schemas"]["ChatCompletionCachedContent"] | null;
/** Data */
data?: string;
/** Index */
index?: number;
/**
* Type
* @constant
@ -28800,6 +28802,8 @@ export interface components {
cache_control?: {
[key: string]: unknown;
} | components["schemas"]["ChatCompletionCachedContent"] | null;
/** Index */
index?: number;
/** Signature */
signature?: string | null;
/** Thinking */