mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-07 02:59:05 +00:00
fix(anthropic): preserve mid-turn system messages
Generated with AI Co-Authored-By: Claude Code
This commit is contained in:
eugene-yao-zocdoc
parent
f54f92437b
commit
bae58eb4e0
9 changed files with 1380 additions and 94 deletions
|
|
@ -13,6 +13,7 @@ Pattern Overview:
|
|||
"""
|
||||
|
||||
import json
|
||||
from copy import deepcopy
|
||||
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple, cast
|
||||
|
||||
from litellm._logging import verbose_proxy_logger
|
||||
|
|
@ -24,7 +25,6 @@ from litellm.llms.base_llm.guardrail_translation.base_translation import BaseTra
|
|||
from litellm.llms.base_llm.guardrail_translation.utils import (
|
||||
effective_skip_system_message_for_guardrail,
|
||||
effective_skip_tool_message_for_guardrail,
|
||||
openai_messages_without_system,
|
||||
openai_messages_without_tool,
|
||||
)
|
||||
from litellm.proxy.pass_through_endpoints.llm_provider_handlers.anthropic_passthrough_logging_handler import (
|
||||
|
|
@ -59,14 +59,10 @@ if TYPE_CHECKING:
|
|||
|
||||
|
||||
class AnthropicMessagesHandler(BaseTranslation):
|
||||
"""
|
||||
Handler for processing Anthropic messages with guardrails.
|
||||
"""Process Anthropic messages with guardrails.
|
||||
|
||||
This class provides methods to:
|
||||
1. Process input messages (pre-call hook)
|
||||
2. Process output responses (post-call hook)
|
||||
|
||||
Methods can be overridden to customize behavior for different message formats.
|
||||
In-sequence system entries are untrusted client input. This handler scans and preserves
|
||||
them through guardrail rewrites; downstream provider handling is out of scope.
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
|
|
@ -279,14 +275,26 @@ class AnthropicMessagesHandler(BaseTranslation):
|
|||
skip_system = effective_skip_system_message_for_guardrail(guardrail_to_apply)
|
||||
skip_tool = effective_skip_tool_message_for_guardrail(guardrail_to_apply)
|
||||
|
||||
chat_completion_compatible_request = self._translate_to_openai(data)
|
||||
# Exclude only the trusted top-level prompt. In-sequence system entries are untrusted
|
||||
# and must stay aligned with texts_to_check for positional masking. When the top-level
|
||||
# prompt is included, the pre-existing count mismatch disables positional masking.
|
||||
translation_source = { # mutable-ok: API message payload
|
||||
key: value for key, value in data.items() if key != "system"
|
||||
} # mutable-ok: API message payload
|
||||
chat_completion_compatible_request = self._translate_to_openai(translation_source)
|
||||
|
||||
structured_messages = cast(
|
||||
List[AllMessageValues],
|
||||
chat_completion_compatible_request.get("messages", []),
|
||||
)
|
||||
if skip_system:
|
||||
structured_messages = openai_messages_without_system(structured_messages)
|
||||
has_midturn_system_message = any(
|
||||
str(message.get("role") or "").lower() == "system" for message in structured_messages
|
||||
)
|
||||
hoisted_system_message: AllMessageValues | None = None
|
||||
if not skip_system:
|
||||
hoisted_system_message = self._hoisted_top_level_system_message(data)
|
||||
if hoisted_system_message is not None:
|
||||
structured_messages.insert(0, hoisted_system_message)
|
||||
if skip_tool:
|
||||
structured_messages = openai_messages_without_tool(structured_messages)
|
||||
|
||||
|
|
@ -346,7 +354,12 @@ class AnthropicMessagesHandler(BaseTranslation):
|
|||
guardrailed_structured_messages is not None
|
||||
and guardrailed_structured_messages is not original_structured_messages
|
||||
):
|
||||
self._write_back_structured_messages(data, guardrailed_structured_messages)
|
||||
self._write_back_structured_messages(
|
||||
data,
|
||||
guardrailed_structured_messages,
|
||||
hoisted_system_message=hoisted_system_message,
|
||||
preserve_system_messages=has_midturn_system_message,
|
||||
)
|
||||
else:
|
||||
# Step 3: Map guardrail responses back to original message structure
|
||||
await self._apply_guardrail_responses_to_input(
|
||||
|
|
@ -359,36 +372,120 @@ class AnthropicMessagesHandler(BaseTranslation):
|
|||
|
||||
return data
|
||||
|
||||
@staticmethod
|
||||
def _write_back_structured_messages(data: dict, structured_messages: list) -> None:
|
||||
"""Convert compressed structured_messages back to Anthropic format and write to data.
|
||||
def _hoisted_top_level_system_message(
|
||||
self, data: dict
|
||||
) -> AllMessageValues | None: # mutable-ok: API message payload
|
||||
"""Return the system message produced by translating the top-level prompt."""
|
||||
system = data.get("system")
|
||||
if not system:
|
||||
return None
|
||||
probe = self._translate_to_openai(
|
||||
{ # mutable-ok: API message payload
|
||||
"model": data.get("model") or "",
|
||||
"messages": [], # mutable-ok: API message payload
|
||||
"system": system,
|
||||
}
|
||||
)
|
||||
hoisted = probe.get("messages") or [] # mutable-ok: API message payload
|
||||
return hoisted[0] if hoisted else None
|
||||
|
||||
``anthropic_messages_pt`` merges every run of consecutive user/tool rows
|
||||
into a single message, so a turn carrying only tool results and the user
|
||||
turn that follows it come back fused, and the request the model sees no
|
||||
longer has the boundaries the client sent. Converting a row at a time
|
||||
would keep them apart but breaks tool pairing: an assistant row whose
|
||||
tool results sit outside its own call reads as an orphaned tool call,
|
||||
and under ``modify_params`` the sanitizer answers it with a synthetic
|
||||
"tool execution skipped" result and drops the real one. Converting each
|
||||
assistant row together with the tool rows that answer it, and every
|
||||
other row on its own, satisfies both.
|
||||
"""
|
||||
@staticmethod
|
||||
def _openai_system_message_to_anthropic(
|
||||
message: dict[str, Any],
|
||||
) -> dict[str, Any] | None: # mutable-ok: API message payload
|
||||
"""Convert an OpenAI system message to the client's Anthropic-shaped entry."""
|
||||
content = message.get("content")
|
||||
if isinstance(content, str):
|
||||
return (
|
||||
{"role": "system", "content": content} if content else None # mutable-ok: API message payload
|
||||
) # mutable-ok: API message payload
|
||||
if not isinstance(content, list):
|
||||
return None
|
||||
blocks: list[dict[str, Any]] = [] # mutable-ok: API message payload
|
||||
for block in content:
|
||||
if not isinstance(block, dict) or block.get("type") != "text":
|
||||
continue
|
||||
text = block.get("text")
|
||||
if not isinstance(text, str) or not text:
|
||||
continue
|
||||
anthropic_block: dict[str, Any] = { # mutable-ok: API message payload
|
||||
"type": "text",
|
||||
"text": text,
|
||||
} # mutable-ok: API message payload
|
||||
cache_control = block.get("cache_control")
|
||||
if cache_control:
|
||||
anthropic_block["cache_control"] = deepcopy(cache_control)
|
||||
blocks.append(anthropic_block)
|
||||
return (
|
||||
{"role": "system", "content": blocks} if blocks else None # mutable-ok: API message payload
|
||||
) # mutable-ok: API message payload
|
||||
|
||||
@staticmethod
|
||||
def _is_hoisted_top_level_system(message: Any, hoisted_system_message: Any) -> bool:
|
||||
"""Match the hoisted prompt by identity, or by value after serialization."""
|
||||
if hoisted_system_message is None:
|
||||
return False
|
||||
if message is hoisted_system_message:
|
||||
return True
|
||||
return (
|
||||
isinstance(message, dict) and isinstance(hoisted_system_message, dict) and message == hoisted_system_message
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _write_back_structured_messages(
|
||||
data: dict, # mutable-ok: API message payload
|
||||
structured_messages: list, # mutable-ok: API message payload
|
||||
hoisted_system_message: Any = None,
|
||||
preserve_system_messages: bool = False,
|
||||
) -> None:
|
||||
"""Write a guardrail's structured-message rewrite back without losing corrections."""
|
||||
from litellm.litellm_core_utils.prompt_templates.factory import (
|
||||
anthropic_messages_pt,
|
||||
group_tool_exchanges,
|
||||
)
|
||||
|
||||
def _is_system(message: Any) -> bool:
|
||||
return isinstance(message, dict) and str(message.get("role") or "").lower() == "system"
|
||||
|
||||
model = str(data.get("model") or "")
|
||||
non_system = [m for m in structured_messages if m.get("role") != "system"]
|
||||
groups = tuple([non_system[index] for index in group] for group in group_tool_exchanges(non_system)) or (
|
||||
non_system,
|
||||
)
|
||||
converted = [
|
||||
message
|
||||
for group in groups
|
||||
for message in anthropic_messages_pt(messages=group, model=model, llm_provider="anthropic")
|
||||
]
|
||||
converted: list = [] # mutable-ok: API message payload
|
||||
|
||||
def _convert_run(run: list) -> None: # mutable-ok: API message payload
|
||||
for group in group_tool_exchanges(run):
|
||||
converted.extend(
|
||||
anthropic_messages_pt(
|
||||
messages=[ # mutable-ok: API message payload
|
||||
run[index] for index in group
|
||||
], # mutable-ok: API message payload
|
||||
model=model,
|
||||
llm_provider="anthropic",
|
||||
)
|
||||
)
|
||||
|
||||
run: list = [] # mutable-ok: API message payload
|
||||
hoisted_dropped = False
|
||||
for message in structured_messages:
|
||||
if not _is_system(message):
|
||||
run.append(message)
|
||||
continue
|
||||
_convert_run(run)
|
||||
run = [] # mutable-ok: API message payload
|
||||
if not hoisted_dropped and AnthropicMessagesHandler._is_hoisted_top_level_system(
|
||||
message, hoisted_system_message
|
||||
):
|
||||
hoisted_dropped = True
|
||||
continue
|
||||
if preserve_system_messages:
|
||||
anthropic_system = AnthropicMessagesHandler._openai_system_message_to_anthropic(message)
|
||||
if anthropic_system is not None:
|
||||
converted.append(anthropic_system)
|
||||
_convert_run(run)
|
||||
if not any(not _is_system(message) for message in converted):
|
||||
converted.extend(
|
||||
anthropic_messages_pt(
|
||||
messages=[], model=model, llm_provider="anthropic"
|
||||
) # mutable-ok: API message payload
|
||||
) # mutable-ok: API message payload
|
||||
for msg in converted:
|
||||
content = msg.get("content")
|
||||
if isinstance(content, list):
|
||||
|
|
@ -397,6 +494,29 @@ class AnthropicMessagesHandler(BaseTranslation):
|
|||
block.pop("cache_control", None)
|
||||
data["messages"] = converted
|
||||
|
||||
@staticmethod
|
||||
def _extract_midturn_system_text(
|
||||
message: dict[str, Any], # mutable-ok: API message payload
|
||||
msg_idx: int,
|
||||
texts_to_check: list[str], # mutable-ok: API message payload
|
||||
task_mappings: list[tuple[int, int | None]], # mutable-ok: API message payload
|
||||
) -> None:
|
||||
content = message.get("content")
|
||||
if isinstance(content, str):
|
||||
if content:
|
||||
texts_to_check.append(content)
|
||||
task_mappings.append((msg_idx, None))
|
||||
return
|
||||
if not isinstance(content, list):
|
||||
return
|
||||
for content_idx, content_item in enumerate(content):
|
||||
if not isinstance(content_item, dict) or content_item.get("type") != "text":
|
||||
continue
|
||||
text_str = content_item.get("text")
|
||||
if isinstance(text_str, str) and text_str:
|
||||
texts_to_check.append(text_str)
|
||||
task_mappings.append((msg_idx, content_idx))
|
||||
|
||||
def extract_request_tool_names(self, data: dict) -> List[str]:
|
||||
"""Extract tool names from Anthropic messages request (tools[].name)."""
|
||||
names: List[str] = []
|
||||
|
|
@ -415,15 +535,18 @@ class AnthropicMessagesHandler(BaseTranslation):
|
|||
skip_system_message: bool = False,
|
||||
skip_tool_message: bool = False,
|
||||
) -> None:
|
||||
"""
|
||||
Extract text content and images from a message.
|
||||
|
||||
Override this method to customize text/image extraction logic.
|
||||
"""
|
||||
role = str(message.get("role") or "").lower()
|
||||
if skip_system_message and role == "system":
|
||||
"""Extract text content and images from a message."""
|
||||
role = str(message.get("role") or "")
|
||||
if role == "system":
|
||||
# Match the adapter's filtering so positional guardrail write-back stays aligned.
|
||||
self._extract_midturn_system_text(
|
||||
message=message,
|
||||
msg_idx=msg_idx,
|
||||
texts_to_check=texts_to_check,
|
||||
task_mappings=task_mappings,
|
||||
)
|
||||
return
|
||||
if skip_tool_message and role == "tool":
|
||||
if skip_tool_message and role.lower() == "tool":
|
||||
return
|
||||
|
||||
content = message.get("content", None)
|
||||
|
|
|
|||
|
|
@ -85,12 +85,12 @@ from litellm.llms.anthropic.experimental_pass_through.context_management import
|
|||
)
|
||||
from litellm.types.llms.anthropic import (
|
||||
ANTHROPIC_HOSTED_TOOLS,
|
||||
AllAnthropicPassThroughMessageValues,
|
||||
AllAnthropicToolsValues,
|
||||
AnthopicMessagesAssistantMessageParam,
|
||||
AnthropicFinishReason,
|
||||
AnthropicMessagesRequest,
|
||||
AnthropicMessagesSystemMessageParam,
|
||||
AnthropicMessagesToolChoice,
|
||||
AnthropicMessagesUserMessageParam,
|
||||
AnthropicResponseContentBlockRedactedThinking,
|
||||
AnthropicResponseContentBlockText,
|
||||
AnthropicResponseContentBlockThinking,
|
||||
|
|
@ -354,12 +354,7 @@ class LiteLLMAnthropicMessagesAdapter:
|
|||
|
||||
def translate_anthropic_messages_to_openai(
|
||||
self,
|
||||
messages: List[
|
||||
Union[
|
||||
AnthropicMessagesUserMessageParam,
|
||||
AnthopicMessagesAssistantMessageParam,
|
||||
]
|
||||
],
|
||||
messages: List[AllAnthropicPassThroughMessageValues], # mutable-ok: API message payload
|
||||
model: Optional[str] = None,
|
||||
) -> List:
|
||||
new_messages: List[AllMessageValues] = []
|
||||
|
|
@ -367,6 +362,11 @@ class LiteLLMAnthropicMessagesAdapter:
|
|||
user_message: Optional[ChatCompletionUserMessage] = None
|
||||
tool_message_list: List[ChatCompletionToolMessage] = []
|
||||
new_user_content_list: List[Union[ChatCompletionTextObject, ChatCompletionImageObject]] = []
|
||||
if m["role"] == "system":
|
||||
system_message = self._translate_midturn_system_message_to_openai(m, model)
|
||||
if system_message is not None:
|
||||
new_messages.append(system_message)
|
||||
continue
|
||||
## USER MESSAGE ##
|
||||
if m["role"] == "user":
|
||||
## translate user message
|
||||
|
|
@ -867,6 +867,29 @@ class LiteLLMAnthropicMessagesAdapter:
|
|||
for def_schema in schema[key].values():
|
||||
LiteLLMAnthropicMessagesAdapter._add_additional_properties_false(def_schema)
|
||||
|
||||
def _translate_midturn_system_message_to_openai(
|
||||
self,
|
||||
message: AnthropicMessagesSystemMessageParam,
|
||||
model: str | None,
|
||||
) -> ChatCompletionSystemMessage | None:
|
||||
"""Translate an in-sequence system entry without changing its role or position."""
|
||||
content = message.get("content")
|
||||
if isinstance(content, str):
|
||||
return ChatCompletionSystemMessage(role="system", content=content) if content else None
|
||||
if not isinstance(content, list):
|
||||
return None
|
||||
text_parts: list[ChatCompletionTextObject] = [] # mutable-ok: API message payload
|
||||
for block in content:
|
||||
if not isinstance(block, dict) or block.get("type") != "text":
|
||||
continue
|
||||
text = block.get("text")
|
||||
if not text:
|
||||
continue
|
||||
text_obj = ChatCompletionTextObject(type="text", text=text)
|
||||
self._add_cache_control_if_applicable(block, text_obj, model)
|
||||
text_parts.append(text_obj)
|
||||
return ChatCompletionSystemMessage(role="system", content=text_parts) if text_parts else None
|
||||
|
||||
def _add_system_message_to_messages(
|
||||
self,
|
||||
new_messages: List[AllMessageValues],
|
||||
|
|
@ -1068,13 +1091,8 @@ class LiteLLMAnthropicMessagesAdapter:
|
|||
tool_name_mapping: Dict[str, str] = {}
|
||||
|
||||
## CONVERT ANTHROPIC MESSAGES TO OPENAI
|
||||
messages_list: List[Union[AnthropicMessagesUserMessageParam, AnthopicMessagesAssistantMessageParam]] = cast(
|
||||
List[
|
||||
Union[
|
||||
AnthropicMessagesUserMessageParam,
|
||||
AnthopicMessagesAssistantMessageParam,
|
||||
]
|
||||
],
|
||||
messages_list = cast(
|
||||
List[AllAnthropicPassThroughMessageValues],
|
||||
anthropic_message_request["messages"],
|
||||
)
|
||||
new_messages = self.translate_anthropic_messages_to_openai(
|
||||
|
|
|
|||
|
|
@ -6,6 +6,7 @@ path used for OpenAI and Azure models.
|
|||
"""
|
||||
|
||||
import json
|
||||
from collections.abc import Iterable
|
||||
from typing import Any, Dict, List, Optional, Union, cast
|
||||
|
||||
from litellm.litellm_core_utils.reasoning_effort_utils import (
|
||||
|
|
@ -15,15 +16,15 @@ from litellm.llms.anthropic.experimental_pass_through.utils import (
|
|||
is_reasoning_auto_summary_enabled,
|
||||
)
|
||||
from litellm.types.llms.anthropic import (
|
||||
AllAnthropicPassThroughMessageValues,
|
||||
AllAnthropicToolsValues,
|
||||
AnthopicMessagesAssistantMessageParam,
|
||||
AnthropicFinishReason,
|
||||
AnthropicMessagesRequest,
|
||||
AnthropicMessagesToolChoice,
|
||||
AnthropicMessagesUserMessageParam,
|
||||
AnthropicResponseContentBlockText,
|
||||
AnthropicResponseContentBlockThinking,
|
||||
AnthropicResponseContentBlockToolUse,
|
||||
AnthropicSystemMessageContent,
|
||||
)
|
||||
from litellm.types.llms.anthropic_messages.anthropic_response import (
|
||||
AnthropicMessagesResponse,
|
||||
|
|
@ -54,19 +55,32 @@ class LiteLLMAnthropicToResponsesAPIAdapter:
|
|||
return source.get("url")
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
def _translate_midturn_system_content_to_responses(
|
||||
content: Union[str, Iterable[AnthropicSystemMessageContent]],
|
||||
) -> list[dict[str, str]]: # mutable-ok: API message payload
|
||||
"""Convert in-sequence system content to Responses input-text parts."""
|
||||
if isinstance(content, str):
|
||||
return (
|
||||
[{"type": "input_text", "text": content}] if content else [] # mutable-ok: API message payload
|
||||
) # mutable-ok: API message payload
|
||||
if not isinstance(content, list):
|
||||
return [] # mutable-ok: API message payload
|
||||
return [ # mutable-ok: API message payload
|
||||
{"type": "input_text", "text": text} # mutable-ok: API message payload
|
||||
for block in content
|
||||
if isinstance(block, dict) and block.get("type") == "text" and (text := block.get("text"))
|
||||
]
|
||||
|
||||
def translate_messages_to_responses_input(
|
||||
self,
|
||||
messages: List[
|
||||
Union[
|
||||
AnthropicMessagesUserMessageParam,
|
||||
AnthopicMessagesAssistantMessageParam,
|
||||
]
|
||||
],
|
||||
messages: List[AllAnthropicPassThroughMessageValues], # mutable-ok: API message payload
|
||||
) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Convert Anthropic messages list to Responses API `input` items.
|
||||
|
||||
Mapping:
|
||||
system text -> message(role=system, input_text)
|
||||
user text -> message(role=user, input_text)
|
||||
user image -> message(role=user, input_image)
|
||||
user tool_result -> function_call_output
|
||||
|
|
@ -76,6 +90,18 @@ class LiteLLMAnthropicToResponsesAPIAdapter:
|
|||
input_items: List[Dict[str, Any]] = []
|
||||
|
||||
for m in messages:
|
||||
if m["role"] == "system":
|
||||
system_parts = self._translate_midturn_system_content_to_responses(m.get("content"))
|
||||
if system_parts:
|
||||
input_items.append(
|
||||
{ # mutable-ok: API message payload
|
||||
"type": "message",
|
||||
"role": "system",
|
||||
"content": system_parts,
|
||||
}
|
||||
)
|
||||
continue
|
||||
|
||||
role = m["role"]
|
||||
content = m.get("content")
|
||||
|
||||
|
|
@ -287,12 +313,7 @@ class LiteLLMAnthropicToResponsesAPIAdapter:
|
|||
"""
|
||||
model: str = anthropic_request["model"]
|
||||
messages_list = cast(
|
||||
List[
|
||||
Union[
|
||||
AnthropicMessagesUserMessageParam,
|
||||
AnthopicMessagesAssistantMessageParam,
|
||||
]
|
||||
],
|
||||
List[AllAnthropicPassThroughMessageValues],
|
||||
anthropic_request["messages"],
|
||||
)
|
||||
|
||||
|
|
|
|||
|
|
@ -11,12 +11,24 @@ from litellm.types.proxy.guardrails.guardrail_hooks.akto import (
|
|||
from litellm.types.proxy.guardrails.guardrail_hooks.block_code_execution import (
|
||||
BlockCodeExecutionGuardrailConfigModel,
|
||||
)
|
||||
from litellm.types.proxy.guardrails.guardrail_hooks.cisco_ai_defense import (
|
||||
CiscoAIDefenseGuardrailConfigModel,
|
||||
)
|
||||
from litellm.types.proxy.guardrails.guardrail_hooks.compresr import (
|
||||
CompresrGuardrailConfigModel,
|
||||
)
|
||||
from litellm.types.proxy.guardrails.guardrail_hooks.enkryptai import (
|
||||
EnkryptAIGuardrailConfigs,
|
||||
)
|
||||
from litellm.types.proxy.guardrails.guardrail_hooks.grayswan import (
|
||||
GraySwanGuardrailConfigModel,
|
||||
)
|
||||
from litellm.types.proxy.guardrails.guardrail_hooks.headroom import (
|
||||
HeadroomGuardrailConfigModel,
|
||||
)
|
||||
from litellm.types.proxy.guardrails.guardrail_hooks.hiddenlayer import (
|
||||
HiddenlayerGuardrailConfigModel,
|
||||
)
|
||||
from litellm.types.proxy.guardrails.guardrail_hooks.ibm import (
|
||||
IBMGuardrailsBaseConfigModel,
|
||||
)
|
||||
|
|
@ -29,38 +41,26 @@ from litellm.types.proxy.guardrails.guardrail_hooks.ovalix import (
|
|||
from litellm.types.proxy.guardrails.guardrail_hooks.promptguard import (
|
||||
PromptGuardConfigModel,
|
||||
)
|
||||
from litellm.types.proxy.guardrails.guardrail_hooks.xecguard import (
|
||||
XecGuardConfigModel,
|
||||
from litellm.types.proxy.guardrails.guardrail_hooks.qohash import (
|
||||
QostodianNexusConfigModel,
|
||||
)
|
||||
from litellm.types.proxy.guardrails.guardrail_hooks.qualifire import (
|
||||
QualifireGuardrailConfigModel,
|
||||
)
|
||||
from litellm.types.proxy.guardrails.guardrail_hooks.tool_permission import (
|
||||
ToolPermissionGuardrailConfigModel,
|
||||
)
|
||||
from litellm.types.proxy.guardrails.guardrail_hooks.hiddenlayer import (
|
||||
HiddenlayerGuardrailConfigModel,
|
||||
)
|
||||
from litellm.types.proxy.guardrails.guardrail_hooks.qohash import (
|
||||
QostodianNexusConfigModel,
|
||||
)
|
||||
from litellm.types.proxy.guardrails.guardrail_hooks.repelloai import (
|
||||
RepelloAIGuardrailConfigModel,
|
||||
)
|
||||
from litellm.types.proxy.guardrails.guardrail_hooks.vigil_guard import (
|
||||
VigilGuardGuardrailConfigModel,
|
||||
)
|
||||
from litellm.types.proxy.guardrails.guardrail_hooks.cisco_ai_defense import (
|
||||
CiscoAIDefenseGuardrailConfigModel,
|
||||
)
|
||||
from litellm.types.proxy.guardrails.guardrail_hooks.singulr import (
|
||||
SingulrGuardrailConfigModel,
|
||||
)
|
||||
from litellm.types.proxy.guardrails.guardrail_hooks.headroom import (
|
||||
HeadroomGuardrailConfigModel,
|
||||
from litellm.types.proxy.guardrails.guardrail_hooks.tool_permission import (
|
||||
ToolPermissionGuardrailConfigModel,
|
||||
)
|
||||
from litellm.types.proxy.guardrails.guardrail_hooks.compresr import (
|
||||
CompresrGuardrailConfigModel,
|
||||
from litellm.types.proxy.guardrails.guardrail_hooks.vigil_guard import (
|
||||
VigilGuardGuardrailConfigModel,
|
||||
)
|
||||
from litellm.types.proxy.guardrails.guardrail_hooks.xecguard import (
|
||||
XecGuardConfigModel,
|
||||
)
|
||||
|
||||
"""
|
||||
|
|
@ -743,7 +743,10 @@ class BaseLitellmParams(ContentFilterConfigModel): # works for new and patch up
|
|||
"When True, unified guardrails skip system-role messages when building "
|
||||
"evaluation inputs (texts and structured_messages). When False, system "
|
||||
"messages are included even if litellm_settings sets a global skip. When "
|
||||
"None, use the global litellm.skip_system_message_in_guardrail setting."
|
||||
"None, use the global litellm.skip_system_message_in_guardrail setting. "
|
||||
"For Anthropic /v1/messages, the flag applies only to the trusted top-level "
|
||||
"system prompt. In-sequence system entries are untrusted client input and remain "
|
||||
"in texts and structured_messages."
|
||||
),
|
||||
)
|
||||
|
||||
|
|
|
|||
|
|
@ -365,8 +365,20 @@ class AnthropicSystemMessageContent(TypedDict, total=False):
|
|||
cache_control: Optional[Union[dict, ChatCompletionCachedContent]]
|
||||
|
||||
|
||||
class AnthropicMessagesSystemMessageParam(TypedDict, total=False):
|
||||
role: Required[Literal["system"]]
|
||||
content: Required[Union[str, Iterable[AnthropicSystemMessageContent]]]
|
||||
|
||||
|
||||
AllAnthropicMessageValues = Union[AnthropicMessagesUserMessageParam, AnthopicMessagesAssistantMessageParam]
|
||||
|
||||
# System is not a native Anthropic message role; only pass-through adapters use this union.
|
||||
AllAnthropicPassThroughMessageValues = Union[
|
||||
AnthropicMessagesUserMessageParam,
|
||||
AnthopicMessagesAssistantMessageParam,
|
||||
AnthropicMessagesSystemMessageParam,
|
||||
]
|
||||
|
||||
|
||||
class AnthropicMessagesRequestOptionalParams(TypedDict, total=False):
|
||||
max_tokens: Optional[int]
|
||||
|
|
|
|||
|
|
@ -75,6 +75,51 @@ class MockRecordingGuardrail(CustomGuardrail):
|
|||
return inputs
|
||||
|
||||
|
||||
class MockMaskingGuardrail(CustomGuardrail):
|
||||
"""Capture request inputs and mask one known prohibited value."""
|
||||
|
||||
def __init__(self, skip_system_message_in_guardrail: Optional[bool] = True):
|
||||
super().__init__(guardrail_name="masking-test")
|
||||
self.skip_system_message_in_guardrail = skip_system_message_in_guardrail
|
||||
self.inputs: Optional[GenericGuardrailAPIInputs] = None
|
||||
|
||||
async def apply_guardrail(
|
||||
self,
|
||||
inputs: GenericGuardrailAPIInputs,
|
||||
request_data: dict,
|
||||
input_type: Literal["request", "response"],
|
||||
logging_obj: Optional[Any] = None,
|
||||
) -> GenericGuardrailAPIInputs:
|
||||
self.inputs = inputs.copy()
|
||||
masked_inputs = inputs.copy()
|
||||
masked_inputs["texts"] = [
|
||||
"[MASKED]" if text == "prohibited correction" else text for text in inputs.get("texts", [])
|
||||
]
|
||||
return masked_inputs
|
||||
|
||||
|
||||
class MockCompactingGuardrail(CustomGuardrail):
|
||||
"""Stand in for a compaction guardrail that rewrites `structured_messages` wholesale."""
|
||||
|
||||
def __init__(self, replacement_messages: list):
|
||||
super().__init__(guardrail_name="compacting-test")
|
||||
self.replacement_messages = replacement_messages
|
||||
self.inputs: Optional[GenericGuardrailAPIInputs] = None
|
||||
|
||||
async def apply_guardrail(
|
||||
self,
|
||||
inputs: GenericGuardrailAPIInputs,
|
||||
request_data: dict,
|
||||
input_type: Literal["request", "response"],
|
||||
logging_obj: Optional[Any] = None,
|
||||
) -> GenericGuardrailAPIInputs:
|
||||
self.inputs = inputs.copy()
|
||||
rewritten = inputs.copy()
|
||||
# A new list object -- this is what signals a rewrite to the handler.
|
||||
rewritten["structured_messages"] = list(self.replacement_messages)
|
||||
return rewritten
|
||||
|
||||
|
||||
class TestAnthropicMessagesHandlerStreamingRequestData:
|
||||
"""Post-call guardrails on streaming /v1/messages receive the response and identity metadata"""
|
||||
|
||||
|
|
@ -210,6 +255,660 @@ class TestAnthropicMessagesHandlerInputProcessing:
|
|||
assert data.get("litellm_metadata", {}).get("guardrails")
|
||||
assert guardrail.dynamic_params == {"policy_id": "policy-123"}
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_midturn_system_correction_is_guardrailed_when_top_level_system_is_skipped(
|
||||
self,
|
||||
):
|
||||
handler = AnthropicMessagesHandler()
|
||||
guardrail = MockMaskingGuardrail()
|
||||
data = {
|
||||
"model": "claude-3-5-sonnet-20241022",
|
||||
"system": "trusted top-level system prompt",
|
||||
"messages": [
|
||||
{"role": "user", "content": "safe text"},
|
||||
{
|
||||
"role": "system",
|
||||
"content": [
|
||||
{"type": "unsupported", "text": "discarded text"},
|
||||
{"type": "text", "text": "prohibited correction"},
|
||||
],
|
||||
},
|
||||
],
|
||||
}
|
||||
|
||||
await handler.process_input_messages(data=data, guardrail_to_apply=guardrail)
|
||||
|
||||
assert guardrail.inputs is not None
|
||||
assert guardrail.inputs["texts"] == ["safe text", "prohibited correction"]
|
||||
assert "trusted top-level system prompt" not in guardrail.inputs["texts"]
|
||||
assert data["messages"][1]["content"][0]["text"] == "discarded text"
|
||||
assert data["messages"][1]["content"][1]["text"] == "[MASKED]"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_string_midturn_system_correction_is_guardrailed(self):
|
||||
handler = AnthropicMessagesHandler()
|
||||
guardrail = MockMaskingGuardrail()
|
||||
data = {
|
||||
"model": "claude-3-5-sonnet-20241022",
|
||||
"messages": [{"role": "system", "content": "prohibited correction"}],
|
||||
}
|
||||
|
||||
await handler.process_input_messages(data=data, guardrail_to_apply=guardrail)
|
||||
|
||||
assert guardrail.inputs is not None
|
||||
assert guardrail.inputs["texts"] == ["prohibited correction"]
|
||||
assert data["messages"][0]["content"] == "[MASKED]"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_unsupported_midturn_system_content_is_not_guardrailed(self):
|
||||
handler = AnthropicMessagesHandler()
|
||||
guardrail = MockMaskingGuardrail()
|
||||
data = {
|
||||
"model": "claude-3-5-sonnet-20241022",
|
||||
"messages": [
|
||||
{
|
||||
"role": "system",
|
||||
"content": [{"type": "image", "source": {"type": "url"}}],
|
||||
}
|
||||
],
|
||||
}
|
||||
|
||||
await handler.process_input_messages(data=data, guardrail_to_apply=guardrail)
|
||||
|
||||
assert guardrail.inputs is None
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_skip_system_message_excludes_only_hoisted_top_level_system(self):
|
||||
handler = AnthropicMessagesHandler()
|
||||
guardrail = MockMaskingGuardrail()
|
||||
data = {
|
||||
"model": "claude-3-5-sonnet-20241022",
|
||||
"system": "trusted top-level system prompt",
|
||||
"messages": [
|
||||
{"role": "user", "content": "safe text"},
|
||||
{"role": "system", "content": "prohibited correction"},
|
||||
{"role": "user", "content": "continue"},
|
||||
],
|
||||
}
|
||||
|
||||
await handler.process_input_messages(data=data, guardrail_to_apply=guardrail)
|
||||
|
||||
assert guardrail.inputs is not None
|
||||
structured = guardrail.inputs["structured_messages"]
|
||||
assert [m["role"] for m in structured] == ["user", "system", "user"]
|
||||
assert structured[1]["content"] == "prohibited correction"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_default_skip_false_scans_midturn_system_and_hoists_top_level_system(
|
||||
self,
|
||||
):
|
||||
handler = AnthropicMessagesHandler()
|
||||
guardrail = MockMaskingGuardrail(skip_system_message_in_guardrail=None)
|
||||
data = {
|
||||
"model": "claude-3-5-sonnet-20241022",
|
||||
"system": "trusted top-level system prompt",
|
||||
"messages": [
|
||||
{"role": "user", "content": "safe text"},
|
||||
{"role": "system", "content": "prohibited correction"},
|
||||
],
|
||||
}
|
||||
|
||||
await handler.process_input_messages(data=data, guardrail_to_apply=guardrail)
|
||||
|
||||
assert guardrail.inputs is not None
|
||||
assert guardrail.inputs["texts"] == ["safe text", "prohibited correction"]
|
||||
structured = guardrail.inputs["structured_messages"]
|
||||
assert [m["role"] for m in structured] == ["system", "user", "system"]
|
||||
assert structured[0]["content"] == "trusted top-level system prompt"
|
||||
assert data["messages"][1]["content"] == "[MASKED]"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_bedrock_masking_slice_is_unavailable_when_top_level_system_is_included(
|
||||
self,
|
||||
):
|
||||
from litellm.proxy.guardrails.guardrail_hooks.bedrock_guardrails import (
|
||||
BedrockGuardrail,
|
||||
)
|
||||
|
||||
handler = AnthropicMessagesHandler()
|
||||
guardrail = MockMaskingGuardrail(skip_system_message_in_guardrail=None)
|
||||
data = {
|
||||
"model": "claude-3-5-sonnet-20241022",
|
||||
"system": "trusted top-level system prompt",
|
||||
"messages": [
|
||||
{"role": "user", "content": "safe text"},
|
||||
{"role": "system", "content": "prohibited correction"},
|
||||
{"role": "user", "content": "latest question"},
|
||||
],
|
||||
}
|
||||
|
||||
await handler.process_input_messages(data=data, guardrail_to_apply=guardrail)
|
||||
|
||||
assert guardrail.inputs is not None
|
||||
texts = guardrail.inputs["texts"]
|
||||
structured = guardrail.inputs["structured_messages"]
|
||||
|
||||
bedrock = BedrockGuardrail(guardrailIdentifier="gi", guardrailVersion="1")
|
||||
assert sum(bedrock._count_message_texts(m) for m in structured) == len(texts) + 1
|
||||
latest_user_index = bedrock._find_latest_message_index(structured, target_role="user")
|
||||
assert (
|
||||
bedrock._locate_message_texts_slice(
|
||||
structured_messages=structured,
|
||||
target_index=latest_user_index,
|
||||
texts=texts,
|
||||
)
|
||||
is None
|
||||
)
|
||||
assert (
|
||||
bedrock._merge_masked_texts(
|
||||
masked_texts=["{MASKED}"],
|
||||
texts=texts,
|
||||
scanned_slice=None,
|
||||
scanned_role_subset=True,
|
||||
)
|
||||
== texts
|
||||
)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("skip_system_message_in_guardrail", [True, None])
|
||||
async def test_midturn_system_text_extraction_matches_translation_in_both_skip_modes(
|
||||
self,
|
||||
skip_system_message_in_guardrail: Optional[bool],
|
||||
):
|
||||
from litellm.proxy.guardrails.guardrail_hooks.bedrock_guardrails import (
|
||||
BedrockGuardrail,
|
||||
)
|
||||
|
||||
handler = AnthropicMessagesHandler()
|
||||
guardrail = MockMaskingGuardrail(skip_system_message_in_guardrail=skip_system_message_in_guardrail)
|
||||
data = {
|
||||
"model": "claude-3-5-sonnet-20241022",
|
||||
"messages": [
|
||||
{"role": "user", "content": "safe text"},
|
||||
{
|
||||
"role": "system",
|
||||
"content": [
|
||||
{"type": "text", "text": ""},
|
||||
{"type": "image", "source": {"type": "url", "url": "https://example.com/a.png"}},
|
||||
{"type": "text", "text": "prohibited correction"},
|
||||
],
|
||||
},
|
||||
],
|
||||
}
|
||||
|
||||
await handler.process_input_messages(data=data, guardrail_to_apply=guardrail)
|
||||
|
||||
assert guardrail.inputs is not None
|
||||
texts = guardrail.inputs["texts"]
|
||||
structured = guardrail.inputs["structured_messages"]
|
||||
assert texts == ["safe text", "prohibited correction"]
|
||||
bedrock = BedrockGuardrail(guardrailIdentifier="gi", guardrailVersion="1")
|
||||
assert sum(bedrock._count_message_texts(m) for m in structured) == len(texts)
|
||||
assert data["messages"][1]["content"][2]["text"] == "[MASKED]"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_bedrock_masking_slice_stays_aligned_with_midturn_system(self):
|
||||
from litellm.proxy.guardrails.guardrail_hooks.bedrock_guardrails import (
|
||||
BedrockGuardrail,
|
||||
)
|
||||
|
||||
handler = AnthropicMessagesHandler()
|
||||
guardrail = MockMaskingGuardrail()
|
||||
data = {
|
||||
"model": "claude-3-5-sonnet-20241022",
|
||||
"system": "trusted top-level system prompt",
|
||||
"messages": [
|
||||
{"role": "user", "content": "safe text"},
|
||||
{
|
||||
"role": "system",
|
||||
"content": [
|
||||
{"type": "text", "text": "prohibited correction"},
|
||||
{"type": "text", "text": "second correction"},
|
||||
],
|
||||
},
|
||||
{"role": "user", "content": "latest question"},
|
||||
],
|
||||
}
|
||||
|
||||
await handler.process_input_messages(data=data, guardrail_to_apply=guardrail)
|
||||
|
||||
assert guardrail.inputs is not None
|
||||
texts = guardrail.inputs["texts"]
|
||||
structured = guardrail.inputs["structured_messages"]
|
||||
|
||||
bedrock = BedrockGuardrail(guardrailIdentifier="gi", guardrailVersion="1")
|
||||
total = sum(bedrock._count_message_texts(m) for m in structured)
|
||||
assert total == len(texts)
|
||||
|
||||
latest_user_index = bedrock._find_latest_message_index(structured, target_role="user")
|
||||
assert latest_user_index == 2
|
||||
scanned_slice = bedrock._locate_message_texts_slice(
|
||||
structured_messages=structured,
|
||||
target_index=latest_user_index,
|
||||
texts=texts,
|
||||
)
|
||||
assert scanned_slice == (3, 1)
|
||||
|
||||
merged = bedrock._merge_masked_texts(
|
||||
masked_texts=["{MASKED}"],
|
||||
texts=texts,
|
||||
scanned_slice=scanned_slice,
|
||||
scanned_role_subset=True,
|
||||
)
|
||||
assert merged == [
|
||||
"safe text",
|
||||
"prohibited correction",
|
||||
"second correction",
|
||||
"{MASKED}",
|
||||
]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_compaction_rewrite_keeps_midturn_system_messages(self):
|
||||
handler = AnthropicMessagesHandler()
|
||||
guardrail = MockCompactingGuardrail(
|
||||
replacement_messages=[
|
||||
{"role": "user", "content": "compacted history"},
|
||||
{
|
||||
"role": "system",
|
||||
"content": [{"type": "text", "text": "use the corrected result"}],
|
||||
},
|
||||
{"role": "user", "content": "continue"},
|
||||
]
|
||||
)
|
||||
data = {
|
||||
"model": "claude-3-5-sonnet-20241022",
|
||||
"system": "trusted top-level system prompt",
|
||||
"messages": [
|
||||
{"role": "user", "content": "original history"},
|
||||
{"role": "system", "content": "use the corrected result"},
|
||||
{"role": "user", "content": "continue"},
|
||||
],
|
||||
}
|
||||
|
||||
await handler.process_input_messages(data=data, guardrail_to_apply=guardrail)
|
||||
|
||||
assert [m["role"] for m in data["messages"]] == ["user", "system", "user"]
|
||||
assert data["messages"][1]["content"] == [{"type": "text", "text": "use the corrected result"}]
|
||||
assert data["messages"][0]["content"] == [{"type": "text", "text": "compacted history"}]
|
||||
assert data["messages"][2]["content"] == [{"type": "text", "text": "continue"}]
|
||||
assert data["system"] == "trusted top-level system prompt"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_compaction_rewrite_does_not_duplicate_hoisted_top_level_system(self):
|
||||
handler = AnthropicMessagesHandler()
|
||||
guardrail = MockCompactingGuardrail(
|
||||
replacement_messages=[
|
||||
{"role": "system", "content": "trusted top-level system prompt"},
|
||||
{"role": "user", "content": "compacted history"},
|
||||
{"role": "system", "content": "use the corrected result"},
|
||||
]
|
||||
)
|
||||
guardrail.skip_system_message_in_guardrail = None
|
||||
data = {
|
||||
"model": "claude-3-5-sonnet-20241022",
|
||||
"system": "trusted top-level system prompt",
|
||||
"messages": [
|
||||
{"role": "user", "content": "original history"},
|
||||
{"role": "system", "content": "use the corrected result"},
|
||||
],
|
||||
}
|
||||
|
||||
await handler.process_input_messages(data=data, guardrail_to_apply=guardrail)
|
||||
|
||||
assert [m["role"] for m in data["messages"]] == ["user", "system"]
|
||||
assert data["messages"][1]["content"] == "use the corrected result"
|
||||
assert data["system"] == "trusted top-level system prompt"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_compaction_rewrite_keeps_leading_midturn_system_when_system_is_skipped(
|
||||
self,
|
||||
):
|
||||
handler = AnthropicMessagesHandler()
|
||||
guardrail = MockCompactingGuardrail(
|
||||
replacement_messages=[
|
||||
{"role": "system", "content": "use the corrected result"},
|
||||
{"role": "user", "content": "compacted history"},
|
||||
]
|
||||
)
|
||||
guardrail.skip_system_message_in_guardrail = True
|
||||
data = {
|
||||
"model": "claude-3-5-sonnet-20241022",
|
||||
"system": "trusted top-level system prompt",
|
||||
"messages": [
|
||||
{"role": "system", "content": "use the corrected result"},
|
||||
{"role": "user", "content": "original history"},
|
||||
],
|
||||
}
|
||||
|
||||
await handler.process_input_messages(data=data, guardrail_to_apply=guardrail)
|
||||
|
||||
assert [m["role"] for m in data["messages"]] == ["system", "user"]
|
||||
assert data["messages"][0]["content"] == "use the corrected result"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_compaction_rewrite_keeps_leading_correction_when_top_level_system_hoists_nothing(
|
||||
self,
|
||||
):
|
||||
handler = AnthropicMessagesHandler()
|
||||
guardrail = MockCompactingGuardrail(
|
||||
replacement_messages=[
|
||||
{"role": "system", "content": "use the corrected result"},
|
||||
{"role": "user", "content": "compacted history"},
|
||||
]
|
||||
)
|
||||
guardrail.skip_system_message_in_guardrail = None
|
||||
data = {
|
||||
"model": "claude-3-5-sonnet-20241022",
|
||||
"system": [{"type": "image", "source": {"type": "url", "url": "https://example.com/a.png"}}],
|
||||
"messages": [
|
||||
{"role": "system", "content": "use the corrected result"},
|
||||
{"role": "user", "content": "original history"},
|
||||
],
|
||||
}
|
||||
|
||||
await handler.process_input_messages(data=data, guardrail_to_apply=guardrail)
|
||||
|
||||
assert [m["role"] for m in data["messages"]] == ["system", "user"]
|
||||
assert data["messages"][0]["content"] == "use the corrected result"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_compaction_rewrite_keeps_leading_correction_when_hoisted_prompt_is_dropped(
|
||||
self,
|
||||
):
|
||||
handler = AnthropicMessagesHandler()
|
||||
guardrail = MockCompactingGuardrail(
|
||||
replacement_messages=[
|
||||
{"role": "system", "content": "CLIENT CORRECTION"},
|
||||
{"role": "user", "content": "compacted history"},
|
||||
]
|
||||
)
|
||||
guardrail.skip_system_message_in_guardrail = None
|
||||
data = {
|
||||
"model": "claude-3-5-sonnet-20241022",
|
||||
"system": "TRUSTED",
|
||||
"messages": [
|
||||
{"role": "system", "content": "CLIENT CORRECTION"},
|
||||
{"role": "user", "content": "original history"},
|
||||
],
|
||||
}
|
||||
|
||||
await handler.process_input_messages(data=data, guardrail_to_apply=guardrail)
|
||||
|
||||
assert guardrail.inputs is not None
|
||||
assert guardrail.inputs["structured_messages"][0] == {
|
||||
"role": "system",
|
||||
"content": "TRUSTED",
|
||||
}
|
||||
assert [m["role"] for m in data["messages"]] == ["system", "user"]
|
||||
assert data["messages"][0]["content"] == "CLIENT CORRECTION"
|
||||
assert data["system"] == "TRUSTED"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_compaction_rewrite_drops_hoisted_prompt_matched_by_content_copy(self):
|
||||
import json
|
||||
|
||||
handler = AnthropicMessagesHandler()
|
||||
guardrail = MockCompactingGuardrail(
|
||||
replacement_messages=[
|
||||
json.loads(json.dumps({"role": "system", "content": "TRUSTED"})),
|
||||
{"role": "user", "content": "compacted history"},
|
||||
{"role": "system", "content": "CLIENT CORRECTION"},
|
||||
]
|
||||
)
|
||||
guardrail.skip_system_message_in_guardrail = None
|
||||
data = {
|
||||
"model": "claude-3-5-sonnet-20241022",
|
||||
"system": "TRUSTED",
|
||||
"messages": [
|
||||
{"role": "user", "content": "original history"},
|
||||
{"role": "system", "content": "CLIENT CORRECTION"},
|
||||
],
|
||||
}
|
||||
|
||||
await handler.process_input_messages(data=data, guardrail_to_apply=guardrail)
|
||||
|
||||
assert [m["role"] for m in data["messages"]] == ["user", "system"]
|
||||
assert data["messages"][1]["content"] == "CLIENT CORRECTION"
|
||||
assert data["system"] == "TRUSTED"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_compaction_rewrite_preserves_cache_control_on_system_blocks(self):
|
||||
"""
|
||||
`cache_control` on an in-sequence system text block survives the write-back, and is
|
||||
copied rather than aliased into the guardrail's own returned list.
|
||||
"""
|
||||
handler = AnthropicMessagesHandler()
|
||||
source_cache_control = {"type": "ephemeral"}
|
||||
guardrail = MockCompactingGuardrail(
|
||||
replacement_messages=[
|
||||
{"role": "user", "content": "compacted history"},
|
||||
{
|
||||
"role": "system",
|
||||
"content": [
|
||||
{
|
||||
"type": "text",
|
||||
"text": "use the corrected result",
|
||||
"cache_control": source_cache_control,
|
||||
}
|
||||
],
|
||||
},
|
||||
]
|
||||
)
|
||||
guardrail.skip_system_message_in_guardrail = True
|
||||
data = {
|
||||
"model": "claude-3-5-sonnet-20241022",
|
||||
"messages": [
|
||||
{"role": "user", "content": "original history"},
|
||||
{"role": "system", "content": "use the corrected result"},
|
||||
],
|
||||
}
|
||||
|
||||
await handler.process_input_messages(data=data, guardrail_to_apply=guardrail)
|
||||
|
||||
assert data["messages"][1]["content"] == [
|
||||
{
|
||||
"type": "text",
|
||||
"text": "use the corrected result",
|
||||
"cache_control": {"type": "ephemeral"},
|
||||
}
|
||||
]
|
||||
assert data["messages"][1]["content"][0]["cache_control"] is not source_cache_control
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_compaction_rewrite_rstrips_trailing_assistant_in_each_run(self):
|
||||
handler = AnthropicMessagesHandler()
|
||||
guardrail = MockCompactingGuardrail(
|
||||
replacement_messages=[
|
||||
{"role": "user", "content": "compacted history"},
|
||||
{"role": "assistant", "content": "earlier "},
|
||||
{"role": "system", "content": "use the corrected result"},
|
||||
{"role": "user", "content": "continue"},
|
||||
{"role": "assistant", "content": "prefill "},
|
||||
]
|
||||
)
|
||||
guardrail.skip_system_message_in_guardrail = True
|
||||
data = {
|
||||
"model": "claude-3-5-sonnet-20241022",
|
||||
"messages": [
|
||||
{"role": "user", "content": "original history"},
|
||||
{"role": "system", "content": "use the corrected result"},
|
||||
],
|
||||
}
|
||||
|
||||
await handler.process_input_messages(data=data, guardrail_to_apply=guardrail)
|
||||
|
||||
assert [m["role"] for m in data["messages"]] == [
|
||||
"user",
|
||||
"assistant",
|
||||
"system",
|
||||
"user",
|
||||
"assistant",
|
||||
]
|
||||
assert data["messages"][1]["content"] == [{"type": "text", "text": "earlier"}]
|
||||
assert data["messages"][-1]["content"] == [{"type": "text", "text": "prefill"}]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_compaction_rewrite_drops_text_free_system_message(self):
|
||||
handler = AnthropicMessagesHandler()
|
||||
guardrail = MockCompactingGuardrail(
|
||||
replacement_messages=[
|
||||
{"role": "user", "content": "compacted history"},
|
||||
{"role": "system", "content": [{"type": "text", "text": ""}]},
|
||||
{"role": "system", "content": ""},
|
||||
{"role": "user", "content": "continue"},
|
||||
]
|
||||
)
|
||||
guardrail.skip_system_message_in_guardrail = True
|
||||
data = {
|
||||
"model": "claude-3-5-sonnet-20241022",
|
||||
"messages": [
|
||||
{"role": "user", "content": "original history"},
|
||||
{"role": "system", "content": "use the corrected result"},
|
||||
{"role": "user", "content": "continue"},
|
||||
],
|
||||
}
|
||||
|
||||
await handler.process_input_messages(data=data, guardrail_to_apply=guardrail)
|
||||
|
||||
assert [m["role"] for m in data["messages"]] == ["user", "user"]
|
||||
assert data["messages"][0]["content"] == [{"type": "text", "text": "compacted history"}]
|
||||
assert data["messages"][1]["content"] == [{"type": "text", "text": "continue"}]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_noncanonical_system_role_casing_is_still_scanned(self):
|
||||
handler = AnthropicMessagesHandler()
|
||||
guardrail = MockMaskingGuardrail()
|
||||
data = {
|
||||
"model": "claude-3-5-sonnet-20241022",
|
||||
"messages": [
|
||||
{"role": "user", "content": "safe text"},
|
||||
{"role": "System", "content": "prohibited correction"},
|
||||
],
|
||||
}
|
||||
|
||||
await handler.process_input_messages(data=data, guardrail_to_apply=guardrail)
|
||||
|
||||
assert guardrail.inputs is not None
|
||||
assert "prohibited correction" in guardrail.inputs["texts"]
|
||||
assert data["messages"][1]["content"] == "[MASKED]"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_tool_result_turns_have_a_preexisting_alignment_gap(self):
|
||||
from litellm.proxy.guardrails.guardrail_hooks.bedrock_guardrails import (
|
||||
BedrockGuardrail,
|
||||
)
|
||||
|
||||
handler = AnthropicMessagesHandler()
|
||||
bedrock = BedrockGuardrail(guardrailIdentifier="gi", guardrailVersion="1")
|
||||
tool_loop = [
|
||||
{"role": "user", "content": "call the tool"},
|
||||
{
|
||||
"role": "assistant",
|
||||
"content": [{"type": "tool_use", "id": "tu_1", "name": "get", "input": {"a": 1}}],
|
||||
},
|
||||
{
|
||||
"role": "user",
|
||||
"content": [
|
||||
{
|
||||
"type": "tool_result",
|
||||
"tool_use_id": "tu_1",
|
||||
"content": [{"type": "text", "text": "tool output"}],
|
||||
}
|
||||
],
|
||||
},
|
||||
]
|
||||
|
||||
async def _slice_for(messages: list):
|
||||
guardrail = MockMaskingGuardrail()
|
||||
data = {"model": "claude-3-5-sonnet-20241022", "messages": messages}
|
||||
await handler.process_input_messages(data=data, guardrail_to_apply=guardrail)
|
||||
assert guardrail.inputs is not None
|
||||
texts = guardrail.inputs["texts"]
|
||||
structured = guardrail.inputs["structured_messages"]
|
||||
target_index = bedrock._find_latest_message_index(structured, target_role="user")
|
||||
return (
|
||||
sum(bedrock._count_message_texts(m) for m in structured) - len(texts),
|
||||
bedrock._locate_message_texts_slice(
|
||||
structured_messages=structured,
|
||||
target_index=target_index,
|
||||
texts=texts,
|
||||
),
|
||||
)
|
||||
|
||||
with_system = await _slice_for(
|
||||
tool_loop
|
||||
+ [
|
||||
{"role": "system", "content": "use the corrected result"},
|
||||
{"role": "user", "content": "latest question"},
|
||||
]
|
||||
)
|
||||
without_system = await _slice_for(tool_loop + [{"role": "user", "content": "latest question"}])
|
||||
|
||||
assert with_system == without_system == (1, None)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_compaction_rewrite_to_only_system_messages_is_rejected(self):
|
||||
import litellm
|
||||
|
||||
handler = AnthropicMessagesHandler()
|
||||
guardrail = MockCompactingGuardrail(
|
||||
replacement_messages=[{"role": "system", "content": "use the corrected result"}]
|
||||
)
|
||||
guardrail.skip_system_message_in_guardrail = True
|
||||
data = {
|
||||
"model": "claude-3-5-sonnet-20241022",
|
||||
"messages": [
|
||||
{"role": "user", "content": "original history"},
|
||||
{"role": "system", "content": "use the corrected result"},
|
||||
],
|
||||
}
|
||||
|
||||
with patch.object(litellm, "modify_params", False):
|
||||
with pytest.raises(litellm.BadRequestError, match="at least one non-system message"):
|
||||
await handler.process_input_messages(data=data, guardrail_to_apply=guardrail)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_compaction_rewrite_to_only_system_messages_repaired_with_modify_params(
|
||||
self,
|
||||
):
|
||||
import litellm
|
||||
|
||||
handler = AnthropicMessagesHandler()
|
||||
guardrail = MockCompactingGuardrail(
|
||||
replacement_messages=[{"role": "system", "content": "use the corrected result"}]
|
||||
)
|
||||
guardrail.skip_system_message_in_guardrail = True
|
||||
data = {
|
||||
"model": "claude-3-5-sonnet-20241022",
|
||||
"messages": [
|
||||
{"role": "user", "content": "original history"},
|
||||
{"role": "system", "content": "use the corrected result"},
|
||||
],
|
||||
}
|
||||
|
||||
with patch.object(litellm, "modify_params", True):
|
||||
await handler.process_input_messages(data=data, guardrail_to_apply=guardrail)
|
||||
|
||||
assert [m["role"] for m in data["messages"]] == ["system", "user"]
|
||||
assert data["messages"][0]["content"] == "use the corrected result"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_compaction_rewrite_without_system_messages_is_unchanged(self):
|
||||
handler = AnthropicMessagesHandler()
|
||||
guardrail = MockCompactingGuardrail(replacement_messages=[{"role": "user", "content": "compacted history"}])
|
||||
data = {
|
||||
"model": "claude-3-5-sonnet-20241022",
|
||||
"messages": [
|
||||
{"role": "user", "content": "a"},
|
||||
{"role": "assistant", "content": "b"},
|
||||
{"role": "user", "content": "c"},
|
||||
],
|
||||
}
|
||||
|
||||
await handler.process_input_messages(data=data, guardrail_to_apply=guardrail)
|
||||
|
||||
assert data["messages"] == [{"role": "user", "content": [{"type": "text", "text": "compacted history"}]}]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_process_output_streaming_response_empty_choices(self):
|
||||
"""Test that streaming response with empty choices doesn't raise IndexError
|
||||
|
|
|
|||
|
|
@ -413,6 +413,224 @@ def test_translate_anthropic_messages_to_openai_tool_message_placement():
|
|||
), "Tool message should be placed before user message"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("system_content", "expected_content"),
|
||||
[
|
||||
("Use the corrected result.", "Use the corrected result."),
|
||||
(
|
||||
[{"type": "text", "text": "Use the corrected result."}],
|
||||
[{"type": "text", "text": "Use the corrected result."}],
|
||||
),
|
||||
(
|
||||
[
|
||||
{
|
||||
"type": "image",
|
||||
"source": {"type": "url", "url": "https://example.com/a.png"},
|
||||
},
|
||||
{"type": "text", "text": "Use the corrected result."},
|
||||
],
|
||||
[{"type": "text", "text": "Use the corrected result."}],
|
||||
),
|
||||
(
|
||||
[
|
||||
{"type": "text", "text": "First correction."},
|
||||
{"type": "text", "text": "Second correction."},
|
||||
],
|
||||
[
|
||||
{"type": "text", "text": "First correction."},
|
||||
{"type": "text", "text": "Second correction."},
|
||||
],
|
||||
),
|
||||
],
|
||||
)
|
||||
def test_translate_anthropic_messages_to_openai_preserves_midturn_system_correction(
|
||||
system_content: object,
|
||||
expected_content: object,
|
||||
):
|
||||
messages = [
|
||||
{
|
||||
"role": "assistant",
|
||||
"content": [
|
||||
{
|
||||
"type": "tool_use",
|
||||
"id": "toolu_01234",
|
||||
"name": "get_weather",
|
||||
"input": {"location": "Boston"},
|
||||
}
|
||||
],
|
||||
},
|
||||
{
|
||||
"role": "user",
|
||||
"content": [
|
||||
{
|
||||
"type": "tool_result",
|
||||
"tool_use_id": "toolu_01234",
|
||||
"content": "Rainy, 55°F",
|
||||
}
|
||||
],
|
||||
},
|
||||
{"role": "system", "content": system_content},
|
||||
{"role": "user", "content": "Continue."},
|
||||
]
|
||||
|
||||
result = LiteLLMAnthropicMessagesAdapter().translate_anthropic_messages_to_openai(
|
||||
messages=messages,
|
||||
model="claude-3-5-sonnet-20240620",
|
||||
)
|
||||
|
||||
assert result == [
|
||||
{
|
||||
"role": "assistant",
|
||||
"content": None,
|
||||
"thinking_blocks": None,
|
||||
"tool_calls": [
|
||||
{
|
||||
"id": "toolu_01234",
|
||||
"type": "function",
|
||||
"function": {
|
||||
"name": "get_weather",
|
||||
"arguments": '{"location": "Boston"}',
|
||||
},
|
||||
}
|
||||
],
|
||||
},
|
||||
{
|
||||
"role": "tool",
|
||||
"tool_call_id": "toolu_01234",
|
||||
"content": "Rainy, 55°F",
|
||||
},
|
||||
{"role": "system", "content": expected_content},
|
||||
{"role": "user", "content": "Continue."},
|
||||
]
|
||||
|
||||
|
||||
def test_translate_anthropic_messages_to_openai_preserves_midturn_system_cache_control():
|
||||
"""
|
||||
`cache_control` on an in-sequence system text block survives, matching how the
|
||||
hoisted top-level `system` prompt and user text blocks are already handled.
|
||||
"""
|
||||
messages = [
|
||||
{
|
||||
"role": "system",
|
||||
"content": [
|
||||
{
|
||||
"type": "text",
|
||||
"text": "Use the corrected result.",
|
||||
"cache_control": {"type": "ephemeral"},
|
||||
}
|
||||
],
|
||||
}
|
||||
]
|
||||
|
||||
result = LiteLLMAnthropicMessagesAdapter().translate_anthropic_messages_to_openai(
|
||||
messages=messages,
|
||||
model="claude-3-5-sonnet-20240620",
|
||||
)
|
||||
|
||||
assert result == [
|
||||
{
|
||||
"role": "system",
|
||||
"content": [
|
||||
{
|
||||
"type": "text",
|
||||
"text": "Use the corrected result.",
|
||||
"cache_control": {"type": "ephemeral"},
|
||||
}
|
||||
],
|
||||
}
|
||||
]
|
||||
|
||||
|
||||
def test_translate_anthropic_messages_to_openai_drops_midturn_system_cache_control_for_non_claude():
|
||||
"""
|
||||
`cache_control` goes through the same `_add_cache_control_if_applicable` gate as the
|
||||
hoisted top-level prompt and user text blocks, so a non-Claude *requested model name*
|
||||
does not get it. That gate is a best-effort check of the requested name before routing
|
||||
(behind the proxy it is often a public alias), not a guarantee about the backend that
|
||||
ultimately serves the request.
|
||||
"""
|
||||
messages = [
|
||||
{
|
||||
"role": "system",
|
||||
"content": [
|
||||
{
|
||||
"type": "text",
|
||||
"text": "Use the corrected result.",
|
||||
"cache_control": {"type": "ephemeral"},
|
||||
}
|
||||
],
|
||||
}
|
||||
]
|
||||
|
||||
result = LiteLLMAnthropicMessagesAdapter().translate_anthropic_messages_to_openai(
|
||||
messages=messages,
|
||||
model="gpt-4o",
|
||||
)
|
||||
|
||||
assert result == [
|
||||
{
|
||||
"role": "system",
|
||||
"content": [{"type": "text", "text": "Use the corrected result."}],
|
||||
}
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"system_content",
|
||||
[
|
||||
"",
|
||||
[{"type": "text", "text": ""}],
|
||||
[
|
||||
{
|
||||
"type": "image",
|
||||
"source": {"type": "url", "url": "https://example.com/a.png"},
|
||||
}
|
||||
],
|
||||
None,
|
||||
],
|
||||
)
|
||||
def test_translate_anthropic_messages_to_openai_drops_empty_midturn_system(
|
||||
system_content: object,
|
||||
):
|
||||
messages = [{"role": "system", "content": system_content}]
|
||||
|
||||
result = LiteLLMAnthropicMessagesAdapter().translate_anthropic_messages_to_openai(
|
||||
messages=messages,
|
||||
model="claude-3-5-sonnet-20240620",
|
||||
)
|
||||
|
||||
assert result == []
|
||||
|
||||
|
||||
def test_translate_anthropic_to_openai_orders_top_level_and_midturn_system():
|
||||
"""
|
||||
Request level: the trusted top-level prompt is hoisted to index 0 exactly once and the
|
||||
in-sequence correction keeps its own position and `role: "system"` -- no duplication of
|
||||
either, and no reordering of the surrounding turns.
|
||||
"""
|
||||
openai_request, _ = LiteLLMAnthropicMessagesAdapter().translate_anthropic_to_openai(
|
||||
anthropic_message_request={
|
||||
"model": "claude-3-5-sonnet-20240620",
|
||||
"max_tokens": 100,
|
||||
"system": "Trusted top-level prompt.",
|
||||
"messages": [
|
||||
{"role": "user", "content": "First question."},
|
||||
{"role": "assistant", "content": "First answer."},
|
||||
{"role": "system", "content": "Use the corrected result."},
|
||||
{"role": "user", "content": "Continue."},
|
||||
],
|
||||
}
|
||||
)
|
||||
|
||||
assert openai_request["messages"] == [
|
||||
{"role": "system", "content": "Trusted top-level prompt."},
|
||||
{"role": "user", "content": "First question."},
|
||||
{"role": "assistant", "content": "First answer.", "thinking_blocks": None},
|
||||
{"role": "system", "content": "Use the corrected result."},
|
||||
{"role": "user", "content": "Continue."},
|
||||
]
|
||||
|
||||
|
||||
def test_translate_openai_content_to_anthropic_empty_function_arguments():
|
||||
"""Test that empty function arguments are handled safely and don't cause JSON parsing errors."""
|
||||
|
||||
|
|
|
|||
|
|
@ -2475,3 +2475,57 @@ def test_endpoint_runs_failure_hook_on_500_context_management_error():
|
|||
body = response.json()
|
||||
assert body["type"] == "error"
|
||||
failure_hook.assert_awaited_once()
|
||||
|
||||
|
||||
def test_count_effective_tokens_counts_midturn_system_correction():
|
||||
from litellm.llms.anthropic.experimental_pass_through.context_management.editors.compact import (
|
||||
_count_effective_tokens,
|
||||
)
|
||||
|
||||
base: List[Dict[str, Any]] = [
|
||||
{"role": "user", "content": "hello"},
|
||||
{"role": "assistant", "content": "hi"},
|
||||
]
|
||||
correction = {
|
||||
"role": "system",
|
||||
"content": [{"type": "text", "text": "use the corrected result " * 20}],
|
||||
}
|
||||
|
||||
without_correction = _count_effective_tokens(
|
||||
model=MODEL, effective_messages=base, compaction_block=None, tools=None
|
||||
)
|
||||
with_correction = _count_effective_tokens(
|
||||
model=MODEL,
|
||||
effective_messages=base + [correction],
|
||||
compaction_block=None,
|
||||
tools=None,
|
||||
)
|
||||
|
||||
assert with_correction > without_correction
|
||||
|
||||
|
||||
def test_build_summary_messages_keeps_midturn_system_correction_in_place():
|
||||
from litellm.llms.anthropic.experimental_pass_through.context_management.editors.compact import (
|
||||
_build_summary_messages,
|
||||
)
|
||||
|
||||
summary_messages = _build_summary_messages(
|
||||
effective_messages=[
|
||||
{"role": "user", "content": "original question"},
|
||||
{"role": "system", "content": "use the corrected result"},
|
||||
{"role": "assistant", "content": "acknowledged"},
|
||||
],
|
||||
prompt="summarize the conversation",
|
||||
system="caller system prompt",
|
||||
)
|
||||
|
||||
assert [m["role"] for m in summary_messages] == [
|
||||
"system",
|
||||
"user",
|
||||
"system",
|
||||
"assistant",
|
||||
"user",
|
||||
]
|
||||
assert summary_messages[0]["content"] == "caller system prompt"
|
||||
assert summary_messages[2]["content"] == "use the corrected result"
|
||||
assert summary_messages[-1]["content"] == "summarize the conversation"
|
||||
|
|
|
|||
|
|
@ -9,6 +9,8 @@ import sys
|
|||
from typing import Any, Dict, List
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
|
||||
sys.path.insert(0, os.path.abspath("../../../../../../.."))
|
||||
|
||||
from litellm.constants import (
|
||||
|
|
@ -221,6 +223,106 @@ class TestTranslateMessagesToResponsesInput:
|
|||
{"type": "input_text", "text": "Second part."},
|
||||
]
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"system_content",
|
||||
[
|
||||
"Use the corrected result.",
|
||||
[{"type": "text", "text": "Use the corrected result."}],
|
||||
[
|
||||
{"type": "image", "source": {"type": "url", "url": "https://example.com/a.png"}},
|
||||
{"type": "text", "text": "Use the corrected result."},
|
||||
],
|
||||
],
|
||||
)
|
||||
def test_midturn_system_correction_stays_system_in_sequence(self, system_content: object):
|
||||
messages = [
|
||||
{
|
||||
"role": "assistant",
|
||||
"content": [
|
||||
{
|
||||
"type": "tool_use",
|
||||
"id": "toolu_01234",
|
||||
"name": "get_weather",
|
||||
"input": {"location": "Boston"},
|
||||
}
|
||||
],
|
||||
},
|
||||
{
|
||||
"role": "user",
|
||||
"content": [
|
||||
{
|
||||
"type": "tool_result",
|
||||
"tool_use_id": "toolu_01234",
|
||||
"content": "Rainy, 55°F",
|
||||
}
|
||||
],
|
||||
},
|
||||
{"role": "system", "content": system_content},
|
||||
{"role": "user", "content": "Continue."},
|
||||
]
|
||||
|
||||
result = _translate_messages(messages)
|
||||
|
||||
assert result == [
|
||||
{
|
||||
"type": "function_call",
|
||||
"call_id": "toolu_01234",
|
||||
"name": "get_weather",
|
||||
"arguments": '{"location": "Boston"}',
|
||||
},
|
||||
{
|
||||
"type": "function_call_output",
|
||||
"call_id": "toolu_01234",
|
||||
"output": "Rainy, 55°F",
|
||||
},
|
||||
{
|
||||
"type": "message",
|
||||
"role": "system",
|
||||
"content": [{"type": "input_text", "text": "Use the corrected result."}],
|
||||
},
|
||||
{
|
||||
"type": "message",
|
||||
"role": "user",
|
||||
"content": [{"type": "input_text", "text": "Continue."}],
|
||||
},
|
||||
]
|
||||
|
||||
def test_midturn_system_correction_keeps_multiple_text_blocks(self):
|
||||
messages = [
|
||||
{
|
||||
"role": "system",
|
||||
"content": [
|
||||
{"type": "text", "text": "First correction."},
|
||||
{"type": "text", "text": "Second correction."},
|
||||
],
|
||||
}
|
||||
]
|
||||
|
||||
assert _translate_messages(messages) == [
|
||||
{
|
||||
"type": "message",
|
||||
"role": "system",
|
||||
"content": [
|
||||
{"type": "input_text", "text": "First correction."},
|
||||
{"type": "input_text", "text": "Second correction."},
|
||||
],
|
||||
}
|
||||
]
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"system_content",
|
||||
[
|
||||
"",
|
||||
[{"type": "text", "text": ""}],
|
||||
[{"type": "image", "source": {"type": "url", "url": "https://example.com/a.png"}}],
|
||||
None,
|
||||
],
|
||||
)
|
||||
def test_empty_or_unsupported_midturn_system_correction_is_dropped(self, system_content: object):
|
||||
messages = [{"role": "system", "content": system_content}]
|
||||
|
||||
assert _translate_messages(messages) == []
|
||||
|
||||
def test_user_base64_image(self):
|
||||
"""User message with base64 image source becomes input_image with data URL."""
|
||||
messages = [
|
||||
|
|
@ -722,6 +824,42 @@ class TestTranslateRequestBroaderCoverage:
|
|||
kwargs = _ADAPTER.translate_request(req)
|
||||
assert kwargs["instructions"] == "You are a helpful assistant."
|
||||
|
||||
def test_top_level_system_and_midturn_correction_are_not_duplicated(self):
|
||||
"""
|
||||
Request level: the trusted top-level prompt goes to `instructions` only, and the
|
||||
in-sequence correction stays a `role: "system"` input item in its original position.
|
||||
Neither appears twice, and the surrounding turns keep their order.
|
||||
"""
|
||||
req = _make_request(
|
||||
system="Trusted top-level prompt.",
|
||||
messages=[
|
||||
{"role": "user", "content": "First question."},
|
||||
{"role": "system", "content": "Use the corrected result."},
|
||||
{"role": "user", "content": "Continue."},
|
||||
],
|
||||
)
|
||||
|
||||
kwargs = _ADAPTER.translate_request(req)
|
||||
|
||||
assert kwargs["instructions"] == "Trusted top-level prompt."
|
||||
assert kwargs["input"] == [
|
||||
{
|
||||
"type": "message",
|
||||
"role": "user",
|
||||
"content": [{"type": "input_text", "text": "First question."}],
|
||||
},
|
||||
{
|
||||
"type": "message",
|
||||
"role": "system",
|
||||
"content": [{"type": "input_text", "text": "Use the corrected result."}],
|
||||
},
|
||||
{
|
||||
"type": "message",
|
||||
"role": "user",
|
||||
"content": [{"type": "input_text", "text": "Continue."}],
|
||||
},
|
||||
]
|
||||
|
||||
def test_system_list_of_text_blocks_joined(self):
|
||||
req = _make_request(
|
||||
system=[
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue