scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-17 23:51:30 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 66

fix(responses): stream emulated file_search results instead of 500ing

Browse source
This commit is contained in:
Devin AI 2026-07-27 11:12:59 +00:00
parent 24123269cc
commit 789ea140d9
3 changed files with 174 additions and 33 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

83
litellm/responses/file_search/emulated_handler.py
View file

@ -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,
)

35
litellm/responses/streaming_iterator.py
View file

@ -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()

89
tests/test_litellm/llms/test_file_search_responses.py
View file

@ -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."