From e1a2bfb63abcb878446540bc1180c07e3496fb34 Mon Sep 17 00:00:00 2001 From: Cole McIntosh Date: Thu, 7 Aug 2025 09:49:33 -0600 Subject: [PATCH 1/5] Fix Ollama GPT-OSS streaming with 'thinking' field - Handle chunks containing 'thinking' field with empty 'response' - Treat these as intermediate chunks that don't contain user content - Add comprehensive tests for chunk parsing scenarios - Resolves APIConnectionError for GPT-OSS model streaming Fixes #13340 --- .../llms/ollama/completion/transformation.py | 9 +++ .../test_ollama_completion_transformation.py | 77 ++++++++++++++++++- 2 files changed, 85 insertions(+), 1 deletion(-) diff --git a/litellm/llms/ollama/completion/transformation.py b/litellm/llms/ollama/completion/transformation.py index aa1da616d8..cec27b02d6 100644 --- a/litellm/llms/ollama/completion/transformation.py +++ b/litellm/llms/ollama/completion/transformation.py @@ -459,6 +459,15 @@ class OllamaTextCompletionResponseIterator(BaseModelResponseIterator): finish_reason="stop", usage=None, ) + elif "thinking" in chunk and not chunk["response"]: + # Handle GPT-OSS models that include 'thinking' field with empty response + # These are intermediate chunks that don't contain user-facing content + return GenericStreamingChunk( + text="", + is_finished=is_finished, + finish_reason=None, + usage=None, + ) else: raise Exception(f"Unable to parse ollama chunk - {chunk}") except Exception as e: diff --git a/tests/test_litellm/llms/ollama/test_ollama_completion_transformation.py b/tests/test_litellm/llms/ollama/test_ollama_completion_transformation.py index f0b5c00d01..241558cf3e 100644 --- a/tests/test_litellm/llms/ollama/test_ollama_completion_transformation.py +++ b/tests/test_litellm/llms/ollama/test_ollama_completion_transformation.py @@ -10,7 +10,10 @@ sys.path.insert( 0, os.path.abspath("../../../../..") ) # Adds the parent directory to the system path -from litellm.llms.ollama.completion.transformation import OllamaConfig +from litellm.llms.ollama.completion.transformation import ( + OllamaConfig, + OllamaTextCompletionResponseIterator, +) from litellm.types.utils import Message, ModelResponse @@ -155,3 +158,75 @@ class TestOllamaConfig: assert result.choices[0]["message"].content == expected_content assert result.choices[0]["finish_reason"] == "stop" # No usage assertions here as we don't need to test them in every case + + +class TestOllamaTextCompletionResponseIterator: + def test_chunk_parser_with_thinking_field(self): + """Test that chunks with 'thinking' field and empty 'response' are handled correctly.""" + iterator = OllamaTextCompletionResponseIterator( + streaming_response=iter([]), sync_stream=True, json_mode=False + ) + + # Test chunk with thinking field - this is the problematic case from the issue + chunk_with_thinking = { + "model": "gpt-oss:20b", + "created_at": "2025-08-06T14:34:31.5276077Z", + "response": "", + "thinking": "User", + "done": False, + } + + result = iterator.chunk_parser(chunk_with_thinking) + + # Should return empty text and not be finished + assert result["text"] == "" + assert result["is_finished"] is False + assert result["finish_reason"] is None + assert result["usage"] is None + + def test_chunk_parser_normal_response(self): + """Test that normal response chunks still work.""" + iterator = OllamaTextCompletionResponseIterator( + streaming_response=iter([]), sync_stream=True, json_mode=False + ) + + # Test normal chunk with response + normal_chunk = { + "model": "llama2", + "created_at": "2025-08-06T14:34:31.5276077Z", + "response": "Hello world", + "done": False, + } + + result = iterator.chunk_parser(normal_chunk) + + assert result["text"] == "Hello world" + assert result["is_finished"] is False + assert result["finish_reason"] == "stop" + assert result["usage"] is None + + def test_chunk_parser_done_chunk(self): + """Test that done chunks work correctly.""" + iterator = OllamaTextCompletionResponseIterator( + streaming_response=iter([]), sync_stream=True, json_mode=False + ) + + # Test done chunk + done_chunk = { + "model": "llama2", + "created_at": "2025-08-06T14:34:31.5276077Z", + "response": "", + "done": True, + "prompt_eval_count": 10, + "eval_count": 5, + } + + result = iterator.chunk_parser(done_chunk) + + assert result["text"] == "" + assert result["is_finished"] is True + assert result["finish_reason"] == "stop" + assert result["usage"] is not None + assert result["usage"]["prompt_tokens"] == 10 + assert result["usage"]["completion_tokens"] == 5 + assert result["usage"]["total_tokens"] == 15 From 938e9ace54296811144935aa42cdfc8323507b28 Mon Sep 17 00:00:00 2001 From: Cole McIntosh Date: Thu, 7 Aug 2025 10:55:33 -0600 Subject: [PATCH 2/5] Fix unclosed aiohttp client session warnings during concurrent requests Fixed MyPy type error in Ollama completion transformation where finish_reason was set to None instead of expected string type. Changed finish_reason=None to finish_reason="" to match GenericStreamingChunk TypedDict requirements. Also updated corresponding test to expect empty string instead of None. --- litellm/llms/ollama/completion/transformation.py | 2 +- .../llms/ollama/test_ollama_completion_transformation.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/litellm/llms/ollama/completion/transformation.py b/litellm/llms/ollama/completion/transformation.py index cec27b02d6..3826a47038 100644 --- a/litellm/llms/ollama/completion/transformation.py +++ b/litellm/llms/ollama/completion/transformation.py @@ -465,7 +465,7 @@ class OllamaTextCompletionResponseIterator(BaseModelResponseIterator): return GenericStreamingChunk( text="", is_finished=is_finished, - finish_reason=None, + finish_reason="", usage=None, ) else: diff --git a/tests/test_litellm/llms/ollama/test_ollama_completion_transformation.py b/tests/test_litellm/llms/ollama/test_ollama_completion_transformation.py index 241558cf3e..36f282cff8 100644 --- a/tests/test_litellm/llms/ollama/test_ollama_completion_transformation.py +++ b/tests/test_litellm/llms/ollama/test_ollama_completion_transformation.py @@ -181,7 +181,7 @@ class TestOllamaTextCompletionResponseIterator: # Should return empty text and not be finished assert result["text"] == "" assert result["is_finished"] is False - assert result["finish_reason"] is None + assert result["finish_reason"] == "" assert result["usage"] is None def test_chunk_parser_normal_response(self): From d5d7e00d340ca08da2c8af72b7f6b2a01f05878e Mon Sep 17 00:00:00 2001 From: Cole McIntosh Date: Mon, 11 Aug 2025 07:01:44 -0600 Subject: [PATCH 3/5] Enhance chunk parsing for Ollama streaming responses Updated the chunk_parser method to return a ModelResponseStream when handling 'thinking' field content, allowing UIs to render reasoning information. Adjusted tests to verify the new behavior, ensuring that reasoning content is correctly returned in the response. --- .../llms/ollama/completion/transformation.py | 20 +++++++++++-------- .../test_ollama_completion_transformation.py | 11 +++++----- 2 files changed, 17 insertions(+), 14 deletions(-) diff --git a/litellm/llms/ollama/completion/transformation.py b/litellm/llms/ollama/completion/transformation.py index 3826a47038..3565f090b0 100644 --- a/litellm/llms/ollama/completion/transformation.py +++ b/litellm/llms/ollama/completion/transformation.py @@ -24,6 +24,8 @@ from litellm.types.utils import ( ModelResponse, ModelResponseStream, ProviderField, + StreamingChoices, + Delta, ) from ..common_utils import OllamaError, _convert_image @@ -423,7 +425,7 @@ class OllamaTextCompletionResponseIterator(BaseModelResponseIterator): ) -> Union[GenericStreamingChunk, ModelResponseStream]: return self.chunk_parser(json.loads(str_line)) - def chunk_parser(self, chunk: dict) -> GenericStreamingChunk: + def chunk_parser(self, chunk: dict) -> Union[GenericStreamingChunk, ModelResponseStream]: try: if "error" in chunk: raise Exception(f"Ollama Error - {chunk}") @@ -460,13 +462,15 @@ class OllamaTextCompletionResponseIterator(BaseModelResponseIterator): usage=None, ) elif "thinking" in chunk and not chunk["response"]: - # Handle GPT-OSS models that include 'thinking' field with empty response - # These are intermediate chunks that don't contain user-facing content - return GenericStreamingChunk( - text="", - is_finished=is_finished, - finish_reason="", - usage=None, + # Return reasoning content as ModelResponseStream so UIs can render it + thinking_content = chunk.get("thinking") or "" + return ModelResponseStream( + choices=[ + StreamingChoices( + index=0, + delta=Delta(reasoning_content=thinking_content), + ) + ] ) else: raise Exception(f"Unable to parse ollama chunk - {chunk}") diff --git a/tests/test_litellm/llms/ollama/test_ollama_completion_transformation.py b/tests/test_litellm/llms/ollama/test_ollama_completion_transformation.py index 36f282cff8..985d51f99d 100644 --- a/tests/test_litellm/llms/ollama/test_ollama_completion_transformation.py +++ b/tests/test_litellm/llms/ollama/test_ollama_completion_transformation.py @@ -14,7 +14,7 @@ from litellm.llms.ollama.completion.transformation import ( OllamaConfig, OllamaTextCompletionResponseIterator, ) -from litellm.types.utils import Message, ModelResponse +from litellm.types.utils import Message, ModelResponse, ModelResponseStream class TestOllamaConfig: @@ -178,11 +178,10 @@ class TestOllamaTextCompletionResponseIterator: result = iterator.chunk_parser(chunk_with_thinking) - # Should return empty text and not be finished - assert result["text"] == "" - assert result["is_finished"] is False - assert result["finish_reason"] == "" - assert result["usage"] is None + # Should return a ModelResponseStream with reasoning content + assert isinstance(result, ModelResponseStream) + assert result.choices and result.choices[0].delta is not None + assert getattr(result.choices[0].delta, "reasoning_content") == "User" def test_chunk_parser_normal_response(self): """Test that normal response chunks still work.""" From e6ca91869a7eef75b0e31e38788b0a0ccf38e377 Mon Sep 17 00:00:00 2001 From: Cole McIntosh Date: Mon, 11 Aug 2025 07:38:16 -0600 Subject: [PATCH 4/5] merge from upstream --- litellm/caching/caching_handler.py | 72 ++++++------------------------ 1 file changed, 14 insertions(+), 58 deletions(-) diff --git a/litellm/caching/caching_handler.py b/litellm/caching/caching_handler.py index f41b745bb1..dcc59b2071 100644 --- a/litellm/caching/caching_handler.py +++ b/litellm/caching/caching_handler.py @@ -1,5 +1,5 @@ """ -This contains LLMCachingHandler +This contains LLMCachingHandler This exposes two methods: - async_get_cache @@ -18,7 +18,6 @@ import asyncio import datetime import inspect import threading -from functools import lru_cache, wraps from typing import ( TYPE_CHECKING, Any, @@ -36,13 +35,11 @@ from pydantic import BaseModel import litellm from litellm._logging import print_verbose, verbose_logger -from litellm._service_logger import ServiceLogging -from litellm.caching import InMemoryCache from litellm.caching.caching import S3Cache +from litellm.types.caching import CachedEmbedding from litellm.litellm_core_utils.logging_utils import ( _assemble_complete_response_from_streaming_chunks, ) -from litellm.types.caching import CachedEmbedding from litellm.types.rerank import RerankResponse from litellm.types.utils import ( CallTypes, @@ -71,12 +68,7 @@ class CachingHandlerResponse(BaseModel): cached_result: Optional[Any] = None final_embedding_cached_response: Optional[EmbeddingResponse] = None - embedding_all_elements_cache_hit: bool = ( - False # this is set to True when all elements in the list have a cache hit in the embedding cache, if true return the final_embedding_cached_response no need to make an API call - ) - - -in_memory_cache_obj = InMemoryCache() + embedding_all_elements_cache_hit: bool = False # this is set to True when all elements in the list have a cache hit in the embedding cache, if true return the final_embedding_cached_response no need to make an API call class LLMCachingHandler: @@ -86,20 +78,11 @@ class LLMCachingHandler: request_kwargs: Dict[str, Any], start_time: datetime.datetime, ): - from litellm.caching import DualCache, RedisCache - self.async_streaming_chunks: List[ModelResponse] = [] self.sync_streaming_chunks: List[ModelResponse] = [] self.request_kwargs = request_kwargs self.original_function = original_function self.start_time = start_time - if litellm.cache is not None and isinstance(litellm.cache.cache, RedisCache): - self.dual_cache: Optional[DualCache] = DualCache( - redis_cache=litellm.cache.cache, - in_memory_cache=in_memory_cache_obj, - ) - else: - self.dual_cache = None pass async def _async_get_cache( @@ -132,16 +115,10 @@ class LLMCachingHandler: Raises: None """ - from litellm.litellm_core_utils.core_helpers import ( - _get_parent_otel_span_from_kwargs, - ) from litellm.utils import CustomStreamWrapper - kwargs = kwargs.copy() args = args or () - parent_otel_span = _get_parent_otel_span_from_kwargs(kwargs) - kwargs["parent_otel_span"] = parent_otel_span final_embedding_cached_response: Optional[EmbeddingResponse] = None embedding_all_elements_cache_hit: bool = False cached_result: Optional[Any] = None @@ -329,15 +306,13 @@ class LLMCachingHandler: else: raise ValueError("input must be a string or a list") - def _extract_model_from_cached_results( - self, non_null_list: List[Tuple[int, CachedEmbedding]] - ) -> Optional[str]: + def _extract_model_from_cached_results(self, non_null_list: List[Tuple[int, CachedEmbedding]]) -> Optional[str]: """ Helper method to extract the model name from cached results. - + Args: non_null_list: List of (idx, cr) tuples where cr is the cached result dict - + Returns: Optional[str]: The model name if found, None otherwise """ @@ -583,12 +558,7 @@ class LLMCachingHandler: preset_cache_key = litellm.cache.get_cache_key( **{**new_kwargs, "input": i} ) - tasks.append( - litellm.cache.async_get_cache( - cache_key=preset_cache_key, - dynamic_cache_object=self.dual_cache, - ) - ) + tasks.append(litellm.cache.async_get_cache(cache_key=preset_cache_key)) cached_result = await asyncio.gather(*tasks) ## check if cached result is None ## if cached_result is not None and isinstance(cached_result, list): @@ -597,14 +567,9 @@ class LLMCachingHandler: cached_result = None else: if litellm.cache._supports_async() is True: - ## check if dual cache is supported ## - cached_result = await litellm.cache.async_get_cache( - dynamic_cache_object=self.dual_cache, **new_kwargs - ) + cached_result = await litellm.cache.async_get_cache(**new_kwargs) else: # for s3 caching. [NOT RECOMMENDED IN PROD - this will slow down responses since boto3 is sync] - cached_result = litellm.cache.get_cache( - dynamic_cache_object=self.dual_cache, **new_kwargs - ) + cached_result = litellm.cache.get_cache(**new_kwargs) return cached_result def _convert_cached_result_to_model_response( @@ -770,9 +735,6 @@ class LLMCachingHandler: Raises: None """ - from litellm.litellm_core_utils.core_helpers import ( - _get_parent_otel_span_from_kwargs, - ) if litellm.cache is None: return @@ -784,8 +746,6 @@ class LLMCachingHandler: args, ) ) - parent_otel_span = _get_parent_otel_span_from_kwargs(new_kwargs) - new_kwargs["parent_otel_span"] = parent_otel_span # [OPTIONAL] ADD TO CACHE if self._should_store_result_in_cache( original_function=original_function, kwargs=new_kwargs @@ -804,9 +764,7 @@ class LLMCachingHandler: ) # s3 doesn't support bulk writing. Exclude. ): asyncio.create_task( - litellm.cache.async_add_cache_pipeline( - result, dynamic_cache_object=self.dual_cache, **new_kwargs - ) + litellm.cache.async_add_cache_pipeline(result, **new_kwargs) ) elif isinstance(litellm.cache.cache, S3Cache): threading.Thread( @@ -817,9 +775,7 @@ class LLMCachingHandler: else: asyncio.create_task( litellm.cache.async_add_cache( - result.model_dump_json(), - dynamic_cache_object=self.dual_cache, - **new_kwargs, + result.model_dump_json(), **new_kwargs ) ) else: @@ -977,9 +933,9 @@ class LLMCachingHandler: } if litellm.cache is not None: - litellm_params["preset_cache_key"] = ( - litellm.cache._get_preset_cache_key_from_kwargs(**kwargs) - ) + litellm_params[ + "preset_cache_key" + ] = litellm.cache._get_preset_cache_key_from_kwargs(**kwargs) else: litellm_params["preset_cache_key"] = None From 8197fd74d5eb674d18bcd55d1bb8fc41f77b39df Mon Sep 17 00:00:00 2001 From: Cole McIntosh Date: Mon, 11 Aug 2025 07:54:47 -0600 Subject: [PATCH 5/5] Revert "merge from upstream" This reverts commit e6ca91869a7eef75b0e31e38788b0a0ccf38e377. --- litellm/caching/caching_handler.py | 72 ++++++++++++++++++++++++------ 1 file changed, 58 insertions(+), 14 deletions(-) diff --git a/litellm/caching/caching_handler.py b/litellm/caching/caching_handler.py index dcc59b2071..f41b745bb1 100644 --- a/litellm/caching/caching_handler.py +++ b/litellm/caching/caching_handler.py @@ -1,5 +1,5 @@ """ -This contains LLMCachingHandler +This contains LLMCachingHandler This exposes two methods: - async_get_cache @@ -18,6 +18,7 @@ import asyncio import datetime import inspect import threading +from functools import lru_cache, wraps from typing import ( TYPE_CHECKING, Any, @@ -35,11 +36,13 @@ from pydantic import BaseModel import litellm from litellm._logging import print_verbose, verbose_logger +from litellm._service_logger import ServiceLogging +from litellm.caching import InMemoryCache from litellm.caching.caching import S3Cache -from litellm.types.caching import CachedEmbedding from litellm.litellm_core_utils.logging_utils import ( _assemble_complete_response_from_streaming_chunks, ) +from litellm.types.caching import CachedEmbedding from litellm.types.rerank import RerankResponse from litellm.types.utils import ( CallTypes, @@ -68,7 +71,12 @@ class CachingHandlerResponse(BaseModel): cached_result: Optional[Any] = None final_embedding_cached_response: Optional[EmbeddingResponse] = None - embedding_all_elements_cache_hit: bool = False # this is set to True when all elements in the list have a cache hit in the embedding cache, if true return the final_embedding_cached_response no need to make an API call + embedding_all_elements_cache_hit: bool = ( + False # this is set to True when all elements in the list have a cache hit in the embedding cache, if true return the final_embedding_cached_response no need to make an API call + ) + + +in_memory_cache_obj = InMemoryCache() class LLMCachingHandler: @@ -78,11 +86,20 @@ class LLMCachingHandler: request_kwargs: Dict[str, Any], start_time: datetime.datetime, ): + from litellm.caching import DualCache, RedisCache + self.async_streaming_chunks: List[ModelResponse] = [] self.sync_streaming_chunks: List[ModelResponse] = [] self.request_kwargs = request_kwargs self.original_function = original_function self.start_time = start_time + if litellm.cache is not None and isinstance(litellm.cache.cache, RedisCache): + self.dual_cache: Optional[DualCache] = DualCache( + redis_cache=litellm.cache.cache, + in_memory_cache=in_memory_cache_obj, + ) + else: + self.dual_cache = None pass async def _async_get_cache( @@ -115,10 +132,16 @@ class LLMCachingHandler: Raises: None """ + from litellm.litellm_core_utils.core_helpers import ( + _get_parent_otel_span_from_kwargs, + ) from litellm.utils import CustomStreamWrapper + kwargs = kwargs.copy() args = args or () + parent_otel_span = _get_parent_otel_span_from_kwargs(kwargs) + kwargs["parent_otel_span"] = parent_otel_span final_embedding_cached_response: Optional[EmbeddingResponse] = None embedding_all_elements_cache_hit: bool = False cached_result: Optional[Any] = None @@ -306,13 +329,15 @@ class LLMCachingHandler: else: raise ValueError("input must be a string or a list") - def _extract_model_from_cached_results(self, non_null_list: List[Tuple[int, CachedEmbedding]]) -> Optional[str]: + def _extract_model_from_cached_results( + self, non_null_list: List[Tuple[int, CachedEmbedding]] + ) -> Optional[str]: """ Helper method to extract the model name from cached results. - + Args: non_null_list: List of (idx, cr) tuples where cr is the cached result dict - + Returns: Optional[str]: The model name if found, None otherwise """ @@ -558,7 +583,12 @@ class LLMCachingHandler: preset_cache_key = litellm.cache.get_cache_key( **{**new_kwargs, "input": i} ) - tasks.append(litellm.cache.async_get_cache(cache_key=preset_cache_key)) + tasks.append( + litellm.cache.async_get_cache( + cache_key=preset_cache_key, + dynamic_cache_object=self.dual_cache, + ) + ) cached_result = await asyncio.gather(*tasks) ## check if cached result is None ## if cached_result is not None and isinstance(cached_result, list): @@ -567,9 +597,14 @@ class LLMCachingHandler: cached_result = None else: if litellm.cache._supports_async() is True: - cached_result = await litellm.cache.async_get_cache(**new_kwargs) + ## check if dual cache is supported ## + cached_result = await litellm.cache.async_get_cache( + dynamic_cache_object=self.dual_cache, **new_kwargs + ) else: # for s3 caching. [NOT RECOMMENDED IN PROD - this will slow down responses since boto3 is sync] - cached_result = litellm.cache.get_cache(**new_kwargs) + cached_result = litellm.cache.get_cache( + dynamic_cache_object=self.dual_cache, **new_kwargs + ) return cached_result def _convert_cached_result_to_model_response( @@ -735,6 +770,9 @@ class LLMCachingHandler: Raises: None """ + from litellm.litellm_core_utils.core_helpers import ( + _get_parent_otel_span_from_kwargs, + ) if litellm.cache is None: return @@ -746,6 +784,8 @@ class LLMCachingHandler: args, ) ) + parent_otel_span = _get_parent_otel_span_from_kwargs(new_kwargs) + new_kwargs["parent_otel_span"] = parent_otel_span # [OPTIONAL] ADD TO CACHE if self._should_store_result_in_cache( original_function=original_function, kwargs=new_kwargs @@ -764,7 +804,9 @@ class LLMCachingHandler: ) # s3 doesn't support bulk writing. Exclude. ): asyncio.create_task( - litellm.cache.async_add_cache_pipeline(result, **new_kwargs) + litellm.cache.async_add_cache_pipeline( + result, dynamic_cache_object=self.dual_cache, **new_kwargs + ) ) elif isinstance(litellm.cache.cache, S3Cache): threading.Thread( @@ -775,7 +817,9 @@ class LLMCachingHandler: else: asyncio.create_task( litellm.cache.async_add_cache( - result.model_dump_json(), **new_kwargs + result.model_dump_json(), + dynamic_cache_object=self.dual_cache, + **new_kwargs, ) ) else: @@ -933,9 +977,9 @@ class LLMCachingHandler: } if litellm.cache is not None: - litellm_params[ - "preset_cache_key" - ] = litellm.cache._get_preset_cache_key_from_kwargs(**kwargs) + litellm_params["preset_cache_key"] = ( + litellm.cache._get_preset_cache_key_from_kwargs(**kwargs) + ) else: litellm_params["preset_cache_key"] = None