diff --git a/litellm/responses/file_search/emulated_handler.py b/litellm/responses/file_search/emulated_handler.py index 6e27ad66f54..b7db9c8acd7 100644 --- a/litellm/responses/file_search/emulated_handler.py +++ b/litellm/responses/file_search/emulated_handler.py @@ -18,6 +18,7 @@ from typing import Any, Dict, Iterable, List, Optional, Tuple, Union, cast from litellm._internal_context import is_internal_call from litellm._logging import verbose_logger +from litellm.responses.streaming_iterator import SyntheticResponsesAPIStreamingIterator from litellm.types.llms.openai import ResponseOutputItem, ResponsesAPIResponse from litellm.types.vector_stores import VectorStoreSearchResult @@ -381,20 +382,41 @@ async def _call_aresponses(input, model, tools, **kwargs): # pragma: no cover def _prepare_emulated_file_search_call( - kwargs: Dict[str, Any], -) -> Tuple[bool, Dict[str, Any]]: + kwargs: dict[str, Any], +) -> tuple[bool, bool, dict[str, Any]]: include_items: List[str] = list(kwargs.get("include") or []) include_search_results = "file_search_call.results" in include_items - original_stream = kwargs.get("stream") + stream_requested = bool(kwargs.get("stream")) updated_kwargs = kwargs - if original_stream: + if stream_requested: verbose_logger.debug( - "Streaming is not yet supported for emulated file_search. Disabling stream for this request." + "Emulated file_search runs the provider calls non-streaming; " + "the synthesized response is replayed as Responses API stream events." ) updated_kwargs = {**kwargs, "stream": False} - return include_search_results, updated_kwargs + return include_search_results, stream_requested, updated_kwargs + + +def _as_stream_if_requested( + response: ResponsesAPIResponse, + stream_requested: bool, + kwargs: dict[str, Any], +) -> Union[ResponsesAPIResponse, "SyntheticResponsesAPIStreamingIterator"]: + if not stream_requested: + return response + + logging_obj = kwargs.get("litellm_logging_obj") + if logging_obj is None: + verbose_logger.warning("Emulated file_search: no logging object available, returning a non-streaming response.") + return response + + return SyntheticResponsesAPIStreamingIterator( + response=response, + logging_obj=logging_obj, + custom_llm_provider=kwargs.get("custom_llm_provider"), + ) def _extract_tool_call_fields(tool_call: Any, fallback_call_id: str) -> Tuple[str, str]: @@ -495,15 +517,16 @@ async def aresponses_with_emulated_file_search( tools: Optional[Iterable[ToolParam]] = None, # Pass-through params — forwarded as-is to the underlying aresponses call **kwargs: Any, -) -> ResponsesAPIResponse: +) -> Union[ResponsesAPIResponse, SyntheticResponsesAPIStreamingIterator]: """ Emulated file_search for providers that don't support it natively. Replaces file_search tools with a function tool, intercepts the tool call, - runs vector search, and synthesizes an OpenAI-format response. + runs vector search, and synthesizes an OpenAI-format response. When the caller asked + for stream=true, that response is replayed as Responses API stream events. """ # Determine whether caller wants search_results populated in the output. - _include_search_results, kwargs = _prepare_emulated_file_search_call(kwargs=kwargs) + _include_search_results, _stream_requested, kwargs = _prepare_emulated_file_search_call(kwargs=kwargs) # 1. Replace file_search tools with function tool transformed_tools, all_vs_ids = _replace_file_search_tools(tools) @@ -547,15 +570,19 @@ async def aresponses_with_emulated_file_search( # Return as-is wrapped in OpenAI format. call_id = f"fs_{uuid.uuid4().hex[:24]}" response_text = _extract_text_from_responses_output(first_response) - return _synthesize_responses_api_response( - original_response=first_response, - file_search_call_output=_build_file_search_call_output( - call_id=call_id, - queries=[str(input)], - results=None, - include_search_results=False, + return _as_stream_if_requested( + _synthesize_responses_api_response( + original_response=first_response, + file_search_call_output=_build_file_search_call_output( + call_id=call_id, + queries=[str(input)], + results=None, + include_search_results=False, + ), + message_output=_build_message_output(response_text, []), ), - message_output=_build_message_output(response_text, []), + stream_requested=_stream_requested, + kwargs=kwargs, ) # 4. Execute each file_search tool call @@ -593,14 +620,18 @@ async def aresponses_with_emulated_file_search( # 7. Synthesize OpenAI-format output response_text = _extract_text_from_responses_output(final_response) - return _synthesize_responses_api_response( - original_response=final_response, - file_search_call_output=_build_file_search_call_output( - call_id=file_search_call_id, - queries=all_queries or [str(input)], - results=all_results, - include_search_results=_include_search_results, + return _as_stream_if_requested( + _synthesize_responses_api_response( + original_response=final_response, + file_search_call_output=_build_file_search_call_output( + call_id=file_search_call_id, + queries=all_queries or [str(input)], + results=all_results, + include_search_results=_include_search_results, + ), + message_output=_build_message_output(response_text, all_results), + first_response=first_response, ), - message_output=_build_message_output(response_text, all_results), - first_response=first_response, + stream_requested=_stream_requested, + kwargs=kwargs, ) diff --git a/litellm/responses/streaming_iterator.py b/litellm/responses/streaming_iterator.py index 357a7ecefe6..8c258042d34 100644 --- a/litellm/responses/streaming_iterator.py +++ b/litellm/responses/streaming_iterator.py @@ -901,13 +901,17 @@ class MockResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator): return evt -class CachedResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator): +class SyntheticResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator): + """Replay an already-materialized ResponsesAPIResponse as Responses-API SSE events.""" + def __init__( self, response: Any, logging_obj: LiteLLMLoggingObj, - request_data: Optional[Dict[str, Any]] = None, - call_type: Optional[str] = None, + custom_llm_provider: str | None = None, + litellm_metadata: dict[str, Any] | None = None, + request_data: dict[str, Any] | None = None, + call_type: str | None = None, ): BaseResponsesAPIStreamingIterator.__init__( self, @@ -915,13 +919,11 @@ class CachedResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator): model=getattr(response, "model", ""), responses_api_provider_config=None, logging_obj=logging_obj, - litellm_metadata=None, - custom_llm_provider="cached_response", + litellm_metadata=litellm_metadata, + custom_llm_provider=custom_llm_provider, request_data=request_data, call_type=call_type, ) - self._completed_response_cache_hit = True - self._persist_completed_response_before_logging = False self._events: List[Any] = [] self._idx = 0 self._set_events_from_response(transformed=response, logging_obj=logging_obj) @@ -968,6 +970,25 @@ class CachedResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator): return evt +class CachedResponsesAPIStreamingIterator(SyntheticResponsesAPIStreamingIterator): + def __init__( + self, + response: Any, + logging_obj: LiteLLMLoggingObj, + request_data: dict[str, Any] | None = None, + call_type: str | None = None, + ): + super().__init__( + response=response, + logging_obj=logging_obj, + custom_llm_provider="cached_response", + request_data=request_data, + call_type=call_type, + ) + self._completed_response_cache_hit = True + self._persist_completed_response_before_logging = False + + def _dump_response_object(obj: Any) -> Dict[str, Any]: if hasattr(obj, "model_dump"): return obj.model_dump() diff --git a/tests/test_litellm/llms/test_file_search_responses.py b/tests/test_litellm/llms/test_file_search_responses.py index 2f7ad3874fa..751e8468d4c 100644 --- a/tests/test_litellm/llms/test_file_search_responses.py +++ b/tests/test_litellm/llms/test_file_search_responses.py @@ -935,3 +935,92 @@ class TestEmulatedFileSearchHandler: f"Sub-call {i} must run with is_internal_call=True to suppress " "billing callbacks in wrapper_async" ) + + @pytest.mark.asyncio + async def test_H16_stream_true_returns_async_iterable_of_response_events(self): + """stream=true must yield Responses API SSE events, not a bare ResponsesAPIResponse. + + Regression for https://github.com/BerriAI/litellm/issues/34767 where the proxy + crashed with "'async for' requires an object with __aiter__ method, got + ResponsesAPIResponse". + """ + from datetime import datetime + + from litellm.responses.file_search.emulated_handler import ( + aresponses_with_emulated_file_search, + ) + + class _LoggingObj: + def __init__(self): + self.start_time = datetime.now() + self.completion_start_time = None + self.model_call_details = {"litellm_params": {}} + + async def dispatch_success_handlers(self, *args, **kwargs): + return None + + def success_handler(self, *args, **kwargs): + return None + + async def async_success_handler(self, *args, **kwargs): + return None + + def _update_completion_start_time(self, completion_start_time): + self.completion_start_time = completion_start_time + + first_resp = self._make_mock_responses_api_response(include_function_call=True) + final_resp = self._make_mock_responses_api_response(text="Premium wifi is $10.") + + search_result = MagicMock() + search_result.file_id = "file-h16" + search_result.filename = "plans.pdf" + search_result.score = 0.8 + search_result.content = [{"type": "text", "text": "premium wifi costs $10"}] + mock_search_response = MagicMock() + mock_search_response.data = [search_result] + + captured_kwargs: List[Dict[str, Any]] = [] + + async def _capture(**kwargs): + captured_kwargs.append(kwargs) + return first_resp if len(captured_kwargs) == 1 else final_resp + + with ( + patch( + "litellm.responses.file_search.emulated_handler._call_aresponses", + new=AsyncMock(side_effect=_capture), + ), + patch( + "litellm.vector_stores.main.asearch", + new=AsyncMock(return_value=mock_search_response), + ), + ): + result = await aresponses_with_emulated_file_search( + input="how much is the premium wifi plan?", + model="anthropic/claude-3-5-sonnet", + tools=[{"type": "file_search", "vector_store_ids": ["vs_h16"]}], + stream=True, + litellm_logging_obj=_LoggingObj(), + ) + + assert hasattr(result, "__aiter__"), "streaming request must return an async iterable" + events = [event async for event in result] + + assert all(sub_kwargs.get("stream") is False for sub_kwargs in captured_kwargs) + event_types = [getattr(getattr(event, "type", ""), "value", "") for event in events] + assert event_types[0] == "response.created" + assert event_types[-1] == "response.completed" + assert any(event_type == "response.output_text.delta" for event_type in event_types) + + completed = events[-1].response + def _get(item, key): + return item[key] if isinstance(item, dict) else getattr(item, key, None) + + assert _get(completed.output[0], "type") == "file_search_call" + assert _get(completed.output[1], "type") == "message" + streamed_text = "".join( + str(getattr(event, "delta", "")) + for event in events + if getattr(getattr(event, "type", ""), "value", "") == "response.output_text.delta" + ) + assert streamed_text == "Premium wifi is $10."