scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-13 23:11:40 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 101

fix(anthropic): stream reasoning_content into a thinking block in /v1/messages

Browse source 
Route Responses API reasoning_summary_text.delta events into a dedicated
thinking content block instead of the text block, and emit message_start
exactly once. Fixes empty assistant content in the Anthropic SDK / Claude
Code when a LiteLLM proxy fronts an OpenAI-compatible reasoning backend.

Closes #32357
This commit is contained in:
Devin AI 2026-07-07 19:34:27 +00:00
parent 8c0e3c0509
commit c42f313099
2 changed files with 162 additions and 73 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

119
litellm/llms/anthropic/experimental_pass_through/responses_adapters/streaming_iterator.py
View file

@ -3,7 +3,7 @@
import json
import traceback
from collections import deque
from typing import Any, AsyncIterator, Dict
from typing import Any, AsyncIterator, Dict, Optional
from litellm import verbose_logger
from litellm._uuid import uuid
@ -38,6 +38,7 @@ class AnthropicResponsesStreamWrapper:
self._pending_tool_ids: Dict[str, str] = {} # item_id -> call_id / name accumulator
self._sent_message_start = False
self._sent_message_stop = False
self._open_block_type: Optional[str] = None
self._chunk_queue: deque = deque()
def _make_message_start(self) -> Dict[str, Any]:
@ -64,6 +65,38 @@ class AnthropicResponsesStreamWrapper:
self._current_block_index += 1
return self._current_block_index
def _close_open_block(self) -> None:
if self._open_block_type is not None:
self._chunk_queue.append(
{
"type": "content_block_stop",
"index": self._current_block_index,
}
)
self._open_block_type = None
def _open_content_block(self, content_block: Dict[str, Any], item_id: Optional[str] = None) -> int:
block_idx = self._next_block_index()
if item_id:
self._item_id_to_block_index[item_id] = block_idx
self._open_block_type = content_block["type"]
self._chunk_queue.append(
{
"type": "content_block_start",
"index": block_idx,
"content_block": content_block,
}
)
return block_idx
def _ensure_block_of_type(
self, block_type: str, content_block: Dict[str, Any], item_id: Optional[str] = None
) -> int:
if self._open_block_type != block_type:
self._close_open_block()
return self._open_content_block(content_block, item_id)
return self._current_block_index
def _process_event(self, event: Any) -> None:
"""Convert one Responses API event into zero or more Anthropic chunks queued for emission."""
event_type = getattr(event, "type", None)
@ -75,6 +108,8 @@ class AnthropicResponsesStreamWrapper:
# ---- message_start ----
if event_type == "response.created":
if self._sent_message_start:
return
self._sent_message_start = True
self._chunk_queue.append(self._make_message_start())
return
@ -88,69 +123,29 @@ class AnthropicResponsesStreamWrapper:
item_id = getattr(item, "id", None) or (item.get("id") if isinstance(item, dict) else None)
if item_type == "message":
block_idx = self._next_block_index()
if item_id:
self._item_id_to_block_index[item_id] = block_idx
self._chunk_queue.append(
{
"type": "content_block_start",
"index": block_idx,
"content_block": {"type": "text", "text": ""},
}
)
elif item_type == "function_call":
return
if item_type == "function_call":
call_id = (
getattr(item, "call_id", None) or (item.get("call_id") if isinstance(item, dict) else None) or ""
)
name = getattr(item, "name", None) or (item.get("name") if isinstance(item, dict) else None) or ""
block_idx = self._next_block_index()
if item_id:
self._item_id_to_block_index[item_id] = block_idx
self._pending_tool_ids[item_id] = call_id
self._chunk_queue.append(
{
"type": "content_block_start",
"index": block_idx,
"content_block": {
"type": "tool_use",
"id": call_id,
"name": name,
"input": {},
},
}
self._close_open_block()
self._open_content_block(
{"type": "tool_use", "id": call_id, "name": name, "input": {}},
item_id,
)
elif item_type == "reasoning":
block_idx = self._next_block_index()
if item_id:
self._item_id_to_block_index[item_id] = block_idx
self._chunk_queue.append(
{
"type": "content_block_start",
"index": block_idx,
"content_block": {"type": "thinking", "thinking": ""},
}
)
self._close_open_block()
self._open_content_block({"type": "thinking", "thinking": ""}, item_id)
return
# ---- text delta ----
if event_type == "response.output_text.delta":
item_id = getattr(event, "item_id", None) or (event.get("item_id") if isinstance(event, dict) else None)
delta = getattr(event, "delta", "") or (event.get("delta", "") if isinstance(event, dict) else "")
block_idx = self._item_id_to_block_index.get(item_id, -1) if item_id else self._current_block_index
if block_idx < 0:
# Some providers (e.g. LMStudio) skip response.output_item.added,
# so no text block is open yet; synthesize content_block_start
# instead of emitting a delta with index -1
block_idx = self._next_block_index()
if item_id:
self._item_id_to_block_index[item_id] = block_idx
self._chunk_queue.append(
{
"type": "content_block_start",
"index": block_idx,
"content_block": {"type": "text", "text": ""},
}
)
block_idx = self._ensure_block_of_type("text", {"type": "text", "text": ""}, item_id)
self._chunk_queue.append(
{
"type": "content_block_delta",
@ -164,11 +159,7 @@ class AnthropicResponsesStreamWrapper:
if event_type == "response.reasoning_summary_text.delta":
item_id = getattr(event, "item_id", None) or (event.get("item_id") if isinstance(event, dict) else None)
delta = getattr(event, "delta", "") or (event.get("delta", "") if isinstance(event, dict) else "")
block_idx = (
self._item_id_to_block_index.get(item_id, self._current_block_index)
if item_id
else self._current_block_index
)
block_idx = self._ensure_block_of_type("thinking", {"type": "thinking", "thinking": ""}, item_id)
self._chunk_queue.append(
{
"type": "content_block_delta",
@ -198,21 +189,7 @@ class AnthropicResponsesStreamWrapper:
# ---- output item done -> content_block_stop ----
if event_type == "response.output_item.done":
item = getattr(event, "item", None) or (event.get("item") if isinstance(event, dict) else None)
item_id = (
getattr(item, "id", None) or (item.get("id") if isinstance(item, dict) else None) if item else None
)
block_idx = (
self._item_id_to_block_index.get(item_id, self._current_block_index)
if item_id
else self._current_block_index
)
self._chunk_queue.append(
{
"type": "content_block_stop",
"index": block_idx,
}
)
self._close_open_block()
return
# ---- response completed -> message_delta + message_stop ----
@ -264,6 +241,8 @@ class AnthropicResponsesStreamWrapper:
if cache_read_tokens:
usage_delta["cache_read_input_tokens"] = cache_read_tokens
self._close_open_block()
self._chunk_queue.append(
{
"type": "message_delta",

116
tests/test_litellm/llms/anthropic/experimental_pass_through/responses_adapters/test_responses_adapters_streaming_iterator.py
View file

@ -6,6 +6,8 @@ Tests for AnthropicResponsesStreamWrapper
import os
import sys
import pytest
sys.path.insert(
0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../../../../../.."))
)
@ -22,6 +24,27 @@ def _process_all(events: list) -> list:
return list(wrapper._chunk_queue)
class _FakeResponsesStream:
def __init__(self, events: list) -> None:
self._it = iter(events)
def __aiter__(self):
return self
async def __anext__(self):
try:
return next(self._it)
except StopIteration:
raise StopAsyncIteration
async def _drain(events: list) -> list:
wrapper = AnthropicResponsesStreamWrapper(
responses_stream=_FakeResponsesStream(events), model="m"
)
return [chunk async for chunk in wrapper]
class TestProcessEventTextDeltaWithoutOutputItemAdded:
"""Streams that skip response.output_item.added (e.g. LMStudio) must still
open a text block before any delta and never emit index -1."""
@ -59,9 +82,15 @@ class TestProcessEventTextDeltaWithoutOutputItemAdded:
{"type": "response.output_text.delta", "item_id": "m1", "delta": "Hi"},
]
)
assert chunks[1]["type"] == "content_block_start"
assert chunks[1]["content_block"] == {"type": "text", "text": ""}
assert [c["index"] for c in chunks[1:]] == [1, 1]
assert [c["type"] for c in chunks] == [
"content_block_start",
"content_block_stop",
"content_block_start",
"content_block_delta",
]
assert chunks[0]["content_block"] == {"type": "thinking", "thinking": ""}
assert chunks[2]["content_block"] == {"type": "text", "text": ""}
assert [c["index"] for c in chunks] == [0, 0, 1, 1]
def test_process_event_registered_item_id_does_not_synthesize_start(self):
chunks = _process_all(
@ -77,3 +106,84 @@ class TestProcessEventTextDeltaWithoutOutputItemAdded:
("content_block_start", 0),
("content_block_delta", 0),
]
def _reasoning_first_bridge_events() -> list:
"""The event sequence a chat-completions -> Responses API bridge emits for
an OpenAI-compatible reasoning backend: a message output item is announced
first (from the role-only chunk), reasoning arrives as
reasoning_summary_text.delta events with unrelated item_ids, then the text
answer arrives on the message item."""
return [
{"type": "response.created"},
{"type": "response.in_progress"},
{"type": "response.output_item.added", "item": {"type": "message", "id": "msg_1"}},
{"type": "response.content_part.added", "item_id": "msg_1"},
{"type": "response.reasoning_summary_text.delta", "item_id": "rs_a", "delta": "I"},
{"type": "response.reasoning_summary_text.delta", "item_id": "rs_b", "delta": " am"},
{"type": "response.reasoning_summary_text.delta", "item_id": "rs_c", "delta": " thinking"},
{"type": "response.output_text.delta", "item_id": "msg_1", "delta": "OK"},
{"type": "response.output_text.done", "item_id": "msg_1"},
{"type": "response.content_part.done", "item_id": "msg_1"},
{"type": "response.output_item.done", "item": {"type": "message", "id": "msg_1"}},
{"type": "response.completed"},
]
class TestReasoningContentIsNotStreamedIntoTextBlock:
"""Regression tests for https://github.com/BerriAI/litellm/issues/32357"""
@pytest.mark.asyncio
async def test_message_start_emitted_exactly_once(self):
chunks = await _drain(_reasoning_first_bridge_events())
assert [c["type"] for c in chunks].count("message_start") == 1
@pytest.mark.asyncio
async def test_reasoning_streams_into_its_own_thinking_block(self):
chunks = await _drain(_reasoning_first_bridge_events())
assert [c["type"] for c in chunks] == [
"message_start",
"content_block_start",
"content_block_delta",
"content_block_delta",
"content_block_delta",
"content_block_stop",
"content_block_start",
"content_block_delta",
"content_block_stop",
"message_delta",
"message_stop",
]
thinking_block = chunks[1]
assert thinking_block["content_block"] == {"type": "thinking", "thinking": ""}
assert thinking_block["index"] == 0
assert [c["delta"]["type"] for c in chunks[2:5]] == [
"thinking_delta",
"thinking_delta",
"thinking_delta",
]
assert all(c["index"] == 0 for c in chunks[2:5])
text_block = chunks[6]
assert text_block["content_block"] == {"type": "text", "text": ""}
assert text_block["index"] == 1
assert chunks[7]["delta"] == {"type": "text_delta", "text": "OK"}
assert chunks[7]["index"] == 1
@pytest.mark.asyncio
async def test_no_thinking_delta_is_ever_emitted_into_a_text_block(self):
chunks = await _drain(_reasoning_first_bridge_events())
text_block_indexes = {
c["index"]
for c in chunks
if c["type"] == "content_block_start" and c["content_block"]["type"] == "text"
}
thinking_delta_indexes = {
c["index"]
for c in chunks
if c["type"] == "content_block_delta" and c["delta"]["type"] == "thinking_delta"
}
assert text_block_indexes.isdisjoint(thinking_delta_indexes)