From 2856c56d20e366340f82c90f287875fda3a302a3 Mon Sep 17 00:00:00 2001 From: Rajasimhareddybolla Date: Sun, 25 Oct 2026 17:38:43 +0530 Subject: [PATCH 1/3] fix(openai_like): preserve reasoning_content, multi-tool calls, and usage in streaming Fixes #48238 When streaming responses from OpenAI-like endpoints (including Vertex Model Garden dedicated endpoints 'vertex_ai/openai/...'), reasoning chunks and secondary tool call deltas were previously dropped because: 1. ModelResponseIterator.chunk_parser only checked delta.content and tool_calls[0]. 2. GenericStreamingChunk lacked reasoning_content and tool_calls fields. 3. CustomStreamWrapper considered reasoning-only chunks (with content=='') empty, dropping them before reaching the client. This patch: - Extends GenericStreamingChunk to support reasoning_content and tool_calls. - Updates ModelResponseIterator to extract reasoning_content and preserve all indexed tool calls. - Updates CustomStreamWrapper to treat reasoning_content chunks as non-empty and emit them with content=None. - Adds comprehensive unit and streaming regression tests. --- .../litellm_core_utils/streaming_handler.py | 34 +++- litellm/llms/databricks/streaming_utils.py | 45 ++++-- litellm/types/utils.py | 2 + .../test_databricks_streaming_utils.py | 70 ++++++++ .../chat/test_openai_like_chat_handler.py | 152 +++++++++++++++++- 5 files changed, 286 insertions(+), 17 deletions(-) diff --git a/litellm/litellm_core_utils/streaming_handler.py b/litellm/litellm_core_utils/streaming_handler.py index 2df470edc59..12ba041aff8 100644 --- a/litellm/litellm_core_utils/streaming_handler.py +++ b/litellm/litellm_core_utils/streaming_handler.py @@ -971,6 +971,11 @@ class CustomStreamWrapper: if ( "content" in completion_obj and (isinstance(completion_obj["content"], str) and len(completion_obj["content"]) > 0) + or ( + "reasoning_content" in completion_obj + and completion_obj["reasoning_content"] is not None + and len(completion_obj["reasoning_content"]) > 0 + ) or ( "tool_calls" in completion_obj and completion_obj["tool_calls"] is not None @@ -1214,7 +1219,13 @@ class CustomStreamWrapper: ) else: ## else - completion_obj["content"] = model_response_str + if model_response_str or ( + completion_obj.get("reasoning_content") is None + and completion_obj.get("tool_calls") is None + ): + completion_obj["content"] = model_response_str + else: + completion_obj["content"] = None if self.sent_first_chunk is False: completion_obj["role"] = "assistant" self.sent_first_chunk = True @@ -1345,7 +1356,9 @@ class CustomStreamWrapper: if self.received_finish_reason is not None: _chunk_has_content: Final = isinstance(chunk, dict) and ( bool(chunk.get("text", "")) + or bool(chunk.get("reasoning_content", "")) or chunk.get("tool_use") is not None + or bool(chunk.get("tool_calls")) # Usage-only final chunks are valid and needed to surface # finish_reason/usage to downstream translators. or chunk.get("usage") is not None @@ -1354,6 +1367,8 @@ class CustomStreamWrapper: raise StopIteration anthropic_response_obj: Final[GChunk] = cast(GChunk, chunk) completion_obj["content"] = anthropic_response_obj["text"] + if "reasoning_content" in anthropic_response_obj and anthropic_response_obj["reasoning_content"] is not None: + completion_obj["reasoning_content"] = anthropic_response_obj["reasoning_content"] if anthropic_response_obj["is_finished"]: self.received_finish_reason = anthropic_response_obj["finish_reason"] @@ -1367,7 +1382,9 @@ class CustomStreamWrapper: litellm.Usage(**anthropic_response_obj["usage"]), ) - if "tool_use" in anthropic_response_obj and anthropic_response_obj["tool_use"] is not None: + if "tool_calls" in anthropic_response_obj and anthropic_response_obj["tool_calls"] is not None: + completion_obj["tool_calls"] = anthropic_response_obj["tool_calls"] + elif "tool_use" in anthropic_response_obj and anthropic_response_obj["tool_use"] is not None: completion_obj["tool_calls"] = [anthropic_response_obj["tool_use"]] if ( @@ -2776,6 +2793,14 @@ def convert_generic_chunk_to_model_response_stream( ) -> ModelResponseStream: from litellm.types.utils import Delta + tool_calls = chunk.get("tool_calls", None) + if tool_calls is None and chunk.get("tool_use") is not None: + tool_calls = [chunk["tool_use"]] + + content = chunk["text"] + if not content and (chunk.get("reasoning_content") is not None or tool_calls is not None): + content = None + model_response_stream: Final = ModelResponseStream( id=str(uuid.uuid4()), model="", @@ -2783,8 +2808,9 @@ def convert_generic_chunk_to_model_response_stream( StreamingChoices( index=chunk.get("index", 0), delta=Delta( - content=chunk["text"], - tool_calls=chunk.get("tool_use", None), + content=content, + reasoning_content=chunk.get("reasoning_content", None), + tool_calls=tool_calls, ), ) ], diff --git a/litellm/llms/databricks/streaming_utils.py b/litellm/llms/databricks/streaming_utils.py index eca87f9dd5f..a95578a41d7 100644 --- a/litellm/llms/databricks/streaming_utils.py +++ b/litellm/llms/databricks/streaming_utils.py @@ -33,6 +33,8 @@ class ModelResponseIterator: text = "" tool_use: ChatCompletionToolCallChunk | None = None + tool_calls: list[ChatCompletionToolCallChunk] | None = None + reasoning_content: str | None = None is_finished = False finish_reason = "" usage: ChatCompletionUsageBlock | None = None @@ -54,21 +56,34 @@ class ModelResponseIterator: if processed_chunk.choices[0].delta.content is not None: text = processed_chunk.choices[0].delta.content + delta_reasoning: Final[str | None] = getattr(processed_chunk.choices[0].delta, "reasoning_content", None) + if delta_reasoning is not None: + reasoning_content = delta_reasoning + if ( processed_chunk.choices[0].delta.tool_calls is not None and len(processed_chunk.choices[0].delta.tool_calls) > 0 - and processed_chunk.choices[0].delta.tool_calls[0].function is not None - and processed_chunk.choices[0].delta.tool_calls[0].function.arguments is not None ): - tool_use = ChatCompletionToolCallChunk( - id=processed_chunk.choices[0].delta.tool_calls[0].id, - type="function", - function=ChatCompletionToolCallFunctionChunk( - name=processed_chunk.choices[0].delta.tool_calls[0].function.name, - arguments=processed_chunk.choices[0].delta.tool_calls[0].function.arguments, - ), - index=processed_chunk.choices[0].delta.tool_calls[0].index, - ) + tool_calls = [] + for tc in processed_chunk.choices[0].delta.tool_calls: + tc_fn = getattr(tc, "function", None) + fn_chunk = ( + ChatCompletionToolCallFunctionChunk( + name=getattr(tc_fn, "name", None), + arguments=getattr(tc_fn, "arguments", "") or "", + ) + if tc_fn is not None + else ChatCompletionToolCallFunctionChunk(name=None, arguments="") + ) + call_chunk = ChatCompletionToolCallChunk( + id=getattr(tc, "id", None), + type="function", + function=fn_chunk, + index=getattr(tc, "index", 0) if getattr(tc, "index", None) is not None else 0, + ) + tool_calls.append(call_chunk) + if tool_calls: + tool_use = tool_calls[0] if processed_chunk.choices[0].finish_reason is not None: is_finished = True @@ -78,7 +93,7 @@ class ModelResponseIterator: if usage_chunk is not None: usage = _usage_block(usage_chunk) - return GenericStreamingChunk( + generic_chunk: Final[GenericStreamingChunk] = GenericStreamingChunk( text=text, tool_use=tool_use, is_finished=is_finished, @@ -86,6 +101,12 @@ class ModelResponseIterator: usage=usage, index=0, ) + if reasoning_content is not None: + generic_chunk["reasoning_content"] = reasoning_content + if tool_calls is not None: + generic_chunk["tool_calls"] = tool_calls + + return generic_chunk except json.JSONDecodeError: raise ValueError(f"Failed to decode JSON from chunk: {chunk}") diff --git a/litellm/types/utils.py b/litellm/types/utils.py index cc7ef851111..1bdb2919ba7 100644 --- a/litellm/types/utils.py +++ b/litellm/types/utils.py @@ -498,6 +498,8 @@ class ModelInfo(ModelInfoBase, total=False): class GenericStreamingChunk(TypedDict, total=False): text: Required[str] tool_use: ChatCompletionToolCallChunk | None + tool_calls: list[ChatCompletionToolCallChunk] | None + reasoning_content: str | None is_finished: Required[bool] finish_reason: Required[str] usage: Required[ChatCompletionUsageBlock | None] diff --git a/tests/unit/llms/databricks/test_databricks_streaming_utils.py b/tests/unit/llms/databricks/test_databricks_streaming_utils.py index 5c5527f7be8..0f52afbab3a 100644 --- a/tests/unit/llms/databricks/test_databricks_streaming_utils.py +++ b/tests/unit/llms/databricks/test_databricks_streaming_utils.py @@ -222,3 +222,73 @@ async def test_async_stream_without_done_marker_ends_after_the_last_chunk(): chunks: Final = [chunk async for chunk in ModelResponseIterator(streaming_response=lines(), sync_stream=False)] assert [(chunk["text"], chunk["finish_reason"]) for chunk in chunks] == [("144", ""), ("", "stop")] + + +def test_chunk_parser_extracts_reasoning_content(): + """Reasoning-only delta chunks must preserve reasoning_content.""" + iterator = ModelResponseIterator(streaming_response=None, sync_stream=True) + chunk = { + "id": "chatcmpl-reasoning", + "object": "chat.completion.chunk", + "created": 1, + "model": "glm-4", + "choices": [ + { + "index": 0, + "delta": {"role": "assistant", "reasoning_content": "Thinking about prime numbers..."}, + "finish_reason": None, + } + ], + } + + result = iterator.chunk_parser(chunk=chunk) + + assert result["text"] == "" + assert result.get("reasoning_content") == "Thinking about prime numbers..." + + +def test_chunk_parser_extracts_multiple_tool_calls_with_indices(): + """Multiple tool call deltas in a chunk must preserve all tool calls and their indices.""" + iterator = ModelResponseIterator(streaming_response=None, sync_stream=True) + chunk = { + "id": "chatcmpl-tools", + "object": "chat.completion.chunk", + "created": 1, + "model": "glm-4", + "choices": [ + { + "index": 0, + "delta": { + "tool_calls": [ + { + "index": 0, + "id": "call_1", + "type": "function", + "function": {"name": "get_weather", "arguments": '{"city": "Paris"}'}, + }, + { + "index": 1, + "id": "call_2", + "type": "function", + "function": {"name": "get_time", "arguments": '{"city": "Paris"}'}, + }, + ] + }, + "finish_reason": None, + } + ], + } + + result = iterator.chunk_parser(chunk=chunk) + + assert result.get("tool_calls") is not None + assert len(result["tool_calls"]) == 2 + assert result["tool_calls"][0]["id"] == "call_1" + assert result["tool_calls"][0]["function"]["name"] == "get_weather" + assert result["tool_calls"][0]["index"] == 0 + assert result["tool_calls"][1]["id"] == "call_2" + assert result["tool_calls"][1]["function"]["name"] == "get_time" + assert result["tool_calls"][1]["index"] == 1 + # Backward compatibility with tool_use + assert result["tool_use"]["id"] == "call_1" + diff --git a/tests/unit/llms/openai_like/chat/test_openai_like_chat_handler.py b/tests/unit/llms/openai_like/chat/test_openai_like_chat_handler.py index 75b60b2dd6c..defa02daad9 100644 --- a/tests/unit/llms/openai_like/chat/test_openai_like_chat_handler.py +++ b/tests/unit/llms/openai_like/chat/test_openai_like_chat_handler.py @@ -1,12 +1,12 @@ import json from typing import Final - from unittest.mock import Mock import httpx from litellm.llms.custom_httpx.http_handler import HTTPHandler from litellm.llms.openai_like.chat.handler import make_sync_call +from litellm.utils import CustomStreamWrapper _FAKE_STREAMED_REPLY: Final = { "id": "chatcmpl-fake-stream", @@ -73,3 +73,153 @@ def test_fake_stream_keeps_reasoning_every_tool_call_and_token_detail(): assert [call.function.name for call in delta.tool_calls] == ["get_weather", "get_time"] assert chunks[0].usage.prompt_tokens_details.cached_tokens == 64 assert chunks[0].usage.completion_tokens_details.reasoning_tokens == 1920 + + +def test_real_stream_keeps_reasoning_multi_tool_calls_and_terminal_usage(): + events = [ + "data: " + + json.dumps( + { + "id": "c1", + "object": "chat.completion.chunk", + "created": 1, + "model": "glm-4", + "choices": [ + { + "index": 0, + "delta": {"role": "assistant", "reasoning_content": "Thinking step 1"}, + "finish_reason": None, + } + ], + } + ), + "data: " + + json.dumps( + { + "id": "c2", + "object": "chat.completion.chunk", + "created": 1, + "model": "glm-4", + "choices": [ + { + "index": 0, + "delta": { + "tool_calls": [ + { + "index": 0, + "id": "call_1", + "type": "function", + "function": {"name": "get_weather", "arguments": '{"city": "Paris"}'}, + }, + { + "index": 1, + "id": "call_2", + "type": "function", + "function": {"name": "get_time", "arguments": '{"city": "Paris"}'}, + }, + ] + }, + "finish_reason": None, + } + ], + } + ), + "data: " + + json.dumps( + { + "id": "c3", + "object": "chat.completion.chunk", + "created": 1, + "model": "glm-4", + "choices": [ + { + "index": 0, + "delta": {"content": "Here is the result"}, + "finish_reason": None, + } + ], + } + ), + "data: " + + json.dumps( + { + "id": "c4", + "object": "chat.completion.chunk", + "created": 1, + "model": "glm-4", + "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}], + } + ), + "data: " + + json.dumps( + { + "id": "c5", + "object": "chat.completion.chunk", + "created": 1, + "model": "glm-4", + "choices": [], + "usage": { + "prompt_tokens": 10, + "completion_tokens": 64, + "total_tokens": 74, + "completion_tokens_details": {"reasoning_tokens": 30}, + }, + } + ), + "data: [DONE]", + ] + sse_bytes = ("\n\n".join(events) + "\n\n").encode("utf-8") + + def respond(request: httpx.Request) -> httpx.Response: + return httpx.Response(200, headers={"content-type": "text/event-stream"}, content=sse_bytes) + + mock_logging = Mock() + mock_logging.model_call_details = {"litellm_params": {}} + mock_logging.stream_options = {"include_usage": True} + + raw_stream = make_sync_call( + client=HTTPHandler(client=httpx.Client(transport=httpx.MockTransport(respond))), + api_base="https://openai-like.example/v1/chat/completions", + headers={}, + data=json.dumps({"model": "glm-4", "messages": []}), + model="glm-4", + messages=[], + logging_obj=mock_logging, + fake_stream=False, + ) + wrapper = CustomStreamWrapper( + completion_stream=raw_stream, + model="glm-4", + custom_llm_provider="vertex_ai", + logging_obj=mock_logging, + stream_options={"include_usage": True}, + ) + + chunks = list(wrapper) + assert len(chunks) == 5 + + # 1. Reasoning-only delta must reach the client with reasoning_content and content=None + reasoning_chunk = chunks[0] + assert reasoning_chunk.choices[0].delta.reasoning_content == "Thinking step 1" + assert reasoning_chunk.choices[0].delta.content is None + + # 2. Multi tool call delta must preserve both tool calls and their indices + tool_chunk = chunks[1] + assert len(tool_chunk.choices[0].delta.tool_calls) == 2 + assert tool_chunk.choices[0].delta.tool_calls[0].function.name == "get_weather" + assert tool_chunk.choices[0].delta.tool_calls[0].index == 0 + assert tool_chunk.choices[0].delta.tool_calls[1].function.name == "get_time" + assert tool_chunk.choices[0].delta.tool_calls[1].index == 1 + + # 3. Content delta + assert chunks[2].choices[0].delta.content == "Here is the result" + + # 4. Finish reason + assert chunks[3].choices[0].finish_reason == "stop" + + # 5. Terminal usage chunk with provider usage details preserved + assert chunks[4].usage is not None + assert chunks[4].usage.completion_tokens == 64 + assert chunks[4].usage.prompt_tokens == 10 + assert chunks[4].usage.completion_tokens_details.reasoning_tokens == 30 + From 30d64b1bc97ebfe56929f846a4f8cb8e71af96bf Mon Sep 17 00:00:00 2001 From: Rajasimhareddybolla Date: Sun, 25 Oct 2026 18:05:48 +0530 Subject: [PATCH 2/3] style(lint): satisfy ruff formatting and LIT type discipline gates --- .../litellm_core_utils/streaming_handler.py | 36 ++++++----- litellm/llms/databricks/streaming_utils.py | 61 +++++++------------ litellm/types/utils.py | 4 +- .../test_databricks_streaming_utils.py | 5 +- .../chat/test_openai_like_chat_handler.py | 1 - 5 files changed, 50 insertions(+), 57 deletions(-) diff --git a/litellm/litellm_core_utils/streaming_handler.py b/litellm/litellm_core_utils/streaming_handler.py index 12ba041aff8..32e6e97570f 100644 --- a/litellm/litellm_core_utils/streaming_handler.py +++ b/litellm/litellm_core_utils/streaming_handler.py @@ -1220,12 +1220,11 @@ class CustomStreamWrapper: else: ## else if model_response_str or ( - completion_obj.get("reasoning_content") is None - and completion_obj.get("tool_calls") is None + completion_obj.get("reasoning_content") is None and completion_obj.get("tool_calls") is None ): - completion_obj["content"] = model_response_str + completion_obj["content"] = model_response_str # rebind-ok: builds delta payload else: - completion_obj["content"] = None + completion_obj["content"] = None # rebind-ok: builds delta payload if self.sent_first_chunk is False: completion_obj["role"] = "assistant" self.sent_first_chunk = True @@ -1367,8 +1366,13 @@ class CustomStreamWrapper: raise StopIteration anthropic_response_obj: Final[GChunk] = cast(GChunk, chunk) completion_obj["content"] = anthropic_response_obj["text"] - if "reasoning_content" in anthropic_response_obj and anthropic_response_obj["reasoning_content"] is not None: - completion_obj["reasoning_content"] = anthropic_response_obj["reasoning_content"] + if ( + "reasoning_content" in anthropic_response_obj + and anthropic_response_obj["reasoning_content"] is not None + ): + completion_obj["reasoning_content"] = anthropic_response_obj[ # rebind-ok: forwards reasoning + "reasoning_content" + ] if anthropic_response_obj["is_finished"]: self.received_finish_reason = anthropic_response_obj["finish_reason"] @@ -1383,9 +1387,9 @@ class CustomStreamWrapper: ) if "tool_calls" in anthropic_response_obj and anthropic_response_obj["tool_calls"] is not None: - completion_obj["tool_calls"] = anthropic_response_obj["tool_calls"] + completion_obj["tool_calls"] = anthropic_response_obj["tool_calls"] # rebind-ok: forwards tool calls elif "tool_use" in anthropic_response_obj and anthropic_response_obj["tool_use"] is not None: - completion_obj["tool_calls"] = [anthropic_response_obj["tool_use"]] + completion_obj["tool_calls"] = [anthropic_response_obj["tool_use"]] # rebind-ok: forwards tool calls if ( "provider_specific_fields" in anthropic_response_obj @@ -2793,13 +2797,17 @@ def convert_generic_chunk_to_model_response_stream( ) -> ModelResponseStream: from litellm.types.utils import Delta - tool_calls = chunk.get("tool_calls", None) - if tool_calls is None and chunk.get("tool_use") is not None: - tool_calls = [chunk["tool_use"]] + raw_tool_calls: Final = chunk.get("tool_calls", None) + tool_calls: Final = ( + [chunk["tool_use"]] if raw_tool_calls is None and chunk.get("tool_use") is not None else raw_tool_calls + ) - content = chunk["text"] - if not content and (chunk.get("reasoning_content") is not None or tool_calls is not None): - content = None + raw_content: Final = chunk["text"] + content: Final = ( + None + if not raw_content and (chunk.get("reasoning_content") is not None or tool_calls is not None) + else raw_content + ) model_response_stream: Final = ModelResponseStream( id=str(uuid.uuid4()), diff --git a/litellm/llms/databricks/streaming_utils.py b/litellm/llms/databricks/streaming_utils.py index a95578a41d7..af9e7bc3c50 100644 --- a/litellm/llms/databricks/streaming_utils.py +++ b/litellm/llms/databricks/streaming_utils.py @@ -1,4 +1,5 @@ import json +from collections.abc import Sequence from typing import Final import litellm @@ -31,14 +32,6 @@ class ModelResponseIterator: try: processed_chunk: Final = litellm.ModelResponseStream(**chunk) - text = "" - tool_use: ChatCompletionToolCallChunk | None = None - tool_calls: list[ChatCompletionToolCallChunk] | None = None - reasoning_content: str | None = None - is_finished = False - finish_reason = "" - usage: ChatCompletionUsageBlock | None = None - # Usage-only final chunk (OpenAI ``stream_options.include_usage``) # arrives with an empty ``choices`` list — return usage without # indexing ``choices[0]``. @@ -53,21 +46,16 @@ class ModelResponseIterator: index=0, ) - if processed_chunk.choices[0].delta.content is not None: - text = processed_chunk.choices[0].delta.content + choice_0: Final = processed_chunk.choices[0] + choice_delta: Final = choice_0.delta + text: Final[str] = choice_delta.content if choice_delta.content is not None else "" + reasoning_content: Final[str | None] = getattr(choice_delta, "reasoning_content", None) - delta_reasoning: Final[str | None] = getattr(processed_chunk.choices[0].delta, "reasoning_content", None) - if delta_reasoning is not None: - reasoning_content = delta_reasoning - - if ( - processed_chunk.choices[0].delta.tool_calls is not None - and len(processed_chunk.choices[0].delta.tool_calls) > 0 - ): - tool_calls = [] - for tc in processed_chunk.choices[0].delta.tool_calls: - tc_fn = getattr(tc, "function", None) - fn_chunk = ( + tool_calls_list: list[ChatCompletionToolCallChunk] = [] # mutable-ok: accumulates delta calls + if choice_delta.tool_calls is not None: + for tc in choice_delta.tool_calls: + tc_fn: Final = getattr(tc, "function", None) + fn_chunk: Final = ( ChatCompletionToolCallFunctionChunk( name=getattr(tc_fn, "name", None), arguments=getattr(tc_fn, "arguments", "") or "", @@ -75,38 +63,35 @@ class ModelResponseIterator: if tc_fn is not None else ChatCompletionToolCallFunctionChunk(name=None, arguments="") ) - call_chunk = ChatCompletionToolCallChunk( + call_chunk: Final = ChatCompletionToolCallChunk( id=getattr(tc, "id", None), type="function", function=fn_chunk, index=getattr(tc, "index", 0) if getattr(tc, "index", None) is not None else 0, ) - tool_calls.append(call_chunk) - if tool_calls: - tool_use = tool_calls[0] + tool_calls_list.append(call_chunk) - if processed_chunk.choices[0].finish_reason is not None: - is_finished = True - finish_reason = processed_chunk.choices[0].finish_reason + tool_calls: Final[Sequence[ChatCompletionToolCallChunk] | None] = tool_calls_list or None + tool_use: Final[ChatCompletionToolCallChunk | None] = tool_calls_list[0] if tool_calls_list else None + + is_finished: Final[bool] = choice_0.finish_reason is not None + finish_reason: Final[str] = choice_0.finish_reason or "" usage_chunk: Final[Usage | None] = getattr(processed_chunk, "usage", None) - if usage_chunk is not None: - usage = _usage_block(usage_chunk) + usage: Final[ChatCompletionUsageBlock | None] = ( + _usage_block(usage_chunk) if usage_chunk is not None else None + ) - generic_chunk: Final[GenericStreamingChunk] = GenericStreamingChunk( + return GenericStreamingChunk( text=text, tool_use=tool_use, + tool_calls=tool_calls, + reasoning_content=reasoning_content, is_finished=is_finished, finish_reason=finish_reason, usage=usage, index=0, ) - if reasoning_content is not None: - generic_chunk["reasoning_content"] = reasoning_content - if tool_calls is not None: - generic_chunk["tool_calls"] = tool_calls - - return generic_chunk except json.JSONDecodeError: raise ValueError(f"Failed to decode JSON from chunk: {chunk}") diff --git a/litellm/types/utils.py b/litellm/types/utils.py index 1bdb2919ba7..9a90b884507 100644 --- a/litellm/types/utils.py +++ b/litellm/types/utils.py @@ -498,8 +498,8 @@ class ModelInfo(ModelInfoBase, total=False): class GenericStreamingChunk(TypedDict, total=False): text: Required[str] tool_use: ChatCompletionToolCallChunk | None - tool_calls: list[ChatCompletionToolCallChunk] | None - reasoning_content: str | None + tool_calls: ReadOnly[Sequence[ChatCompletionToolCallChunk] | None] + reasoning_content: ReadOnly[str | None] is_finished: Required[bool] finish_reason: Required[str] usage: Required[ChatCompletionUsageBlock | None] diff --git a/tests/unit/llms/databricks/test_databricks_streaming_utils.py b/tests/unit/llms/databricks/test_databricks_streaming_utils.py index 0f52afbab3a..b405505ba52 100644 --- a/tests/unit/llms/databricks/test_databricks_streaming_utils.py +++ b/tests/unit/llms/databricks/test_databricks_streaming_utils.py @@ -182,7 +182,9 @@ async def test_async_stream_skips_blank_sse_lines_and_keeps_the_usage_chunk(): chunks: Final = [chunk async for chunk in ModelResponseIterator(streaming_response=lines(), sync_stream=False)] - assert [(chunk["text"], chunk["finish_reason"], (chunk["usage"] or {}).get("prompt_tokens")) for chunk in chunks] == [ + assert [ + (chunk["text"], chunk["finish_reason"], (chunk["usage"] or {}).get("prompt_tokens")) for chunk in chunks + ] == [ ("144", "", None), ("", "stop", None), ("", "", None), @@ -291,4 +293,3 @@ def test_chunk_parser_extracts_multiple_tool_calls_with_indices(): assert result["tool_calls"][1]["index"] == 1 # Backward compatibility with tool_use assert result["tool_use"]["id"] == "call_1" - diff --git a/tests/unit/llms/openai_like/chat/test_openai_like_chat_handler.py b/tests/unit/llms/openai_like/chat/test_openai_like_chat_handler.py index defa02daad9..c43f91edf9d 100644 --- a/tests/unit/llms/openai_like/chat/test_openai_like_chat_handler.py +++ b/tests/unit/llms/openai_like/chat/test_openai_like_chat_handler.py @@ -222,4 +222,3 @@ def test_real_stream_keeps_reasoning_multi_tool_calls_and_terminal_usage(): assert chunks[4].usage.completion_tokens == 64 assert chunks[4].usage.prompt_tokens == 10 assert chunks[4].usage.completion_tokens_details.reasoning_tokens == 30 - From c6e4d372cd8ef5732a7a8b2a75c423bcf0aa70bc Mon Sep 17 00:00:00 2001 From: Rajasimhareddybolla Date: Sun, 25 Oct 2026 18:26:23 +0530 Subject: [PATCH 3/3] fix(types): resolve basedpyright diagnostics and add alternating incremental multi-tool streaming test --- .../litellm_core_utils/streaming_handler.py | 25 +- litellm/llms/databricks/streaming_utils.py | 6 +- .../chat/test_openai_like_chat_handler.py | 383 ++++++++++++++++++ 3 files changed, 397 insertions(+), 17 deletions(-) diff --git a/litellm/litellm_core_utils/streaming_handler.py b/litellm/litellm_core_utils/streaming_handler.py index 32e6e97570f..92bc0e9e6a7 100644 --- a/litellm/litellm_core_utils/streaming_handler.py +++ b/litellm/litellm_core_utils/streaming_handler.py @@ -1366,13 +1366,9 @@ class CustomStreamWrapper: raise StopIteration anthropic_response_obj: Final[GChunk] = cast(GChunk, chunk) completion_obj["content"] = anthropic_response_obj["text"] - if ( - "reasoning_content" in anthropic_response_obj - and anthropic_response_obj["reasoning_content"] is not None - ): - completion_obj["reasoning_content"] = anthropic_response_obj[ # rebind-ok: forwards reasoning - "reasoning_content" - ] + reasoning_chunk: Final = anthropic_response_obj.get("reasoning_content") + if reasoning_chunk is not None: + completion_obj["reasoning_content"] = reasoning_chunk # rebind-ok: forwards reasoning if anthropic_response_obj["is_finished"]: self.received_finish_reason = anthropic_response_obj["finish_reason"] @@ -1386,10 +1382,12 @@ class CustomStreamWrapper: litellm.Usage(**anthropic_response_obj["usage"]), ) - if "tool_calls" in anthropic_response_obj and anthropic_response_obj["tool_calls"] is not None: - completion_obj["tool_calls"] = anthropic_response_obj["tool_calls"] # rebind-ok: forwards tool calls - elif "tool_use" in anthropic_response_obj and anthropic_response_obj["tool_use"] is not None: - completion_obj["tool_calls"] = [anthropic_response_obj["tool_use"]] # rebind-ok: forwards tool calls + tool_calls_chunk: Final = anthropic_response_obj.get("tool_calls") + legacy_tool_chunk: Final = anthropic_response_obj.get("tool_use") + if tool_calls_chunk is not None: + completion_obj["tool_calls"] = tool_calls_chunk # rebind-ok: forwards tool calls + elif legacy_tool_chunk is not None: + completion_obj["tool_calls"] = [legacy_tool_chunk] # rebind-ok: forwards tool calls if ( "provider_specific_fields" in anthropic_response_obj @@ -2798,9 +2796,8 @@ def convert_generic_chunk_to_model_response_stream( from litellm.types.utils import Delta raw_tool_calls: Final = chunk.get("tool_calls", None) - tool_calls: Final = ( - [chunk["tool_use"]] if raw_tool_calls is None and chunk.get("tool_use") is not None else raw_tool_calls - ) + legacy_tool_use: Final = chunk.get("tool_use", None) + tool_calls: Final = [legacy_tool_use] if raw_tool_calls is None and legacy_tool_use is not None else raw_tool_calls raw_content: Final = chunk["text"] content: Final = ( diff --git a/litellm/llms/databricks/streaming_utils.py b/litellm/llms/databricks/streaming_utils.py index af9e7bc3c50..f9757eef4f8 100644 --- a/litellm/llms/databricks/streaming_utils.py +++ b/litellm/llms/databricks/streaming_utils.py @@ -54,8 +54,8 @@ class ModelResponseIterator: tool_calls_list: list[ChatCompletionToolCallChunk] = [] # mutable-ok: accumulates delta calls if choice_delta.tool_calls is not None: for tc in choice_delta.tool_calls: - tc_fn: Final = getattr(tc, "function", None) - fn_chunk: Final = ( + tc_fn = getattr(tc, "function", None) + fn_chunk = ( ChatCompletionToolCallFunctionChunk( name=getattr(tc_fn, "name", None), arguments=getattr(tc_fn, "arguments", "") or "", @@ -63,7 +63,7 @@ class ModelResponseIterator: if tc_fn is not None else ChatCompletionToolCallFunctionChunk(name=None, arguments="") ) - call_chunk: Final = ChatCompletionToolCallChunk( + call_chunk = ChatCompletionToolCallChunk( id=getattr(tc, "id", None), type="function", function=fn_chunk, diff --git a/tests/unit/llms/openai_like/chat/test_openai_like_chat_handler.py b/tests/unit/llms/openai_like/chat/test_openai_like_chat_handler.py index c43f91edf9d..d85f9bdb38f 100644 --- a/tests/unit/llms/openai_like/chat/test_openai_like_chat_handler.py +++ b/tests/unit/llms/openai_like/chat/test_openai_like_chat_handler.py @@ -222,3 +222,386 @@ def test_real_stream_keeps_reasoning_multi_tool_calls_and_terminal_usage(): assert chunks[4].usage.completion_tokens == 64 assert chunks[4].usage.prompt_tokens == 10 assert chunks[4].usage.completion_tokens_details.reasoning_tokens == 30 + + +def test_real_stream_incremental_alternating_tool_calls_and_negative_controls(): + events: Final[list[str]] = [ + # Frame 1: index 0 initial (id, name, arg prefix) + "data: " + + json.dumps( + { + "id": "c1", + "object": "chat.completion.chunk", + "created": 1, + "model": "glm-4", + "choices": [ + { + "index": 0, + "delta": { + "tool_calls": [ + { + "index": 0, + "id": "call_weather", + "type": "function", + "function": {"name": "get_weather", "arguments": '{"city": "'}, + } + ] + }, + "finish_reason": None, + } + ], + } + ), + # Frame 2: index 1 initial (id, name, arg prefix) + "data: " + + json.dumps( + { + "id": "c2", + "object": "chat.completion.chunk", + "created": 1, + "model": "glm-4", + "choices": [ + { + "index": 0, + "delta": { + "tool_calls": [ + { + "index": 1, + "id": "call_time", + "type": "function", + "function": {"name": "get_time", "arguments": '{"location": "'}, + } + ] + }, + "finish_reason": None, + } + ], + } + ), + # Frame 3: index 0 suffix (id and name omitted) + "data: " + + json.dumps( + { + "id": "c3", + "object": "chat.completion.chunk", + "created": 1, + "model": "glm-4", + "choices": [ + { + "index": 0, + "delta": { + "tool_calls": [ + { + "index": 0, + "function": {"arguments": 'Paris"}'}, + } + ] + }, + "finish_reason": None, + } + ], + } + ), + # Frame 4: index 1 suffix (id and name omitted) + "data: " + + json.dumps( + { + "id": "c4", + "object": "chat.completion.chunk", + "created": 1, + "model": "glm-4", + "choices": [ + { + "index": 0, + "delta": { + "tool_calls": [ + { + "index": 1, + "function": {"arguments": 'Europe/Paris"}'}, + } + ] + }, + "finish_reason": None, + } + ], + } + ), + # Frame 5: normal finish_reason length as reporter observed on Vertex + "data: " + + json.dumps( + { + "id": "c5", + "object": "chat.completion.chunk", + "created": 1, + "model": "glm-4", + "choices": [{"index": 0, "delta": {}, "finish_reason": "length"}], + } + ), + # Frame 6: empty-choice usage-only terminal completion_tokens=64 + "data: " + + json.dumps( + { + "id": "c6", + "object": "chat.completion.chunk", + "created": 1, + "model": "glm-4", + "choices": [], + "usage": { + "prompt_tokens": 12, + "completion_tokens": 64, + "total_tokens": 76, + "completion_tokens_details": {"reasoning_tokens": 18}, + }, + } + ), + "data: [DONE]", + ] + + sse_bytes: Final[bytes] = ("\n\n".join(events) + "\n\n").encode("utf-8") + + def respond(request: httpx.Request) -> httpx.Response: + return httpx.Response(200, headers={"content-type": "text/event-stream"}, content=sse_bytes) + + mock_logging: Final = Mock() + mock_logging.model_call_details = {"litellm_params": {}} + mock_logging.stream_options = {"include_usage": True} + + raw_stream: Final = make_sync_call( + client=HTTPHandler(client=httpx.Client(transport=httpx.MockTransport(respond))), + api_base="https://openai-like.example/v1/chat/completions", + headers={}, + data=json.dumps({"model": "glm-4", "messages": []}), + model="glm-4", + messages=[], + logging_obj=mock_logging, + fake_stream=False, + ) + wrapper: Final = CustomStreamWrapper( + completion_stream=raw_stream, + model="glm-4", + custom_llm_provider="vertex_ai", + logging_obj=mock_logging, + stream_options={"include_usage": True}, + ) + + chunks: Final = list(wrapper) + assert len(chunks) == 6 + + # Assert per-frame index/id/argument preservation + # Frame 0: index 0 initial + assert chunks[0].choices[0].delta.tool_calls[0].index == 0 + assert chunks[0].choices[0].delta.tool_calls[0].id == "call_weather" + assert chunks[0].choices[0].delta.tool_calls[0].function.name == "get_weather" + assert chunks[0].choices[0].delta.tool_calls[0].function.arguments == '{"city": "' + + # Frame 1: index 1 initial + assert chunks[1].choices[0].delta.tool_calls[0].index == 1 + assert chunks[1].choices[0].delta.tool_calls[0].id == "call_time" + assert chunks[1].choices[0].delta.tool_calls[0].function.name == "get_time" + assert chunks[1].choices[0].delta.tool_calls[0].function.arguments == '{"location": "' + + # Frame 2: index 0 continuation (id/name absent) + assert chunks[2].choices[0].delta.tool_calls[0].index == 0 + assert chunks[2].choices[0].delta.tool_calls[0].id is None + assert chunks[2].choices[0].delta.tool_calls[0].function.name is None + assert chunks[2].choices[0].delta.tool_calls[0].function.arguments == 'Paris"}' + + # Frame 3: index 1 continuation (id/name absent) + assert chunks[3].choices[0].delta.tool_calls[0].index == 1 + assert chunks[3].choices[0].delta.tool_calls[0].id is None + assert chunks[3].choices[0].delta.tool_calls[0].function.name is None + assert chunks[3].choices[0].delta.tool_calls[0].function.arguments == 'Europe/Paris"}' + + # Frame 4: finish_reason length + assert chunks[4].choices[0].finish_reason == "length" + + # Frame 5: usage chunk + assert chunks[5].usage is not None + assert chunks[5].usage.completion_tokens == 64 + assert chunks[5].usage.prompt_tokens == 12 + + # Reassembling in client: final separately ordered complete arguments + assembled_calls: Final[dict[int, dict[str, str]]] = {} + for c in chunks: + if c.choices and c.choices[0].delta and c.choices[0].delta.tool_calls: + for tc in c.choices[0].delta.tool_calls: + idx = tc.index + if idx not in assembled_calls: + assembled_calls[idx] = {"id": tc.id or "", "name": tc.function.name or "", "arguments": ""} + if tc.id: + assembled_calls[idx]["id"] = tc.id + if tc.function.name: + assembled_calls[idx]["name"] = tc.function.name + assembled_calls[idx]["arguments"] += tc.function.arguments + + assert assembled_calls[0]["id"] == "call_weather" + assert assembled_calls[0]["name"] == "get_weather" + assert assembled_calls[0]["arguments"] == '{"city": "Paris"}' + + assert assembled_calls[1]["id"] == "call_time" + assert assembled_calls[1]["name"] == "get_time" + assert assembled_calls[1]["arguments"] == '{"location": "Europe/Paris"}' + + +def test_real_stream_content_only_and_no_provider_usage_negative_controls(): + # Negative control 1: content-only stream emits no ghost tool calls, reasoning, or provider usage + events_content: Final[list[str]] = [ + "data: " + + json.dumps( + { + "id": "c1", + "object": "chat.completion.chunk", + "created": 1, + "model": "glm-4", + "choices": [ + { + "index": 0, + "delta": {"content": "Direct text reply."}, + "finish_reason": None, + } + ], + } + ), + "data: " + + json.dumps( + { + "id": "c2", + "object": "chat.completion.chunk", + "created": 1, + "model": "glm-4", + "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}], + } + ), + "data: [DONE]", + ] + sse_bytes_content: Final[bytes] = ("\n\n".join(events_content) + "\n\n").encode("utf-8") + + def respond_content(request: httpx.Request) -> httpx.Response: + return httpx.Response(200, headers={"content-type": "text/event-stream"}, content=sse_bytes_content) + + mock_logging_content: Final = Mock() + mock_logging_content.model_call_details = {"litellm_params": {}} + mock_logging_content.stream_options = {} + + stream_content: Final = make_sync_call( + client=HTTPHandler(client=httpx.Client(transport=httpx.MockTransport(respond_content))), + api_base="https://openai-like.example/v1/chat/completions", + headers={}, + data=json.dumps({"model": "glm-4", "messages": []}), + model="glm-4", + messages=[], + logging_obj=mock_logging_content, + fake_stream=False, + ) + wrapper_content: Final = CustomStreamWrapper( + completion_stream=stream_content, + model="glm-4", + custom_llm_provider="vertex_ai", + logging_obj=mock_logging_content, + stream_options={}, + ) + chunks_content: Final = list(wrapper_content) + assert len(chunks_content) == 2 + assert chunks_content[0].choices[0].delta.content == "Direct text reply." + assert chunks_content[0].choices[0].delta.tool_calls is None + assert getattr(chunks_content[0].choices[0].delta, "reasoning_content", None) is None + assert getattr(chunks_content[0], "usage", None) is None + assert chunks_content[1].choices[0].finish_reason == "stop" + + # Negative control 2: no-provider-usage stream terminates cleanly without synthetic usage chunks + events_no_usage: Final[list[str]] = [ + "data: " + + json.dumps( + { + "id": "c1", + "object": "chat.completion.chunk", + "created": 1, + "model": "glm-4", + "choices": [ + { + "index": 0, + "delta": { + "tool_calls": [ + { + "index": 0, + "id": "call_single", + "type": "function", + "function": {"name": "lookup", "arguments": "{}"}, + } + ] + }, + "finish_reason": None, + } + ], + } + ), + "data: " + + json.dumps( + { + "id": "c2", + "object": "chat.completion.chunk", + "created": 1, + "model": "glm-4", + "choices": [{"index": 0, "delta": {}, "finish_reason": "tool_calls"}], + } + ), + "data: [DONE]", + ] + sse_bytes_no_usage: Final[bytes] = ("\n\n".join(events_no_usage) + "\n\n").encode("utf-8") + + def respond_no_usage(request: httpx.Request) -> httpx.Response: + return httpx.Response(200, headers={"content-type": "text/event-stream"}, content=sse_bytes_no_usage) + + mock_logging_no_usage: Final = Mock() + mock_logging_no_usage.model_call_details = {"litellm_params": {}} + mock_logging_no_usage.stream_options = {} + + stream_no_usage: Final = make_sync_call( + client=HTTPHandler(client=httpx.Client(transport=httpx.MockTransport(respond_no_usage))), + api_base="https://openai-like.example/v1/chat/completions", + headers={}, + data=json.dumps({"model": "glm-4", "messages": []}), + model="glm-4", + messages=[], + logging_obj=mock_logging_no_usage, + fake_stream=False, + ) + wrapper_no_usage: Final = CustomStreamWrapper( + completion_stream=stream_no_usage, + model="glm-4", + custom_llm_provider="vertex_ai", + logging_obj=mock_logging_no_usage, + stream_options={}, + ) + chunks_no_usage: Final = list(wrapper_no_usage) + assert len(chunks_no_usage) == 2 + assert chunks_no_usage[0].choices[0].delta.tool_calls[0].id == "call_single" + assert getattr(chunks_no_usage[0], "usage", None) is None + assert getattr(chunks_no_usage[1], "usage", None) is None + assert chunks_no_usage[1].choices[0].finish_reason == "tool_calls" + + +def test_legacy_tool_use_fallback_covered(): + from litellm.litellm_core_utils.streaming_handler import convert_generic_chunk_to_model_response_stream + from litellm.types.llms.openai import ( + ChatCompletionToolCallChunk, + ChatCompletionToolCallFunctionChunk, + ) + from litellm.types.utils import GenericStreamingChunk + + legacy_chunk: Final[GenericStreamingChunk] = GenericStreamingChunk( + text="", + tool_use=ChatCompletionToolCallChunk( + id="call_legacy", + type="function", + function=ChatCompletionToolCallFunctionChunk(name="old_tool", arguments="{}"), + index=0, + ), + is_finished=False, + finish_reason="", + usage=None, + index=0, + ) + res_stream: Final = convert_generic_chunk_to_model_response_stream(legacy_chunk) + assert res_stream.choices[0].delta.tool_calls is not None + assert res_stream.choices[0].delta.tool_calls[0].id == "call_legacy"