Merge pull request #25579 from BerriAI/feat/anthropic-advisor-tool
feat(advisor): advisor tool orchestration loop for non-Anthropic providers
This commit is contained in:
ishaan-berri
GitHub
commit
329a526b9d
@ -14,12 +14,48 @@ The advisor tool is in beta. Include `anthropic-beta: advisor-tool-2026-03-01` i
|
||||
|
||||
## Supported Providers
|
||||
|
||||
| Provider | Chat Completions API | Messages API |
|
||||
|----------|---------------------|--------------|
|
||||
| **Anthropic API** | ✅ | ✅ |
|
||||
| **Azure Anthropic** | ❌ (coming soon) | ❌ (coming soon) |
|
||||
| **Google Cloud Vertex AI** | ❌ (coming soon) | ❌ (coming soon) |
|
||||
| **Amazon Bedrock** | ❌ (coming soon) | ❌ (coming soon) |
|
||||
| Provider | Chat Completions API | Messages API | Notes |
|
||||
|----------|---------------------|--------------|-------|
|
||||
| **Anthropic API** | ✅ | ✅ | Native — runs server-side |
|
||||
| **OpenAI / Azure OpenAI** | ✅ | ✅ | LiteLLM orchestration loop |
|
||||
| **Amazon Bedrock** | ✅ | ✅ | LiteLLM orchestration loop |
|
||||
| **Google Vertex AI** | ✅ | ✅ | LiteLLM orchestration loop |
|
||||
| **Groq / Mistral / others** | ✅ | ✅ | LiteLLM orchestration loop |
|
||||
|
||||
## How it works (LiteLLM native orchestration)
|
||||
|
||||
For non-Anthropic providers, LiteLLM implements the advisor loop itself. The API you call is identical — LiteLLM handles everything transparently.
|
||||
|
||||
When a request arrives with an `advisor_20260301` tool and a non-Anthropic provider, `AdvisorOrchestrationHandler` intercepts it. It translates the advisor tool into a regular function tool the provider understands, then runs an orchestration loop:
|
||||
|
||||
```mermaid
|
||||
flowchart TD
|
||||
A["Your request\ntools: advisor_20260301\nmodel: e.g. openai/gpt-4.1-mini"] --> B["AdvisorOrchestrationHandler\ntranslates advisor → regular fn tool"]
|
||||
|
||||
B --> C["EXECUTOR CALL\nopenai / bedrock / vertex / etc."]
|
||||
|
||||
C --> D{"executor calls\nadvisor tool?"}
|
||||
|
||||
D -->|"yes — tool_use\nname=advisor"| E{"max_uses\nexceeded?"}
|
||||
|
||||
E -->|no| F["ADVISOR SUB-CALL\nclaude-opus-4-6\nfull transcript forwarded\nno tools"]
|
||||
|
||||
F --> G["Inject advice as\ntool_result into history"]
|
||||
|
||||
G --> C
|
||||
|
||||
E -->|yes| H["AdvisorMaxIterationsError"]
|
||||
|
||||
D -->|"no — end_turn\nor other stop reason"| I["Clean final response\nno advisor blocks in output"]
|
||||
```
|
||||
|
||||
**What LiteLLM does for you:**
|
||||
|
||||
- Strips `advisor_20260301` from the outgoing request — the provider only sees a standard function tool named `advisor`
|
||||
- When the executor calls it, intercepts before the result reaches you, runs the advisor sub-call, and injects the advice
|
||||
- Strips any `advisor_tool_result` / `server_tool_use` blocks from message history on re-send so non-Anthropic providers never see Anthropic-specific types
|
||||
- Wraps the final response in an SSE stream if you requested `stream=True`
|
||||
- Enforces `max_uses` as a hard cap — `AdvisorMaxIterationsError` is raised if exceeded; `max_uses=0` disables the advisor entirely
|
||||
|
||||
## Model Compatibility
|
||||
|
||||
@ -305,6 +341,37 @@ response = client.beta.messages.create(
|
||||
print(response)
|
||||
```
|
||||
|
||||
#### Non-Anthropic Provider (LiteLLM orchestration loop)
|
||||
|
||||
```python showLineNumbers title="Advisor Tool with OpenAI executor"
|
||||
import asyncio
|
||||
import litellm
|
||||
|
||||
async def main():
|
||||
# executor: openai/gpt-4.1-mini | advisor: claude-opus-4-6
|
||||
# LiteLLM runs the orchestration loop automatically
|
||||
response = await litellm.anthropic.messages.acreate(
|
||||
model="openai/gpt-4.1-mini",
|
||||
messages=[
|
||||
{"role": "user", "content": "Implement a Python LRU cache with O(1) get and put."}
|
||||
],
|
||||
tools=[
|
||||
{
|
||||
"type": "advisor_20260301",
|
||||
"name": "advisor",
|
||||
"model": "claude-opus-4-6",
|
||||
"max_uses": 3,
|
||||
}
|
||||
],
|
||||
max_tokens=1024,
|
||||
custom_llm_provider="openai",
|
||||
)
|
||||
# Final response is clean — no advisor tool_use blocks
|
||||
print(response["content"][0]["text"])
|
||||
|
||||
asyncio.run(main())
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Response Structure
|
||||
|
||||
@ -1421,7 +1421,7 @@ APSCHEDULER_REPLACE_EXISTING = os.getenv(
|
||||
"1",
|
||||
] # always replace existing jobs
|
||||
|
||||
# The number of tag entries are higher than number of user, team entries. This leads to a higher QPS.
|
||||
# The number of tag entries are higher than number of user, team entries. This leads to a higher QPS.
|
||||
# This will run tag spcific tasks at a later time to smooth QPS
|
||||
DAILY_TAG_SPEND_BATCH_MULTIPLIER = 2.3
|
||||
|
||||
@ -1581,3 +1581,16 @@ MAX_PAYLOAD_SIZE_FOR_DEBUG_LOG = int(
|
||||
MAX_COMPETITOR_NAMES = int(os.getenv("MAX_COMPETITOR_NAMES", 100))
|
||||
COMPETITOR_LLM_TEMPERATURE = float(os.getenv("COMPETITOR_LLM_TEMPERATURE", 0.3))
|
||||
DEFAULT_COMPETITOR_DISCOVERY_MODEL = "gpt-4o-mini"
|
||||
|
||||
# Advisor tool orchestration
|
||||
# Providers that support advisor_20260301 natively (no LiteLLM orchestration needed).
|
||||
# Add vertex_ai here once verified.
|
||||
ADVISOR_NATIVE_PROVIDERS: frozenset = frozenset({"anthropic"})
|
||||
# Hard cap on advisor iterations per request to prevent runaway loops.
|
||||
ADVISOR_MAX_USES: int = 5
|
||||
# Description injected into the synthetic advisor tool definition sent to non-native providers.
|
||||
ADVISOR_TOOL_DESCRIPTION: str = (
|
||||
"Consult a highly intelligent advisor model when you need expert guidance, "
|
||||
"want to verify your reasoning, or face a complex decision. "
|
||||
"Describe your question or challenge clearly in the 'question' field."
|
||||
)
|
||||
|
||||
@ -639,15 +639,23 @@ class AnthropicModelInfo(BaseLLMModelInfo):
|
||||
return AnthropicTokenCounter()
|
||||
|
||||
|
||||
def strip_advisor_blocks_from_messages(messages: List[Any]) -> List[Any]:
|
||||
def strip_advisor_blocks_from_messages(
|
||||
messages: List[Any], replace_with_text: bool = False
|
||||
) -> List[Any]:
|
||||
"""
|
||||
Remove server_tool_use (name='advisor') and advisor_tool_result blocks from
|
||||
assistant message content when the advisor tool is absent from the request.
|
||||
Remove (or replace) server_tool_use (name='advisor') and advisor_tool_result blocks
|
||||
from assistant message content.
|
||||
|
||||
Prevents Anthropic 400 invalid_request_error: if advisor_tool_result blocks
|
||||
exist in history but the advisor tool is not in the tools array, the API rejects
|
||||
the request. This happens when the user has removed the advisor tool for cost
|
||||
control or on a follow-up turn.
|
||||
|
||||
Args:
|
||||
messages: Conversation history to process (mutated in-place).
|
||||
replace_with_text: When True, replace the advisor exchange with an
|
||||
<advisor_feedback> text block so the executor retains the semantic
|
||||
context of what the advisor said. When False (default), strip silently.
|
||||
"""
|
||||
for message in messages:
|
||||
if not isinstance(message, dict) or message.get("role") != "assistant":
|
||||
@ -655,7 +663,9 @@ def strip_advisor_blocks_from_messages(messages: List[Any]) -> List[Any]:
|
||||
content = message.get("content")
|
||||
if not isinstance(content, list):
|
||||
continue
|
||||
advisor_ids: set = set()
|
||||
|
||||
# Collect advisor server_tool_use ids and their advice text (for replace mode).
|
||||
advisor_id_to_text: dict = {}
|
||||
for block in content:
|
||||
if (
|
||||
isinstance(block, dict)
|
||||
@ -664,26 +674,65 @@ def strip_advisor_blocks_from_messages(messages: List[Any]) -> List[Any]:
|
||||
):
|
||||
bid = block.get("id")
|
||||
if bid:
|
||||
advisor_ids.add(bid)
|
||||
if not advisor_ids:
|
||||
advisor_id_to_text[bid] = None # text filled in below
|
||||
|
||||
if not advisor_id_to_text:
|
||||
continue
|
||||
message["content"] = [
|
||||
block
|
||||
for block in content
|
||||
if not (
|
||||
isinstance(block, dict)
|
||||
and (
|
||||
(
|
||||
block.get("type") == "server_tool_use"
|
||||
and block.get("name") == "advisor"
|
||||
|
||||
# If replacing, collect the advisor response text from advisor_tool_result blocks.
|
||||
if replace_with_text:
|
||||
for block in content:
|
||||
if (
|
||||
isinstance(block, dict)
|
||||
and block.get("type") == "advisor_tool_result"
|
||||
and block.get("tool_use_id") in advisor_id_to_text
|
||||
):
|
||||
raw = block.get("content") or ""
|
||||
text = (
|
||||
raw
|
||||
if isinstance(raw, str)
|
||||
else next(
|
||||
(
|
||||
b.get("text", "")
|
||||
for b in raw
|
||||
if isinstance(b, dict) and b.get("type") == "text"
|
||||
),
|
||||
"",
|
||||
)
|
||||
)
|
||||
or (
|
||||
block.get("type") == "advisor_tool_result"
|
||||
and block.get("tool_use_id") in advisor_ids
|
||||
)
|
||||
)
|
||||
advisor_id_to_text[block["tool_use_id"]] = text
|
||||
|
||||
new_content = []
|
||||
for block in content:
|
||||
if not isinstance(block, dict):
|
||||
new_content.append(block)
|
||||
continue
|
||||
is_advisor_use = (
|
||||
block.get("type") == "server_tool_use"
|
||||
and block.get("name") == "advisor"
|
||||
and block.get("id") in advisor_id_to_text
|
||||
)
|
||||
]
|
||||
is_advisor_result = (
|
||||
block.get("type") == "advisor_tool_result"
|
||||
and block.get("tool_use_id") in advisor_id_to_text
|
||||
)
|
||||
if is_advisor_use:
|
||||
if replace_with_text:
|
||||
advice = advisor_id_to_text.get(block.get("id")) or ""
|
||||
if advice:
|
||||
new_content.append(
|
||||
{
|
||||
"type": "text",
|
||||
"text": f"<advisor_feedback>\n{advice}\n</advisor_feedback>",
|
||||
}
|
||||
)
|
||||
# else: drop silently
|
||||
elif is_advisor_result:
|
||||
pass # always drop — replaced above (or stripped)
|
||||
else:
|
||||
new_content.append(block)
|
||||
|
||||
message["content"] = new_content
|
||||
return messages
|
||||
|
||||
|
||||
|
||||
@ -26,6 +26,7 @@ from litellm.utils import ProviderConfigManager, client
|
||||
|
||||
from ..adapters.handler import LiteLLMMessagesToCompletionTransformationHandler
|
||||
from ..responses_adapters.handler import LiteLLMMessagesToResponsesAPIHandler
|
||||
from .interceptors import get_messages_interceptors
|
||||
from .utils import AnthropicMessagesRequestUtils, mock_response
|
||||
|
||||
# Providers that are routed directly to the OpenAI Responses API instead of
|
||||
@ -236,6 +237,23 @@ async def anthropic_messages(
|
||||
if short_circuit_response is not None:
|
||||
return short_circuit_response
|
||||
|
||||
# Run registered MessagesInterceptors (e.g. advisor orchestration loop).
|
||||
# api_key and api_base are explicit params (not in **kwargs) so pass them
|
||||
# explicitly so interceptor sub-calls can route to the same backend.
|
||||
for interceptor in get_messages_interceptors():
|
||||
if interceptor.can_handle(tools, custom_llm_provider):
|
||||
return await interceptor.handle(
|
||||
model=model,
|
||||
messages=messages,
|
||||
tools=tools,
|
||||
stream=original_stream,
|
||||
max_tokens=max_tokens,
|
||||
custom_llm_provider=custom_llm_provider,
|
||||
api_key=api_key,
|
||||
api_base=api_base,
|
||||
**kwargs,
|
||||
)
|
||||
|
||||
loop = asyncio.get_event_loop()
|
||||
kwargs["is_async"] = True
|
||||
|
||||
|
||||
@ -0,0 +1,62 @@
|
||||
# Messages Interceptors
|
||||
|
||||
Interceptors are short-circuit handlers for the `/v1/messages` path. They run **before** the normal backend call and can fully replace it with their own response.
|
||||
|
||||
## When to add an interceptor
|
||||
|
||||
Use an interceptor when you need to **replace the backend call entirely** with your own logic — for example, running an orchestration loop, synthesizing a response from multiple sub-calls, or short-circuiting to a non-LLM backend.
|
||||
|
||||
Use a **pre-request hook** (`_execute_pre_request_hooks` / `CustomLogger.async_pre_request_hook`) instead when you only need to **mutate request parameters** (tools, stream flag, metadata) before the normal call proceeds.
|
||||
|
||||
| Scenario | Use |
|
||||
|---|---|
|
||||
| Replace the backend call with a loop or synthetic response | Interceptor |
|
||||
| Translate or strip tools before the call | Pre-request hook |
|
||||
| Feature that is always active (built-in LiteLLM behavior) | Interceptor |
|
||||
| Optional integration that operators register | `CustomLogger` callback |
|
||||
|
||||
## How to add a new interceptor
|
||||
|
||||
1. Create `your_feature.py` in this directory.
|
||||
2. Implement `MessagesInterceptor` from `base.py`:
|
||||
- `can_handle(tools, custom_llm_provider) -> bool` — return True when your interceptor owns this request.
|
||||
- `async handle(...) -> Union[AnthropicMessagesResponse, AsyncIterator]` — do your work and return the response.
|
||||
3. Register it in `__init__.py` by appending to `_interceptors`.
|
||||
|
||||
```python
|
||||
# your_feature.py
|
||||
from .base import MessagesInterceptor
|
||||
|
||||
class MyFeatureHandler(MessagesInterceptor):
|
||||
def can_handle(self, tools, custom_llm_provider):
|
||||
return some_condition(tools, custom_llm_provider)
|
||||
|
||||
async def handle(self, *, model, messages, tools, stream, max_tokens,
|
||||
custom_llm_provider, **kwargs):
|
||||
...
|
||||
return response
|
||||
```
|
||||
|
||||
```python
|
||||
# __init__.py
|
||||
from .your_feature import MyFeatureHandler
|
||||
|
||||
_interceptors = [
|
||||
AdvisorOrchestrationHandler(),
|
||||
MyFeatureHandler(), # add here
|
||||
]
|
||||
```
|
||||
|
||||
## Existing interceptors
|
||||
|
||||
### `AdvisorOrchestrationHandler`
|
||||
|
||||
Handles `advisor_20260301` tool for providers that don't support it natively (all non-Anthropic providers for now).
|
||||
|
||||
**Triggers when:** `advisor_20260301` is in `tools` AND `custom_llm_provider` is not in `ADVISOR_NATIVE_PROVIDERS`.
|
||||
|
||||
**What it does:**
|
||||
- Translates the advisor tool to a regular function tool the provider understands.
|
||||
- Runs the executor model; when it calls the `advisor` tool, runs the advisor model and injects the result as a `tool_result`.
|
||||
- Loops until the executor produces a final text response or `max_uses` is exceeded.
|
||||
- Wraps the final response in `FakeAnthropicMessagesStreamIterator` if the caller requested streaming.
|
||||
@ -0,0 +1,17 @@
|
||||
from typing import List
|
||||
|
||||
from .advisor import AdvisorOrchestrationHandler
|
||||
from .base import MessagesInterceptor
|
||||
|
||||
_interceptors: List[MessagesInterceptor] = [
|
||||
AdvisorOrchestrationHandler(),
|
||||
]
|
||||
|
||||
|
||||
def get_messages_interceptors() -> List[MessagesInterceptor]:
|
||||
"""Return the list of active MessagesInterceptors.
|
||||
|
||||
Order matters: interceptors are tried in list order; the first one whose
|
||||
``can_handle()`` returns True wins.
|
||||
"""
|
||||
return _interceptors
|
||||
@ -0,0 +1,351 @@
|
||||
"""
|
||||
Advisor Orchestration Handler
|
||||
|
||||
Implements the advisor tool loop for providers that don't support
|
||||
advisor_20260301 natively (i.e. everything except Anthropic direct for now).
|
||||
|
||||
How it works:
|
||||
1. Detects advisor_20260301 in tools + non-native provider → intercepts.
|
||||
2. Translates the advisor tool to a regular function tool the provider understands.
|
||||
3. Calls the executor model (non-streaming).
|
||||
4. If the executor makes a tool_use call named "advisor", runs the advisor model
|
||||
and injects the result as a tool_result before re-calling the executor.
|
||||
5. Repeats until the executor produces a final text response or max_uses is hit.
|
||||
6. Wraps in FakeAnthropicMessagesStreamIterator if the caller requested streaming.
|
||||
"""
|
||||
|
||||
import uuid
|
||||
from typing import Any, AsyncIterator, Dict, List, Optional, Union
|
||||
|
||||
import litellm.constants as _c
|
||||
from litellm.llms.anthropic.common_utils import strip_advisor_blocks_from_messages
|
||||
from litellm.types.llms.anthropic_messages.anthropic_response import (
|
||||
AnthropicMessagesResponse,
|
||||
)
|
||||
from litellm.types.llms.anthropic import ANTHROPIC_ADVISOR_TOOL_TYPE
|
||||
|
||||
ADVISOR_MAX_USES: int = _c.ADVISOR_MAX_USES
|
||||
ADVISOR_NATIVE_PROVIDERS: frozenset = _c.ADVISOR_NATIVE_PROVIDERS
|
||||
ADVISOR_TOOL_DESCRIPTION: str = _c.ADVISOR_TOOL_DESCRIPTION
|
||||
|
||||
from .base import MessagesInterceptor
|
||||
|
||||
|
||||
class AdvisorMaxIterationsError(Exception):
|
||||
"""Raised when the advisor loop exceeds max_uses."""
|
||||
|
||||
|
||||
class AdvisorOrchestrationHandler(MessagesInterceptor):
|
||||
"""Orchestrates the advisor tool loop for non-native providers."""
|
||||
|
||||
def can_handle(
|
||||
self,
|
||||
tools: Optional[List[Dict]],
|
||||
custom_llm_provider: Optional[str],
|
||||
) -> bool:
|
||||
if not tools:
|
||||
return False
|
||||
has_advisor = any(t.get("type") == ANTHROPIC_ADVISOR_TOOL_TYPE for t in tools)
|
||||
is_non_native = custom_llm_provider not in ADVISOR_NATIVE_PROVIDERS
|
||||
return has_advisor and is_non_native
|
||||
|
||||
async def handle(
|
||||
self,
|
||||
*,
|
||||
model: str,
|
||||
messages: List[Dict],
|
||||
tools: Optional[List[Dict]],
|
||||
stream: Optional[bool],
|
||||
max_tokens: int,
|
||||
custom_llm_provider: Optional[str],
|
||||
**kwargs,
|
||||
) -> Union[AnthropicMessagesResponse, AsyncIterator]:
|
||||
from litellm.llms.anthropic.experimental_pass_through.messages.fake_stream_iterator import (
|
||||
FakeAnthropicMessagesStreamIterator,
|
||||
)
|
||||
|
||||
# Extract advisor tool config.
|
||||
advisor_tool = next(
|
||||
(t for t in (tools or []) if t.get("type") == ANTHROPIC_ADVISOR_TOOL_TYPE),
|
||||
None,
|
||||
)
|
||||
if advisor_tool is None:
|
||||
raise ValueError(
|
||||
f"handle() called but no {ANTHROPIC_ADVISOR_TOOL_TYPE} tool found in tools list"
|
||||
)
|
||||
advisor_model: str = advisor_tool.get("model") or ""
|
||||
if not advisor_model:
|
||||
raise ValueError(
|
||||
"advisor tool definition must include a 'model' field specifying the advisor model"
|
||||
)
|
||||
_raw_max_uses = advisor_tool.get("max_uses")
|
||||
max_uses: int = ADVISOR_MAX_USES if _raw_max_uses is None else int(_raw_max_uses)
|
||||
# Optional routing overrides for the advisor sub-call (e.g. proxy routing).
|
||||
# If not set in the tool definition, litellm resolves from env vars.
|
||||
advisor_api_key: Optional[str] = advisor_tool.get("api_key")
|
||||
advisor_api_base: Optional[str] = advisor_tool.get("api_base")
|
||||
|
||||
# Build the synthetic tool definition the provider will receive.
|
||||
synthetic_advisor_tool = _make_synthetic_advisor_tool()
|
||||
|
||||
# Executor tools = all original tools with advisor replaced by the synthetic one.
|
||||
executor_tools: List[Dict] = [
|
||||
(
|
||||
synthetic_advisor_tool
|
||||
if t.get("type") == ANTHROPIC_ADVISOR_TOOL_TYPE
|
||||
else t
|
||||
)
|
||||
for t in (tools or [])
|
||||
]
|
||||
|
||||
# Strip prior advisor blocks from history, preserving advice text as context.
|
||||
current_messages: List[Dict] = strip_advisor_blocks_from_messages(
|
||||
[dict(m) for m in messages], replace_with_text=True
|
||||
)
|
||||
|
||||
parent_request_id: str = str(
|
||||
kwargs.pop("litellm_call_id", None) or uuid.uuid4()
|
||||
)
|
||||
metadata_base: Dict = dict(kwargs.pop("metadata", None) or {})
|
||||
iteration = 0
|
||||
|
||||
while True:
|
||||
# --- Executor call (always non-streaming) ---
|
||||
executor_response: AnthropicMessagesResponse = await _call_messages_handler(
|
||||
model=model,
|
||||
messages=current_messages,
|
||||
tools=executor_tools,
|
||||
stream=False,
|
||||
max_tokens=max_tokens,
|
||||
custom_llm_provider=custom_llm_provider,
|
||||
metadata={
|
||||
**metadata_base,
|
||||
"advisor_sub_call": False,
|
||||
"parent_request_id": parent_request_id,
|
||||
},
|
||||
**kwargs,
|
||||
)
|
||||
|
||||
advisor_use_block = _find_advisor_tool_use(executor_response)
|
||||
|
||||
if advisor_use_block is None:
|
||||
# No more advisor calls — this is the final response.
|
||||
if stream:
|
||||
return FakeAnthropicMessagesStreamIterator(executor_response)
|
||||
return executor_response
|
||||
|
||||
iteration += 1
|
||||
if iteration > max_uses:
|
||||
raise AdvisorMaxIterationsError(
|
||||
f"Advisor orchestration loop exceeded max_uses={max_uses}. "
|
||||
"Increase max_uses in the advisor tool definition or cap the request."
|
||||
)
|
||||
|
||||
# --- Build advisor context ---
|
||||
advisor_messages = _build_advisor_context(
|
||||
current_messages, executor_response, advisor_use_block
|
||||
)
|
||||
|
||||
# --- Advisor sub-call (always non-streaming, no tools) ---
|
||||
advisor_response: AnthropicMessagesResponse = await _call_messages_handler(
|
||||
model=advisor_model,
|
||||
messages=advisor_messages,
|
||||
tools=None,
|
||||
stream=False,
|
||||
max_tokens=max_tokens,
|
||||
custom_llm_provider=None, # let litellm resolve from model name
|
||||
metadata={
|
||||
**metadata_base,
|
||||
"advisor_sub_call": True,
|
||||
"parent_request_id": parent_request_id,
|
||||
},
|
||||
api_key=advisor_api_key,
|
||||
api_base=advisor_api_base,
|
||||
)
|
||||
|
||||
advisor_text = _extract_response_text(advisor_response)
|
||||
|
||||
# --- Inject advisor result and continue loop ---
|
||||
current_messages = _inject_advisor_turn(
|
||||
current_messages,
|
||||
executor_response,
|
||||
advisor_use_block,
|
||||
advisor_text,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _make_synthetic_advisor_tool() -> Dict:
|
||||
"""Build a regular tool definition the executor provider can understand."""
|
||||
return {
|
||||
"name": "advisor",
|
||||
"description": ADVISOR_TOOL_DESCRIPTION,
|
||||
"input_schema": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"question": {
|
||||
"type": "string",
|
||||
"description": "The question or challenge you want guidance on.",
|
||||
}
|
||||
},
|
||||
"required": ["question"],
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def _find_advisor_tool_use(response: Any) -> Optional[Dict]:
|
||||
"""Return the first tool_use block with name='advisor', or None."""
|
||||
content = response.get("content") if isinstance(response, dict) else []
|
||||
if not isinstance(content, list):
|
||||
return None
|
||||
for block in content:
|
||||
if (
|
||||
isinstance(block, dict)
|
||||
and block.get("type") == "tool_use"
|
||||
and block.get("name") == "advisor"
|
||||
):
|
||||
return block
|
||||
return None
|
||||
|
||||
|
||||
def _extract_response_text(response: Any) -> str:
|
||||
"""Extract concatenated text from all text blocks in a response."""
|
||||
content = response.get("content") if isinstance(response, dict) else []
|
||||
if not isinstance(content, list):
|
||||
return ""
|
||||
parts = [
|
||||
b.get("text", "")
|
||||
for b in content
|
||||
if isinstance(b, dict) and b.get("type") == "text"
|
||||
]
|
||||
return "\n".join(parts).strip()
|
||||
|
||||
|
||||
_PROVIDER_SPECIFIC_KEYS = frozenset({"provider_specific_fields"})
|
||||
|
||||
|
||||
def _build_advisor_context(
|
||||
messages: List[Dict],
|
||||
executor_response: Any,
|
||||
advisor_use_block: Dict,
|
||||
) -> List[Dict]:
|
||||
"""
|
||||
Build the message list for the advisor sub-call.
|
||||
|
||||
Passes the full conversation + any text the executor produced so far, then
|
||||
poses the advisor question as the last user turn.
|
||||
|
||||
tool_use blocks are excluded because Anthropic requires tool_use to be
|
||||
immediately followed by tool_result — not the advisor question.
|
||||
"""
|
||||
question = (advisor_use_block.get("input") or {}).get("question") or (
|
||||
"Please provide guidance on the current task."
|
||||
)
|
||||
raw_content = (
|
||||
executor_response.get("content") if isinstance(executor_response, dict) else []
|
||||
) or []
|
||||
# Keep only text blocks — strip tool_use and provider-specific fields.
|
||||
executor_text_blocks = [
|
||||
{k: v for k, v in block.items() if k not in _PROVIDER_SPECIFIC_KEYS}
|
||||
for block in raw_content
|
||||
if isinstance(block, dict) and block.get("type") == "text"
|
||||
]
|
||||
result = list(messages)
|
||||
if executor_text_blocks:
|
||||
result.append({"role": "assistant", "content": executor_text_blocks})
|
||||
result.append({"role": "user", "content": question})
|
||||
return result
|
||||
|
||||
|
||||
def _inject_advisor_turn(
|
||||
messages: List[Dict],
|
||||
executor_response: Any,
|
||||
advisor_use_block: Dict,
|
||||
advisor_text: str,
|
||||
) -> List[Dict]:
|
||||
"""
|
||||
Append the executor's response (as an assistant turn) and the advisor
|
||||
result (as a user tool_result turn) so the executor can continue.
|
||||
"""
|
||||
executor_content = (
|
||||
executor_response.get("content") if isinstance(executor_response, dict) else []
|
||||
) or []
|
||||
tool_use_id = advisor_use_block.get("id", "")
|
||||
return [
|
||||
*messages,
|
||||
{"role": "assistant", "content": executor_content},
|
||||
{
|
||||
"role": "user",
|
||||
"content": [
|
||||
{
|
||||
"type": "tool_result",
|
||||
"tool_use_id": tool_use_id,
|
||||
"content": advisor_text,
|
||||
}
|
||||
],
|
||||
},
|
||||
]
|
||||
|
||||
|
||||
def _inject_max_uses_error(
|
||||
messages: List[Dict],
|
||||
executor_response: Any,
|
||||
advisor_use_block: Dict,
|
||||
) -> List[Dict]:
|
||||
"""
|
||||
Inject a max_uses_exceeded error tool_result so the executor continues
|
||||
without further advisor calls (mirrors Anthropic's server-side behaviour).
|
||||
"""
|
||||
executor_content = (
|
||||
executor_response.get("content") if isinstance(executor_response, dict) else []
|
||||
) or []
|
||||
tool_use_id = advisor_use_block.get("id", "")
|
||||
return [
|
||||
*messages,
|
||||
{"role": "assistant", "content": executor_content},
|
||||
{
|
||||
"role": "user",
|
||||
"content": [
|
||||
{
|
||||
"type": "tool_result",
|
||||
"tool_use_id": tool_use_id,
|
||||
"content": "Advisor unavailable: max_uses limit reached. Continue without advisor guidance.",
|
||||
}
|
||||
],
|
||||
},
|
||||
]
|
||||
|
||||
|
||||
async def _call_messages_handler(
|
||||
model: str,
|
||||
messages: List[Dict],
|
||||
tools: Optional[List[Dict]],
|
||||
stream: bool,
|
||||
max_tokens: int,
|
||||
custom_llm_provider: Optional[str],
|
||||
**kwargs,
|
||||
) -> Any:
|
||||
"""
|
||||
Call anthropic_messages() — the public async /messages entry point — for
|
||||
orchestration sub-calls (executor or advisor).
|
||||
|
||||
Using the public function (decorated with @client) ensures logging, retries,
|
||||
and provider resolution all work correctly, identical to a direct user call.
|
||||
"""
|
||||
from litellm.llms.anthropic.experimental_pass_through.messages.handler import (
|
||||
anthropic_messages,
|
||||
)
|
||||
|
||||
return await anthropic_messages(
|
||||
model=model,
|
||||
messages=messages,
|
||||
tools=tools,
|
||||
stream=stream,
|
||||
max_tokens=max_tokens,
|
||||
custom_llm_provider=custom_llm_provider,
|
||||
**kwargs,
|
||||
)
|
||||
@ -0,0 +1,41 @@
|
||||
from abc import ABC, abstractmethod
|
||||
from typing import AsyncIterator, Dict, List, Optional, Union
|
||||
|
||||
from litellm.types.llms.anthropic_messages.anthropic_response import (
|
||||
AnthropicMessagesResponse,
|
||||
)
|
||||
|
||||
|
||||
class MessagesInterceptor(ABC):
|
||||
"""
|
||||
Base class for /messages short-circuit interceptors.
|
||||
|
||||
An interceptor can fully replace the normal backend call when it detects
|
||||
a pattern it owns (e.g. advisor orchestration, web-search short-circuit).
|
||||
``can_handle`` is checked first; if True, ``handle`` is called and its
|
||||
return value is returned directly to the caller.
|
||||
|
||||
See interceptors/README.md for when to add an interceptor vs. a pre-request hook.
|
||||
"""
|
||||
|
||||
@abstractmethod
|
||||
def can_handle(
|
||||
self,
|
||||
tools: Optional[List[Dict]],
|
||||
custom_llm_provider: Optional[str],
|
||||
) -> bool:
|
||||
"""Return True if this interceptor should handle the request."""
|
||||
|
||||
@abstractmethod
|
||||
async def handle(
|
||||
self,
|
||||
*,
|
||||
model: str,
|
||||
messages: List[Dict],
|
||||
tools: Optional[List[Dict]],
|
||||
stream: Optional[bool],
|
||||
max_tokens: int,
|
||||
custom_llm_provider: Optional[str],
|
||||
**kwargs,
|
||||
) -> Union[AnthropicMessagesResponse, AsyncIterator]:
|
||||
"""Execute the interception and return the response."""
|
||||
@ -0,0 +1,185 @@
|
||||
"""
|
||||
Integration tests for advisor orchestration through the full /messages handler.
|
||||
|
||||
These tests exercise the real dispatch path:
|
||||
anthropic_messages() → interceptor registry → AdvisorOrchestrationHandler.handle()
|
||||
|
||||
The only thing mocked is _call_messages_handler (the outbound LLM call), so the
|
||||
interceptor detection, loop logic, and message assembly all run for real.
|
||||
"""
|
||||
|
||||
from typing import Dict
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
ADVISOR_TOOL = {
|
||||
"type": "advisor_20260301",
|
||||
"name": "advisor",
|
||||
"model": "claude-opus-4-6",
|
||||
}
|
||||
|
||||
MESSAGES = [{"role": "user", "content": "Write a Python function to check if a number is prime."}]
|
||||
|
||||
|
||||
def _text_resp(text: str, model: str = "gpt-4o-mini") -> Dict:
|
||||
return {
|
||||
"id": "msg_int_test",
|
||||
"type": "message",
|
||||
"role": "assistant",
|
||||
"model": model,
|
||||
"content": [{"type": "text", "text": text}],
|
||||
"stop_reason": "end_turn",
|
||||
"usage": {"input_tokens": 10, "output_tokens": 20},
|
||||
}
|
||||
|
||||
|
||||
def _advisor_call_resp(question: str = "How do I approach this?", tool_id: str = "tid_01") -> Dict:
|
||||
return {
|
||||
"id": "msg_int_test",
|
||||
"type": "message",
|
||||
"role": "assistant",
|
||||
"model": "gpt-4o-mini",
|
||||
"content": [
|
||||
{
|
||||
"type": "tool_use",
|
||||
"id": tool_id,
|
||||
"name": "advisor",
|
||||
"input": {"question": question},
|
||||
}
|
||||
],
|
||||
"stop_reason": "tool_use",
|
||||
"usage": {"input_tokens": 10, "output_tokens": 15},
|
||||
}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 1. Full dispatch: interceptor fires and orchestration loop completes
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_full_dispatch_interceptor_fires_and_loop_completes():
|
||||
"""
|
||||
Call anthropic_messages() with an openai model + advisor_20260301 tool.
|
||||
The interceptor must fire, run the loop (1 advisor call), and return a
|
||||
clean final response with no advisor tool_use blocks.
|
||||
"""
|
||||
from litellm.llms.anthropic.experimental_pass_through.messages.handler import (
|
||||
anthropic_messages,
|
||||
)
|
||||
|
||||
call_count = 0
|
||||
|
||||
async def mock_handler(model, messages, tools, stream, max_tokens, **kwargs):
|
||||
nonlocal call_count
|
||||
call_count += 1
|
||||
if call_count == 1:
|
||||
return _advisor_call_resp() # executor: calls advisor
|
||||
if call_count == 2:
|
||||
return _text_resp("Use trial division.", model="claude-opus-4-6") # advisor
|
||||
return _text_resp("def is_prime(n): ...") # executor: final
|
||||
|
||||
with patch(
|
||||
"litellm.llms.anthropic.experimental_pass_through.messages.interceptors.advisor._call_messages_handler",
|
||||
side_effect=mock_handler,
|
||||
):
|
||||
result = await anthropic_messages(
|
||||
model="openai/gpt-4o-mini",
|
||||
messages=MESSAGES,
|
||||
tools=[ADVISOR_TOOL],
|
||||
stream=False,
|
||||
max_tokens=512,
|
||||
custom_llm_provider="openai",
|
||||
)
|
||||
|
||||
# 3 internal calls: executor → advisor → executor-final
|
||||
assert call_count == 3
|
||||
|
||||
assert isinstance(result, dict)
|
||||
content = result.get("content", [])
|
||||
text_blocks = [b for b in content if b.get("type") == "text"]
|
||||
advisor_uses = [b for b in content if b.get("type") == "tool_use" and b.get("name") == "advisor"]
|
||||
|
||||
assert len(text_blocks) >= 1, "Final response must have text"
|
||||
assert len(advisor_uses) == 0, "No advisor tool_use blocks must appear in final output"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 2. max_uses enforced through the full handler path
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_max_uses_enforced_through_full_handler():
|
||||
"""
|
||||
AdvisorMaxIterationsError propagates out of anthropic_messages() when
|
||||
the executor keeps calling the advisor past max_uses.
|
||||
"""
|
||||
from litellm.llms.anthropic.experimental_pass_through.messages.handler import (
|
||||
anthropic_messages,
|
||||
)
|
||||
from litellm.llms.anthropic.experimental_pass_through.messages.interceptors.advisor import (
|
||||
AdvisorMaxIterationsError,
|
||||
)
|
||||
|
||||
advisor_tool_capped = {**ADVISOR_TOOL, "max_uses": 1}
|
||||
|
||||
async def mock_handler(model, messages, tools, stream, max_tokens, **kwargs):
|
||||
# Advisor always returns text; executor always calls advisor
|
||||
if tools is None:
|
||||
return _text_resp("Some advice.", model="claude-opus-4-6")
|
||||
return _advisor_call_resp()
|
||||
|
||||
with patch(
|
||||
"litellm.llms.anthropic.experimental_pass_through.messages.interceptors.advisor._call_messages_handler",
|
||||
side_effect=mock_handler,
|
||||
):
|
||||
with pytest.raises(AdvisorMaxIterationsError):
|
||||
await anthropic_messages(
|
||||
model="openai/gpt-4o-mini",
|
||||
messages=MESSAGES,
|
||||
tools=[advisor_tool_capped],
|
||||
stream=False,
|
||||
max_tokens=512,
|
||||
custom_llm_provider="openai",
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 3. Anthropic provider bypasses interceptor — no orchestration loop runs
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_anthropic_provider_bypasses_interceptor():
|
||||
"""
|
||||
With custom_llm_provider='anthropic', the interceptor must NOT fire.
|
||||
The advisor_20260301 tool is forwarded as-is to the underlying handler.
|
||||
"""
|
||||
from litellm.llms.anthropic.experimental_pass_through.messages.handler import (
|
||||
anthropic_messages,
|
||||
)
|
||||
|
||||
direct_response = _text_resp("Native anthropic response.")
|
||||
|
||||
# Patch the non-interceptor code path — anthropic_messages_handler
|
||||
with patch(
|
||||
"litellm.llms.anthropic.experimental_pass_through.messages.handler.anthropic_messages_handler",
|
||||
return_value=direct_response,
|
||||
) as mock_native:
|
||||
result = await anthropic_messages(
|
||||
model="claude-sonnet-4-6",
|
||||
messages=MESSAGES,
|
||||
tools=[ADVISOR_TOOL],
|
||||
stream=False,
|
||||
max_tokens=512,
|
||||
custom_llm_provider="anthropic",
|
||||
)
|
||||
|
||||
# Native handler was called (not the orchestration loop)
|
||||
mock_native.assert_called_once()
|
||||
# Response passes through unmodified
|
||||
content = result.get("content", []) if isinstance(result, dict) else []
|
||||
text_blocks = [b for b in content if b.get("type") == "text"]
|
||||
assert any("Native anthropic" in b.get("text", "") for b in text_blocks)
|
||||
@ -0,0 +1,518 @@
|
||||
"""
|
||||
Tests for advisor orchestration on non-Anthropic providers.
|
||||
|
||||
Tests:
|
||||
1. can_handle edge cases
|
||||
2. Anthropic native: interceptor does NOT trigger (routing confirmed)
|
||||
3. Orchestration loop logic (mocked backend): single advisor call, multi-turn, max_uses cap
|
||||
4. strip_advisor_blocks_from_messages with replace_with_text=True
|
||||
"""
|
||||
|
||||
from typing import Dict
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
ADVISOR_TOOL = {
|
||||
"type": "advisor_20260301",
|
||||
"name": "advisor",
|
||||
"model": "claude-opus-4-6",
|
||||
}
|
||||
|
||||
MESSAGES = [
|
||||
{
|
||||
"role": "user",
|
||||
"content": "Write a Python function that checks if a number is prime.",
|
||||
}
|
||||
]
|
||||
|
||||
|
||||
def _make_text_response(text: str, model: str = "openai/gpt-4o-mini") -> Dict:
|
||||
return {
|
||||
"id": "msg_test",
|
||||
"type": "message",
|
||||
"role": "assistant",
|
||||
"model": model,
|
||||
"content": [{"type": "text", "text": text}],
|
||||
"stop_reason": "end_turn",
|
||||
"usage": {"input_tokens": 10, "output_tokens": 20},
|
||||
}
|
||||
|
||||
|
||||
def _make_advisor_tool_use_response(
|
||||
question: str = "How should I approach this?",
|
||||
tool_id: str = "toolu_advisor_01",
|
||||
model: str = "openai/gpt-4o-mini",
|
||||
) -> Dict:
|
||||
return {
|
||||
"id": "msg_test",
|
||||
"type": "message",
|
||||
"role": "assistant",
|
||||
"model": model,
|
||||
"content": [
|
||||
{
|
||||
"type": "tool_use",
|
||||
"id": tool_id,
|
||||
"name": "advisor",
|
||||
"input": {"question": question},
|
||||
}
|
||||
],
|
||||
"stop_reason": "tool_use",
|
||||
"usage": {"input_tokens": 10, "output_tokens": 15},
|
||||
}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 1. can_handle edge cases
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_can_handle_edge_cases():
|
||||
from litellm.llms.anthropic.experimental_pass_through.messages.interceptors.advisor import (
|
||||
AdvisorOrchestrationHandler,
|
||||
)
|
||||
|
||||
h = AdvisorOrchestrationHandler()
|
||||
|
||||
assert h.can_handle([ADVISOR_TOOL], "openai")
|
||||
assert h.can_handle([ADVISOR_TOOL], "bedrock")
|
||||
assert h.can_handle([ADVISOR_TOOL], "gemini")
|
||||
assert not h.can_handle([ADVISOR_TOOL], "anthropic")
|
||||
assert not h.can_handle([], "openai")
|
||||
assert not h.can_handle(None, "openai")
|
||||
assert not h.can_handle([{"type": "function", "name": "bash"}], "openai")
|
||||
# provider=None: unknown → should intercept (treat as non-native)
|
||||
assert h.can_handle([ADVISOR_TOOL], None)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 2. Anthropic native: interceptor must NOT trigger
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_anthropic_native_interceptor_skipped():
|
||||
"""
|
||||
For provider=anthropic, can_handle() must return False.
|
||||
The interceptor must never call handle().
|
||||
"""
|
||||
from litellm.llms.anthropic.experimental_pass_through.messages.interceptors.advisor import (
|
||||
AdvisorOrchestrationHandler,
|
||||
)
|
||||
|
||||
h = AdvisorOrchestrationHandler()
|
||||
assert not h.can_handle(
|
||||
[ADVISOR_TOOL], "anthropic"
|
||||
), "Interceptor must NOT trigger for anthropic provider"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 3. Orchestration loop: no advisor call needed (executor returns text directly)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_loop_no_advisor_call():
|
||||
"""Executor returns text on first try — no advisor call, loop exits immediately."""
|
||||
from litellm.llms.anthropic.experimental_pass_through.messages.interceptors.advisor import (
|
||||
AdvisorOrchestrationHandler,
|
||||
_call_messages_handler,
|
||||
)
|
||||
|
||||
final_text = "def is_prime(n): return n > 1 and all(n % i for i in range(2, n))"
|
||||
executor_response = _make_text_response(final_text)
|
||||
|
||||
with patch(
|
||||
"litellm.llms.anthropic.experimental_pass_through.messages.interceptors.advisor._call_messages_handler",
|
||||
new_callable=AsyncMock,
|
||||
return_value=executor_response,
|
||||
) as mock_call:
|
||||
h = AdvisorOrchestrationHandler()
|
||||
result = await h.handle(
|
||||
model="openai/gpt-4o-mini",
|
||||
messages=MESSAGES,
|
||||
tools=[ADVISOR_TOOL],
|
||||
stream=False,
|
||||
max_tokens=512,
|
||||
custom_llm_provider="openai",
|
||||
)
|
||||
|
||||
# Only one call (executor), no advisor call
|
||||
assert mock_call.call_count == 1
|
||||
content = result.get("content", [])
|
||||
texts = [b for b in content if b.get("type") == "text"]
|
||||
assert len(texts) == 1
|
||||
assert final_text in texts[0]["text"]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 4. Orchestration loop: one advisor call then final text
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_loop_one_advisor_call():
|
||||
"""
|
||||
Executor calls advisor once → advisor responds → executor produces final text.
|
||||
Total calls: 3 (executor, advisor, executor-final).
|
||||
"""
|
||||
from litellm.llms.anthropic.experimental_pass_through.messages.interceptors.advisor import (
|
||||
AdvisorOrchestrationHandler,
|
||||
)
|
||||
|
||||
advisor_tool_use_resp = _make_advisor_tool_use_response(
|
||||
question="Should I use a sieve or trial division?",
|
||||
tool_id="toolu_01",
|
||||
)
|
||||
advisor_advice_resp = _make_text_response(
|
||||
"Use trial division for simplicity — only check up to sqrt(n).",
|
||||
model="claude-opus-4-6",
|
||||
)
|
||||
final_resp = _make_text_response(
|
||||
"def is_prime(n):\n import math\n if n < 2: return False\n for i in range(2, int(math.sqrt(n))+1):\n if n % i == 0: return False\n return True"
|
||||
)
|
||||
|
||||
call_count = 0
|
||||
|
||||
async def mock_call(model, messages, tools, stream, max_tokens, **kwargs):
|
||||
nonlocal call_count
|
||||
call_count += 1
|
||||
if call_count == 1:
|
||||
return advisor_tool_use_resp # executor: calls advisor
|
||||
if call_count == 2:
|
||||
return advisor_advice_resp # advisor: returns advice
|
||||
return final_resp # executor: final answer
|
||||
|
||||
with patch(
|
||||
"litellm.llms.anthropic.experimental_pass_through.messages.interceptors.advisor._call_messages_handler",
|
||||
side_effect=mock_call,
|
||||
):
|
||||
h = AdvisorOrchestrationHandler()
|
||||
result = await h.handle(
|
||||
model="openai/gpt-4o-mini",
|
||||
messages=MESSAGES,
|
||||
tools=[ADVISOR_TOOL],
|
||||
stream=False,
|
||||
max_tokens=512,
|
||||
custom_llm_provider="openai",
|
||||
)
|
||||
|
||||
assert call_count == 3
|
||||
content = result.get("content", [])
|
||||
texts = [b for b in content if b.get("type") == "text"]
|
||||
assert len(texts) == 1
|
||||
assert "is_prime" in texts[0]["text"]
|
||||
|
||||
# No advisor tool_use blocks in final response
|
||||
advisor_uses = [
|
||||
b for b in content if b.get("type") == "tool_use" and b.get("name") == "advisor"
|
||||
]
|
||||
assert len(advisor_uses) == 0
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 5. max_uses cap
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_loop_max_uses_raises():
|
||||
"""Loop exceeding max_uses must raise AdvisorMaxIterationsError."""
|
||||
from litellm.llms.anthropic.experimental_pass_through.messages.interceptors.advisor import (
|
||||
AdvisorMaxIterationsError,
|
||||
AdvisorOrchestrationHandler,
|
||||
)
|
||||
|
||||
advisor_tool_with_max = {**ADVISOR_TOOL, "max_uses": 2}
|
||||
# Always return an advisor tool_use → loop never terminates naturally
|
||||
advisor_tool_use_resp = _make_advisor_tool_use_response()
|
||||
advisor_advice_resp = _make_text_response("Here is my advice.")
|
||||
|
||||
call_count = 0
|
||||
|
||||
async def mock_call(model, messages, tools, stream, max_tokens, **kwargs):
|
||||
nonlocal call_count
|
||||
call_count += 1
|
||||
# Executor calls always return advisor tool_use; advisor always returns text
|
||||
if tools is None:
|
||||
return advisor_advice_resp
|
||||
return advisor_tool_use_resp
|
||||
|
||||
with patch(
|
||||
"litellm.llms.anthropic.experimental_pass_through.messages.interceptors.advisor._call_messages_handler",
|
||||
side_effect=mock_call,
|
||||
):
|
||||
h = AdvisorOrchestrationHandler()
|
||||
with pytest.raises(AdvisorMaxIterationsError):
|
||||
await h.handle(
|
||||
model="openai/gpt-4o-mini",
|
||||
messages=MESSAGES,
|
||||
tools=[advisor_tool_with_max],
|
||||
stream=False,
|
||||
max_tokens=512,
|
||||
custom_llm_provider="openai",
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 6. Streaming: final response wrapped in FakeAnthropicMessagesStreamIterator
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_loop_streaming_wraps_response():
|
||||
"""stream=True: final response must be wrapped in FakeAnthropicMessagesStreamIterator."""
|
||||
from litellm.llms.anthropic.experimental_pass_through.messages.fake_stream_iterator import (
|
||||
FakeAnthropicMessagesStreamIterator,
|
||||
)
|
||||
from litellm.llms.anthropic.experimental_pass_through.messages.interceptors.advisor import (
|
||||
AdvisorOrchestrationHandler,
|
||||
)
|
||||
|
||||
executor_response = _make_text_response("Hello, world!")
|
||||
|
||||
with patch(
|
||||
"litellm.llms.anthropic.experimental_pass_through.messages.interceptors.advisor._call_messages_handler",
|
||||
new_callable=AsyncMock,
|
||||
return_value=executor_response,
|
||||
):
|
||||
h = AdvisorOrchestrationHandler()
|
||||
result = await h.handle(
|
||||
model="openai/gpt-4o-mini",
|
||||
messages=MESSAGES,
|
||||
tools=[ADVISOR_TOOL],
|
||||
stream=True,
|
||||
max_tokens=512,
|
||||
custom_llm_provider="openai",
|
||||
)
|
||||
|
||||
assert isinstance(result, FakeAnthropicMessagesStreamIterator)
|
||||
|
||||
chunks = []
|
||||
async for chunk in result:
|
||||
chunks.append(chunk)
|
||||
|
||||
assert len(chunks) > 0
|
||||
first = chunks[0].decode() if isinstance(chunks[0], bytes) else str(chunks[0])
|
||||
assert "message_start" in first
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 7. Multi-turn: prior advisor blocks replaced with text in history
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_prior_advisor_blocks_replaced_in_history():
|
||||
"""
|
||||
History containing server_tool_use + advisor_tool_result blocks gets
|
||||
collapsed to <advisor_feedback> text before forwarding to the executor.
|
||||
"""
|
||||
from litellm.llms.anthropic.experimental_pass_through.messages.interceptors.advisor import (
|
||||
AdvisorOrchestrationHandler,
|
||||
)
|
||||
|
||||
messages_with_history = [
|
||||
*MESSAGES,
|
||||
{
|
||||
"role": "assistant",
|
||||
"content": [
|
||||
{
|
||||
"type": "server_tool_use",
|
||||
"id": "srvtool_01",
|
||||
"name": "advisor",
|
||||
"input": {},
|
||||
},
|
||||
{
|
||||
"type": "advisor_tool_result",
|
||||
"tool_use_id": "srvtool_01",
|
||||
"content": "Use trial division up to sqrt(n).",
|
||||
},
|
||||
{"type": "text", "text": "I will now write the function."},
|
||||
],
|
||||
},
|
||||
{"role": "user", "content": "Actually make it more efficient."},
|
||||
]
|
||||
|
||||
captured_messages = []
|
||||
|
||||
async def mock_call(model, messages, tools, stream, max_tokens, **kwargs):
|
||||
captured_messages.extend(messages)
|
||||
return _make_text_response("Here is the efficient version.")
|
||||
|
||||
with patch(
|
||||
"litellm.llms.anthropic.experimental_pass_through.messages.interceptors.advisor._call_messages_handler",
|
||||
side_effect=mock_call,
|
||||
):
|
||||
h = AdvisorOrchestrationHandler()
|
||||
await h.handle(
|
||||
model="openai/gpt-4o-mini",
|
||||
messages=messages_with_history,
|
||||
tools=[ADVISOR_TOOL],
|
||||
stream=False,
|
||||
max_tokens=512,
|
||||
custom_llm_provider="openai",
|
||||
)
|
||||
|
||||
# Find the assistant message in forwarded history
|
||||
assistant_msgs = [m for m in captured_messages if m.get("role") == "assistant"]
|
||||
assert len(assistant_msgs) >= 1
|
||||
content = assistant_msgs[0].get("content", [])
|
||||
types = [b.get("type") for b in content if isinstance(b, dict)]
|
||||
|
||||
# server_tool_use and advisor_tool_result must be gone
|
||||
assert "server_tool_use" not in types
|
||||
assert "advisor_tool_result" not in types
|
||||
|
||||
# Text block with advisor feedback must be present
|
||||
text_blocks = [b for b in content if b.get("type") == "text"]
|
||||
feedback_blocks = [
|
||||
b for b in text_blocks if "advisor_feedback" in b.get("text", "")
|
||||
]
|
||||
assert len(feedback_blocks) >= 1
|
||||
assert "trial division" in feedback_blocks[0]["text"]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 8. Advisor tool is translated to a regular tool for the executor
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_advisor_tool_translated_for_executor():
|
||||
"""
|
||||
The executor must receive a regular tool definition (not advisor_20260301 type).
|
||||
"""
|
||||
from litellm.llms.anthropic.experimental_pass_through.messages.interceptors.advisor import (
|
||||
AdvisorOrchestrationHandler,
|
||||
)
|
||||
|
||||
captured_tools = []
|
||||
|
||||
async def mock_call(model, messages, tools, stream, max_tokens, **kwargs):
|
||||
if tools:
|
||||
captured_tools.extend(tools)
|
||||
return _make_text_response("Done.")
|
||||
|
||||
with patch(
|
||||
"litellm.llms.anthropic.experimental_pass_through.messages.interceptors.advisor._call_messages_handler",
|
||||
side_effect=mock_call,
|
||||
):
|
||||
h = AdvisorOrchestrationHandler()
|
||||
await h.handle(
|
||||
model="openai/gpt-4o-mini",
|
||||
messages=MESSAGES,
|
||||
tools=[ADVISOR_TOOL],
|
||||
stream=False,
|
||||
max_tokens=512,
|
||||
custom_llm_provider="openai",
|
||||
)
|
||||
|
||||
assert len(captured_tools) > 0
|
||||
advisor_tool = next(t for t in captured_tools if t.get("name") == "advisor")
|
||||
# Must NOT have the advisor_20260301 type (provider won't understand it)
|
||||
assert advisor_tool.get("type") != "advisor_20260301"
|
||||
# Must have a description and input_schema
|
||||
assert "description" in advisor_tool
|
||||
assert "input_schema" in advisor_tool
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 9. max_uses=0 means zero advisor calls allowed — first call raises immediately
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_max_uses_zero_raises_on_first_advisor_call():
|
||||
"""max_uses=0 must cause AdvisorMaxIterationsError on the first advisor call."""
|
||||
from litellm.llms.anthropic.experimental_pass_through.messages.interceptors.advisor import (
|
||||
AdvisorMaxIterationsError,
|
||||
AdvisorOrchestrationHandler,
|
||||
)
|
||||
|
||||
advisor_tool_with_zero = {**ADVISOR_TOOL, "max_uses": 0}
|
||||
advisor_tool_use_resp = _make_advisor_tool_use_response()
|
||||
|
||||
async def mock_call(model, messages, tools, stream, max_tokens, **kwargs):
|
||||
return advisor_tool_use_resp # executor always tries to call advisor
|
||||
|
||||
with patch(
|
||||
"litellm.llms.anthropic.experimental_pass_through.messages.interceptors.advisor._call_messages_handler",
|
||||
side_effect=mock_call,
|
||||
):
|
||||
h = AdvisorOrchestrationHandler()
|
||||
with pytest.raises(AdvisorMaxIterationsError):
|
||||
await h.handle(
|
||||
model="openai/gpt-4o-mini",
|
||||
messages=MESSAGES,
|
||||
tools=[advisor_tool_with_zero],
|
||||
stream=False,
|
||||
max_tokens=512,
|
||||
custom_llm_provider="openai",
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 10. Missing model in advisor tool definition raises ValueError from handle()
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_missing_advisor_model_raises_value_error():
|
||||
"""handle() must raise ValueError when the advisor tool has no model field."""
|
||||
from litellm.llms.anthropic.experimental_pass_through.messages.interceptors.advisor import (
|
||||
AdvisorOrchestrationHandler,
|
||||
)
|
||||
|
||||
advisor_tool_no_model = {"type": "advisor_20260301", "name": "advisor"}
|
||||
|
||||
h = AdvisorOrchestrationHandler()
|
||||
with pytest.raises(ValueError, match="model"):
|
||||
await h.handle(
|
||||
model="openai/gpt-4o-mini",
|
||||
messages=MESSAGES,
|
||||
tools=[advisor_tool_no_model],
|
||||
stream=False,
|
||||
max_tokens=512,
|
||||
custom_llm_provider="openai",
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 11. max_uses not set → falls back to ADVISOR_MAX_USES default
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_max_uses_none_falls_back_to_default():
|
||||
"""When max_uses is absent, the handler uses ADVISOR_MAX_USES from constants."""
|
||||
import litellm.constants as _c
|
||||
from litellm.llms.anthropic.experimental_pass_through.messages.interceptors.advisor import (
|
||||
AdvisorMaxIterationsError,
|
||||
AdvisorOrchestrationHandler,
|
||||
)
|
||||
|
||||
advisor_tool_use_resp = _make_advisor_tool_use_response()
|
||||
advisor_advice_resp = _make_text_response("Here is advice.")
|
||||
|
||||
async def mock_call(model, messages, tools, stream, max_tokens, **kwargs):
|
||||
if tools is None:
|
||||
return advisor_advice_resp
|
||||
return advisor_tool_use_resp
|
||||
|
||||
with patch(
|
||||
"litellm.llms.anthropic.experimental_pass_through.messages.interceptors.advisor._call_messages_handler",
|
||||
side_effect=mock_call,
|
||||
):
|
||||
h = AdvisorOrchestrationHandler()
|
||||
with pytest.raises(AdvisorMaxIterationsError) as exc_info:
|
||||
await h.handle(
|
||||
model="openai/gpt-4o-mini",
|
||||
messages=MESSAGES,
|
||||
tools=[ADVISOR_TOOL], # no max_uses — should use default
|
||||
stream=False,
|
||||
max_tokens=512,
|
||||
custom_llm_provider="openai",
|
||||
)
|
||||
|
||||
assert str(_c.ADVISOR_MAX_USES) in str(exc_info.value)
|
||||
Loading…
Reference in New Issue
Block a user