tiennm99/litellm
1
0
Fork 0
mirror of https://github.com/tiennm99/litellm.git synced 2026-08-06 14:24:50 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: streaming code_interpreter_results dropped for multiple code executions

Browse Source 
stream_chunk_builder uses "last value wins" for list-valued
provider_specific_fields keys. _build_code_interpreter_results was
emitting only new items (incremental), so earlier results were silently
dropped when multiple sequential code executions occurred.

- Emit cumulative list from _build_code_interpreter_results, matching
  web_search_results pattern
- Assemble server_tool_use input from input_json_delta deltas at
  content_block_stop (Anthropic streams input: {} in start block)
- Handle dict items in _extract_tool_result_output_items after
  model_dump() serialization in stream_chunk_builder
- Simplify _merge_provider_specific_fields to last-value-wins for lists,
  matching stream_chunk_builder semantics
This commit is contained in:
Andrzej Pomirski
2026-03-18 12:37:13 +01:00
parent 92b89353ae
commit 2bf8751f6b
4 changed files with 165 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
litellm/llms/anthropic/chat/handler.py
+35 -10
View File
@@ -545,7 +545,7 @@ class ModelResponseIterator:
# Track server tool use inputs and results for code_interpreter_results
self._server_tool_inputs: Dict[str, Any] = {}
self.tool_results: List[Dict[str, Any]] = []
self._last_code_interpreter_results_count: int = 0
self._current_server_tool_id: Optional[str] = None
def check_empty_tool_call_args(self) -> bool:
"""
@@ -695,13 +695,14 @@ class ModelResponseIterator:
Called during streaming to produce provider-neutral code_interpreter_results
alongside the raw tool_results, so the Responses API layer doesn't need
Anthropic-specific knowledge.
Returns the full cumulative list each time (not incremental), matching
how web_search_results works. stream_chunk_builder uses "last value
wins" for list-valued provider_specific_fields keys, so the last
emission must contain every result.
"""
# Only convert tool_results added since the last call to avoid
# duplicates when _merge_provider_specific_fields extends the list.
new_results = self.tool_results[self._last_code_interpreter_results_count :]
self._last_code_interpreter_results_count = len(self.tool_results)
results = []
for tr in new_results:
for tr in self.tool_results:
call_id = tr.get("tool_use_id", "")
content = tr.get("content", {})
if isinstance(content, dict):
@@ -793,17 +794,23 @@ class ModelResponseIterator:
),
index=self.tool_index,
)
# Track server tool use inputs for code_interpreter_results
# Track server tool use inputs for code_interpreter_results.
# The initial input in content_block_start is typically {}
# for streaming; the full input arrives via input_json_delta
# and is assembled at content_block_stop.
if (
content_block_start["content_block"]["type"]
== "server_tool_use"
):
self._current_server_tool_id = content_block_start[
"content_block"
]["id"]
tool_input = content_block_start["content_block"].get(
"input", {}
)
self._server_tool_inputs[
content_block_start["content_block"]["id"]
] = tool_input
self._server_tool_inputs[self._current_server_tool_id] = (
tool_input
)
# Include caller information if present (for programmatic tool calling)
if "caller" in content_block_start["content_block"]:
caller_data = content_block_start["content_block"]["caller"]
@@ -886,6 +893,24 @@ class ModelResponseIterator:
),
index=self.tool_index,
)
# Update server_tool_inputs with fully assembled input
# from input_json_delta chunks (content_block_start has {})
if (
self.current_content_block_type == "server_tool_use"
and self._current_server_tool_id
):
args = ""
for block in self.content_blocks:
if block["delta"]["type"] == "input_json_delta":
args += block["delta"].get("partial_json", "")
if args:
try:
self._server_tool_inputs[
self._current_server_tool_id
] = json.loads(args)
except (json.JSONDecodeError, TypeError):
pass
self._current_server_tool_id = None
# Reset response_format tool tracking when block stops
self.is_response_format_tool = False
# Reset current content block type
litellm/responses/litellm_completion_transformation/streaming_iterator.py
+9 -10
View File
@@ -481,17 +481,16 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
return event
def _merge_provider_specific_fields(self, src: dict) -> None:
"""Merge provider_specific_fields, extending list values instead of replacing."""
"""Merge provider_specific_fields using last-value-wins for lists.
List-valued keys (web_search_results, tool_results,
code_interpreter_results, etc.) are emitted cumulatively — each
emission contains the full list so far. Using "last value wins"
matches stream_chunk_builder's semantics and avoids quadratic
growth from repeated extend calls.
"""
for key, val in src.items():
existing = self._accumulated_provider_specific_fields.get(key)
if (
existing is not None
and isinstance(val, list)
and isinstance(existing, list)
):
existing.extend(val)
else:
self._accumulated_provider_specific_fields[key] = val
self._accumulated_provider_specific_fields[key] = val
def create_litellm_model_response(self) -> Optional[ModelResponse]:
response = cast(
litellm/responses/litellm_completion_transformation/transformation.py
+4 -1
View File
@@ -1738,7 +1738,10 @@ class LiteLLMCompletionResponsesConfig:
)
)
if tool_result_items:
result_by_id = {item.id: item for item in tool_result_items}
result_by_id = {
(item.get("id") if isinstance(item, dict) else item.id): item
for item in tool_result_items
}
replaced_ids = set(result_by_id.keys())
responses_output = [
(
tests/test_litellm/llms/anthropic/chat/test_anthropic_chat_handler.py
+117 -16
View File
@@ -1245,9 +1245,9 @@ def test_streaming_code_execution_produces_code_interpreter_results():
def test_streaming_multiple_code_executions_no_duplicates():
"""
Test that multiple code executions in a single streaming response produce
exactly one code_interpreter_result per execution — no duplicates from
_build_code_interpreter_results rebuilding the full list.
Test that multiple code executions in a single streaming response emit
cumulative code_interpreter_results on each chunk (matching stream_chunk_builder's
"last value wins" contract). The final emission must contain ALL results.
"""
chunks = [
{
@@ -1323,24 +1323,125 @@ def test_streaming_multiple_code_executions_no_duplicates():
iterator = ModelResponseIterator(None, sync_stream=True)
# Collect ALL code_interpreter_results emitted across all chunks
all_results = []
# Collect each emission of code_interpreter_results
emissions = []
for chunk in chunks:
parsed = iterator.chunk_parser(chunk)
psf = None
if parsed.choices and parsed.choices[0].delta:
psf = getattr(parsed.choices[0].delta, "provider_specific_fields", None)
if psf and "code_interpreter_results" in psf:
all_results.extend(psf["code_interpreter_results"])
emissions.append(psf["code_interpreter_results"])
# Should have exactly 2 results, one per execution — no duplicates
assert len(all_results) == 2, (
f"Expected 2 code_interpreter_results, got {len(all_results)}. "
f"IDs: {[r.id for r in all_results]}"
# Should have 2 emissions (one per tool_result block)
assert len(emissions) == 2, f"Expected 2 emissions, got {len(emissions)}"
# First emission: cumulative list with 1 result
assert len(emissions[0]) == 1
assert emissions[0][0].id == "srvtoolu_01AAA"
assert emissions[0][0].code == "echo first"
assert emissions[0][0].outputs[0].logs == "first\n"
# Second (final) emission: cumulative list with BOTH results
# This is what stream_chunk_builder will pick as "last value wins"
assert len(emissions[1]) == 2, (
f"Expected final emission to have 2 results, got {len(emissions[1])}. "
f"IDs: {[r.id for r in emissions[1]]}"
)
assert all_results[0].id == "srvtoolu_01AAA"
assert all_results[0].code == "echo first"
assert all_results[0].outputs[0].logs == "first\n"
assert all_results[1].id == "srvtoolu_01BBB"
assert all_results[1].code == "echo second"
assert all_results[1].outputs[0].logs == "second\n"
assert emissions[1][0].id == "srvtoolu_01AAA"
assert emissions[1][0].code == "echo first"
assert emissions[1][0].outputs[0].logs == "first\n"
assert emissions[1][1].id == "srvtoolu_01BBB"
assert emissions[1][1].code == "echo second"
assert emissions[1][1].outputs[0].logs == "second\n"
def test_streaming_code_execution_input_assembled_from_deltas():
"""
In real Anthropic streaming, content_block_start for server_tool_use has
input: {}. The actual input arrives via input_json_delta deltas and must
be assembled at content_block_stop so the code field is populated.
This test uses realistic chunk shapes (empty input in start, partial JSON
in deltas) to exercise the input assembly path.
"""
chunks = [
{
"type": "message_start",
"message": {
"id": "msg_01XYZ",
"type": "message",
"role": "assistant",
"content": [],
"usage": {"input_tokens": 100, "output_tokens": 1},
},
},
# server_tool_use with empty input (real streaming behaviour)
{
"type": "content_block_start",
"index": 0,
"content_block": {
"type": "server_tool_use",
"id": "srvtoolu_01AAA",
"name": "code_execution",
"input": {},
},
},
# Input arrives via deltas, split across two chunks
{
"type": "content_block_delta",
"index": 0,
"delta": {
"type": "input_json_delta",
"partial_json": '{"comma',
},
},
{
"type": "content_block_delta",
"index": 0,
"delta": {
"type": "input_json_delta",
"partial_json": 'nd": "echo hello"}',
},
},
{"type": "content_block_stop", "index": 0},
# Tool result
{
"type": "content_block_start",
"index": 1,
"content_block": {
"type": "code_execution_tool_result",
"tool_use_id": "srvtoolu_01AAA",
"content": {
"type": "code_execution_result",
"stdout": "hello\n",
"stderr": "",
"return_code": 0,
},
},
},
{"type": "content_block_stop", "index": 1},
{
"type": "message_delta",
"delta": {"stop_reason": "end_turn"},
"usage": {"output_tokens": 50},
},
]
iterator = ModelResponseIterator(None, sync_stream=True)
code_results = None
for chunk in chunks:
parsed = iterator.chunk_parser(chunk)
psf = None
if parsed.choices and parsed.choices[0].delta:
psf = getattr(parsed.choices[0].delta, "provider_specific_fields", None)
if psf and "code_interpreter_results" in psf:
code_results = psf["code_interpreter_results"]
# The code field must contain the assembled input, not be empty
assert code_results is not None, "No code_interpreter_results emitted"
assert len(code_results) == 1
assert code_results[0].id == "srvtoolu_01AAA"
assert code_results[0].code == "echo hello"
assert code_results[0].outputs[0].logs == "hello\n"