fix(responses): stream tool call events in completion bridge (#19368)
Emit Responses API streaming events for tool calls when the underlying chat stream contains tool_call deltas, and recover tool calls into the stream when they only appear in the final response.
This commit is contained in:
victorigualada
GitHub
parent
0dfc3fad5a
commit
581d086c20
@ -21,6 +21,8 @@ from litellm.types.llms.openai import (
|
||||
OutputTextAnnotationAddedEvent,
|
||||
OutputTextDeltaEvent,
|
||||
OutputTextDoneEvent,
|
||||
FunctionCallArgumentsDeltaEvent,
|
||||
FunctionCallArgumentsDoneEvent,
|
||||
ReasoningSummaryTextDeltaEvent,
|
||||
ResponseCompletedEvent,
|
||||
ResponseCreatedEvent,
|
||||
@ -79,6 +81,161 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
|
||||
Union[ModelResponse, TextCompletionResponse]
|
||||
] = None
|
||||
self.final_text: str = ""
|
||||
self._pending_tool_events: List[BaseLiteLLMOpenAIResponseObject] = []
|
||||
self._tool_output_index_by_call_id: dict[str, int] = {}
|
||||
self._tool_args_by_call_id: dict[str, str] = {}
|
||||
self._next_tool_output_index: int = 1 # output_index=0 reserved for the message item
|
||||
self._final_tool_events_queued: bool = False
|
||||
|
||||
def _get_or_assign_tool_output_index(self, call_id: str) -> int:
|
||||
existing = self._tool_output_index_by_call_id.get(call_id)
|
||||
if existing is not None:
|
||||
return existing
|
||||
idx = self._next_tool_output_index
|
||||
self._next_tool_output_index += 1
|
||||
self._tool_output_index_by_call_id[call_id] = idx
|
||||
return idx
|
||||
|
||||
def _queue_tool_call_delta_events(self, tool_calls: object) -> None:
|
||||
"""
|
||||
Convert chat-completions streaming `tool_calls` deltas into Responses API streaming events.
|
||||
|
||||
We emit:
|
||||
- response.output_item.added (function_call)
|
||||
- response.function_call_arguments.delta
|
||||
"""
|
||||
if not isinstance(tool_calls, list):
|
||||
return
|
||||
|
||||
for tc in tool_calls:
|
||||
call_id_raw = tc.get("id") if isinstance(tc, dict) else getattr(tc, "id", None)
|
||||
if not call_id_raw:
|
||||
continue
|
||||
call_id = str(call_id_raw)
|
||||
|
||||
fn = tc.get("function") if isinstance(tc, dict) else getattr(tc, "function", None)
|
||||
fn_name = ""
|
||||
fn_args_delta = ""
|
||||
if isinstance(fn, dict):
|
||||
fn_name = str(fn.get("name") or "")
|
||||
fn_args_delta = str(fn.get("arguments") or "")
|
||||
else:
|
||||
fn_name = str(getattr(fn, "name", "") or "")
|
||||
fn_args_delta = str(getattr(fn, "arguments", "") or "")
|
||||
|
||||
output_index = self._get_or_assign_tool_output_index(call_id)
|
||||
|
||||
if call_id not in self._tool_args_by_call_id:
|
||||
self._tool_args_by_call_id[call_id] = ""
|
||||
self._pending_tool_events.append(
|
||||
OutputItemAddedEvent(
|
||||
type=ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED,
|
||||
output_index=output_index,
|
||||
item=BaseLiteLLMOpenAIResponseObject(
|
||||
**{
|
||||
"type": "function_call",
|
||||
"id": call_id,
|
||||
"call_id": call_id,
|
||||
"name": fn_name,
|
||||
"arguments": "",
|
||||
"status": "in_progress",
|
||||
}
|
||||
),
|
||||
)
|
||||
)
|
||||
|
||||
if fn_args_delta:
|
||||
self._tool_args_by_call_id[call_id] += fn_args_delta
|
||||
self._pending_tool_events.append(
|
||||
FunctionCallArgumentsDeltaEvent(
|
||||
type=ResponsesAPIStreamEvents.FUNCTION_CALL_ARGUMENTS_DELTA,
|
||||
item_id=call_id,
|
||||
output_index=output_index,
|
||||
delta=fn_args_delta,
|
||||
)
|
||||
)
|
||||
|
||||
def _queue_final_tool_call_done_events(self, litellm_complete_object: ModelResponse) -> None:
|
||||
"""
|
||||
Ensure tool calls that were not streamed as deltas still get emitted before response.completed.
|
||||
"""
|
||||
if self._final_tool_events_queued:
|
||||
return
|
||||
self._final_tool_events_queued = True
|
||||
|
||||
try:
|
||||
message = litellm_complete_object.choices[0].message # type: ignore
|
||||
tool_calls = getattr(message, "tool_calls", None)
|
||||
except Exception:
|
||||
tool_calls = None
|
||||
|
||||
if not tool_calls or not isinstance(tool_calls, list):
|
||||
return
|
||||
|
||||
for tc in tool_calls:
|
||||
call_id_raw = tc.get("id") if isinstance(tc, dict) else getattr(tc, "id", None)
|
||||
if not call_id_raw:
|
||||
continue
|
||||
call_id = str(call_id_raw)
|
||||
output_index = self._get_or_assign_tool_output_index(call_id)
|
||||
|
||||
fn = tc.get("function") if isinstance(tc, dict) else getattr(tc, "function", None)
|
||||
fn_name = ""
|
||||
fn_args = ""
|
||||
if isinstance(fn, dict):
|
||||
fn_name = str(fn.get("name") or "")
|
||||
fn_args = str(fn.get("arguments") or "")
|
||||
else:
|
||||
fn_name = str(getattr(fn, "name", "") or "")
|
||||
fn_args = str(getattr(fn, "arguments", "") or "")
|
||||
|
||||
# If we never sent output_item.added for this call_id, emit it now.
|
||||
if call_id not in self._tool_args_by_call_id:
|
||||
self._tool_args_by_call_id[call_id] = ""
|
||||
self._pending_tool_events.append(
|
||||
OutputItemAddedEvent(
|
||||
type=ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED,
|
||||
output_index=output_index,
|
||||
item=BaseLiteLLMOpenAIResponseObject(
|
||||
**{
|
||||
"type": "function_call",
|
||||
"id": call_id,
|
||||
"call_id": call_id,
|
||||
"name": fn_name,
|
||||
"arguments": "",
|
||||
"status": "in_progress",
|
||||
}
|
||||
),
|
||||
)
|
||||
)
|
||||
|
||||
final_args = fn_args or self._tool_args_by_call_id.get(call_id, "")
|
||||
self._pending_tool_events.append(
|
||||
FunctionCallArgumentsDoneEvent(
|
||||
type=ResponsesAPIStreamEvents.FUNCTION_CALL_ARGUMENTS_DONE,
|
||||
item_id=call_id,
|
||||
output_index=output_index,
|
||||
arguments=final_args,
|
||||
)
|
||||
)
|
||||
|
||||
self._pending_tool_events.append(
|
||||
OutputItemDoneEvent(
|
||||
type=ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE,
|
||||
output_index=output_index,
|
||||
sequence_number=1,
|
||||
item=BaseLiteLLMOpenAIResponseObject(
|
||||
**{
|
||||
"type": "function_call",
|
||||
"id": call_id,
|
||||
"call_id": call_id,
|
||||
"name": fn_name,
|
||||
"arguments": final_args,
|
||||
"status": "completed",
|
||||
}
|
||||
),
|
||||
)
|
||||
)
|
||||
|
||||
def _default_response_created_event_data(self) -> dict:
|
||||
response_created_event_data = {
|
||||
@ -310,6 +467,12 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
|
||||
):
|
||||
self.litellm_model_response = self.create_litellm_model_response()
|
||||
if self.litellm_model_response:
|
||||
# If tool calls exist, emit tool events before finishing/response.completed.
|
||||
if isinstance(self.litellm_model_response, ModelResponse):
|
||||
self._queue_final_tool_call_done_events(self.litellm_model_response)
|
||||
if self._pending_tool_events:
|
||||
return self._pending_tool_events.pop(0)
|
||||
|
||||
done_event = self.return_default_done_events(self.litellm_model_response)
|
||||
if done_event:
|
||||
return done_event
|
||||
@ -462,13 +625,27 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
|
||||
content_index=0,
|
||||
delta=delta_content,
|
||||
)
|
||||
|
||||
# Priority 3: Handle tool call deltas (if any) -> queue events and emit them
|
||||
if (
|
||||
chunk.choices
|
||||
and hasattr(chunk.choices[0].delta, "tool_calls")
|
||||
and chunk.choices[0].delta.tool_calls
|
||||
):
|
||||
self._queue_tool_call_delta_events(chunk.choices[0].delta.tool_calls)
|
||||
if self._pending_tool_events:
|
||||
return self._pending_tool_events.pop(0)
|
||||
|
||||
# Priority 3: If we have pending annotation events, emit the next one
|
||||
# Priority 4: If we have pending annotation events, emit the next one
|
||||
# This happens when the current chunk has no text/reasoning content
|
||||
if hasattr(self, '_pending_annotation_events') and self._pending_annotation_events:
|
||||
event = self._pending_annotation_events.pop(0)
|
||||
return event
|
||||
|
||||
# Priority 5: If we have pending tool events (from earlier chunk), emit the next one
|
||||
if self._pending_tool_events:
|
||||
return self._pending_tool_events.pop(0)
|
||||
|
||||
return None
|
||||
|
||||
def _get_delta_string_from_streaming_choices(
|
||||
|
||||
@ -0,0 +1,116 @@
|
||||
"""
|
||||
Tests for streaming tool-calls in Responses API transformation.
|
||||
|
||||
Ensures that when the underlying chat-completions stream includes tool_calls deltas,
|
||||
LiteLLM emits Responses API streaming events (output_item.added + function_call_arguments.*).
|
||||
|
||||
Also ensures that tool calls that only appear in the final built response still get emitted
|
||||
before response.completed.
|
||||
"""
|
||||
|
||||
from unittest.mock import AsyncMock
|
||||
|
||||
from litellm.responses.litellm_completion_transformation.streaming_iterator import (
|
||||
LiteLLMCompletionStreamingIterator,
|
||||
)
|
||||
from litellm.types.llms.openai import ResponsesAPIStreamEvents
|
||||
from litellm.types.utils import Delta, ModelResponse, ModelResponseStream, StreamingChoices
|
||||
|
||||
|
||||
def test_tool_call_delta_is_emitted_as_responses_events():
|
||||
iterator = LiteLLMCompletionStreamingIterator(
|
||||
model="test-model",
|
||||
litellm_custom_stream_wrapper=AsyncMock(),
|
||||
request_input="Test input",
|
||||
responses_api_request={},
|
||||
)
|
||||
|
||||
# A streaming chunk with tool_calls delta but no text
|
||||
chunk = ModelResponseStream(
|
||||
id="chunk-1",
|
||||
created=123,
|
||||
model="test-model",
|
||||
object="chat.completion.chunk",
|
||||
choices=[
|
||||
StreamingChoices(
|
||||
finish_reason=None,
|
||||
index=0,
|
||||
delta=Delta(
|
||||
role="assistant",
|
||||
content="",
|
||||
tool_calls=[
|
||||
{
|
||||
"id": "call_1",
|
||||
"type": "function",
|
||||
"function": {"name": "do_thing", "arguments": '{"x":1}'},
|
||||
}
|
||||
],
|
||||
),
|
||||
)
|
||||
],
|
||||
)
|
||||
|
||||
evt1 = iterator._transform_chat_completion_chunk_to_response_api_chunk(chunk)
|
||||
assert evt1 is not None
|
||||
assert evt1.type == ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED
|
||||
assert evt1.output_index == 1
|
||||
|
||||
evt2 = iterator._transform_chat_completion_chunk_to_response_api_chunk(chunk)
|
||||
assert evt2 is not None
|
||||
assert evt2.type == ResponsesAPIStreamEvents.FUNCTION_CALL_ARGUMENTS_DELTA
|
||||
assert evt2.item_id == "call_1"
|
||||
assert evt2.output_index == 1
|
||||
assert evt2.delta == '{"x":1}'
|
||||
|
||||
|
||||
def test_tool_calls_present_only_in_final_response_are_emitted_before_completed():
|
||||
iterator = LiteLLMCompletionStreamingIterator(
|
||||
model="test-model",
|
||||
litellm_custom_stream_wrapper=AsyncMock(),
|
||||
request_input="Test input",
|
||||
responses_api_request={},
|
||||
)
|
||||
|
||||
# Construct a final ModelResponse with tool_calls on the message.
|
||||
# We bypass the stream builder and directly set iterator.litellm_model_response.
|
||||
response = ModelResponse(
|
||||
id="resp-1",
|
||||
created=123,
|
||||
model="test-model",
|
||||
object="chat.completion",
|
||||
choices=[
|
||||
{
|
||||
"index": 0,
|
||||
"finish_reason": "tool_calls",
|
||||
"message": {
|
||||
"role": "assistant",
|
||||
"content": None,
|
||||
"tool_calls": [
|
||||
{
|
||||
"id": "call_2",
|
||||
"type": "function",
|
||||
"function": {"name": "do_thing", "arguments": '{"y":2}'},
|
||||
"index": 0,
|
||||
}
|
||||
],
|
||||
},
|
||||
}
|
||||
],
|
||||
)
|
||||
iterator.litellm_model_response = response
|
||||
|
||||
# First common_done_event_logic call should yield tool events, not response.completed.
|
||||
evt1 = iterator.common_done_event_logic(sync_mode=True)
|
||||
assert evt1.type == ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED
|
||||
assert evt1.output_index == 1
|
||||
|
||||
evt2 = iterator.common_done_event_logic(sync_mode=True)
|
||||
assert evt2.type == ResponsesAPIStreamEvents.FUNCTION_CALL_ARGUMENTS_DONE
|
||||
assert evt2.item_id == "call_2"
|
||||
assert evt2.output_index == 1
|
||||
assert evt2.arguments == '{"y":2}'
|
||||
|
||||
evt3 = iterator.common_done_event_logic(sync_mode=True)
|
||||
assert evt3.type == ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE
|
||||
assert evt3.output_index == 1
|
||||
|
||||
Loading…
Reference in New Issue
Block a user