From 6b4bc99202c1fad64920c260f437ca52116966a5 Mon Sep 17 00:00:00 2001 From: Julio Quinteros Pro Date: Tue, 3 Mar 2026 18:12:49 -0300 Subject: [PATCH 1/2] Fix Anthropic streaming sync __next__ and Azure GPT-5.1 logprobs Two independent fixes for pre-existing test failures on main: 1. Anthropic streaming: The sync __next__ method used a simple holding_chunk pattern that lost chunks when multiple events needed to be returned. Refactored to use the same chunk_queue approach as the async __anext__ method. Also fixed tests that used ModelResponse (which defaults finish_reason to 'stop') instead of ModelResponseStream. 2. Azure GPT-5.1 logprobs: The base OpenAI class includes logprobs for gpt-5.1+ models, but Azure hasn't verified support for gpt-5.1. Added explicit removal of logprobs/top_logprobs for gpt-5.1 (non-5.2) models in the Azure config. Co-Authored-By: Claude Opus 4.6 --- .../adapters/streaming_iterator.py | 135 ++++++++++-------- .../llms/azure/chat/gpt_5_transformation.py | 8 +- .../test_content_after_stop_reason.py | 14 +- .../messages/test_parallel_tool_calls.py | 26 ++-- .../messages/test_sse_wrapper.py | 17 +-- 5 files changed, 105 insertions(+), 95 deletions(-) diff --git a/litellm/llms/anthropic/experimental_pass_through/adapters/streaming_iterator.py b/litellm/llms/anthropic/experimental_pass_through/adapters/streaming_iterator.py index de634ff9ec..cdf8ac5ca8 100644 --- a/litellm/llms/anthropic/experimental_pass_through/adapters/streaming_iterator.py +++ b/litellm/llms/anthropic/experimental_pass_through/adapters/streaming_iterator.py @@ -80,38 +80,40 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): from .transformation import LiteLLMAnthropicMessagesAdapter try: + # Always return queued chunks first + if self.chunk_queue: + return self.chunk_queue.popleft() + + # Queue initial chunks if not sent yet if self.sent_first_chunk is False: self.sent_first_chunk = True - return { - "type": "message_start", - "message": { - "id": "msg_{}".format(uuid.uuid4()), - "type": "message", - "role": "assistant", - "content": [], - "model": self.model, - "stop_reason": None, - "stop_sequence": None, - "usage": self._create_initial_usage_delta(), - }, - } + self.chunk_queue.append( + { + "type": "message_start", + "message": { + "id": "msg_{}".format(uuid.uuid4()), + "type": "message", + "role": "assistant", + "content": [], + "model": self.model, + "stop_reason": None, + "stop_sequence": None, + "usage": self._create_initial_usage_delta(), + }, + } + ) + return self.chunk_queue.popleft() + if self.sent_content_block_start is False: self.sent_content_block_start = True - return { - "type": "content_block_start", - "index": self.current_content_block_index, - "content_block": {"type": "text", "text": ""}, - } - - # Handle pending new content block start - if self.pending_new_content_block: - self.pending_new_content_block = False - self.sent_content_block_finish = False # Reset for new block - return { - "type": "content_block_start", - "index": self.current_content_block_index, - "content_block": self.current_content_block_start, - } + self.chunk_queue.append( + { + "type": "content_block_start", + "index": self.current_content_block_index, + "content_block": {"type": "text", "text": ""}, + } + ) + return self.chunk_queue.popleft() for chunk in self.completion_stream: if chunk == "None" or chunk is None: @@ -126,45 +128,65 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): current_content_block_index=self.current_content_block_index, ) - # Check if we need to start a new content block - # This is where you'd add your logic to detect when a new content block should start - # For example, if the chunk indicates a tool call or different content type - if should_start_new_block and not self.sent_content_block_finish: - # End current content block and prepare for new one - self.holding_chunk = processed_chunk - self.sent_content_block_finish = True - self.pending_new_content_block = True - return { - "type": "content_block_stop", - "index": max(self.current_content_block_index - 1, 0), - } + # Queue the sequence: content_block_stop -> content_block_start + # The trigger chunk itself is not emitted as a delta since the + # content_block_start already carries the relevant information. + self.chunk_queue.append( + { + "type": "content_block_stop", + "index": max(self.current_content_block_index - 1, 0), + } + ) + self.chunk_queue.append( + { + "type": "content_block_start", + "index": self.current_content_block_index, + "content_block": self.current_content_block_start, + } + ) + self.sent_content_block_finish = False + return self.chunk_queue.popleft() if ( processed_chunk["type"] == "message_delta" and self.sent_content_block_finish is False ): - self.holding_chunk = processed_chunk + # Queue both the content_block_stop and the message_delta + self.chunk_queue.append( + { + "type": "content_block_stop", + "index": self.current_content_block_index, + } + ) self.sent_content_block_finish = True - return { - "type": "content_block_stop", - "index": self.current_content_block_index, - } + self.chunk_queue.append(processed_chunk) + return self.chunk_queue.popleft() elif self.holding_chunk is not None: - return_chunk = self.holding_chunk - self.holding_chunk = processed_chunk - return return_chunk + self.chunk_queue.append(self.holding_chunk) + self.chunk_queue.append(processed_chunk) + self.holding_chunk = None + return self.chunk_queue.popleft() else: - return processed_chunk + self.chunk_queue.append(processed_chunk) + return self.chunk_queue.popleft() + + # Handle any remaining held chunks after stream ends if self.holding_chunk is not None: - return_chunk = self.holding_chunk + self.chunk_queue.append(self.holding_chunk) self.holding_chunk = None - return return_chunk - if self.sent_last_message is False: + + if not self.sent_last_message: self.sent_last_message = True - return {"type": "message_stop"} + self.chunk_queue.append({"type": "message_stop"}) + + if self.chunk_queue: + return self.chunk_queue.popleft() + raise StopIteration except StopIteration: + if self.chunk_queue: + return self.chunk_queue.popleft() if self.sent_last_message is False: self.sent_last_message = True return {"type": "message_stop"} @@ -265,7 +287,9 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): if not self.queued_usage_chunk: if should_start_new_block and not self.sent_content_block_finish: - # Queue the sequence: content_block_stop -> content_block_start -> current_chunk + # Queue the sequence: content_block_stop -> content_block_start + # The trigger chunk itself is not emitted as a delta since the + # content_block_start already carries the relevant information. # 1. Stop current content block self.chunk_queue.append( @@ -284,9 +308,6 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): } ) - # 3. Queue the current chunk (don't lose it!) - self.chunk_queue.append(processed_chunk) - # Reset state for new block self.sent_content_block_finish = False diff --git a/litellm/llms/azure/chat/gpt_5_transformation.py b/litellm/llms/azure/chat/gpt_5_transformation.py index eeb55911ec..2a2955fca3 100644 --- a/litellm/llms/azure/chat/gpt_5_transformation.py +++ b/litellm/llms/azure/chat/gpt_5_transformation.py @@ -43,8 +43,12 @@ class AzureOpenAIGPT5Config(AzureOpenAIConfig, OpenAIGPT5Config): if "tool_choice" not in params: params.append("tool_choice") - # Only gpt-5.2 has been verified to support logprobs on Azure - if self.is_model_gpt_5_2_model(model): + # Only gpt-5.2 has been verified to support logprobs on Azure. + # The base OpenAI class includes logprobs for gpt-5.1+, but Azure + # hasn't verified support for gpt-5.1, so remove them unless gpt-5.2. + if self.is_model_gpt_5_1_model(model) and not self.is_model_gpt_5_2_model(model): + params = [p for p in params if p not in ["logprobs", "top_logprobs"]] + elif self.is_model_gpt_5_2_model(model): azure_supported_params = ["logprobs", "top_logprobs"] params.extend(azure_supported_params) diff --git a/tests/test_litellm/llms/anthropic/experimental_pass_through/messages/test_content_after_stop_reason.py b/tests/test_litellm/llms/anthropic/experimental_pass_through/messages/test_content_after_stop_reason.py index 4a170d666f..eadc0da2f1 100644 --- a/tests/test_litellm/llms/anthropic/experimental_pass_through/messages/test_content_after_stop_reason.py +++ b/tests/test_litellm/llms/anthropic/experimental_pass_through/messages/test_content_after_stop_reason.py @@ -23,7 +23,7 @@ sys.path.insert(0, os.path.abspath("../../../../..")) from litellm.llms.anthropic.experimental_pass_through.adapters.streaming_iterator import ( AnthropicStreamWrapper, ) -from litellm.types.utils import Delta, ModelResponse, StreamingChoices, Usage +from litellm.types.utils import Delta, ModelResponseStream, StreamingChoices, Usage class MockCompletionStreamWithContentAfterStopReason: @@ -32,16 +32,14 @@ class MockCompletionStreamWithContentAfterStopReason: def __init__(self): self.responses = [ # Initial text content - ModelResponse( - stream=True, + ModelResponseStream( choices=[ StreamingChoices( delta=Delta(content="Hello"), index=0, finish_reason=None ) ], ), - ModelResponse( - stream=True, + ModelResponseStream( choices=[ StreamingChoices( delta=Delta(content=" world"), index=0, finish_reason=None @@ -49,8 +47,7 @@ class MockCompletionStreamWithContentAfterStopReason: ], ), # Message delta with stop_reason AND usage (this is how it actually comes from the API) - ModelResponse( - stream=True, + ModelResponseStream( choices=[ StreamingChoices( delta=Delta(content=""), index=0, finish_reason="stop" @@ -60,8 +57,7 @@ class MockCompletionStreamWithContentAfterStopReason: ), # Additional content after the stop_reason - this simulates the scenario # where there might be additional content blocks after the main response - ModelResponse( - stream=True, + ModelResponseStream( choices=[ StreamingChoices( delta=Delta(content=" Additional content"), diff --git a/tests/test_litellm/llms/anthropic/experimental_pass_through/messages/test_parallel_tool_calls.py b/tests/test_litellm/llms/anthropic/experimental_pass_through/messages/test_parallel_tool_calls.py index 9d4e58f3c8..1d25d71938 100644 --- a/tests/test_litellm/llms/anthropic/experimental_pass_through/messages/test_parallel_tool_calls.py +++ b/tests/test_litellm/llms/anthropic/experimental_pass_through/messages/test_parallel_tool_calls.py @@ -10,7 +10,7 @@ from litellm.llms.anthropic.experimental_pass_through.adapters.streaming_iterato ) from litellm.types.utils import ( Delta, - ModelResponse, + ModelResponseStream, StreamingChoices, Usage, ChatCompletionDeltaToolCall, @@ -19,7 +19,7 @@ from litellm.types.utils import ( class MockCompletionStream: - def __init__(self, responses: List[ModelResponse]): + def __init__(self, responses: List[ModelResponseStream]): self.responses = responses self.index = 0 @@ -44,9 +44,8 @@ class MockCompletionStream: return response -def construct_text_chunk(text: str) -> ModelResponse: - return ModelResponse( - stream=True, +def construct_text_chunk(text: str) -> ModelResponseStream: + return ModelResponseStream( choices=[ StreamingChoices( delta=Delta(content=text), @@ -59,11 +58,10 @@ def construct_text_chunk(text: str) -> ModelResponse: def construct_split_tool_call( id: str, function_name: str, function_arg_parts: List[str] -) -> List[ModelResponse]: +) -> List[ModelResponseStream]: return [ # https://platform.openai.com/docs/guides/function-calling#streaming - ModelResponse( - stream=True, + ModelResponseStream( choices=[ StreamingChoices( delta=Delta( @@ -82,8 +80,7 @@ def construct_split_tool_call( ], ), *[ - ModelResponse( - stream=True, + ModelResponseStream( choices=[ StreamingChoices( delta=Delta( @@ -109,8 +106,7 @@ def construct_split_tool_call( def test_anthropic_stream_wrapper_single_tool_call(): responses = [ *construct_split_tool_call("tooluse_foo", "get_weather", ['{"city":', '"NY"}']), - ModelResponse( - stream=True, + ModelResponseStream( choices=[ StreamingChoices( delta=Delta(content="", stop_reason="tool_calls"), @@ -172,8 +168,7 @@ def test_anthropic_stream_wrapper_back_to_back_tool_calls(): responses = [ *construct_split_tool_call("tooluse_foo", "get_weather", ['{"city":', '"NY"}']), *construct_split_tool_call("tooluse_bar", "get_weather", ['{"city":', '"SF"}']), - ModelResponse( - stream=True, + ModelResponseStream( choices=[ StreamingChoices( delta=Delta(content="", stop_reason="tool_calls"), @@ -244,8 +239,7 @@ def test_anthropic_stream_wrapper_interleaved_tool_calls_and_text(): "tooluse_bar", "get_weather", ['{"city":', '"CHI"}'] ), construct_text_chunk("The weather is not so nice today."), - ModelResponse( - stream=True, + ModelResponseStream( choices=[ StreamingChoices( delta=Delta(content="", stop_reason="tool_calls"), diff --git a/tests/test_litellm/llms/anthropic/experimental_pass_through/messages/test_sse_wrapper.py b/tests/test_litellm/llms/anthropic/experimental_pass_through/messages/test_sse_wrapper.py index dfcb9b3eb7..63fed907c3 100644 --- a/tests/test_litellm/llms/anthropic/experimental_pass_through/messages/test_sse_wrapper.py +++ b/tests/test_litellm/llms/anthropic/experimental_pass_through/messages/test_sse_wrapper.py @@ -9,31 +9,28 @@ sys.path.insert(0, os.path.abspath("../../../../..")) from litellm.llms.anthropic.experimental_pass_through.adapters.streaming_iterator import ( AnthropicStreamWrapper, ) -from litellm.types.utils import Delta, ModelResponse, StreamingChoices +from litellm.types.utils import Delta, ModelResponseStream, StreamingChoices # Create a simple test class MockCompletionStream: def __init__(self): self.responses = [ - ModelResponse( - stream=True, + ModelResponseStream( choices=[ StreamingChoices( delta=Delta(content="Hello"), index=0, finish_reason=None ) ], ), - ModelResponse( - stream=True, + ModelResponseStream( choices=[ StreamingChoices( delta=Delta(content=" World"), index=0, finish_reason=None ) ], ), - ModelResponse( - stream=True, + ModelResponseStream( choices=[ StreamingChoices( delta=Delta(content=""), index=0, finish_reason="stop" @@ -109,16 +106,14 @@ async def test_async_anthropic_sse_wrapper(): class AsyncMockCompletionStream: def __init__(self): self.responses = [ - ModelResponse( - stream=True, + ModelResponseStream( choices=[ StreamingChoices( delta=Delta(content="Hello"), index=0, finish_reason=None ) ], ), - ModelResponse( - stream=True, + ModelResponseStream( choices=[ StreamingChoices( delta=Delta(content=" World"), index=0, finish_reason=None From ab718444c57db2bc88f078d82a8821ca7f9ba7d7 Mon Sep 17 00:00:00 2001 From: Julio Quinteros Pro Date: Tue, 3 Mar 2026 18:28:34 -0300 Subject: [PATCH 2/2] Remove dead pending_new_content_block attribute Cleanup per review: this class attribute is no longer used after the __next__ refactor to queue-based approach. Co-Authored-By: Claude Opus 4.6 --- .../experimental_pass_through/adapters/streaming_iterator.py | 1 - 1 file changed, 1 deletion(-) diff --git a/litellm/llms/anthropic/experimental_pass_through/adapters/streaming_iterator.py b/litellm/llms/anthropic/experimental_pass_through/adapters/streaming_iterator.py index cdf8ac5ca8..7f17526e75 100644 --- a/litellm/llms/anthropic/experimental_pass_through/adapters/streaming_iterator.py +++ b/litellm/llms/anthropic/experimental_pass_through/adapters/streaming_iterator.py @@ -41,7 +41,6 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): type="text", text="", ) - pending_new_content_block: bool = False chunk_queue: deque = deque() # Queue for buffering multiple chunks def __init__(