From f9b0fb1c3a5bbc03bcf58b7da5ce54929c39fa1b Mon Sep 17 00:00:00 2001 From: narmstrong-berri Date: Mon, 26 Oct 2026 03:24:33 +0000 Subject: [PATCH] refactor(responses): send bridged chat calls to the Responses executor The chat-to-Responses bridge called litellm.responses() from inside the chat call, so the @client pass and its deployment hooks ran twice per request. It now hands the translated request to the native Responses executor, shared with the /v1/messages bridge in litellm/responses/bridge_executor.py, and keeps responses() for requests that need what only it adds --- .../handler.py | 69 ++++- .../transformation.py | 4 - .../responses_adapters/handler.py | 236 +---------------- litellm/responses/bridge_executor.py | 246 ++++++++++++++++++ ...itellm_responses_transformation_handler.py | 121 ++++++++- ...t_responses_bridge_provider_propagation.py | 17 +- ...erimental_pass_through_messages_handler.py | 79 ++++++ tests/unit/llms/azure/test_azure.py | 8 +- tests/unit/llms/openai/test_openai.py | 6 +- tests/unit/test_router/test_router.py | 4 +- 10 files changed, 524 insertions(+), 266 deletions(-) create mode 100644 litellm/responses/bridge_executor.py diff --git a/litellm/completion_extras/litellm_responses_transformation/handler.py b/litellm/completion_extras/litellm_responses_transformation/handler.py index 3f7017eccc2..082d9efbd74 100644 --- a/litellm/completion_extras/litellm_responses_transformation/handler.py +++ b/litellm/completion_extras/litellm_responses_transformation/handler.py @@ -11,8 +11,10 @@ from typing import ( cast, # noqa: TID251 # optional_params reaches the bridge as an untyped dict ) +from pydantic import TypeAdapter, ValidationError from typing_extensions import TypedDict +import litellm from litellm.types.llms.openai import ResponsesAPIResponse if TYPE_CHECKING: @@ -36,6 +38,48 @@ def _restore_routing_prefix(model: str, custom_llm_provider: str) -> str: return f"{custom_llm_provider}/{model}" +_TOOLS: Final = TypeAdapter(tuple[Mapping[str, object], ...]) +_BRIDGED_REQUEST: Final = TypeAdapter(dict[str, object]) +_TOOLS_ONLY_RESPONSES_SERVES: Final = frozenset({"mcp", "file_search"}) + + +def _needs_responses_surface(request_data: Mapping[str, object]) -> bool: + """`responses()` also checks the cache and dispatches MCP and emulated file_search tools, which the executor skips.""" + if litellm.cache is not None or request_data.get("use_chat_completions_api"): + return True + model: Final = request_data.get("model") + if not isinstance(model, str) or model.startswith("openai/chat_completions/"): + return True + try: + tools: Final = _TOOLS.validate_python(request_data.get("tools") or ()) + except ValidationError: + return True + return any(tool.get("type") in _TOOLS_ONLY_RESPONSES_SERVES for tool in tools) + + +def _executor_logging_obj( + request_data: Mapping[str, object], logging_obj: "LiteLLMLoggingObj" +) -> "LiteLLMLoggingObj | None": + from litellm.responses.bridge_executor import has_native_responses_config, native_logging_obj + + model: Final = request_data.get("model") + if _needs_responses_surface(request_data) or not isinstance(model, str): + return None + if not has_native_responses_config(model, request_data): + return None + return native_logging_obj(logging_obj, request_data, request_data) + + +def _retagged_as_responses( + request_data: Mapping[str, object], logging_obj: "LiteLLMLoggingObj" +) -> Mapping[str, object]: + """`responses()` logs the Responses object it returns, so the shared logging object takes its call type.""" + from litellm.types.utils import CallTypes + + setattr(logging_obj, "call_type", CallTypes.responses.value) + return request_data + + class ResponsesToCompletionBridgeHandler: def __init__(self): from .transformation import LiteLLMResponsesTransformationHandler @@ -172,6 +216,7 @@ class ResponsesToCompletionBridgeHandler: from litellm import responses from litellm.litellm_core_utils.streaming_handler import CustomStreamWrapper + from litellm.responses.bridge_executor import native_responses validated_kwargs: Final = self.validate_input_kwargs(kwargs) model: Final = validated_kwargs["model"] @@ -199,11 +244,16 @@ class ResponsesToCompletionBridgeHandler: # `sanitized_litellm_params` already carries both, so passing them again # would raise a duplicate-keyword TypeError. request_data["custom_llm_provider"] = custom_llm_provider - request_data["model"] = _restore_routing_prefix(model, custom_llm_provider) + routed_model: Final = _restore_routing_prefix(model, custom_llm_provider) + request_data["model"] = routed_model if kwargs.get("cache") is not None: request_data["cache"] = kwargs["cache"] - result: Final = responses( - **request_data, + bridged_request: Final = _BRIDGED_REQUEST.validate_python(request_data) + executor_logging: Final = _executor_logging_obj(bridged_request, logging_obj) + result: Final = ( + responses(**_retagged_as_responses(request_data, logging_obj)) + if executor_logging is None + else native_responses(bridged_request, routed_model, bridged_request, executor_logging) ) from litellm.types.utils import ModelResponse @@ -272,6 +322,7 @@ class ResponsesToCompletionBridgeHandler: async def acompletion(self, *args, **kwargs) -> Union["ModelResponse", "CustomStreamWrapper"]: from litellm import aresponses from litellm.litellm_core_utils.streaming_handler import CustomStreamWrapper + from litellm.responses.bridge_executor import anative_responses validated_kwargs: Final = self.validate_input_kwargs(kwargs) model: Final = validated_kwargs["model"] @@ -301,12 +352,16 @@ class ResponsesToCompletionBridgeHandler: # `sanitized_litellm_params` already carries both, so passing them again # would raise a duplicate-keyword TypeError. request_data["custom_llm_provider"] = custom_llm_provider - request_data["model"] = _restore_routing_prefix(model, custom_llm_provider) + routed_model: Final = _restore_routing_prefix(model, custom_llm_provider) + request_data["model"] = routed_model if kwargs.get("cache") is not None: request_data["cache"] = kwargs["cache"] - result: Final = await aresponses( - **request_data, - aresponses=True, + bridged_request: Final = _BRIDGED_REQUEST.validate_python(request_data) + executor_logging: Final = _executor_logging_obj(bridged_request, logging_obj) + result: Final = ( + await aresponses(**_retagged_as_responses(request_data, logging_obj), aresponses=True) + if executor_logging is None + else await anative_responses(bridged_request, routed_model, bridged_request, executor_logging) ) from litellm.types.utils import ModelResponse diff --git a/litellm/completion_extras/litellm_responses_transformation/transformation.py b/litellm/completion_extras/litellm_responses_transformation/transformation.py index 98322aecf6a..57da7f93d0f 100644 --- a/litellm/completion_extras/litellm_responses_transformation/transformation.py +++ b/litellm/completion_extras/litellm_responses_transformation/transformation.py @@ -1137,10 +1137,6 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge): api_model: Final = model - from litellm.types.utils import CallTypes - - setattr(litellm_logging_obj, "call_type", CallTypes.responses.value) - sanitized_litellm_params: Final = self._build_sanitized_litellm_params(litellm_params) request_data: Final = { diff --git a/litellm/llms/anthropic/pass_through/responses_adapters/handler.py b/litellm/llms/anthropic/pass_through/responses_adapters/handler.py index 68219fc0c65..fed76d8bbd5 100644 --- a/litellm/llms/anthropic/pass_through/responses_adapters/handler.py +++ b/litellm/llms/anthropic/pass_through/responses_adapters/handler.py @@ -5,14 +5,11 @@ Used when the target model is an OpenAI or Azure model. """ from collections.abc import AsyncIterator, Callable, Coroutine, Mapping, Sequence -from typing import Final, NoReturn, TypeAlias +from typing import Final, TypeAlias -import httpx -from aiohttp import ClientSession from pydantic import TypeAdapter, ValidationError import litellm -from litellm.constants import request_timeout from litellm.integrations.anthropic_cache_control_hook import ( AnthropicCacheControlHook, supports_openai_prompt_cache_breakpoint, @@ -20,20 +17,8 @@ from litellm.integrations.anthropic_cache_control_hook import ( from litellm.litellm_core_utils.core_helpers import drops_unsupported_params from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj from litellm.litellm_core_utils.prompt_templates.common_utils import cache_control_as_prompt_cache_breakpoint -from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler, HTTPHandler -from litellm.native.responses import execute_responses -from litellm.responses.main import ( - apply_managed_file_id_mapping, - base_llm_http_handler, - reasoning_for_deployment, - requested_reasoning, - resolve_model_provider_for_responses, - resolve_responses_api_provider_config, - with_provider_affinity_header, -) -from litellm.responses.streaming_iterator import BaseResponsesAPIStreamingIterator -from litellm.responses.utils import ResponsesAPIRequestUtils, auto_reasoning_summary -from litellm.router_utils.reasoning_effort_capability import deployment_reasoning_effort_levels +from litellm.responses.bridge_executor import anative_responses, native_logging_obj, native_responses +from litellm.responses.utils import auto_reasoning_summary from litellm.types.integrations.custom_logger import CONVERTED_STREAM_KEYS from litellm.types.llms.anthropic import ( AllAnthropicMessageValues, @@ -46,7 +31,6 @@ from litellm.types.llms.anthropic_messages.anthropic_response import ( AnthropicMessagesResponse, ) from litellm.types.llms.openai import ResponsesAPIResponse -from litellm.types.router import GenericLiteLLMParams from litellm.types.utils import CallTypes from litellm.utils import ProviderConfigManager @@ -62,9 +46,6 @@ BlockRewrite: TypeAlias = Callable[[Mapping[str, object]], Mapping[str, object]] _BLOCK_LIST: Final = TypeAdapter(list[dict[str, object]]) _ADAPTER: Final = LiteLLMAnthropicToResponsesAPIAdapter() -_STRING_KEYED_DICT: Final = TypeAdapter(dict[str, object]) -_CLIENT_HEADERS: Final = TypeAdapter(dict[str, str]) -_STRINGS: Final = TypeAdapter(tuple[str, ...]) _NEVER_FORWARDED: Final = CONVERTED_STREAM_KEYS | {"anthropic_messages"} @@ -258,156 +239,6 @@ def _build_responses_request( return responses_request -def _optional_str(value: object) -> str | None: - return value if isinstance(value, str) else None - - -def _string_keyed_dict(value: object) -> dict[str, object] | None: - try: - return _STRING_KEYED_DICT.validate_python(value) if value is not None else None - except ValidationError: - return None - - -def _client_headers(value: object) -> Mapping[str, str] | None: - try: - return _CLIENT_HEADERS.validate_python(value) if isinstance(value, dict) else None - except ValidationError: - return None - - -def _shared_session(value: object) -> ClientSession | None: - return value if isinstance(value, ClientSession) else None - - -def _http_client(value: object) -> HTTPHandler | AsyncHTTPHandler | None: - return value if isinstance(value, (HTTPHandler, AsyncHTTPHandler)) else None - - -def _timeout(value: object) -> float | httpx.Timeout: - return value if isinstance(value, (float, int, httpx.Timeout)) and value else request_timeout - - -def _allowed_openai_params(value: object) -> Sequence[str] | None: - try: - return _STRINGS.validate_python(value) if value is not None else None - except ValidationError: - return None - - -def _native_responses_call( - request: Mapping[str, object], - model: str, - forwarded_kwargs: Mapping[str, object], - logging_obj: LiteLLMLoggingObj, - is_async: bool, -) -> ( - ResponsesAPIResponse - | BaseResponsesAPIStreamingIterator - | Coroutine[object, object, ResponsesAPIResponse | BaseResponsesAPIStreamingIterator] -): - """Resolve the deployment and hand the translated request to the native Responses executor.""" - litellm_params: Final = GenericLiteLLMParams(**forwarded_kwargs) - provider_model, custom_llm_provider = resolve_model_provider_for_responses( - model=model, - custom_llm_provider=_optional_str(forwarded_kwargs.get("custom_llm_provider")), - litellm_params=litellm_params, - local_vars={}, - ) - if custom_llm_provider is None: - raise ValueError(f"No provider resolved for {model}") - provider_config: Final = resolve_responses_api_provider_config( - provider_model, custom_llm_provider, forwarded_kwargs.get("model_info"), litellm_params.api_base - ) - if provider_config is None: - raise ValueError(f"{custom_llm_provider} has no Responses API config for {provider_model}") - request_input, request_tools = apply_managed_file_id_mapping( - input=request["input"], - tools=request.get("tools"), - kwargs=dict(forwarded_kwargs), - local_vars={}, - ) - params: Final = { - **forwarded_kwargs, - **request, - "input": request_input, - "tools": request_tools, - "custom_llm_provider": custom_llm_provider, - } - deployment_model_info: Final = forwarded_kwargs.get("model_info") - reasoning: Final = reasoning_for_deployment( - requested_reasoning(params, provider_model, custom_llm_provider), - provider_model, - custom_llm_provider, - deployment_model_info, - forwarded_kwargs, - ) - return execute_responses( - http_handler=base_llm_http_handler, - model=provider_model, - input=request_input, - provider_config=provider_config, - requested_params=ResponsesAPIRequestUtils.get_requested_response_api_optional_param( - {key: value for key, value in {**params, "reasoning": reasoning}.items() if key != "reasoning_effort"} - ), - allowed_openai_params=_allowed_openai_params(forwarded_kwargs.get("allowed_openai_params")), - declared_reasoning_efforts=deployment_reasoning_effort_levels(deployment_model_info) or (), - custom_llm_provider=custom_llm_provider, - litellm_params=litellm_params, - logging_obj=logging_obj, - user=_optional_str(request.get("user")), - extra_headers=with_provider_affinity_header( - ResponsesAPIRequestUtils.merge_client_forwarded_headers( - extra_headers=_string_keyed_dict(forwarded_kwargs.get("extra_headers")), - client_headers=_client_headers(forwarded_kwargs.get("headers")), - ), - litellm_params, - _string_keyed_dict(request.get("metadata")), - forwarded_kwargs, - provider_model, - custom_llm_provider, - ), - extra_body=_string_keyed_dict(forwarded_kwargs.get("extra_body")), - timeout=_timeout(forwarded_kwargs.get("timeout")), - is_async=is_async, - stream=request.get("stream") is True, - client=_http_client(forwarded_kwargs.get("client")), - shared_session=_shared_session(forwarded_kwargs.get("shared_session")), - litellm_metadata=_string_keyed_dict(forwarded_kwargs.get("litellm_metadata")) or {}, - request_kwargs=dict(forwarded_kwargs), - ) - - -_OPENAI_CHAT_COMPLETIONS_MODEL_PREFIX: Final = "openai/chat_completions/" - - -def _requests_chat_completions_api(params: Mapping[str, object]) -> bool: - model: Final = params.get("model") - return bool(params.get("use_chat_completions_api")) or ( - isinstance(model, str) and model.startswith(_OPENAI_CHAT_COMPLETIONS_MODEL_PREFIX) - ) - - -def _native_logging_obj( - logging_obj: LiteLLMLoggingObj | None, forwarded_kwargs: Mapping[str, object], request: Mapping[str, object] -) -> LiteLLMLoggingObj | None: - """The logging object when the native executor can serve the call, None when only `aresponses` can. - - The executor skips the prompt management hooks `aresponses` runs, so a call that needs them re-enters it. - So does a deployment that asks for the chat completions API, which `aresponses` bridges to chat. - """ - if logging_obj is None: - return None - params: Final = {**forwarded_kwargs, **request} - if _requests_chat_completions_api(params): - return None - if logging_obj.should_run_prompt_management_hooks( - non_default_params=params, prompt_id=_optional_str(params.get("prompt_id")) - ): - return None - return logging_obj - - def _reentrant_kwargs( forwarded_kwargs: Mapping[str, object], request: Mapping[str, object], logging_obj: LiteLLMLoggingObj | None ) -> dict[str, object]: @@ -416,55 +247,6 @@ def _reentrant_kwargs( return {**forwarded_kwargs, **request} -def _raise_mapped(error: Exception, model: str, forwarded_kwargs: Mapping[str, object]) -> NoReturn: - raise litellm.exception_type( - model=model, - custom_llm_provider=_optional_str(forwarded_kwargs.get("custom_llm_provider")), - original_exception=error, - extra_kwargs=dict(forwarded_kwargs), - ) from error - - -def _priced( - response: ResponsesAPIResponse | BaseResponsesAPIStreamingIterator, logging_obj: LiteLLMLoggingObj -) -> ResponsesAPIResponse | BaseResponsesAPIStreamingIterator: - """The response, priced as `aresponses` would have priced it before the bridge translates it.""" - if isinstance(response, ResponsesAPIResponse): - logging_obj.price_bridged_provider_response(response) - else: - logging_obj.price_bridged_provider_stream(response) - return response - - -def _native_responses( - request: Mapping[str, object], - model: str, - forwarded_kwargs: Mapping[str, object], - logging_obj: LiteLLMLoggingObj, -) -> ResponsesAPIResponse | BaseResponsesAPIStreamingIterator: - try: - response: Final = _native_responses_call(request, model, forwarded_kwargs, logging_obj, is_async=False) - except Exception as e: # noqa: BLE001 # every provider failure maps to the litellm exception aresponses would raise - _raise_mapped(e, model, forwarded_kwargs) - if isinstance(response, Coroutine): - raise TypeError("The sync Responses executor returned a coroutine") - return _priced(response, logging_obj) - - -async def _native_aresponses( - request: Mapping[str, object], - model: str, - forwarded_kwargs: Mapping[str, object], - logging_obj: LiteLLMLoggingObj, -) -> ResponsesAPIResponse | BaseResponsesAPIStreamingIterator: - try: - response: Final = _native_responses_call(request, model, forwarded_kwargs, logging_obj, is_async=True) - settled: Final = await response if isinstance(response, Coroutine) else response - except Exception as e: # noqa: BLE001 # every provider failure maps to the litellm exception aresponses would raise - _raise_mapped(e, model, forwarded_kwargs) - return _priced(settled, logging_obj) - - class LiteLLMMessagesToResponsesAPIHandler: """ Handles Anthropic /v1/messages requests for OpenAI / Azure models by @@ -515,11 +297,11 @@ class LiteLLMMessagesToResponsesAPIHandler: logging_obj: Final = litellm_logging_obj_from_kwargs(kwargs) safeguards_evaluator: Final = request_safeguards_evaluator(kwargs, messages) - native_logging_obj: Final = _native_logging_obj(logging_obj, forwarded_kwargs, request) + native_logging: Final = native_logging_obj(logging_obj, forwarded_kwargs, request) result: Final = ( await litellm.aresponses(**_reentrant_kwargs(forwarded_kwargs, request, logging_obj)) - if native_logging_obj is None - else await _native_aresponses(request, model, forwarded_kwargs, native_logging_obj) + if native_logging is None + else await anative_responses(request, model, forwarded_kwargs, native_logging) ) if stream: @@ -605,11 +387,11 @@ class LiteLLMMessagesToResponsesAPIHandler: ) forwarded_kwargs: Final = _forwarded_kwargs(kwargs) logging_obj: Final = litellm_logging_obj_from_kwargs(kwargs) - native_logging_obj: Final = _native_logging_obj(logging_obj, forwarded_kwargs, request) + native_logging: Final = native_logging_obj(logging_obj, forwarded_kwargs, request) result: Final = ( litellm.responses(**_reentrant_kwargs(forwarded_kwargs, request, logging_obj)) - if native_logging_obj is None - else _native_responses(request, model, forwarded_kwargs, native_logging_obj) + if native_logging is None + else native_responses(request, model, forwarded_kwargs, native_logging) ) if stream: diff --git a/litellm/responses/bridge_executor.py b/litellm/responses/bridge_executor.py new file mode 100644 index 00000000000..a57e13a649b --- /dev/null +++ b/litellm/responses/bridge_executor.py @@ -0,0 +1,246 @@ +"""Hands a bridged Responses request to the native Responses executor, so the bridge never re-enters `aresponses`.""" + +from collections.abc import Coroutine, Mapping, Sequence +from typing import Final, NoReturn + +import httpx +from aiohttp import ClientSession +from pydantic import TypeAdapter, ValidationError + +import litellm +from litellm.constants import request_timeout +from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj +from litellm.llms.base_llm.responses.transformation import BaseResponsesAPIConfig +from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler, HTTPHandler +from litellm.native.responses import execute_responses +from litellm.responses.main import ( + apply_managed_file_id_mapping, + base_llm_http_handler, + reasoning_for_deployment, + requested_reasoning, + resolve_model_provider_for_responses, + resolve_responses_api_provider_config, + with_provider_affinity_header, +) +from litellm.responses.streaming_iterator import BaseResponsesAPIStreamingIterator +from litellm.responses.utils import ResponsesAPIRequestUtils +from litellm.router_utils.reasoning_effort_capability import deployment_reasoning_effort_levels +from litellm.types.llms.openai import ResponsesAPIResponse +from litellm.types.router import GenericLiteLLMParams + +__all__ = ("anative_responses", "has_native_responses_config", "native_logging_obj", "native_responses") + +_STRING_KEYED_DICT: Final = TypeAdapter(dict[str, object]) +_CLIENT_HEADERS: Final = TypeAdapter(dict[str, str]) +_STRINGS: Final = TypeAdapter(tuple[str, ...]) + + +def _optional_str(value: object) -> str | None: + return value if isinstance(value, str) else None + + +def _string_keyed_dict(value: object) -> dict[str, object] | None: + try: + return _STRING_KEYED_DICT.validate_python(value) if value is not None else None + except ValidationError: + return None + + +def _client_headers(value: object) -> Mapping[str, str] | None: + try: + return _CLIENT_HEADERS.validate_python(value) if isinstance(value, dict) else None + except ValidationError: + return None + + +def _shared_session(value: object) -> ClientSession | None: + return value if isinstance(value, ClientSession) else None + + +def _http_client(value: object) -> HTTPHandler | AsyncHTTPHandler | None: + return value if isinstance(value, (HTTPHandler, AsyncHTTPHandler)) else None + + +def _timeout(value: object) -> float | httpx.Timeout: + return value if isinstance(value, (float, int, httpx.Timeout)) and value else request_timeout + + +def _allowed_openai_params(value: object) -> Sequence[str] | None: + try: + return _STRINGS.validate_python(value) if value is not None else None + except ValidationError: + return None + + +def _responses_target( + model: str, forwarded_kwargs: Mapping[str, object], litellm_params: GenericLiteLLMParams +) -> tuple[str, str, BaseResponsesAPIConfig] | None: + provider_model, custom_llm_provider = resolve_model_provider_for_responses( + model=model, + custom_llm_provider=_optional_str(forwarded_kwargs.get("custom_llm_provider")), + litellm_params=litellm_params, + local_vars={}, + ) + if custom_llm_provider is None: + return None + provider_config: Final = resolve_responses_api_provider_config( + provider_model, custom_llm_provider, forwarded_kwargs.get("model_info"), litellm_params.api_base + ) + return None if provider_config is None else (provider_model, custom_llm_provider, provider_config) + + +def has_native_responses_config(model: str, forwarded_kwargs: Mapping[str, object]) -> bool: + return _responses_target(model, forwarded_kwargs, GenericLiteLLMParams(**forwarded_kwargs)) is not None + + +def _native_responses_call( + request: Mapping[str, object], + model: str, + forwarded_kwargs: Mapping[str, object], + logging_obj: LiteLLMLoggingObj, + is_async: bool, +) -> ( + ResponsesAPIResponse + | BaseResponsesAPIStreamingIterator + | Coroutine[object, object, ResponsesAPIResponse | BaseResponsesAPIStreamingIterator] +): + """Resolve the deployment and hand the translated request to the native Responses executor.""" + litellm_params: Final = GenericLiteLLMParams(**forwarded_kwargs) + target: Final = _responses_target(model, forwarded_kwargs, litellm_params) + if target is None: + raise ValueError(f"No Responses API config for {model}") + provider_model, custom_llm_provider, provider_config = target + request_input, request_tools = apply_managed_file_id_mapping( + input=request["input"], + tools=request.get("tools"), + kwargs=dict(forwarded_kwargs), + local_vars={}, + ) + params: Final = { + **forwarded_kwargs, + **request, + "input": request_input, + "tools": request_tools, + "custom_llm_provider": custom_llm_provider, + } + deployment_model_info: Final = forwarded_kwargs.get("model_info") + reasoning: Final = reasoning_for_deployment( + requested_reasoning(params, provider_model, custom_llm_provider), + provider_model, + custom_llm_provider, + deployment_model_info, + forwarded_kwargs, + ) + return execute_responses( + http_handler=base_llm_http_handler, + model=provider_model, + input=request_input, + provider_config=provider_config, + requested_params=ResponsesAPIRequestUtils.get_requested_response_api_optional_param( + {key: value for key, value in {**params, "reasoning": reasoning}.items() if key != "reasoning_effort"} + ), + allowed_openai_params=_allowed_openai_params(forwarded_kwargs.get("allowed_openai_params")), + declared_reasoning_efforts=deployment_reasoning_effort_levels(deployment_model_info) or (), + custom_llm_provider=custom_llm_provider, + litellm_params=litellm_params, + logging_obj=logging_obj, + user=_optional_str(request.get("user")), + extra_headers=with_provider_affinity_header( + ResponsesAPIRequestUtils.merge_client_forwarded_headers( + extra_headers=_string_keyed_dict(forwarded_kwargs.get("extra_headers")), + client_headers=_client_headers(forwarded_kwargs.get("headers")), + ), + litellm_params, + _string_keyed_dict(request.get("metadata")), + forwarded_kwargs, + provider_model, + custom_llm_provider, + ), + extra_body=_string_keyed_dict(forwarded_kwargs.get("extra_body")), + timeout=_timeout(forwarded_kwargs.get("timeout")), + is_async=is_async, + stream=request.get("stream") is True, + client=_http_client(forwarded_kwargs.get("client")), + shared_session=_shared_session(forwarded_kwargs.get("shared_session")), + litellm_metadata=_string_keyed_dict(forwarded_kwargs.get("litellm_metadata")) or {}, + request_kwargs=dict(forwarded_kwargs), + ) + + +_OPENAI_CHAT_COMPLETIONS_MODEL_PREFIX: Final = "openai/chat_completions/" + + +def _requests_chat_completions_api(params: Mapping[str, object]) -> bool: + model: Final = params.get("model") + return bool(params.get("use_chat_completions_api")) or ( + isinstance(model, str) and model.startswith(_OPENAI_CHAT_COMPLETIONS_MODEL_PREFIX) + ) + + +def native_logging_obj( + logging_obj: LiteLLMLoggingObj | None, forwarded_kwargs: Mapping[str, object], request: Mapping[str, object] +) -> LiteLLMLoggingObj | None: + """The logging object when the native executor can serve the call, None when only `aresponses` can. + + The executor skips the prompt management hooks `aresponses` runs, so a call that needs them re-enters it. + So does a deployment that asks for the chat completions API, which `aresponses` bridges to chat. + """ + if logging_obj is None: + return None + params: Final = {**forwarded_kwargs, **request} + if _requests_chat_completions_api(params): + return None + if logging_obj.should_run_prompt_management_hooks( + non_default_params=params, prompt_id=_optional_str(params.get("prompt_id")) + ): + return None + return logging_obj + + +def _raise_mapped(error: Exception, model: str, forwarded_kwargs: Mapping[str, object]) -> NoReturn: + raise litellm.exception_type( + model=model, + custom_llm_provider=_optional_str(forwarded_kwargs.get("custom_llm_provider")), + original_exception=error, + extra_kwargs=dict(forwarded_kwargs), + ) from error + + +def _priced( + response: ResponsesAPIResponse | BaseResponsesAPIStreamingIterator, logging_obj: LiteLLMLoggingObj +) -> ResponsesAPIResponse | BaseResponsesAPIStreamingIterator: + """The response, priced as `aresponses` would have priced it before the bridge translates it.""" + if isinstance(response, ResponsesAPIResponse): + logging_obj.price_bridged_provider_response(response) + else: + logging_obj.price_bridged_provider_stream(response) + return response + + +def native_responses( + request: Mapping[str, object], + model: str, + forwarded_kwargs: Mapping[str, object], + logging_obj: LiteLLMLoggingObj, +) -> ResponsesAPIResponse | BaseResponsesAPIStreamingIterator: + try: + response: Final = _native_responses_call(request, model, forwarded_kwargs, logging_obj, is_async=False) + except Exception as e: # noqa: BLE001 # every provider failure maps to the litellm exception aresponses would raise + _raise_mapped(e, model, forwarded_kwargs) + if isinstance(response, Coroutine): + raise TypeError("The sync Responses executor returned a coroutine") + return _priced(response, logging_obj) + + +async def anative_responses( + request: Mapping[str, object], + model: str, + forwarded_kwargs: Mapping[str, object], + logging_obj: LiteLLMLoggingObj, +) -> ResponsesAPIResponse | BaseResponsesAPIStreamingIterator: + try: + response: Final = _native_responses_call(request, model, forwarded_kwargs, logging_obj, is_async=True) + settled: Final = await response if isinstance(response, Coroutine) else response + except Exception as e: # noqa: BLE001 # every provider failure maps to the litellm exception aresponses would raise + _raise_mapped(e, model, forwarded_kwargs) + return _priced(settled, logging_obj) diff --git a/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_handler.py b/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_handler.py index c5d7ca96a21..1c4636b6e12 100644 --- a/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_handler.py +++ b/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_handler.py @@ -1,15 +1,19 @@ from datetime import datetime from unittest.mock import AsyncMock, MagicMock, patch +import httpx import pytest - +import respx import litellm +from litellm.caching.caching import Cache from litellm.completion_extras.litellm_responses_transformation.handler import ( ResponsesToCompletionBridgeHandler, ) +from litellm.integrations.custom_logger import CustomLogger from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLogging from litellm.litellm_core_utils.streaming_handler import CustomStreamWrapper +from litellm.responses import bridge_executor from litellm.types.router import GenericLiteLLMParams from litellm.types.utils import ModelResponse @@ -41,6 +45,11 @@ def test_is_preformatted_cached_chat_stream_false_wrong_type(): ) +@pytest.fixture +def _cache_on(monkeypatch): + monkeypatch.setattr(litellm, "cache", MagicMock(spec=Cache)) + + def _bridge_kwargs(stream: bool): logging_obj = LiteLLMLogging( litellm_call_id="test-call", @@ -63,7 +72,7 @@ def _bridge_kwargs(stream: bool): } -def test_completion_returns_cached_model_response_directly(): +def test_completion_returns_cached_model_response_directly(_cache_on): """Non-streaming bridge cache hit: responses() returns a ModelResponse -> bridge returns it as-is.""" cached = ModelResponse(id="chatcmpl-cached-nonstream", model="gpt-5.4") bridge = ResponsesToCompletionBridgeHandler() @@ -82,7 +91,7 @@ def test_completion_returns_cached_model_response_directly(): @pytest.mark.asyncio -async def test_acompletion_returns_cached_model_response_directly(): +async def test_acompletion_returns_cached_model_response_directly(_cache_on): cached = ModelResponse(id="chatcmpl-cached-nonstream-async", model="gpt-5.4") bridge = ResponsesToCompletionBridgeHandler() @@ -99,7 +108,7 @@ async def test_acompletion_returns_cached_model_response_directly(): assert result is cached -def test_completion_skips_rewrapping_preformatted_cached_chat_stream(): +def test_completion_skips_rewrapping_preformatted_cached_chat_stream(_cache_on): """Streaming bridge cache hit returning CustomStreamWrapper(cached_response) -> bridge skips re-wrapping.""" stream = MagicMock(spec=CustomStreamWrapper) stream.custom_llm_provider = "cached_response" @@ -124,7 +133,7 @@ def test_completion_skips_rewrapping_preformatted_cached_chat_stream(): assert result is stream -def test_completion_preserves_top_level_stream_flag_in_responses_request(): +def test_completion_preserves_top_level_stream_flag_in_responses_request(_cache_on): stream = MagicMock(spec=CustomStreamWrapper) stream.custom_llm_provider = "cached_response" bridge = ResponsesToCompletionBridgeHandler() @@ -152,7 +161,7 @@ def test_completion_preserves_top_level_stream_flag_in_responses_request(): @pytest.mark.asyncio -async def test_acompletion_skips_rewrapping_preformatted_cached_chat_stream(): +async def test_acompletion_skips_rewrapping_preformatted_cached_chat_stream(_cache_on): stream = MagicMock(spec=CustomStreamWrapper) stream.custom_llm_provider = "cached_response" bridge = ResponsesToCompletionBridgeHandler() @@ -177,7 +186,7 @@ async def test_acompletion_skips_rewrapping_preformatted_cached_chat_stream(): @pytest.mark.asyncio -async def test_acompletion_preserves_top_level_stream_flag_in_responses_request(): +async def test_acompletion_preserves_top_level_stream_flag_in_responses_request(_cache_on): stream = MagicMock(spec=CustomStreamWrapper) stream.custom_llm_provider = "cached_response" bridge = ResponsesToCompletionBridgeHandler() @@ -219,7 +228,7 @@ def _completed_chat_response() -> ModelResponse: @pytest.mark.asyncio -async def test_acompletion_streams_completed_model_response(): +async def test_acompletion_streams_completed_model_response(_cache_on): """A streaming request whose bridge call comes back already completed must still be handed back as an async-iterable stream. Returning the bare ModelResponse crashed the proxy's SSE generator with "'async for' requires an object with __aiter__ method". @@ -245,7 +254,7 @@ async def test_acompletion_streams_completed_model_response(): assert [c for c in chunks if c.choices[0].finish_reason], "stream never emitted a finish_reason" -def test_completion_streams_completed_model_response(): +def test_completion_streams_completed_model_response(_cache_on): completed = _completed_chat_response() bridge = ResponsesToCompletionBridgeHandler() @@ -288,7 +297,7 @@ def _upstream_model_for(handed_model: str, custom_llm_provider: str) -> str: def test_completion_keeps_provider_native_model_id_through_responses( custom_llm_provider, bridge_model, expected_upstream_model ): - """responses() resolves the provider itself, so the bridge must not hand it an already-stripped model.""" + """The Responses executor resolves the provider itself, so the bridge must not hand it an already-stripped model.""" cached = ModelResponse(id="chatcmpl-cached", model=bridge_model) bridge = ResponsesToCompletionBridgeHandler() kwargs = _bridge_kwargs(stream=False) @@ -301,11 +310,11 @@ def test_completion_keeps_provider_native_model_id_through_responses( "transform_request", return_value={"model": bridge_model, "input": "hi"}, ), - patch("litellm.responses", return_value=cached) as responses_call, + patch.object(bridge_executor, "native_responses", return_value=cached) as responses_call, ): bridge.completion(**kwargs) - handed_model = responses_call.call_args.kwargs["model"] + handed_model = responses_call.call_args.args[1] assert _upstream_model_for(handed_model, custom_llm_provider) == expected_upstream_model @@ -329,9 +338,93 @@ async def test_acompletion_keeps_provider_native_model_id_through_responses( "transform_request", return_value={"model": bridge_model, "input": "hi"}, ), - patch("litellm.aresponses", new=AsyncMock(return_value=cached)) as responses_call, + patch.object(bridge_executor, "anative_responses", new=AsyncMock(return_value=cached)) as responses_call, ): await bridge.acompletion(**kwargs) - handed_model = responses_call.call_args.kwargs["model"] + handed_model = responses_call.call_args.args[1] assert _upstream_model_for(handed_model, custom_llm_provider) == expected_upstream_model + + +class _DeploymentHookCounter(CustomLogger): + def __init__(self) -> None: + super().__init__() + self.calls = 0 + + async def async_pre_call_deployment_hook(self, kwargs, call_type) -> None: + self.calls += 1 + + +@pytest.mark.asyncio +async def test_bridged_chat_call_runs_deployment_hooks_once(monkeypatch): + counter = _DeploymentHookCounter() + monkeypatch.setattr(litellm, "callbacks", [counter]) + monkeypatch.setattr(litellm, "disable_aiohttp_transport", True) + litellm.in_memory_llm_clients_cache.flush_cache() + litellm.register_model( + {"openai/bridge-only-model": {"litellm_provider": "openai", "mode": "responses", "max_tokens": 100}} + ) + + with respx.mock(assert_all_called=False) as upstream: + responses_route = upstream.post("https://example.invalid/v1/responses").mock( + return_value=httpx.Response( + 200, + json={ + "id": "resp_bridge", + "object": "response", + "created_at": 1, + "status": "completed", + "model": "bridge-only-model", + "output": [ + { + "type": "message", + "id": "msg_bridge", + "status": "completed", + "role": "assistant", + "content": [{"type": "output_text", "text": "from upstream", "annotations": []}], + } + ], + "usage": {"input_tokens": 10, "output_tokens": 5, "total_tokens": 15}, + }, + ) + ) + response = await litellm.acompletion( + model="openai/bridge-only-model", + api_base="https://example.invalid/v1", + api_key="x", + messages=[{"role": "user", "content": "hi"}], + ) + + assert response.choices[0].message.content == "from upstream" + assert responses_route.call_count == 1 + assert counter.calls == 1 + + +@pytest.mark.parametrize( + "bridge_model, tools", + [ + ("gpt-5.4", [{"type": "mcp", "server_label": "docs", "server_url": "litellm_proxy"}]), + ("gpt-5.4", [{"type": "file_search", "vector_store_ids": ["vs_1"]}]), + ("chat_completions/gpt-5.4", None), + ], +) +def test_completion_keeps_responses_surface_for_what_only_it_serves(bridge_model, tools): + cached = ModelResponse(id="chatcmpl-surface", model="gpt-5.4") + bridge = ResponsesToCompletionBridgeHandler() + kwargs = _bridge_kwargs(stream=False) + kwargs["model"] = bridge_model + + with ( + patch.object( + bridge.transformation_handler, + "transform_request", + return_value={"model": bridge_model, "input": "hi", "tools": tools}, + ), + patch("litellm.responses", return_value=cached) as responses_call, + patch.object(bridge_executor, "native_responses") as executor_call, + ): + result = bridge.completion(**kwargs) + + assert result is cached + assert responses_call.call_count == 1 + assert executor_call.call_count == 0 diff --git a/tests/unit/completion_extras/test_responses_bridge_provider_propagation.py b/tests/unit/completion_extras/test_responses_bridge_provider_propagation.py index 5aff9397747..210ec951620 100644 --- a/tests/unit/completion_extras/test_responses_bridge_provider_propagation.py +++ b/tests/unit/completion_extras/test_responses_bridge_provider_propagation.py @@ -13,6 +13,7 @@ from litellm.completion_extras.litellm_responses_transformation.handler import ( ResponsesToCompletionBridgeHandler, ) from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLogging +from litellm.responses import bridge_executor from litellm.types.utils import ModelResponse MODEL = "openai.gpt-5.5" @@ -73,11 +74,11 @@ def test_completion_forwards_custom_llm_provider_to_responses(): bridge = ResponsesToCompletionBridgeHandler() cached = ModelResponse(id="chatcmpl-cached", model="gpt-5.5") - with patch("litellm.responses", return_value=cached) as fake_responses: + with patch.object(bridge_executor, "native_responses", return_value=cached) as fake_responses: result = bridge.completion(**_openai_kwargs()) assert result is cached - assert fake_responses.call_args.kwargs["custom_llm_provider"] == "openai" + assert fake_responses.call_args.args[2]["custom_llm_provider"] == "openai" @pytest.mark.asyncio @@ -85,13 +86,13 @@ async def test_acompletion_forwards_custom_llm_provider_to_aresponses(): bridge = ResponsesToCompletionBridgeHandler() cached = ModelResponse(id="chatcmpl-cached", model="gpt-5.5") - async def _fake_aresponses(**kwargs): - _fake_aresponses.kwargs = kwargs + async def _fake_aresponses(request, model, forwarded_kwargs, logging_obj): + _fake_aresponses.kwargs = forwarded_kwargs return cached _fake_aresponses.kwargs = {} - with patch("litellm.aresponses", _fake_aresponses): + with patch.object(bridge_executor, "anative_responses", _fake_aresponses): result = await bridge.acompletion(**_openai_kwargs()) assert result is cached @@ -103,13 +104,13 @@ async def test_acompletion_forwards_aws_region_name_to_aresponses(): bridge = ResponsesToCompletionBridgeHandler() cached = ModelResponse(id="chatcmpl-cached", model=MODEL) - async def _fake_aresponses(**kwargs): - _fake_aresponses.kwargs = kwargs + async def _fake_aresponses(request, model, forwarded_kwargs, logging_obj): + _fake_aresponses.kwargs = forwarded_kwargs return cached _fake_aresponses.kwargs = {} - with patch("litellm.aresponses", _fake_aresponses): + with patch.object(bridge_executor, "anative_responses", _fake_aresponses): result = await bridge.acompletion(**_bedrock_mantle_kwargs()) assert result is cached diff --git a/tests/unit/llms/anthropic/pass_through/messages/test_anthropic_experimental_pass_through_messages_handler.py b/tests/unit/llms/anthropic/pass_through/messages/test_anthropic_experimental_pass_through_messages_handler.py index ca8cef61b80..2440507708d 100644 --- a/tests/unit/llms/anthropic/pass_through/messages/test_anthropic_experimental_pass_through_messages_handler.py +++ b/tests/unit/llms/anthropic/pass_through/messages/test_anthropic_experimental_pass_through_messages_handler.py @@ -4,6 +4,7 @@ import json import os import unittest.mock import uuid +from collections.abc import Mapping from typing import Any, Dict, Final, List from unittest.mock import AsyncMock, MagicMock, patch @@ -2328,3 +2329,81 @@ def test_native_messages_tool_config_point_marks_last_tool_and_drops_the_key(mon assert "cache_control" not in captured["params"]["tools"][0] assert captured["messages"][0]["content"][-1]["cache_control"] == {"type": "ephemeral"} assert tools == tools_before + + +_REASONING_RESPONSE: Final = { + "id": "resp_thinking", + "object": "response", + "created_at": 1, + "status": "completed", + "model": "gpt-5.5", + "output": [ + {"type": "reasoning", "id": "rs_1", "summary": [{"type": "summary_text", "text": "worked it out"}]}, + { + "type": "message", + "id": "msg_1", + "role": "assistant", + "status": "completed", + "content": [{"type": "output_text", "text": "the answer", "annotations": []}], + }, + ], + "parallel_tool_calls": True, + "tool_choice": "auto", + "tools": [], + "usage": {"input_tokens": 3, "output_tokens": 4, "total_tokens": 7}, +} + + +class _DeploymentHookCounter(CustomLogger): + def __init__(self) -> None: + super().__init__() + self.calls = 0 + + async def async_pre_call_deployment_hook(self, kwargs: Mapping[str, object], call_type: object) -> None: + self.calls += 1 + + +@pytest.mark.asyncio +async def test_forced_chat_sends_openai_thinking_to_responses_through_the_chat_bridge( + monkeypatch: pytest.MonkeyPatch, +): + counter: Final = _DeploymentHookCounter() + monkeypatch.setattr(litellm, "callbacks", [counter]) + monkeypatch.setattr(litellm, "use_chat_completions_url_for_anthropic_messages", True) + monkeypatch.setattr(litellm, "disable_aiohttp_transport", True) + litellm.in_memory_llm_clients_cache.flush_cache() + + with respx.mock(assert_all_called=False) as upstream: + responses_route: Final = upstream.post("http://openai.test/v1/responses").mock( + return_value=httpx.Response(200, json=_REASONING_RESPONSE) + ) + chat_route: Final = upstream.post("http://openai.test/v1/chat/completions") + response: Final = await litellm.anthropic.messages.acreate( + model="openai/gpt-5.5", + api_base="http://openai.test/v1", + api_key="sk-test", + max_tokens=1024, + messages=[ + {"role": "user", "content": "weather?"}, + { + "role": "assistant", + "content": [ + {"type": "text", "text": "let me check"}, + {"type": "tool_use", "id": "toolu_1", "name": "get_w", "input": {"c": "a"}}, + ], + }, + {"role": "user", "content": [{"type": "tool_result", "tool_use_id": "toolu_1", "content": "sunny"}]}, + ], + tools=[{"name": "get_w", "description": "d", "input_schema": {"type": "object", "properties": {}}}], + thinking={"type": "enabled", "budget_tokens": 1024}, + ) + + assert isinstance(response, dict) + assert [block["type"] for block in response["content"]] == ["thinking", "text"] + assert response["content"][1]["text"] == "the answer" + assert (responses_route.call_count, chat_route.call_count) == (1, 0) + assert counter.calls == 1 + sent_input: Final = json.loads(responses_route.calls[0].request.content)["input"] + assert [item["type"] for item in sent_input] == ["message", "message", "function_call", "function_call_output"] + assert sent_input[1]["role"] == "assistant" + assert sent_input[3]["output"] == [{"type": "input_text", "text": "sunny"}] diff --git a/tests/unit/llms/azure/test_azure.py b/tests/unit/llms/azure/test_azure.py index 7d5afde68ef..ae164ff64e7 100644 --- a/tests/unit/llms/azure/test_azure.py +++ b/tests/unit/llms/azure/test_azure.py @@ -585,7 +585,9 @@ def test_azure_openai_responses_bridge(): litellm.turn_on_debug() - with patch.object(litellm, "responses") as mock_responses: + from litellm.responses import bridge_executor + + with patch.object(bridge_executor, "native_responses") as mock_responses: try: response = completion( model="azure/responses/test-azure-computer-use-preview", @@ -599,10 +601,10 @@ def test_azure_openai_responses_bridge(): mock_responses.assert_called_once() assert ( - mock_responses.call_args.kwargs["model"] + mock_responses.call_args.args[0]["model"] == "azure/test-azure-computer-use-preview" ) - assert mock_responses.call_args.kwargs["custom_llm_provider"] == "azure" + assert mock_responses.call_args.args[0]["custom_llm_provider"] == "azure" def test_azure_with_content_safety_error(): diff --git a/tests/unit/llms/openai/test_openai.py b/tests/unit/llms/openai/test_openai.py index 11d7c7d6d3c..0d7eac5fd1a 100644 --- a/tests/unit/llms/openai/test_openai.py +++ b/tests/unit/llms/openai/test_openai.py @@ -1933,7 +1933,9 @@ def test_responses_gpt54_with_xhigh_reasoning(): Ensure chat->responses bridge sends the correct request payload for openai/responses/gpt-5.4 with reasoning_effort="xhigh". """ - with patch("litellm.responses") as mock_responses: + from litellm.responses import bridge_executor + + with patch.object(bridge_executor, "native_responses") as mock_responses: mock_responses.side_effect = RuntimeError("stop_after_request_build") with pytest.raises(litellm.APIConnectionError): @@ -1945,7 +1947,7 @@ def test_responses_gpt54_with_xhigh_reasoning(): ) mock_responses.assert_called_once() - request_body = mock_responses.call_args.kwargs + request_body = mock_responses.call_args.args[0] assert request_body["model"] == "openai/gpt-5.4" diff --git a/tests/unit/test_router/test_router.py b/tests/unit/test_router/test_router.py index d645ab4566d..fdb050eb185 100644 --- a/tests/unit/test_router/test_router.py +++ b/tests/unit/test_router/test_router.py @@ -1668,7 +1668,9 @@ def test_arouter_responses_api_bridge(): ## CONFIRM BRIDGE IS CALLED # an empty stream: the Router reads the first chunk before returning, and a stream of mocks never produces one - with patch.object(litellm, "responses", return_value=iter([])) as mock_responses: + from litellm.responses import bridge_executor + + with patch.object(bridge_executor, "native_responses", return_value=iter([])) as mock_responses: result = router.completion( model="[IP-approved] o3-pro", messages=[{"role": "user", "content": "Hello, world!"}],