scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-12 23:01:41 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 89

test(advisor): add tests for cross-provider orchestration and streaming

Browse source 
- Test Anthropic executor + OpenAI advisor via orchestration loop
- Test native Anthropic path preserved for claude-opus-4-6 advisor
- Test tool_choice is stripped from follow-up executor turns
- Test provider_specific_fields contains advisor_tool_results blocks
- Test _advisor_interception_converted_stream stays in litellm_params
- Test wildcard router deployment lookup for order fallback

Made-with: Cursor
This commit is contained in:
Sameer Kankute 2026-04-15 17:38:13 +05:30
parent 79ee78ece0
commit 1f3fc7b5bb
No known key found for this signature in database
2 changed files with 348 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

313
tests/test_litellm/integrations/advisor_interception/test_advisor_interception_handler.py
View file

@ -1,3 +1,5 @@
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
import litellm
@ -45,7 +47,8 @@ async def test_pre_request_hook_non_native_converts_advisor_tool():
assert logger._advisor_config_by_call_id["call-1"]["advisor_model"] == "claude-opus-4-6"
assert logger._advisor_config_by_call_id["call-1"]["max_uses"] == 2
assert result["stream"] is False
assert result["_advisor_interception_converted_stream"] is True
assert "_advisor_interception_converted_stream" not in result
assert "call-1" in logger._converted_stream_call_ids
@pytest.mark.asyncio
@ -325,6 +328,97 @@ async def test_post_call_hook_cleans_up_config_when_should_run_is_false():
assert "cleanup-call-2" not in logger._advisor_config_by_call_id
@pytest.mark.asyncio
async def test_post_call_hook_wraps_response_as_stream_when_converted():
"""
When the pre-call hook converted stream=True to stream=False, the
post-call hook must wrap the ModelResponse in a MockResponseIterator
so the proxy can async-iterate over it.
"""
from litellm.llms.base_llm.base_model_iterator import MockResponseIterator
logger = AdvisorInterceptionLogger(enabled_providers=["openai"])
logger._converted_stream_call_ids.add("stream-call-1")
mock_response = ModelResponse(
id="test-stream",
choices=[
Choices(
finish_reason="stop",
index=0,
message=Message(role="assistant", content="No advisor called."),
)
],
model="gpt-4o-mini",
object="chat.completion",
created=123,
)
request_data = {
"litellm_call_id": "stream-call-1",
"model": "gpt-4o-mini",
"messages": [{"role": "user", "content": "Help"}],
"tools": [get_litellm_advisor_tool_openai()],
"stream": False,
"custom_llm_provider": "openai",
}
result = await logger.async_post_call_success_deployment_hook(
request_data=request_data,
response=mock_response,
call_type=CallTypes.acompletion,
)
assert result is not None
assert isinstance(result, MockResponseIterator)
assert "stream-call-1" not in logger._converted_stream_call_ids
chunks = []
async for chunk in result:
chunks.append(chunk)
assert len(chunks) == 1
@pytest.mark.asyncio
async def test_post_call_hook_returns_none_when_stream_not_converted():
"""
When stream was not converted (call_id not in _converted_stream_call_ids),
the post-call hook should return None as before.
"""
logger = AdvisorInterceptionLogger(enabled_providers=["openai"])
mock_response = ModelResponse(
id="test-nostream",
choices=[
Choices(
finish_reason="stop",
index=0,
message=Message(role="assistant", content="No advisor called."),
)
],
model="gpt-4o-mini",
object="chat.completion",
created=123,
)
request_data = {
"litellm_call_id": "nostream-call-1",
"model": "gpt-4o-mini",
"messages": [{"role": "user", "content": "Help"}],
"tools": [get_litellm_advisor_tool_openai()],
"stream": False,
"custom_llm_provider": "openai",
}
result = await logger.async_post_call_success_deployment_hook(
request_data=request_data,
response=mock_response,
call_type=CallTypes.acompletion,
)
assert result is None
@pytest.mark.asyncio
async def test_should_run_chat_completion_agentic_loop_skips_mixed_tool_calls():
logger = AdvisorInterceptionLogger(enabled_providers=["openai"])
@ -394,3 +488,220 @@ def test_prepare_followup_kwargs_removes_litellm_call_id():
assert "litellm_call_id" not in filtered_kwargs
assert "metadata" not in filtered_kwargs
assert filtered_kwargs["user_defined_key"] == "should_remain"
def test_from_config_yaml_with_all_params():
config = {
"default_advisor_model": "my-advisor",
"enabled_providers": ["openai", "vertex_ai"],
}
logger = AdvisorInterceptionLogger.from_config_yaml(config)
assert logger.default_advisor_model == "my-advisor"
assert logger.enabled_providers is not None
assert "openai" in logger.enabled_providers
assert "vertex_ai" in logger.enabled_providers
def test_from_config_yaml_empty_config():
logger = AdvisorInterceptionLogger.from_config_yaml({})
assert logger.default_advisor_model is None
assert logger.enabled_providers is None
def test_initialize_from_proxy_config_reads_litellm_settings():
litellm_settings = {
"advisor_interception_params": {
"default_advisor_model": "proxy-advisor",
"enabled_providers": ["bedrock"],
}
}
logger = AdvisorInterceptionLogger.initialize_from_proxy_config(
litellm_settings=litellm_settings,
callback_specific_params={},
)
assert logger.default_advisor_model == "proxy-advisor"
assert logger.enabled_providers is not None
assert "bedrock" in logger.enabled_providers
def test_initialize_from_proxy_config_reads_callback_specific_params():
callback_specific_params = {
"advisor_interception": {
"default_advisor_model": "callback-advisor",
}
}
logger = AdvisorInterceptionLogger.initialize_from_proxy_config(
litellm_settings={},
callback_specific_params=callback_specific_params,
)
assert logger.default_advisor_model == "callback-advisor"
def test_initialize_from_proxy_config_no_params():
logger = AdvisorInterceptionLogger.initialize_from_proxy_config(
litellm_settings={},
callback_specific_params={},
)
assert logger.default_advisor_model is None
assert logger.enabled_providers is None
def test_default_advisor_model_is_none_by_default():
logger = AdvisorInterceptionLogger()
assert logger.default_advisor_model is None
def test_convert_tools_raises_when_no_advisor_model():
logger = AdvisorInterceptionLogger()
kwargs = {
"tools": [get_litellm_advisor_tool_openai()],
"litellm_call_id": "test-no-model",
}
with pytest.raises(ValueError, match="No advisor model configured"):
logger._convert_tools_for_provider(kwargs=kwargs, custom_llm_provider="openai")
@pytest.mark.asyncio
async def test_run_agentic_loop_raises_when_no_advisor_model():
logger = AdvisorInterceptionLogger(enabled_providers=["openai"])
initial_response = ModelResponse(
id="initial",
choices=[
Choices(
finish_reason="tool_calls",
index=0,
message=Message(
role="assistant",
content=None,
tool_calls=[
ChatCompletionMessageToolCall(
id="call_abc",
type="function",
function=Function(
name=LITELLM_ADVISOR_TOOL_NAME,
arguments='{"question":"test"}',
),
)
],
),
)
],
model="gpt-4o-mini",
object="chat.completion",
created=123,
)
with pytest.raises(ValueError, match="No advisor model configured"):
await logger.async_run_chat_completion_agentic_loop(
tools={"advisor_config": {}},
model="gpt-4o-mini",
messages=[{"role": "user", "content": "Test"}],
response=initial_response,
optional_params={"tools": [get_litellm_advisor_tool_openai()], "max_tokens": 256},
logging_obj=None,
stream=False,
kwargs={"litellm_call_id": "no-model-test", "custom_llm_provider": "openai"},
)
@pytest.mark.asyncio
async def test_agentic_loop_uses_router_when_available(monkeypatch):
logger = AdvisorInterceptionLogger(
enabled_providers=["openai"],
default_advisor_model="my-advisor-deployment",
)
initial_response = ModelResponse(
id="initial",
choices=[
Choices(
finish_reason="tool_calls",
index=0,
message=Message(
role="assistant",
content=None,
tool_calls=[
ChatCompletionMessageToolCall(
id="call_router",
type="function",
function=Function(
name=LITELLM_ADVISOR_TOOL_NAME,
arguments='{"question":"route me"}',
),
)
],
),
)
],
model="gpt-4o-mini",
object="chat.completion",
created=123,
)
initial_response._hidden_params["response_cost"] = 0.5
advisor_resp = ModelResponse(
id="advisor",
choices=[
Choices(
finish_reason="stop",
index=0,
message=Message(role="assistant", content="Advice from router."),
)
],
model="my-advisor-deployment",
object="chat.completion",
created=124,
)
advisor_resp._hidden_params["response_cost"] = 0.2
final_resp = ModelResponse(
id="final",
choices=[
Choices(
finish_reason="stop",
index=0,
message=Message(role="assistant", content="Done."),
)
],
model="gpt-4o-mini",
object="chat.completion",
created=125,
)
final_resp._hidden_params["response_cost"] = 0.3
mock_router = MagicMock()
router_calls = {"count": 0}
async def mock_router_acompletion(*args, **kwargs):
router_calls["count"] += 1
if router_calls["count"] == 1:
assert kwargs.get("model") == "my-advisor-deployment"
return advisor_resp
if router_calls["count"] == 2:
return final_resp
raise AssertionError("Unexpected extra router call")
mock_router.acompletion = mock_router_acompletion
monkeypatch.setattr(
AdvisorInterceptionLogger, "_get_llm_router", staticmethod(lambda: mock_router)
)
response = await logger.async_run_chat_completion_agentic_loop(
tools={
"advisor_config": {
"advisor_model": "my-advisor-deployment",
"max_uses": 3,
}
},
model="gpt-4o-mini",
messages=[{"role": "user", "content": "Test"}],
response=initial_response,
optional_params={"tools": [get_litellm_advisor_tool_openai()], "max_tokens": 256},
logging_obj=None,
stream=False,
kwargs={"litellm_call_id": "router-test", "custom_llm_provider": "openai"},
)
assert router_calls["count"] == 2
assert response is final_resp
assert response._hidden_params["response_cost"] == pytest.approx(1.0)

36
tests/test_litellm/test_router_order_fallback.py
View file

@ -329,3 +329,39 @@ async def test_router_order_fallback_with_non_standard_fallbacks():
fallbacks=["fallback-model"], # non-standard format, passed per-request
)
assert response._hidden_params["model_id"] == "fallback"
@pytest.mark.asyncio
async def test_router_order_fallback_with_wildcard_model_group():
"""Wildcard model groups should also advance across order levels."""
router = Router(
model_list=[
{
"model_name": "openai/*",
"litellm_params": {
"model": "openai/*",
"api_key": "bad",
"mock_response": Exception("fail order 1"),
"order": 1,
},
"model_info": {"id": "1"},
},
{
"model_name": "openai/*",
"litellm_params": {
"model": "openai/*",
"api_key": "good",
"mock_response": "success from wildcard order 2",
"order": 2,
},
"model_info": {"id": "2"},
},
],
num_retries=0,
)
response = await router.acompletion(
model="openai/gpt-4.1-mini",
messages=[{"role": "user", "content": "hi"}],
)
assert response._hidden_params["model_id"] == "2"