fix(anthropic): keep each streamed reasoning summary with its own item

A summary delta carries no item index, so the next item's summary joined the block that already got its signature. A thinking chunk after an indexed signature now opens a new block.
This commit is contained in:
mateo-berri 2026-11-06 20:45:06 -08:00
parent ba5d7b83b9
commit 27574a429a
2 changed files with 119 additions and 8 deletions

View file

@ -72,14 +72,21 @@ def _optional_attr_sequence(obj: object, name: str) -> Sequence[object]:
return value if value else ()
def _first_thinking_block_index(chunk: "ModelResponseStream") -> int | None:
def _first_thinking_block(chunk: "ModelResponseStream") -> Mapping[str, object] | None:
delta: Final = _optional_attr(chunk.choices[0], "delta") if chunk.choices else None
blocks: Final = _optional_attr_sequence(delta, "thinking_blocks")
first_block: Final = as_str_mapping(blocks[0]) if blocks else None
index: Final = first_block.get("index") if first_block is not None else None
return as_str_mapping(blocks[0]) if blocks else None
def _thinking_block_index(block: Mapping[str, object] | None) -> int | None:
index: Final = block.get("index") if block is not None else None
return index if isinstance(index, int) else None
def _closes_thinking_block(block: Mapping[str, object] | None) -> bool:
return block is not None and _thinking_block_index(block) is not None and bool(block.get("signature"))
def _error_status_and_message(exc: Exception) -> tuple[int, str]:
if isinstance(exc, (BaseLLMException, MidStreamFallbackError, ContextWindowExceededError)):
return exc.status_code, exc.message
@ -474,6 +481,7 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
queued_usage_chunk: bool = False
current_content_block_index: int = 0
current_thinking_block_index: int | None = None
current_thinking_block_closed: bool = False
def __init__(
self,
@ -1362,9 +1370,16 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
original_name: Final = self.tool_name_mapping.get(truncated_name, truncated_name)
tool_block["name"] = original_name
thinking_block_index: Final = _first_thinking_block_index(chunk)
thinking_block: Final = _first_thinking_block(chunk)
thinking_block_index: Final = _thinking_block_index(thinking_block)
previous_thinking_block_index: Final = self.current_thinking_block_index
if thinking_block_index is not None or block_type != self.current_content_block_type:
follows_closed_thinking_block: Final = self.current_thinking_block_closed and block_type == "thinking"
self.current_thinking_block_closed = _closes_thinking_block(thinking_block)
if (
thinking_block_index is not None
or block_type != self.current_content_block_type
or follows_closed_thinking_block
):
self.current_thinking_block_index = thinking_block_index
if block_type != self.current_content_block_type:
@ -1372,7 +1387,7 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
self.current_content_block_start = content_block_start
return True
if (
if follows_closed_thinking_block or (
thinking_block_index is not None
and previous_thinking_block_index is not None
and thinking_block_index != previous_thinking_block_index

View file

@ -5625,8 +5625,13 @@ def _responses_sse(events: Sequence[Mapping[str, object]]) -> bytes:
)
def _reasoning_item(item_id: str, encrypted: str) -> dict[str, object]:
return {"type": "reasoning", "id": item_id, "summary": [], "encrypted_content": encrypted}
def _reasoning_item(item_id: str, encrypted: str, summary: str = "") -> dict[str, object]:
return {
"type": "reasoning",
"id": item_id,
"summary": [{"type": "summary_text", "text": summary}] if summary else [],
"encrypted_content": encrypted,
}
@pytest.mark.asyncio
@ -5688,3 +5693,94 @@ async def test_streamed_consecutive_reasoning_items_each_become_their_own_block(
("redacted_thinking", encrypted_reasoning_signature("gAAA_second")),
("text", None),
]
def _streamed_blocks(frames: str) -> list[tuple[str, str, str | None]]:
events: Final = tuple(
json.loads(line.removeprefix("data: ")) for line in frames.splitlines() if line.startswith("data: ")
)
deltas: Final = tuple(event for event in events if event["type"] == "content_block_delta")
def block_deltas(index: int) -> tuple[Mapping[str, str], ...]:
return tuple(delta["delta"] for delta in deltas if delta["index"] == index)
return [
(
event["content_block"]["type"],
"".join(delta.get("thinking", "") for delta in block_deltas(event["index"])),
next(
(delta["signature"] for delta in block_deltas(event["index"]) if delta["type"] == "signature_delta"),
None,
),
)
for event in events
if event["type"] == "content_block_start"
]
@pytest.mark.asyncio
@pytest.mark.usefixtures("_chat_bridge_httpx")
async def test_streamed_reasoning_summaries_stay_with_their_own_encrypted_item(
respx_mock: respx.MockRouter,
) -> None:
response: Final = {**_RETRY_REPLY, "id": "resp_two_summaries", "status": "in_progress", "output": []}
message: Final = {
"type": "message",
"id": "msg_two_summaries",
"status": "completed",
"role": "assistant",
"content": [{"type": "output_text", "text": "Oslo", "annotations": []}],
}
first: Final = _reasoning_item("rs_1", "gAAA_first", "Oslo is further north.")
second: Final = _reasoning_item("rs_2", "gAAA_second", "Check the forecast.")
completed: Final = {**response, "status": "completed", "output": [first, second, message]}
def summary_delta(output_index: int, item_id: str, text: str) -> dict[str, object]:
return {
"type": "response.reasoning_summary_text.delta",
"output_index": output_index,
"item_id": item_id,
"summary_index": 0,
"delta": text,
}
events: Final = (
{"type": "response.created", "response": response},
{"type": "response.output_item.added", "output_index": 0, "item": _reasoning_item("rs_1", "")},
summary_delta(0, "rs_1", "Oslo is further north."),
{"type": "response.output_item.done", "output_index": 0, "item": first},
{"type": "response.output_item.added", "output_index": 1, "item": _reasoning_item("rs_2", "")},
summary_delta(1, "rs_2", "Check the forecast."),
{"type": "response.output_item.done", "output_index": 1, "item": second},
{"type": "response.output_item.added", "output_index": 2, "item": {**message, "content": []}},
{
"type": "response.output_text.delta",
"output_index": 2,
"item_id": "msg_two_summaries",
"content_index": 0,
"delta": "Oslo",
},
{"type": "response.output_item.done", "output_index": 2, "item": message},
{"type": "response.completed", "response": completed},
)
respx_mock.post(url__regex=_RETRY_URL).mock(
return_value=httpx.Response(200, headers={"content-type": "text/event-stream"}, content=_responses_sse(events))
)
stream: Final = await litellm.anthropic.messages.acreate(
model="azure/gpt-5.4-nano",
api_base=_RETRY_API_BASE,
api_key="fake-key",
api_version="2025-04-01-preview",
max_tokens=64,
tools=_RETRY_TOOLS,
stream=True,
messages=[{"role": "user", "content": "Which city is colder?"}],
)
frames: Final = b"".join([chunk async for chunk in stream]).decode()
assert _streamed_blocks(frames) == [
("thinking", "Oslo is further north.", encrypted_reasoning_signature("gAAA_first")),
("thinking", "Check the forecast.", encrypted_reasoning_signature("gAAA_second")),
("text", "", None),
]