mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-11 22:51:28 +00:00
Merge pull request #40608 from BerriAI/moe/lit-7493-zocdocauto-router-encrypted-codex-sub-agent-task-is
fix(router): classify encrypted delegated tasks with native Responses
This commit is contained in:
moe-berri
GitHub
commit
bc77aa05d2
8 changed files with 520 additions and 18 deletions
|
|
@ -1201,6 +1201,9 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
|
|||
# Cast to Any to match the expected union type for tools list items
|
||||
tools.append(cast(Any, web_search_tool))
|
||||
|
||||
def transform_response_format_to_text_format(self, response_format: object) -> "ResponseText | None":
|
||||
return self._transform_response_format_to_text_format(response_format)
|
||||
|
||||
def _transform_response_format_to_text_format(self, response_format: object) -> "ResponseText | None":
|
||||
"""
|
||||
Transform Chat Completion response_format parameter to Responses API text.format parameter.
|
||||
|
|
|
|||
|
|
@ -62,6 +62,9 @@ class BaseResponsesAPIConfig(ABC):
|
|||
"""
|
||||
return False
|
||||
|
||||
def supports_encrypted_agent_messages(self) -> bool:
|
||||
return False
|
||||
|
||||
def sign_request(
|
||||
self,
|
||||
headers: dict,
|
||||
|
|
|
|||
|
|
@ -110,6 +110,9 @@ class OpenAIResponsesAPIConfig(BaseResponsesAPIConfig):
|
|||
def supports_native_file_search(self) -> bool:
|
||||
return True
|
||||
|
||||
def supports_encrypted_agent_messages(self) -> bool:
|
||||
return self.custom_llm_provider in (LlmProviders.OPENAI, LlmProviders.AZURE)
|
||||
|
||||
@staticmethod
|
||||
def _is_gpt_5_model(model: str) -> bool:
|
||||
"""Return True only for actual OpenAI GPT-5 models.
|
||||
|
|
|
|||
|
|
@ -4,10 +4,11 @@ from collections.abc import Coroutine, Generator, Iterable, Mapping
|
|||
from contextlib import contextmanager
|
||||
from dataclasses import dataclass
|
||||
from functools import partial
|
||||
from typing import TYPE_CHECKING, Any, Final, Literal, Optional, cast
|
||||
from typing import TYPE_CHECKING, Any, Final, Literal, NoReturn, Optional, TypeAlias, cast
|
||||
|
||||
import httpx
|
||||
from pydantic import BaseModel
|
||||
from typing_extensions import assert_never
|
||||
|
||||
import litellm
|
||||
from litellm._logging import verbose_logger
|
||||
|
|
@ -407,6 +408,37 @@ def _bridges_to_chat_completions(
|
|||
return responses_api_provider_config is None or use_chat_completions_api is True
|
||||
|
||||
|
||||
_ResponsesCompatibilityFailure: TypeAlias = Literal["encrypted_task_unsupported"]
|
||||
|
||||
|
||||
def _encrypted_task_support_failure(
|
||||
responses_api_provider_config: BaseResponsesAPIConfig | None, use_chat_completions_api: bool
|
||||
) -> _ResponsesCompatibilityFailure | None:
|
||||
if (
|
||||
responses_api_provider_config is None
|
||||
or _bridges_to_chat_completions(responses_api_provider_config, use_chat_completions_api)
|
||||
or not responses_api_provider_config.supports_encrypted_agent_messages()
|
||||
):
|
||||
return "encrypted_task_unsupported"
|
||||
return None
|
||||
|
||||
|
||||
def _raise_responses_compatibility_failure(
|
||||
failure: _ResponsesCompatibilityFailure, model: str, custom_llm_provider: str | None
|
||||
) -> NoReturn:
|
||||
match failure:
|
||||
case "encrypted_task_unsupported":
|
||||
raise litellm.exception_type(
|
||||
model=model,
|
||||
custom_llm_provider=custom_llm_provider,
|
||||
original_exception=ValueError(
|
||||
"Encrypted task classification requires a compatible native Responses deployment"
|
||||
),
|
||||
)
|
||||
case _:
|
||||
assert_never(failure)
|
||||
|
||||
|
||||
def _deployment_passes_through_responses(model_info: object) -> bool:
|
||||
"""Whether ``model_info.supported_endpoints`` opts the deployment into native ``{api_base}/responses``."""
|
||||
if not isinstance(model_info, dict):
|
||||
|
|
@ -1078,6 +1110,7 @@ def responses(
|
|||
litellm_call_id: Final[str | None] = kwargs.get("litellm_call_id", None)
|
||||
_is_async: Final = kwargs.pop("aresponses", False) is True
|
||||
skip_mcp_handler: Final = kwargs.pop("_skip_mcp_handler", False)
|
||||
require_encrypted_task_support: Final = kwargs.pop("_require_encrypted_task_support", False) is True
|
||||
use_chat_completions_api = _pop_use_chat_completions_api_kw(kwargs)
|
||||
|
||||
client_headers: Final = kwargs.get("headers")
|
||||
|
|
@ -1186,6 +1219,17 @@ def responses(
|
|||
model, custom_llm_provider, deployment_model_info
|
||||
)
|
||||
|
||||
if (
|
||||
require_encrypted_task_support
|
||||
and (
|
||||
compatibility_failure := _encrypted_task_support_failure(
|
||||
responses_api_provider_config, use_chat_completions_api
|
||||
)
|
||||
)
|
||||
is not None
|
||||
):
|
||||
_raise_responses_compatibility_failure(compatibility_failure, model, custom_llm_provider)
|
||||
|
||||
local_vars.update(kwargs)
|
||||
# Map reasoning_effort (from litellm_params/proxy config) to reasoning when not set
|
||||
if reasoning is None and "reasoning_effort" in local_vars:
|
||||
|
|
|
|||
|
|
@ -361,6 +361,19 @@ model_list:
|
|||
keep the classifier deployment or provider default, or set a supported value such as `none` or
|
||||
`low` to override that call.
|
||||
|
||||
When the current ask is a Responses API `agent_message` containing `encrypted_content`, LLM
|
||||
classification preserves the encrypted task and uses native Responses. This also bypasses the
|
||||
local scoring shortcut in `heuristic_first` and `hybrid` modes. The configured classifier must use
|
||||
a native OpenAI or Azure OpenAI Responses deployment with access to the encrypted content. The
|
||||
provider handles the encrypted task, and the classifier still chooses the tier dynamically
|
||||
|
||||
Compatibility is checked after normal deployment selection. A paused incompatible member of the
|
||||
classifier group does not prevent an eligible compatible deployment from classifying the task
|
||||
|
||||
Unsupported classifier deployments and provider decryption errors use the existing
|
||||
`classifier_fallback` policy. No fixed tier is introduced for encrypted tasks. Plaintext asks and
|
||||
requests carrying only historical encrypted reasoning retain the existing classifier path
|
||||
|
||||
Classifier calls have a one-attempt hard deadline. After a timeout, the router opens a process-local
|
||||
circuit for that classifier and sends every session through `classifier_fallback` for
|
||||
`classifier_llm_config.circuit_breaker_cooldown_seconds` (30 seconds by default). When the cooldown
|
||||
|
|
|
|||
|
|
@ -25,7 +25,7 @@ from threading import Lock
|
|||
from types import MappingProxyType
|
||||
from typing import TYPE_CHECKING, Any, Final, Literal, NamedTuple, cast
|
||||
|
||||
from pydantic import BaseModel, create_model
|
||||
from pydantic import BaseModel, TypeAdapter, ValidationError, create_model
|
||||
|
||||
from litellm._logging import verbose_router_logger
|
||||
from litellm.constants import (
|
||||
|
|
@ -57,6 +57,7 @@ from litellm.types.llms.openai import (
|
|||
AllMessageValues,
|
||||
ChatCompletionImageObject,
|
||||
ChatCompletionTextObject,
|
||||
ResponsesAPIResponse,
|
||||
)
|
||||
from litellm.types.utils import (
|
||||
AUTOROUTER_CLASSIFIER_CALL_ORIGIN,
|
||||
|
|
@ -365,7 +366,7 @@ def _parent_session_kwargs(request_kwargs: Mapping[str, Any] | None) -> Mapping[
|
|||
return {k: kwargs[k] for k in ("litellm_session_id", "litellm_trace_id") if kwargs.get(k) is not None}
|
||||
|
||||
|
||||
def _response_cost_or_none(response: ModelResponse) -> float | None:
|
||||
def _response_cost_or_none(response: ModelResponse | ResponsesAPIResponse) -> float | None:
|
||||
hidden_params: Final = response._hidden_params
|
||||
if not isinstance(hidden_params, dict):
|
||||
return None
|
||||
|
|
@ -494,6 +495,42 @@ def _human_text(content: object, marker_pairs: tuple[tuple[str, str], ...] = _DE
|
|||
return _strip_reminder_blocks(_message_text(content), marker_pairs)
|
||||
|
||||
|
||||
def _encrypted_classifier_task(
|
||||
request_kwargs: Mapping[str, object] | None,
|
||||
marker_pairs: tuple[tuple[str, str], ...],
|
||||
) -> dict[str, object] | None:
|
||||
from litellm.litellm_core_utils.prompt_templates.factory import resolve_structured_messages
|
||||
|
||||
raw_input: Final = (request_kwargs or EMPTY_MAPPING).get("input")
|
||||
if not isinstance(raw_input, list) or (request_kwargs or EMPTY_MAPPING).get("messages"):
|
||||
return None
|
||||
try:
|
||||
items: Final = TypeAdapter(tuple[dict[str, object], ...]).validate_python(raw_input)
|
||||
except ValidationError:
|
||||
return None
|
||||
current: Final = next(
|
||||
(
|
||||
item
|
||||
for item in reversed(items)
|
||||
if (messages := resolve_structured_messages(messages=None, request_kwargs={"input": [item]}))
|
||||
and any(_iter_human_asks_newest_first(messages, marker_pairs))
|
||||
),
|
||||
None,
|
||||
)
|
||||
if current is None or current.get("type") != "agent_message" or not isinstance(current.get("content"), list):
|
||||
return None
|
||||
try:
|
||||
parts: Final = TypeAdapter(tuple[dict[str, object], ...]).validate_python(current["content"])
|
||||
except ValidationError:
|
||||
return None
|
||||
if not any(part.get("type") == "encrypted_content" and part.get("encrypted_content") for part in parts):
|
||||
return None
|
||||
return {
|
||||
**current,
|
||||
"content": [part for part in parts if part.get("type") in ("input_text", "encrypted_content")],
|
||||
}
|
||||
|
||||
|
||||
def _iter_human_asks_newest_first(
|
||||
messages: Sequence[Mapping[str, object]],
|
||||
marker_pairs: tuple[tuple[str, str], ...] = _DEFAULT_REMINDER_MARKERS,
|
||||
|
|
@ -1652,6 +1689,10 @@ class ComplexityRouter(CustomLogger):
|
|||
return self._classify_with_heuristic_v2(prompt)
|
||||
if self.config.classifier_type == "custom":
|
||||
return await self._classify_with_plugin(prompt, system_prompt, request_kwargs, raw_messages)
|
||||
if self.config.classifier_type in ("heuristic_first", "hybrid") and _encrypted_classifier_task(
|
||||
request_kwargs, self._reminder_markers_for_request(request_kwargs or EMPTY_MAPPING)
|
||||
):
|
||||
return await self._llm_classifier_outcome(prompt, system_prompt, request_kwargs, messages)
|
||||
if self.config.classifier_type == "heuristic_first" and self.config.classifier_llm_config is not None:
|
||||
return await self._classify_heuristic_first(prompt, system_prompt, request_kwargs, messages)
|
||||
if self.config.classifier_type == "hybrid" and self.config.classifier_llm_config is not None:
|
||||
|
|
@ -1987,8 +2028,9 @@ class ComplexityRouter(CustomLogger):
|
|||
> 1
|
||||
)
|
||||
|
||||
encrypted_task: Final = _encrypted_classifier_task(request_kwargs, marker_pairs)
|
||||
user_payload: Final = self._build_classifier_user_payload(
|
||||
prompt=prompt,
|
||||
prompt="The delegated task in the following agent_message." if encrypted_task is not None else prompt,
|
||||
system_prompt=system_prompt,
|
||||
prior_turns=prior_turns,
|
||||
messages=messages,
|
||||
|
|
@ -2021,34 +2063,37 @@ class ComplexityRouter(CustomLogger):
|
|||
if llm_config.reasoning_effort is not None:
|
||||
classifier_call_params = MappingProxyType({"reasoning_effort": llm_config.reasoning_effort})
|
||||
|
||||
proxy_server_request: Final = {
|
||||
"body": {
|
||||
"model": llm_config.model,
|
||||
"messages": messages_for_call,
|
||||
"response_format": response_format,
|
||||
**classifier_call_params,
|
||||
}
|
||||
}
|
||||
payload: Final = (
|
||||
self._native_classifier_payload(messages_for_call, response_format, encrypted_task)
|
||||
if encrypted_task is not None
|
||||
else {"messages": messages_for_call, "response_format": response_format, **classifier_call_params}
|
||||
)
|
||||
proxy_server_request: Final = {"body": {"model": llm_config.model, **payload}}
|
||||
classify: Final = (
|
||||
self.litellm_router_instance.aresponses
|
||||
if encrypted_task is not None
|
||||
else self.litellm_router_instance.acompletion
|
||||
)
|
||||
|
||||
classifier_timeout_s: Final[float] = llm_config.timeout_ms / 1000
|
||||
response: Final[ModelResponse] = await asyncio.wait_for(
|
||||
self.litellm_router_instance.acompletion(
|
||||
response: Final[ModelResponse | ResponsesAPIResponse] = await asyncio.wait_for(
|
||||
classify(
|
||||
model=llm_config.model,
|
||||
messages=messages_for_call,
|
||||
stream=False,
|
||||
response_format=response_format,
|
||||
timeout=classifier_timeout_s,
|
||||
num_retries=0,
|
||||
disable_fallbacks=True,
|
||||
metadata=metadata,
|
||||
proxy_server_request=proxy_server_request,
|
||||
turn_off_message_logging=turn_off_message_logging,
|
||||
**classifier_call_params,
|
||||
**payload,
|
||||
**_parent_session_kwargs(request_kwargs),
|
||||
),
|
||||
timeout=classifier_timeout_s,
|
||||
)
|
||||
content: Final = response.choices[0].message.content
|
||||
content: Final = (
|
||||
response.output_text if isinstance(response, ResponsesAPIResponse) else response.choices[0].message.content
|
||||
)
|
||||
if not content:
|
||||
raise ValueError("LLM classifier returned empty content")
|
||||
raw_tier: Final = _LabeledTierClassification.model_validate_json(content).tier
|
||||
|
|
@ -2057,6 +2102,33 @@ class ComplexityRouter(CustomLogger):
|
|||
raise ValueError(f"LLM classifier returned an unrecognized tier: {raw_tier!r}")
|
||||
return tier, _response_cost_or_none(response)
|
||||
|
||||
def _native_classifier_payload(
|
||||
self,
|
||||
messages: list[AllMessageValues], # mutable-ok: existing transformation accepts the SDK message list
|
||||
response_format: Mapping[str, object],
|
||||
encrypted_task: Mapping[str, object],
|
||||
) -> Mapping[str, object]:
|
||||
from litellm.completion_extras.litellm_responses_transformation.transformation import (
|
||||
LiteLLMResponsesTransformationHandler,
|
||||
)
|
||||
|
||||
transformation: Final = LiteLLMResponsesTransformationHandler()
|
||||
input_items, instructions = transformation.convert_chat_completion_messages_to_responses_api(messages)
|
||||
llm_config: Final = self.config.classifier_llm_config
|
||||
reasoning: Final = (
|
||||
{"reasoning": {"effort": llm_config.reasoning_effort}}
|
||||
if llm_config is not None and llm_config.reasoning_effort is not None
|
||||
else {}
|
||||
)
|
||||
return {
|
||||
"input": [*input_items, encrypted_task],
|
||||
"instructions": instructions,
|
||||
"text": transformation.transform_response_format_to_text_format(dict(response_format)),
|
||||
"store": False,
|
||||
"_require_encrypted_task_support": True,
|
||||
**reasoning,
|
||||
}
|
||||
|
||||
@staticmethod
|
||||
def _build_classifier_user_payload(
|
||||
prompt: str,
|
||||
|
|
|
|||
|
|
@ -7,10 +7,14 @@ calls so routed requests do not hit a custom api_base /v1/responses endpoint.
|
|||
"""
|
||||
|
||||
from importlib import import_module
|
||||
from typing import Final
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
import litellm
|
||||
from litellm.llms.custom_httpx.http_handler import HTTPHandler
|
||||
from litellm.types.llms.openai import ResponseAPIUsage, ResponsesAPIResponse
|
||||
from litellm.types.utils import Choices, Message, ModelResponse, Usage
|
||||
|
||||
|
|
@ -18,6 +22,26 @@ from litellm.types.utils import Choices, Message, ModelResponse, Usage
|
|||
class TestUseResponsesApiBridgeFlag:
|
||||
"""Test that bridge opt-in forces the chat completions path."""
|
||||
|
||||
@pytest.mark.parametrize("model", ["openai/chat_completions/gpt-6-astra", "xai/test-classifier"])
|
||||
def test_encrypted_classifier_rejection_preserves_public_error(self, model: str) -> None:
|
||||
respond: Final = MagicMock(side_effect=AssertionError("Incompatible classifier sent an upstream request"))
|
||||
with httpx.Client(transport=httpx.MockTransport(respond)) as client:
|
||||
with pytest.raises(
|
||||
litellm.APIConnectionError,
|
||||
match="Encrypted task classification requires a compatible native Responses deployment",
|
||||
) as error:
|
||||
litellm.responses(
|
||||
model=model,
|
||||
input="Delegated task",
|
||||
api_key="test-key",
|
||||
api_base="https://classifier.test/v1",
|
||||
client=HTTPHandler(client=client),
|
||||
_require_encrypted_task_support=True,
|
||||
num_retries=0,
|
||||
)
|
||||
assert error.value.status_code == 500
|
||||
respond.assert_not_called()
|
||||
|
||||
@patch.object(
|
||||
import_module("litellm.responses.main").litellm_completion_transformation_handler, "response_api_handler"
|
||||
)
|
||||
|
|
|
|||
|
|
@ -5,7 +5,10 @@ Tests the rule-based complexity scoring and tier assignment logic.
|
|||
"""
|
||||
|
||||
import asyncio
|
||||
from collections.abc import AsyncIterator
|
||||
import json
|
||||
from copy import deepcopy
|
||||
from functools import partial
|
||||
import logging
|
||||
import sys
|
||||
import time
|
||||
|
|
@ -13,6 +16,7 @@ from typing import Dict, Final, List
|
|||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
import httpx
|
||||
from pydantic import ValidationError
|
||||
|
||||
import litellm
|
||||
|
|
@ -67,6 +71,8 @@ from litellm.types.router import (
|
|||
LiteLLM_Params,
|
||||
TaggedPreRoutingStrategy,
|
||||
)
|
||||
from litellm.types.llms.openai import ResponsesAPIResponse
|
||||
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler
|
||||
|
||||
|
||||
requires_semantic_router = pytest.mark.skipif(
|
||||
|
|
@ -2490,6 +2496,340 @@ class TestTierLabels:
|
|||
assert set(config.tier_boundaries) == {"simple_medium", "medium_complex", "complex_reasoning"}
|
||||
|
||||
|
||||
def _encrypted_agent_task() -> dict[str, object]:
|
||||
return {
|
||||
"type": "agent_message",
|
||||
"author": "/root",
|
||||
"recipient": "/root/child",
|
||||
"content": [
|
||||
{"type": "input_text", "text": "Message Type: NEW_TASK\nTask name: /root/child\nPayload:\nHello"},
|
||||
{"type": "encrypted_content", "encrypted_content": "opaque-provider-task"},
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
def _native_classifier_response(content: str) -> ResponsesAPIResponse:
|
||||
response: Final = ResponsesAPIResponse(
|
||||
id="resp_classifier",
|
||||
created_at=0,
|
||||
status="completed",
|
||||
output=[{"type": "message", "role": "assistant", "content": [{"type": "output_text", "text": content}]}],
|
||||
)
|
||||
response._hidden_params = {"response_cost": 0.0001}
|
||||
return response
|
||||
|
||||
|
||||
def _native_classifier_router(
|
||||
output: str = '{"tier":"REASONING"}',
|
||||
classifier_type: str = "llm",
|
||||
deployment_model: str = "openai/gpt-6-astra",
|
||||
failure: Exception | None = None,
|
||||
native_router: Router | None = None,
|
||||
http_handler: AsyncHTTPHandler | None = None,
|
||||
) -> tuple[ComplexityRouter, MagicMock]:
|
||||
dependency: Final = MagicMock(
|
||||
aresponses=(
|
||||
native_router.factory_function(partial(litellm.aresponses, client=http_handler), call_type="aresponses")
|
||||
if native_router is not None
|
||||
else AsyncMock(return_value=_native_classifier_response(output), side_effect=failure)
|
||||
),
|
||||
acompletion=AsyncMock(return_value=_llm_response('{"tier":"SIMPLE"}')),
|
||||
get_model_list=(
|
||||
native_router.get_model_list
|
||||
if native_router is not None
|
||||
else MagicMock(return_value=[{"litellm_params": {"model": deployment_model}}])
|
||||
),
|
||||
)
|
||||
return (
|
||||
ComplexityRouter(
|
||||
model_name="encrypted-router",
|
||||
litellm_router_instance=dependency,
|
||||
complexity_router_config={
|
||||
"tiers": {"SIMPLE": "cheap-model", "REASONING": "deep-model"},
|
||||
"classifier_type": classifier_type,
|
||||
"classifier_llm_config": {
|
||||
"model": "classifier",
|
||||
"timeout_ms": 5000 if native_router is not None else 100,
|
||||
"reasoning_effort": "low",
|
||||
},
|
||||
"heuristic_first_max_tier": "SIMPLE" if classifier_type == "heuristic_first" else None,
|
||||
"hybrid_boundary_margin": 0.01 if classifier_type == "hybrid" else None,
|
||||
"classifier_fallback": "default_model",
|
||||
"default_model": "deep-model",
|
||||
"session_affinity": False,
|
||||
"deployment_affinity": False,
|
||||
},
|
||||
),
|
||||
dependency,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def native_classifier_http() -> AsyncIterator[tuple[AsyncHTTPHandler, MagicMock]]:
|
||||
respond: Final = MagicMock(
|
||||
return_value=httpx.Response(200, json=_native_classifier_response('{"tier":"REASONING"}').model_dump())
|
||||
)
|
||||
async with httpx.AsyncClient(transport=httpx.MockTransport(respond)) as client:
|
||||
handler: Final = AsyncHTTPHandler()
|
||||
await handler.client.aclose()
|
||||
handler.client = client
|
||||
yield handler, respond
|
||||
|
||||
|
||||
class TestEncryptedTaskClassifier:
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("classifier_type", ["llm", "heuristic_first", "hybrid"])
|
||||
@pytest.mark.parametrize("codex", [True, False])
|
||||
@pytest.mark.parametrize(
|
||||
"reminder",
|
||||
[
|
||||
"<environment_context>cwd=/repo</environment_context>",
|
||||
"<user_instructions>Keep answers concise</user_instructions>",
|
||||
],
|
||||
)
|
||||
async def test_encrypted_task_detection_uses_request_reminder_markers(
|
||||
self, classifier_type: str, codex: bool, reminder: str
|
||||
):
|
||||
router, dependency = _native_classifier_router(classifier_type=classifier_type)
|
||||
task: Final = _encrypted_agent_task()
|
||||
request: Final = {
|
||||
"input": [task, {"role": "user", "content": reminder}],
|
||||
"metadata": {"user_agent": "codex-tui" if codex else "curl/8.7.1"},
|
||||
}
|
||||
original: Final = deepcopy(request)
|
||||
|
||||
result: Final = await router.async_pre_routing_hook(model="encrypted-router", request_kwargs=request)
|
||||
|
||||
assert request == original
|
||||
assert result.model == ("deep-model" if codex else "cheap-model")
|
||||
if codex:
|
||||
assert result.routing_decision["cause"] == "llm_classifier"
|
||||
assert result.routing_decision["tier"] == "REASONING"
|
||||
dependency.aresponses.assert_awaited_once()
|
||||
assert dependency.aresponses.call_args.kwargs["input"][-1] == task
|
||||
dependency.acompletion.assert_not_called()
|
||||
else:
|
||||
dependency.aresponses.assert_not_called()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("classifier_type", ["llm", "heuristic_first", "hybrid"])
|
||||
@pytest.mark.parametrize("tier,model", [("SIMPLE", "cheap-model"), ("REASONING", "deep-model")])
|
||||
async def test_encrypted_task_routes_by_native_verdict(self, classifier_type: str, tier: str, model: str):
|
||||
router, dependency = _native_classifier_router(json.dumps({"tier": tier}), classifier_type)
|
||||
task: Final = _encrypted_agent_task()
|
||||
request: Final = {
|
||||
"input": [
|
||||
{"role": "user", "content": "Prior task context"},
|
||||
task,
|
||||
{"type": "function_call_output", "call_id": "call_1", "output": "Tool output"},
|
||||
{"role": "user", "content": "<system-reminder>Injected reminder</system-reminder>"},
|
||||
],
|
||||
"instructions": "Caller constraints",
|
||||
"tools": [{"type": "function", "name": "execute"}],
|
||||
"previous_response_id": "resp_parent",
|
||||
"litellm_session_id": "parent-session",
|
||||
"litellm_trace_id": "parent-trace",
|
||||
"turn_off_message_logging": True,
|
||||
"litellm_metadata": {"user_api_key_hash": "caller-key-hash"},
|
||||
}
|
||||
original: Final = deepcopy(request)
|
||||
|
||||
result: Final = await router.async_pre_routing_hook(model="encrypted-router", request_kwargs=request)
|
||||
|
||||
assert result.model == model
|
||||
assert result.routing_decision["tier"] == tier
|
||||
assert result.routing_decision["cause"] == "llm_classifier"
|
||||
assert result.routing_decision["classifier_cost"] == 0.0001
|
||||
assert result.messages is None
|
||||
assert request == original
|
||||
dependency.acompletion.assert_not_called()
|
||||
call: Final = dependency.aresponses.call_args.kwargs
|
||||
assert call["input"][-1] == task
|
||||
assert "opaque-provider-task" not in json.dumps(call["input"][:-1])
|
||||
assert "Prior task context" in json.dumps(call["input"][:-1])
|
||||
assert "Caller constraints" in json.dumps(call["input"][:-1])
|
||||
assert "Caller constraints" not in call["instructions"]
|
||||
assert "SIMPLE" in call["instructions"] and "REASONING" in call["instructions"]
|
||||
assert call["text"]["format"]["schema"]["properties"]["tier"]["enum"] == [
|
||||
"SIMPLE",
|
||||
"MEDIUM",
|
||||
"COMPLEX",
|
||||
"REASONING",
|
||||
]
|
||||
assert call["text"]["format"]["strict"] is True
|
||||
assert call["reasoning"] == {"effort": "low"}
|
||||
assert call["store"] is False
|
||||
assert call["_require_encrypted_task_support"] is True
|
||||
assert call["stream"] is False
|
||||
assert "tools" not in call and "previous_response_id" not in call
|
||||
assert "messages" not in call and "response_format" not in call
|
||||
assert call["timeout"] == 0.1 and call["num_retries"] == 0 and call["disable_fallbacks"] is True
|
||||
assert call["litellm_session_id"] == "parent-session"
|
||||
assert call["litellm_trace_id"] == "parent-trace"
|
||||
assert call["turn_off_message_logging"] is True
|
||||
assert call["metadata"]["user_api_key_hash"] == "caller-key-hash"
|
||||
assert call["proxy_server_request"]["body"]["input"] == call["input"]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize(
|
||||
"items",
|
||||
[
|
||||
[
|
||||
{"type": "reasoning", "encrypted_content": "opaque-history", "summary": []},
|
||||
{"role": "user", "content": "hi"},
|
||||
],
|
||||
[_encrypted_agent_task(), {"role": "user", "content": "hi"}],
|
||||
[{**_encrypted_agent_task(), "content": [{"type": "input_text", "text": "hi"}]}],
|
||||
[{"role": "user", "content": "gAAAA is plain text"}],
|
||||
[
|
||||
{"role": "user", "content": "hi"},
|
||||
{"type": "function_call_output", "call_id": "call_1", "output": "opaque-provider-task"},
|
||||
],
|
||||
],
|
||||
ids=[
|
||||
"historical-reasoning",
|
||||
"older-encrypted-task",
|
||||
"plaintext-agent",
|
||||
"ciphertext-looking-text",
|
||||
"tool-output",
|
||||
],
|
||||
)
|
||||
async def test_other_asks_keep_chat_classifier(self, items: list[dict[str, object]]):
|
||||
router, dependency = _native_classifier_router()
|
||||
|
||||
result: Final = await router.async_pre_routing_hook(model="encrypted-router", request_kwargs={"input": items})
|
||||
|
||||
assert result.model == "cheap-model"
|
||||
assert result.routing_decision["cause"] == "llm_classifier"
|
||||
dependency.aresponses.assert_not_called()
|
||||
dependency.acompletion.assert_awaited_once()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("output", ["", "not-json", '{"tier":"UNKNOWN"}'])
|
||||
async def test_invalid_native_verdict_uses_existing_fallback(self, output: str):
|
||||
router, dependency = _native_classifier_router(output=output)
|
||||
|
||||
result: Final = await router.async_pre_routing_hook(
|
||||
model="encrypted-router", request_kwargs={"input": [_encrypted_agent_task()]}
|
||||
)
|
||||
|
||||
assert result.model == "deep-model"
|
||||
assert result.routing_decision["cause"] == "default_model_fallback"
|
||||
dependency.aresponses.assert_awaited_once()
|
||||
dependency.acompletion.assert_not_called()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize(
|
||||
"deployment_model",
|
||||
["anthropic/test-classifier", "openai/chat_completions/gpt-6-astra", "xai/test-classifier"],
|
||||
)
|
||||
async def test_incompatible_classifier_does_not_flatten_encryption(
|
||||
self, deployment_model: str, native_classifier_http: tuple[AsyncHTTPHandler, MagicMock]
|
||||
):
|
||||
handler, respond = native_classifier_http
|
||||
native: Final = Router(
|
||||
model_list=[
|
||||
{
|
||||
"model_name": "classifier",
|
||||
"litellm_params": {
|
||||
"model": deployment_model,
|
||||
"api_key": "test-key",
|
||||
"api_base": "https://classifier.test/v1",
|
||||
},
|
||||
}
|
||||
],
|
||||
num_retries=0,
|
||||
)
|
||||
router, _ = _native_classifier_router(native_router=native, http_handler=handler)
|
||||
|
||||
result: Final = await router.async_pre_routing_hook(
|
||||
model="encrypted-router", request_kwargs={"input": [_encrypted_agent_task()]}
|
||||
)
|
||||
|
||||
assert result.model == "deep-model"
|
||||
assert result.routing_decision["cause"] == "default_model_fallback"
|
||||
respond.assert_not_called()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("blocked", [True, False])
|
||||
async def test_native_classifier_validates_selected_deployment(
|
||||
self, blocked: bool, native_classifier_http: tuple[AsyncHTTPHandler, MagicMock]
|
||||
):
|
||||
handler, respond = native_classifier_http
|
||||
native: Final = Router(
|
||||
model_list=[
|
||||
{
|
||||
"model_name": "classifier",
|
||||
"litellm_params": {"model": "anthropic/test-classifier", "api_key": "test-key", "order": 0},
|
||||
"model_info": {"id": "incompatible", "blocked": blocked},
|
||||
},
|
||||
{
|
||||
"model_name": "classifier",
|
||||
"litellm_params": {
|
||||
"model": "openai/gpt-6-astra",
|
||||
"api_key": "test-key",
|
||||
"order": 1,
|
||||
"api_base": "https://classifier.test/v1",
|
||||
},
|
||||
"model_info": {"id": "compatible"},
|
||||
},
|
||||
],
|
||||
num_retries=0,
|
||||
)
|
||||
router, _ = _native_classifier_router(native_router=native, http_handler=handler)
|
||||
task: Final = _encrypted_agent_task()
|
||||
|
||||
result: Final = await router.async_pre_routing_hook(model="encrypted-router", request_kwargs={"input": [task]})
|
||||
|
||||
assert result.model == "deep-model"
|
||||
assert result.routing_decision["cause"] == ("llm_classifier" if blocked else "default_model_fallback")
|
||||
if blocked:
|
||||
respond.assert_called_once()
|
||||
request: Final = respond.call_args.args[0]
|
||||
assert request.url.path == "/v1/responses"
|
||||
body: Final = json.loads(request.content)
|
||||
assert body["input"][-1] == task
|
||||
assert "_require_encrypted_task_support" not in body
|
||||
else:
|
||||
respond.assert_not_called()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("classifier_type", ["llm", "heuristic_first", "hybrid"])
|
||||
@pytest.mark.parametrize(
|
||||
"input_items",
|
||||
[
|
||||
["unsupported-input-item"],
|
||||
[{**_encrypted_agent_task(), "content": [{"type": "input_text", "text": "hi"}, None]}],
|
||||
],
|
||||
)
|
||||
async def test_encrypted_detection_does_not_reject_other_input_shapes(
|
||||
self, classifier_type: str, input_items: list[object]
|
||||
):
|
||||
router, dependency = _native_classifier_router(classifier_type=classifier_type)
|
||||
|
||||
result: Final = await router.aclassify("hi", request_kwargs={"input": input_items})
|
||||
|
||||
assert result.cause != "default_model_fallback"
|
||||
assert result.tier == ComplexityTier.SIMPLE
|
||||
dependency.aresponses.assert_not_called()
|
||||
if classifier_type == "llm":
|
||||
dependency.acompletion.assert_awaited_once()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("failure", [ValueError("invalid_encrypted_content"), TimeoutError("classifier timed out")])
|
||||
async def test_native_provider_failure_uses_existing_fallback(self, failure: Exception):
|
||||
router, dependency = _native_classifier_router(failure=failure)
|
||||
|
||||
result: Final = await router.async_pre_routing_hook(
|
||||
model="encrypted-router", request_kwargs={"input": [_encrypted_agent_task()]}
|
||||
)
|
||||
|
||||
assert result.model == "deep-model"
|
||||
assert result.routing_decision["cause"] == "default_model_fallback"
|
||||
dependency.aresponses.assert_awaited_once()
|
||||
dependency.acompletion.assert_not_called()
|
||||
|
||||
|
||||
class TestLLMClassifier:
|
||||
"""Test the LLM-based classifier path (aclassify) and its fallback behavior."""
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue