mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-29 06:33:28 +00:00
fix(guardrails): scan Responses tool results in texts-mode guardrails (#48827)
* fix(guardrails): scan Responses tool results in texts-mode guardrails * fix(guardrails): keep guardrail-added tools and non-text parts on Responses tool scans Tool result texts now come from the same conversion that builds the chat tool message, so items without a call_id, dict outputs, and outputs with no text parts stay aligned for Bedrock and PANW. A changed joined text replaces the first text part and keeps the other parts. scan_only_tool_results appends tools the guardrail returns with new names, like chat, and tool_result and computer_call_output items are scanned through the shared type list. * fix(guardrails): skip empty and screenshot tool outputs and keep text part types on Responses scans * fix(guardrails): keep Bedrock latest-turn masking aligned when tool outputs send no text * fix(guardrails): scan each Responses tool output text part on its own * fix(guardrails): split a Responses tool output's structured row when only textless outputs share its call_id * fix(guardrails): pair each Responses tool output with its own structured row --------- Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
This commit is contained in:
mika-lt
GitHub
parent
83ae1bd63e
commit
4e5f585c2a
9 changed files with 1010 additions and 108 deletions
|
|
@ -32,11 +32,11 @@ import copy
|
|||
import json
|
||||
import time
|
||||
import uuid
|
||||
from collections.abc import Mapping, MutableSequence, Sequence
|
||||
from collections.abc import Iterable, Mapping, MutableSequence, Sequence
|
||||
from dataclasses import dataclass
|
||||
from itertools import accumulate, chain, repeat
|
||||
from itertools import accumulate, chain, groupby, repeat
|
||||
from types import MappingProxyType
|
||||
from typing import TYPE_CHECKING, Any, Final, NamedTuple, Protocol, Union, cast
|
||||
from typing import TYPE_CHECKING, Any, Final, NamedTuple, Protocol, TypeVar, Union, cast
|
||||
|
||||
from pydantic import BaseModel, ConfigDict, TypeAdapter, ValidationError
|
||||
from typing_extensions import ReadOnly, TypedDict
|
||||
|
|
@ -55,7 +55,9 @@ from litellm.llms.base_llm.guardrail_translation.base_translation import (
|
|||
)
|
||||
from litellm.llms.base_llm.guardrail_translation.utils import (
|
||||
blocked_responses_stream_usage,
|
||||
effective_scan_only_tool_results_for_guardrail,
|
||||
effective_skip_system_message_for_guardrail,
|
||||
effective_skip_tool_message_for_guardrail,
|
||||
list_field,
|
||||
merge_guardrailed_scoped_messages,
|
||||
role_out_of_guardrail_scope,
|
||||
|
|
@ -65,7 +67,10 @@ from litellm.llms.base_llm.guardrail_translation.utils import (
|
|||
stream_item_items,
|
||||
unappliable_request_rewrite,
|
||||
)
|
||||
from litellm.llms.openai.responses.guardrail_translation.tool_merge import merge_guardrailed_tools
|
||||
from litellm.llms.openai.responses.guardrail_translation.tool_merge import (
|
||||
append_returned_tools,
|
||||
merge_guardrailed_tools,
|
||||
)
|
||||
from litellm.responses.litellm_completion_transformation.compaction_items import (
|
||||
compaction_summary,
|
||||
decode_compaction_item,
|
||||
|
|
@ -74,7 +79,9 @@ from litellm.responses.litellm_completion_transformation.compaction_items import
|
|||
from litellm.responses.litellm_completion_transformation.custom_tools import without_malformed_tool_flags
|
||||
from litellm.responses.litellm_completion_transformation.tool_search import is_server_tool_search_tool
|
||||
from litellm.responses.litellm_completion_transformation.transformation import (
|
||||
TOOL_CALL_OUTPUT_ITEM_TYPES,
|
||||
LiteLLMCompletionResponsesConfig,
|
||||
tool_output_to_chat_content,
|
||||
)
|
||||
from litellm.types.llms.base import LiteLLMBaseModel
|
||||
from litellm.types.llms.openai import (
|
||||
|
|
@ -270,6 +277,7 @@ _TOOL_CALL_PAYLOAD_EVENT_TYPES: Final = _TOOL_CALL_PAYLOAD_DELTA_EVENT_TYPES | f
|
|||
)
|
||||
_OUTPUT_ITEM_EVENT_TYPES: Final = frozenset({"response.output_item.added", "response.output_item.done"})
|
||||
_OUTPUT_TEXT_EVENT_TYPES: Final = frozenset({"response.output_text.delta", "response.output_text.done"})
|
||||
_TOOL_OUTPUT_TEXT_PART_TYPES: Final = frozenset({"input_text", "output_text", "text"})
|
||||
_PATCHABLE_ITEM_FIELDS: Final[Mapping[str, str]] = MappingProxyType(
|
||||
{"function_call_output": "output", "custom_tool_call_output": "output", "message": "content"}
|
||||
)
|
||||
|
|
@ -519,10 +527,109 @@ def scannable_instructions(data: Mapping[str, object], *, skip_system: bool = Fa
|
|||
|
||||
|
||||
def _input_item_role(item: object) -> str:
|
||||
if stream_item_field(item, "type") in TOOL_CALL_OUTPUT_ITEM_TYPES:
|
||||
return "tool"
|
||||
role: Final = item.get("role") if isinstance(item, Mapping) else None
|
||||
return role.lower() if isinstance(role, str) else ""
|
||||
|
||||
|
||||
def _scanned_text_field(item: Mapping[str, object]) -> str:
|
||||
return "output" if item.get("type") in TOOL_CALL_OUTPUT_ITEM_TYPES else "content"
|
||||
|
||||
|
||||
def _tool_output_part_text(part: object) -> str | None:
|
||||
if not isinstance(part, dict):
|
||||
return None
|
||||
fields: Final = _item_fields_adapter.validate_python(part)
|
||||
text: Final = fields.get("text")
|
||||
return text if fields.get("type") in _TOOL_OUTPUT_TEXT_PART_TYPES and isinstance(text, str) else None
|
||||
|
||||
|
||||
def tool_output_texts(item: object) -> tuple[tuple[int | None, str], ...]:
|
||||
if not stream_item_field(item, "call_id") or stream_item_field(item, "type") == "computer_call_output":
|
||||
return ()
|
||||
parts: Final = stream_item_items(item, "output")
|
||||
part_texts: Final = tuple(
|
||||
(index, text) for index, part in enumerate(parts) if (text := _tool_output_part_text(part))
|
||||
)
|
||||
if part_texts:
|
||||
return part_texts
|
||||
tool_content: Final = tool_output_to_chat_content(stream_item_field(item, "output"))
|
||||
return ((None, tool_content),) if isinstance(tool_content, str) and tool_content else ()
|
||||
|
||||
|
||||
_Grouped: Final = TypeVar("_Grouped")
|
||||
|
||||
|
||||
def _pair_call_id(pair: tuple[str, object]) -> str:
|
||||
return pair[0]
|
||||
|
||||
|
||||
def _grouped_by_call_id(pairs: Iterable[tuple[str, _Grouped]]) -> Mapping[str, tuple[_Grouped, ...]]:
|
||||
ordered: Final = sorted(pairs, key=_pair_call_id)
|
||||
return MappingProxyType(
|
||||
{call_id: tuple(value for _, value in group) for call_id, group in groupby(ordered, key=_pair_call_id)}
|
||||
)
|
||||
|
||||
|
||||
def _texts_by_tool_row(
|
||||
messages: Sequence[Mapping[str, object]], input_data: "str | ResponseInputParam"
|
||||
) -> Mapping[int, tuple[str, ...]]:
|
||||
if isinstance(input_data, str):
|
||||
return MappingProxyType({})
|
||||
output_texts_by_call_id: Final = _grouped_by_call_id(
|
||||
(str(stream_item_field(item, "call_id")), tuple(text for _, text in tool_output_texts(item)))
|
||||
for item in input_data
|
||||
if stream_item_field(item, "type") in TOOL_CALL_OUTPUT_ITEM_TYPES
|
||||
)
|
||||
row_indices_by_call_id: Final = _grouped_by_call_id(
|
||||
(str(message.get("tool_call_id")), index)
|
||||
for index, message in enumerate(messages)
|
||||
if message.get("role") == "tool"
|
||||
)
|
||||
return MappingProxyType(
|
||||
dict(
|
||||
chain.from_iterable(
|
||||
zip(indices, output_texts_by_call_id[call_id], strict=True)
|
||||
for call_id, indices in row_indices_by_call_id.items()
|
||||
if len(indices) == len(output_texts_by_call_id.get(call_id, ()))
|
||||
)
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def _is_string_tool_row(message: Mapping[str, object]) -> bool:
|
||||
return message.get("role") == "tool" and isinstance(message.get("content"), str)
|
||||
|
||||
|
||||
def _split_into_text_parts(message: AllMessageValues, texts: tuple[str, ...]) -> AllMessageValues:
|
||||
if not _is_string_tool_row(message) or len(texts) < 2:
|
||||
return message
|
||||
return cast( # cast-ok: a tool row whose content is its text parts as chat text blocks
|
||||
"AllMessageValues", {**message, "content": [{"type": "text", "text": text} for text in texts]}
|
||||
)
|
||||
|
||||
|
||||
def _with_split_rows_restored(
|
||||
rows: Sequence[AllMessageValues], split_rows: Sequence[tuple[AllMessageValues, AllMessageValues]]
|
||||
) -> Sequence[AllMessageValues]:
|
||||
if not isinstance(rows, list):
|
||||
return rows
|
||||
return [next((original for split, original in split_rows if row == split), row) for row in rows]
|
||||
|
||||
|
||||
def _is_textless_tool_row(message: Mapping[str, object], texts: tuple[str, ...] | None) -> bool:
|
||||
return _is_string_tool_row(message) and (not message.get("content") or texts == ())
|
||||
|
||||
|
||||
def _rewritten_whole_text(item: object, field: str, guardrail_response: str) -> object | None:
|
||||
if isinstance(stream_item_field(item, field), str):
|
||||
return guardrail_response
|
||||
if tool_output_texts(item) == ((None, guardrail_response),):
|
||||
return None
|
||||
return guardrail_response
|
||||
|
||||
|
||||
def _patched_request_fields(
|
||||
raw_input: object,
|
||||
instructions: object,
|
||||
|
|
@ -636,17 +743,28 @@ class OpenAIResponsesHandler(BaseTranslation):
|
|||
|
||||
Handles both string input and list of message objects.
|
||||
"""
|
||||
input_data: Final[str | ResponseInputParam | None] = data.get("input")
|
||||
if not isinstance(input_data, (str, list)):
|
||||
raw_input: Final[str | ResponseInputParam | None] = data.get("input")
|
||||
if not isinstance(raw_input, (str, list)):
|
||||
return data
|
||||
input_data: Final = cast("str | ResponseInputParam", raw_input) # cast-ok: fields read via stream_item_field
|
||||
skip_system: Final = effective_skip_system_message_for_guardrail(guardrail_to_apply)
|
||||
skip_tool: Final = effective_skip_tool_message_for_guardrail(guardrail_to_apply)
|
||||
scan_only: Final = effective_scan_only_tool_results_for_guardrail(guardrail_to_apply)
|
||||
structured_messages: Final = self.get_structured_messages(data)
|
||||
scoped_indices: Final = scoped_structured_message_indices(
|
||||
structured_messages or [], scan_only_tool_results=False, skip_system=skip_system, skip_tool=False
|
||||
all_messages: Final = structured_messages or []
|
||||
texts_by_tool_row: Final = _texts_by_tool_row(all_messages, input_data)
|
||||
scoped_indices: Final = tuple(
|
||||
index
|
||||
for index in scoped_structured_message_indices(
|
||||
all_messages, scan_only_tool_results=scan_only, skip_system=skip_system, skip_tool=skip_tool
|
||||
)
|
||||
if not _is_textless_tool_row(all_messages[index], texts_by_tool_row.get(index))
|
||||
)
|
||||
scoped_structured_messages: Final = (
|
||||
[structured_messages[index] for index in scoped_indices] if structured_messages else None
|
||||
split_rows: Final = tuple(
|
||||
(_split_into_text_parts(all_messages[index], texts_by_tool_row.get(index, ())), all_messages[index])
|
||||
for index in scoped_indices
|
||||
)
|
||||
scoped_structured_messages: Final = [split for split, _ in split_rows] if structured_messages else None
|
||||
raw_tools: Final = data.get("tools")
|
||||
original_tools: Final[tuple[Mapping[str, object], ...]] = (
|
||||
tuple(raw_tools) if isinstance(raw_tools, list) else ()
|
||||
|
|
@ -661,7 +779,12 @@ class OpenAIResponsesHandler(BaseTranslation):
|
|||
)
|
||||
)
|
||||
extracted: Final = self._extract_guardrail_inputs(
|
||||
data, input_data, flattened_tool_groups, skip_system=skip_system
|
||||
data,
|
||||
input_data,
|
||||
() if scan_only else flattened_tool_groups,
|
||||
skip_system=skip_system,
|
||||
skip_tool=skip_tool,
|
||||
scan_only=scan_only,
|
||||
)
|
||||
if not extracted.inputs.get("texts"):
|
||||
return data
|
||||
|
|
@ -674,13 +797,14 @@ class OpenAIResponsesHandler(BaseTranslation):
|
|||
logging_obj=litellm_logging_obj,
|
||||
)
|
||||
self._apply_guardrailed_tools_to_data(
|
||||
data, original_tools, flattened_tool_groups, guardrailed_inputs.get("tools")
|
||||
data, original_tools, flattened_tool_groups, guardrailed_inputs.get("tools"), scan_only=scan_only
|
||||
)
|
||||
written_back: Final = self._written_back_request_fields(
|
||||
data,
|
||||
structured_messages or (),
|
||||
scoped_indices,
|
||||
scoped_structured_messages,
|
||||
tuple((split, original) for split, original in split_rows if split is not original),
|
||||
guardrail_to_apply,
|
||||
guardrailed_inputs,
|
||||
)
|
||||
|
|
@ -730,8 +854,10 @@ class OpenAIResponsesHandler(BaseTranslation):
|
|||
flattened_tool_groups: Sequence[Sequence[Mapping[str, object]]],
|
||||
*,
|
||||
skip_system: bool = False,
|
||||
skip_tool: bool = False,
|
||||
scan_only: bool = False,
|
||||
) -> _ExtractedInputs:
|
||||
instructions: Final = scannable_instructions(data, skip_system=skip_system)
|
||||
instructions: Final = scannable_instructions(data, skip_system=skip_system or scan_only)
|
||||
texts_to_check: Final[list[str]] = [] if instructions is None else [instructions]
|
||||
images_to_check: Final[list[str]] = []
|
||||
task_mappings: Final[list[tuple[int, int | None]]] = []
|
||||
|
|
@ -745,11 +871,15 @@ class OpenAIResponsesHandler(BaseTranslation):
|
|||
)
|
||||
)
|
||||
if isinstance(input_data, str):
|
||||
texts_to_check.append(input_data)
|
||||
if not scan_only:
|
||||
texts_to_check.append(input_data)
|
||||
else:
|
||||
for msg_idx, message in enumerate(input_data):
|
||||
if role_out_of_guardrail_scope(
|
||||
_input_item_role(message), skip_system_message=skip_system, skip_tool_message=False
|
||||
_input_item_role(message),
|
||||
skip_system_message=skip_system,
|
||||
skip_tool_message=skip_tool,
|
||||
scan_only_tool_results=scan_only,
|
||||
):
|
||||
continue
|
||||
self._extract_input_text_and_images(
|
||||
|
|
@ -775,12 +905,14 @@ class OpenAIResponsesHandler(BaseTranslation):
|
|||
structured_messages: Sequence[AllMessageValues],
|
||||
scoped_indices: Sequence[int],
|
||||
scoped_structured_messages: Sequence[AllMessageValues] | None,
|
||||
split_rows: Sequence[tuple[AllMessageValues, AllMessageValues]],
|
||||
guardrail_to_apply: "CustomGuardrail",
|
||||
guardrailed_inputs: GenericGuardrailAPIInputs,
|
||||
) -> _RequestFields | None:
|
||||
guardrailed: Final = guardrailed_inputs.get("structured_messages")
|
||||
if guardrailed is None or guardrailed is scoped_structured_messages:
|
||||
returned: Final = guardrailed_inputs.get("structured_messages")
|
||||
if returned is None or returned is scoped_structured_messages:
|
||||
return None
|
||||
guardrailed: Final = _with_split_rows_restored(returned, split_rows)
|
||||
covers_full_request: Final = len(scoped_indices) == len(structured_messages) or (
|
||||
guardrail_to_apply.structured_messages_cover_full_request() and len(guardrailed) == len(structured_messages)
|
||||
)
|
||||
|
|
@ -814,11 +946,14 @@ class OpenAIResponsesHandler(BaseTranslation):
|
|||
original_tools: Sequence[Mapping[str, object]],
|
||||
flattened_tool_groups: Sequence[Sequence[Mapping[str, object]]],
|
||||
guardrailed_tools: Sequence[ChatCompletionToolParam] | None,
|
||||
*,
|
||||
scan_only: bool,
|
||||
) -> None:
|
||||
if guardrailed_tools is None:
|
||||
return
|
||||
merge: Final = append_returned_tools if scan_only else merge_guardrailed_tools
|
||||
data["tools"] = list( # rebind-ok: in-place request rewrite
|
||||
merge_guardrailed_tools(original_tools, flattened_tool_groups, guardrailed_tools)
|
||||
merge(original_tools, flattened_tool_groups, guardrailed_tools)
|
||||
)
|
||||
|
||||
def _extract_input_text_and_images(
|
||||
|
|
@ -839,6 +974,11 @@ class OpenAIResponsesHandler(BaseTranslation):
|
|||
texts_to_check.append(summary)
|
||||
task_mappings.append((msg_idx, None))
|
||||
return
|
||||
if message.get("type") in TOOL_CALL_OUTPUT_ITEM_TYPES:
|
||||
for content_idx, text in tool_output_texts(message):
|
||||
texts_to_check.append(text)
|
||||
task_mappings.append((msg_idx, content_idx))
|
||||
return
|
||||
content: Final = message.get("content", None)
|
||||
if content is None:
|
||||
return
|
||||
|
|
@ -890,18 +1030,20 @@ class OpenAIResponsesHandler(BaseTranslation):
|
|||
)
|
||||
continue
|
||||
|
||||
content = messages[msg_idx].get("content", None)
|
||||
field = _scanned_text_field(messages[msg_idx])
|
||||
content = messages[msg_idx].get(field, None)
|
||||
if content is None:
|
||||
continue
|
||||
|
||||
if isinstance(content, str) and content_idx_optional is None:
|
||||
# Replace string content with guardrail response
|
||||
messages[msg_idx]["content"] = guardrail_response
|
||||
if content_idx_optional is None:
|
||||
rewritten = _rewritten_whole_text(messages[msg_idx], field, guardrail_response)
|
||||
if rewritten is not None:
|
||||
messages[msg_idx][field] = rewritten
|
||||
|
||||
elif isinstance(content, list) and content_idx_optional is not None:
|
||||
elif isinstance(content, list):
|
||||
# Replace specific text item in list content
|
||||
if isinstance(messages[msg_idx]["content"][content_idx_optional], dict):
|
||||
messages[msg_idx]["content"][content_idx_optional]["text"] = guardrail_response
|
||||
if isinstance(messages[msg_idx][field][content_idx_optional], dict):
|
||||
messages[msg_idx][field][content_idx_optional]["text"] = guardrail_response
|
||||
|
||||
async def process_output_response(
|
||||
self,
|
||||
|
|
|
|||
|
|
@ -6,6 +6,7 @@ from typing import Final, TypeAlias
|
|||
from pydantic import BaseModel, TypeAdapter, ValidationError
|
||||
|
||||
from litellm._logging import verbose_logger
|
||||
from litellm.llms.base_llm.guardrail_translation.utils import merge_returned_tools_into_request_tools, openai_tool_name
|
||||
from litellm.responses.litellm_completion_transformation.custom_tools import custom_tool_grammar_suffix
|
||||
from litellm.responses.litellm_completion_transformation.transformation import (
|
||||
NAMESPACE_DESCRIPTION_SEPARATOR,
|
||||
|
|
@ -183,3 +184,18 @@ def merge_guardrailed_tools(
|
|||
tuple(tool for key, tool in zip(guardrailed_keys, guardrailed) if key not in owned_keys)
|
||||
)
|
||||
return tuple(chain(merged_originals, appended))
|
||||
|
||||
|
||||
def append_returned_tools(
|
||||
original_tools: Sequence[Tool],
|
||||
flattened_groups: Sequence[Sequence[Tool]],
|
||||
returned_tools: Iterable[object],
|
||||
) -> tuple[Tool, ...]:
|
||||
request_chat_tools: Final = tuple(chain.from_iterable(flattened_groups))
|
||||
additions: Final = merge_returned_tools_into_request_tools(
|
||||
request_tools=request_chat_tools, returned_tools=_validated_tools(returned_tools), tool_name=openai_tool_name
|
||||
)[len(request_chat_tools) :]
|
||||
return (
|
||||
*original_tools,
|
||||
*LiteLLMCompletionResponsesConfig.transform_chat_completion_tool_params_to_responses_api_tools(additions),
|
||||
)
|
||||
|
|
|
|||
|
|
@ -49,6 +49,7 @@ from litellm.llms.base_llm.guardrail_translation.utils import (
|
|||
from litellm.llms.bedrock.base_aws_llm import BaseAWSLLM, bedrock_bearer_token, run_aws_signing
|
||||
from litellm.llms.bedrock.guardrail_attachments import find_request_attachments
|
||||
from litellm.llms.custom_httpx.http_handler import (
|
||||
AsyncHTTPHandler,
|
||||
get_async_httpx_client,
|
||||
httpxSpecialProvider,
|
||||
)
|
||||
|
|
@ -286,9 +287,12 @@ class BedrockGuardrail(CustomGuardrail, BaseAWSLLM):
|
|||
streaming_sampling_rate: int | None = None,
|
||||
streaming_end_of_stream_only: bool | None = None,
|
||||
streaming_buffer_release_on_scan: bool | None = None,
|
||||
async_handler: AsyncHTTPHandler | None = None,
|
||||
**kwargs,
|
||||
):
|
||||
self.async_handler = get_async_httpx_client(llm_provider=httpxSpecialProvider.GuardrailCallback)
|
||||
self.async_handler = async_handler or get_async_httpx_client(
|
||||
llm_provider=httpxSpecialProvider.GuardrailCallback
|
||||
)
|
||||
self._set_streaming_params(
|
||||
BedrockGuardrailStreamingParams.from_extras(
|
||||
MappingProxyType(
|
||||
|
|
|
|||
|
|
@ -6,20 +6,19 @@ model_list:
|
|||
model: gpt-4
|
||||
api_key: os.environ/OPENAI_API_KEY
|
||||
|
||||
litellm_settings:
|
||||
guardrails:
|
||||
- guardrail_name: "my-generic-guardrail"
|
||||
litellm_params:
|
||||
guardrail: generic_guardrail_api
|
||||
mode: pre_call # Options: pre_call, post_call, during_call, [pre_call, post_call]
|
||||
api_key: os.environ/GENERIC_GUARDRAIL_API_KEY # Optional if using Bearer auth
|
||||
api_base: http://localhost:8080 # Required. Endpoint /beta/litellm_basic_guardrail_api is automatically appended
|
||||
unreachable_fallback: fail_closed # Options: fail_closed (default, raise), fail_open (proceed if endpoint unreachable or upstream returns 502/503/504)
|
||||
default_on: false # Set to true to apply to all requests by default
|
||||
additional_provider_specific_params:
|
||||
# Any additional parameters your guardrail API needs
|
||||
api_version: "v1"
|
||||
custom_param: "value"
|
||||
guardrails:
|
||||
- guardrail_name: "my-generic-guardrail"
|
||||
litellm_params:
|
||||
guardrail: generic_guardrail_api
|
||||
mode: pre_call # Options: pre_call, post_call, during_call, [pre_call, post_call]
|
||||
api_key: os.environ/GENERIC_GUARDRAIL_API_KEY # Optional if using Bearer auth
|
||||
api_base: http://localhost:8080 # Required. Endpoint /beta/litellm_basic_guardrail_api is automatically appended
|
||||
unreachable_fallback: fail_closed # Options: fail_closed (default, raise), fail_open (proceed if endpoint unreachable or upstream returns 502/503/504)
|
||||
default_on: false # Set to true to apply to all requests by default
|
||||
additional_provider_specific_params:
|
||||
# Any additional parameters your guardrail API needs
|
||||
api_version: "v1"
|
||||
custom_param: "value"
|
||||
|
||||
# Usage examples:
|
||||
|
||||
|
|
|
|||
|
|
@ -30,6 +30,7 @@ from litellm.integrations.custom_guardrail import (
|
|||
from litellm.llms.base_llm.guardrail_translation.utils import (
|
||||
effective_scan_only_tool_results_for_guardrail,
|
||||
effective_skip_system_message_for_guardrail,
|
||||
effective_skip_tool_message_for_guardrail,
|
||||
role_out_of_guardrail_scope,
|
||||
)
|
||||
from litellm.llms.custom_httpx.http_handler import (
|
||||
|
|
@ -37,12 +38,16 @@ from litellm.llms.custom_httpx.http_handler import (
|
|||
get_async_httpx_client,
|
||||
httpxSpecialProvider,
|
||||
)
|
||||
from litellm.llms.openai.responses.guardrail_translation.handler import scannable_instructions
|
||||
from litellm.llms.openai.responses.guardrail_translation.handler import (
|
||||
scannable_instructions,
|
||||
tool_output_texts,
|
||||
)
|
||||
from litellm.proxy._types import UserAPIKeyAuth
|
||||
from litellm.proxy.common_utils.callback_utils import (
|
||||
add_guardrail_scan_id,
|
||||
add_guardrail_to_applied_guardrails_header,
|
||||
)
|
||||
from litellm.responses.litellm_completion_transformation.transformation import TOOL_CALL_OUTPUT_ITEM_TYPES
|
||||
from litellm.types.guardrails import GuardrailEventHooks
|
||||
from litellm.types.llms.base import LiteLLMBaseModel
|
||||
from litellm.types.llms.openai import AllMessageValues
|
||||
|
|
@ -112,12 +117,16 @@ class _ResponsesInputItem(LiteLLMBaseModel):
|
|||
type: str | None = None
|
||||
role: str | None = None
|
||||
content: str | tuple[_ResponsesContentPart, ...] | None = None
|
||||
output: object = None
|
||||
call_id: object = None
|
||||
|
||||
def text_count(self, *, skip_system: bool) -> int:
|
||||
if role_out_of_guardrail_scope(
|
||||
(self.role or "").lower(), skip_system_message=skip_system, skip_tool_message=False
|
||||
):
|
||||
def text_count(self, *, skip_system: bool, skip_tool: bool) -> int:
|
||||
is_tool_output: Final = self.type in TOOL_CALL_OUTPUT_ITEM_TYPES
|
||||
role: Final = "tool" if is_tool_output else (self.role or "").lower()
|
||||
if role_out_of_guardrail_scope(role, skip_system_message=skip_system, skip_tool_message=skip_tool):
|
||||
return 0
|
||||
if is_tool_output:
|
||||
return len(tool_output_texts(self))
|
||||
if isinstance(self.content, str):
|
||||
return 1
|
||||
if self.content is None:
|
||||
|
|
@ -1692,13 +1701,14 @@ class PanwPrismaAirsHandler(CustomGuardrail):
|
|||
request_data: Mapping[str, object],
|
||||
*,
|
||||
skip_system: bool,
|
||||
skip_tool: bool,
|
||||
) -> frozenset[int] | None:
|
||||
"""Return the ``texts`` indices flattened from Responses ``reasoning`` input items.
|
||||
|
||||
The Responses translation handler gives those model-authored items the default
|
||||
``user`` role, so the latest-turn selection must not mistake one for a human turn.
|
||||
Empty for requests without a Responses ``input`` item list; None when the raw items
|
||||
(after the leading ``instructions`` text, both minus whatever ``skip_system`` drops)
|
||||
(after the leading ``instructions`` text, both minus whatever ``skip_system`` and ``skip_tool`` drop)
|
||||
do not account for every entry of ``texts``.
|
||||
"""
|
||||
try:
|
||||
|
|
@ -1708,7 +1718,7 @@ class PanwPrismaAirsHandler(CustomGuardrail):
|
|||
if not isinstance(raw_input, tuple):
|
||||
return frozenset()
|
||||
offset: Final = 0 if scannable_instructions(request_data, skip_system=skip_system) is None else 1
|
||||
counts: Final = tuple(item.text_count(skip_system=skip_system) for item in raw_input)
|
||||
counts: Final = tuple(item.text_count(skip_system=skip_system, skip_tool=skip_tool) for item in raw_input)
|
||||
if offset + sum(counts) != len(texts):
|
||||
return None
|
||||
starts: Final = itertools.accumulate(counts, initial=offset)
|
||||
|
|
@ -1738,7 +1748,10 @@ class PanwPrismaAirsHandler(CustomGuardrail):
|
|||
if sources is None:
|
||||
return None
|
||||
reasoning: Final = self._reasoning_item_text_indices(
|
||||
texts, request_data, skip_system=effective_skip_system_message_for_guardrail(self)
|
||||
texts,
|
||||
request_data,
|
||||
skip_system=effective_skip_system_message_for_guardrail(self),
|
||||
skip_tool=effective_skip_tool_message_for_guardrail(self),
|
||||
)
|
||||
if reasoning is None:
|
||||
return None
|
||||
|
|
|
|||
|
|
@ -321,7 +321,7 @@ def _responses_file_part(item: Mapping[str, object]) -> _ResponsesFilePart:
|
|||
)
|
||||
|
||||
|
||||
def _decode_responses_media(item: Mapping[str, object]) -> _ResponsesMediaPart | None:
|
||||
def decode_responses_media(item: Mapping[str, object]) -> _ResponsesMediaPart | None:
|
||||
kind: Final = item.get("type")
|
||||
if kind in ("input_image", "image_url", "computer_screenshot") and any(key in item for key in ("image_url", "url")):
|
||||
image: Final = _responses_image_part(item)
|
||||
|
|
@ -383,6 +383,52 @@ def _attribute_fields(value: object) -> dict[str, object]:
|
|||
return dict(cast("Iterable[tuple[str, object]]", value)) # cast-ok: dict() raises on non-pair values, as before
|
||||
|
||||
|
||||
def tool_output_to_chat_content(
|
||||
output: object,
|
||||
) -> str | list[ToolMessageContentPart]:
|
||||
if output is None:
|
||||
return ""
|
||||
if isinstance(output, str):
|
||||
return output
|
||||
|
||||
single_media: Final = decode_responses_media(output) if isinstance(output, Mapping) else None
|
||||
if single_media is not None:
|
||||
return [single_media]
|
||||
if isinstance(output, list):
|
||||
normalized_blocks: Final[list[ToolMessageContentPart]] = []
|
||||
text_acc: Final[list[str]] = []
|
||||
for part in output:
|
||||
if not isinstance(part, dict):
|
||||
continue
|
||||
part_type = part.get("type")
|
||||
if part_type in ("input_text", "output_text", "text"):
|
||||
txt = part.get("text")
|
||||
if isinstance(txt, str) and txt:
|
||||
text_acc.append(txt)
|
||||
normalized_blocks.append({"type": "text", "text": txt})
|
||||
elif (media := decode_responses_media(part)) is not None:
|
||||
normalized_blocks.append(media)
|
||||
|
||||
if any(b.get("type") != "text" for b in normalized_blocks):
|
||||
return normalized_blocks
|
||||
if text_acc:
|
||||
return "".join(text_acc)
|
||||
try:
|
||||
return json.dumps(output)
|
||||
except Exception:
|
||||
return str(output)
|
||||
|
||||
try:
|
||||
return json.dumps(output)
|
||||
except Exception:
|
||||
return str(output)
|
||||
|
||||
|
||||
TOOL_CALL_OUTPUT_ITEM_TYPES: Final = frozenset(
|
||||
{"function_call_output", "custom_tool_call_output", "computer_call_output", "tool_result"}
|
||||
)
|
||||
|
||||
|
||||
def _input_item_role(input_item: Mapping[str, object]) -> str:
|
||||
return cast(str, input_item.get("role") or "user") # cast-ok: client-supplied role forwarded verbatim, unvalidated
|
||||
|
||||
|
|
@ -2371,12 +2417,7 @@ class LiteLLMCompletionResponsesConfig:
|
|||
"""
|
||||
Check if the input item is a tool call output
|
||||
"""
|
||||
return input_item.get("type") in [
|
||||
"function_call_output",
|
||||
"custom_tool_call_output",
|
||||
"computer_call_output",
|
||||
"tool_result", # Anthropic/MCP format
|
||||
]
|
||||
return input_item.get("type") in TOOL_CALL_OUTPUT_ITEM_TYPES
|
||||
|
||||
@staticmethod
|
||||
def _is_input_item_function_call(input_item: Mapping[str, object]) -> bool:
|
||||
|
|
@ -2400,58 +2441,7 @@ class LiteLLMCompletionResponsesConfig:
|
|||
if not call_id:
|
||||
return []
|
||||
|
||||
def _normalize_function_call_output_to_tool_content(
|
||||
output: object,
|
||||
) -> str | list[ToolMessageContentPart]:
|
||||
"""
|
||||
Preserve structured media for downstream chat adapters and token estimation.
|
||||
Text-only outputs retain their existing string representation.
|
||||
"""
|
||||
if output is None:
|
||||
return ""
|
||||
if isinstance(output, str):
|
||||
return output
|
||||
|
||||
single_media: Final = _decode_responses_media(output) if isinstance(output, Mapping) else None
|
||||
if single_media is not None:
|
||||
return [single_media]
|
||||
# Some adapters represent tool output as a list of "input_*" parts
|
||||
if isinstance(output, list):
|
||||
normalized_blocks: Final[list[ToolMessageContentPart]] = []
|
||||
text_acc: Final[list[str]] = []
|
||||
for part in output:
|
||||
if not isinstance(part, dict):
|
||||
continue
|
||||
part_type = part.get("type")
|
||||
if part_type in ("input_text", "output_text", "text"):
|
||||
txt = part.get("text")
|
||||
if isinstance(txt, str) and txt:
|
||||
text_acc.append(txt)
|
||||
normalized_blocks.append({"type": "text", "text": txt})
|
||||
elif (media := _decode_responses_media(part)) is not None:
|
||||
normalized_blocks.append(media)
|
||||
|
||||
if any(b.get("type") != "text" for b in normalized_blocks):
|
||||
return normalized_blocks
|
||||
if text_acc:
|
||||
return "".join(text_acc)
|
||||
try:
|
||||
# last resort: keep something meaningful for providers that require a string
|
||||
import json as _json
|
||||
|
||||
return _json.dumps(output)
|
||||
except Exception:
|
||||
return str(output)
|
||||
|
||||
# Fallback for dict/number/etc.
|
||||
try:
|
||||
import json as _json
|
||||
|
||||
return _json.dumps(output)
|
||||
except Exception:
|
||||
return str(output)
|
||||
|
||||
tool_content: Final = _normalize_function_call_output_to_tool_content(tool_call_output.get("output"))
|
||||
tool_content: Final = tool_output_to_chat_content(tool_call_output.get("output"))
|
||||
output_cache_control: Final = _input_item_cache_control(tool_call_output)
|
||||
tool_output_message: Final = (
|
||||
ChatCompletionToolMessage(role="tool", content=tool_content, tool_call_id=str(call_id))
|
||||
|
|
@ -2606,7 +2596,7 @@ class LiteLLMCompletionResponsesConfig:
|
|||
content_list.append(item)
|
||||
elif isinstance(item, dict):
|
||||
if (
|
||||
media := _decode_responses_media(
|
||||
media := decode_responses_media(
|
||||
fields := _validate_responses_media(_STR_KEY_DICT_ADAPTER, item)
|
||||
)
|
||||
) is not None:
|
||||
|
|
|
|||
|
|
@ -7,12 +7,14 @@ with guardrail transformations.
|
|||
|
||||
import copy
|
||||
import json
|
||||
import uuid
|
||||
from collections.abc import Callable
|
||||
from typing import Any, Final, List, Literal, Optional, Tuple
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import logging
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
|
||||
|
|
@ -33,6 +35,10 @@ from litellm.llms.openai.responses.guardrail_translation.handler import (
|
|||
)
|
||||
from litellm.llms.openai.responses.guardrail_translation.tool_merge import merge_guardrailed_tools
|
||||
from litellm.proxy.guardrails.guardrail_hooks.generic_guardrail_api import GenericGuardrailAPI
|
||||
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler
|
||||
from litellm.proxy.guardrails.guardrail_hooks.bedrock_guardrails import BedrockGuardrail
|
||||
from litellm.proxy.guardrails.guardrail_hooks.litellm_content_filter.content_filter import ContentFilterGuardrail
|
||||
from litellm.types.guardrails import ContentFilterAction, ContentFilterPattern
|
||||
from litellm.types.llms.openai import ChatCompletionToolCallChunk
|
||||
from litellm.responses.litellm_completion_transformation.compaction_items import compaction_summary
|
||||
from litellm.responses.litellm_completion_transformation.transformation import (
|
||||
|
|
@ -83,6 +89,7 @@ class RecordingMaskingGuardrail(MockGuardrail):
|
|||
super().__init__(guardrail_name=guardrail_name)
|
||||
self.seen_texts: list[list[str]] = []
|
||||
self.seen_message_contents: list[list[object]] = []
|
||||
self.seen_tools: list[object] = []
|
||||
|
||||
async def apply_guardrail(
|
||||
self,
|
||||
|
|
@ -93,6 +100,7 @@ class RecordingMaskingGuardrail(MockGuardrail):
|
|||
) -> GenericGuardrailAPIInputs:
|
||||
self.seen_texts.append(list(inputs.get("texts", [])))
|
||||
self.seen_message_contents.append([m["content"] for m in inputs.get("structured_messages") or []])
|
||||
self.seen_tools.append(inputs.get("tools"))
|
||||
return await super().apply_guardrail(inputs, request_data, input_type, logging_obj)
|
||||
|
||||
|
||||
|
|
@ -588,6 +596,638 @@ class TestSkipSystemMessageScopesInstructions:
|
|||
]
|
||||
|
||||
|
||||
_TOOL_TURNS: Final = (
|
||||
(
|
||||
{"type": "function_call", "call_id": "call_1", "name": "get_weather", "arguments": '{"city": "Paris"}'},
|
||||
"function_call_output",
|
||||
),
|
||||
(
|
||||
{"type": "custom_tool_call", "call_id": "call_1", "name": "run_sql", "input": "select 1"},
|
||||
"custom_tool_call_output",
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def _tool_result_request(call_item: dict[str, str], output_type: str, output: object) -> dict[str, object]:
|
||||
return {
|
||||
"model": "gpt-5.4-mini",
|
||||
"input": [
|
||||
{"role": "user", "content": "Weather?"},
|
||||
call_item,
|
||||
{"type": output_type, "call_id": "call_1", "output": output},
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
def _bedrock_guardrail(
|
||||
name: str, bedrock_api: Callable[[httpx.Request], httpx.Response], **settings: bool
|
||||
) -> BedrockGuardrail:
|
||||
return BedrockGuardrail(
|
||||
guardrail_name=name,
|
||||
guardrailIdentifier="test-guardrail",
|
||||
guardrailVersion="DRAFT",
|
||||
aws_access_key_id="test-access-key",
|
||||
aws_secret_access_key="test-secret-key",
|
||||
aws_region_name="us-east-1",
|
||||
async_handler=AsyncHTTPHandler(transport=httpx.MockTransport(bedrock_api)),
|
||||
**settings,
|
||||
)
|
||||
|
||||
|
||||
def _bedrock_block_texts(request: httpx.Request) -> list[str]:
|
||||
return [block["text"]["text"] for block in json.loads(request.content)["content"]]
|
||||
|
||||
|
||||
class TestToolResultItemsReachTextsGuardrails:
|
||||
"""A Responses tool result is scanned and rewritten in its `output` field the way a chat `tool`
|
||||
message's content is, and skip_tool_message_in_guardrail keeps it out of the scan the same way."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize(("call_item", "output_type"), _TOOL_TURNS, ids=["function", "custom"])
|
||||
async def test_string_output_is_scanned_and_rewritten_in_place(
|
||||
self, call_item: dict[str, str], output_type: str
|
||||
) -> None:
|
||||
guardrail: Final = RecordingMaskingGuardrail(guardrail_name="test")
|
||||
|
||||
result: Final = await OpenAIResponsesHandler().process_input_messages(
|
||||
_tool_result_request(call_item, output_type, "SECRET forecast"), guardrail
|
||||
)
|
||||
|
||||
assert guardrail.seen_texts == [["Weather?", "SECRET forecast"]]
|
||||
assert result["input"] == [
|
||||
{"role": "user", "content": "Weather? [GUARDRAILED]"},
|
||||
call_item,
|
||||
{"type": output_type, "call_id": "call_1", "output": "SECRET forecast [GUARDRAILED]"},
|
||||
]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_content_part_output_is_scanned_and_rewritten_in_its_part(self) -> None:
|
||||
guardrail: Final = RecordingMaskingGuardrail(guardrail_name="test")
|
||||
call_item, output_type = _TOOL_TURNS[0]
|
||||
image_part: Final = {"type": "input_image", "image_url": "https://example.com/map.png"}
|
||||
|
||||
result: Final = await OpenAIResponsesHandler().process_input_messages(
|
||||
_tool_result_request(
|
||||
call_item, output_type, [{"type": "input_text", "text": "SECRET forecast"}, image_part]
|
||||
),
|
||||
guardrail,
|
||||
)
|
||||
|
||||
assert guardrail.seen_texts == [["Weather?", "SECRET forecast"]]
|
||||
assert result["input"][2]["output"] == [
|
||||
{"type": "input_text", "text": "SECRET forecast [GUARDRAILED]"},
|
||||
image_part,
|
||||
]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("part_type", ["input_text", "output_text"])
|
||||
async def test_text_only_parts_are_scanned_and_rewritten_part_by_part(self, part_type: str) -> None:
|
||||
guardrail: Final = RecordingMaskingGuardrail(guardrail_name="test")
|
||||
call_item, output_type = _TOOL_TURNS[0]
|
||||
parts: Final = [
|
||||
{"type": part_type, "text": "SECRET forecast: ", "annotations": []},
|
||||
{"type": part_type, "text": "sunny"},
|
||||
]
|
||||
|
||||
result: Final = await OpenAIResponsesHandler().process_input_messages(
|
||||
_tool_result_request(call_item, output_type, parts), guardrail
|
||||
)
|
||||
|
||||
assert guardrail.seen_texts == [["Weather?", "SECRET forecast: ", "sunny"]]
|
||||
assert result["input"][2]["output"] == [
|
||||
{"type": part_type, "text": "SECRET forecast: [GUARDRAILED]", "annotations": []},
|
||||
{"type": part_type, "text": "sunny [GUARDRAILED]"},
|
||||
]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_pattern_ending_at_a_part_boundary_blocks_like_it_does_on_chat(self) -> None:
|
||||
access_key: Final = "AKIA" + "SYNTHETIC0000000"
|
||||
content_filter: Final = ContentFilterGuardrail(
|
||||
guardrail_name="secrets",
|
||||
patterns=[
|
||||
ContentFilterPattern(
|
||||
pattern_type="prebuilt", pattern_name="aws_access_key", action=ContentFilterAction.BLOCK
|
||||
)
|
||||
],
|
||||
)
|
||||
call_item, output_type = _TOOL_TURNS[0]
|
||||
parts: Final = [
|
||||
{"type": "input_text", "text": f"key={access_key}"},
|
||||
{"type": "input_text", "text": "region=us-east-1"},
|
||||
]
|
||||
|
||||
with pytest.raises(HTTPException) as blocked:
|
||||
await OpenAIResponsesHandler().process_input_messages(
|
||||
_tool_result_request(call_item, output_type, parts), content_filter
|
||||
)
|
||||
|
||||
assert blocked.value.status_code == 400
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("empty_output", ["", None], ids=["empty_string", "null"])
|
||||
async def test_empty_tool_output_sends_no_text_and_stays_as_sent(self, empty_output: str | None) -> None:
|
||||
guardrail: Final = RecordingMaskingGuardrail(guardrail_name="test")
|
||||
call_item, output_type = _TOOL_TURNS[0]
|
||||
|
||||
result: Final = await OpenAIResponsesHandler().process_input_messages(
|
||||
_tool_result_request(call_item, output_type, empty_output), guardrail
|
||||
)
|
||||
|
||||
assert guardrail.seen_texts == [["Weather?"]]
|
||||
assert result["input"][2] == {"type": output_type, "call_id": "call_1", "output": empty_output}
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("latest_user_only", [False, True], ids=["all_texts", "latest_user_only"])
|
||||
@pytest.mark.parametrize(
|
||||
"tool_output",
|
||||
[
|
||||
{"type": "function_call_output", "call_id": "call_1", "output": ""},
|
||||
{"type": "function_call_output", "call_id": "call_1", "output": None},
|
||||
{
|
||||
"type": "computer_call_output",
|
||||
"call_id": "call_1",
|
||||
"output": {"type": "computer_screenshot", "file_id": "file-screen"},
|
||||
},
|
||||
],
|
||||
ids=["empty_string", "null", "file_id_screenshot"],
|
||||
)
|
||||
async def test_bedrock_masking_applies_to_a_request_with_a_textless_tool_output(
|
||||
self, tool_output: dict[str, object], latest_user_only: bool
|
||||
) -> None:
|
||||
sent_texts: Final[list[list[str]]] = []
|
||||
|
||||
def bedrock_masking_emails(request: httpx.Request) -> httpx.Response:
|
||||
texts: Final = _bedrock_block_texts(request)
|
||||
sent_texts.append(texts)
|
||||
outputs: Final = [{"text": text.replace("jane@example.com", "{EMAIL}")} for text in texts]
|
||||
return httpx.Response(200, json={"action": "GUARDRAIL_INTERVENED", "outputs": outputs})
|
||||
|
||||
guardrail: Final = _bedrock_guardrail(
|
||||
"bedrock-pre", bedrock_masking_emails, experimental_use_latest_role_message_only=latest_user_only
|
||||
)
|
||||
|
||||
request: Final = {
|
||||
"model": "gpt-5.4-mini",
|
||||
"input": [
|
||||
{"role": "user", "content": "Look up the order."},
|
||||
{"type": "function_call", "call_id": "call_1", "name": "lookup", "arguments": "{}"},
|
||||
copy.deepcopy(tool_output),
|
||||
{"role": "user", "content": "Email jane@example.com the result."},
|
||||
],
|
||||
}
|
||||
|
||||
result: Final = await OpenAIResponsesHandler().process_input_messages(request, guardrail)
|
||||
|
||||
expected_texts: Final = (
|
||||
[["Email jane@example.com the result."]]
|
||||
if latest_user_only
|
||||
else [["Look up the order.", "Email jane@example.com the result."]]
|
||||
)
|
||||
assert sent_texts == expected_texts
|
||||
assert result["input"][3]["content"] == "Email {EMAIL} the result."
|
||||
assert result["input"][2] == tool_output
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_bedrock_latest_user_masking_applies_with_a_multi_part_tool_output(self) -> None:
|
||||
sent_texts: Final[list[list[str]]] = []
|
||||
|
||||
def bedrock_masking_emails(request: httpx.Request) -> httpx.Response:
|
||||
texts: Final = _bedrock_block_texts(request)
|
||||
sent_texts.append(texts)
|
||||
outputs: Final = [{"text": text.replace("jane@example.com", "{EMAIL}")} for text in texts]
|
||||
return httpx.Response(200, json={"action": "GUARDRAIL_INTERVENED", "outputs": outputs})
|
||||
|
||||
guardrail: Final = _bedrock_guardrail(
|
||||
"bedrock-pre", bedrock_masking_emails, experimental_use_latest_role_message_only=True
|
||||
)
|
||||
tool_output: Final = {
|
||||
"type": "function_call_output",
|
||||
"call_id": "call_1",
|
||||
"output": [{"type": "input_text", "text": "order 42"}, {"type": "input_text", "text": "shipped"}],
|
||||
}
|
||||
request: Final = {
|
||||
"model": "gpt-5.4-mini",
|
||||
"input": [
|
||||
{"role": "user", "content": "Look up the order."},
|
||||
{"type": "function_call", "call_id": "call_1", "name": "lookup", "arguments": "{}"},
|
||||
copy.deepcopy(tool_output),
|
||||
{"role": "user", "content": "Email jane@example.com the result."},
|
||||
],
|
||||
}
|
||||
|
||||
result: Final = await OpenAIResponsesHandler().process_input_messages(request, guardrail)
|
||||
|
||||
assert sent_texts == [["Email jane@example.com the result."]]
|
||||
assert result["input"][3]["content"] == "Email {EMAIL} the result."
|
||||
assert result["input"][2] == tool_output
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_bedrock_only_scan_new_messages_scans_only_the_new_tool_result(self) -> None:
|
||||
sent_texts: Final[list[list[str]]] = []
|
||||
|
||||
def bedrock_allowing(request: httpx.Request) -> httpx.Response:
|
||||
sent_texts.append(_bedrock_block_texts(request))
|
||||
return httpx.Response(200, json={"action": "NONE", "outputs": []})
|
||||
|
||||
guardrail: Final = _bedrock_guardrail("bedrock-incremental", bedrock_allowing, only_scan_new_messages=True)
|
||||
|
||||
first_turn: Final = [
|
||||
{"role": "user", "content": "Weather in Paris and Rome?"},
|
||||
{"type": "function_call", "call_id": "call_1", "name": "get_weather", "arguments": '{"city": "Paris"}'},
|
||||
{"type": "function_call_output", "call_id": "call_1", "output": "Paris: sunny"},
|
||||
]
|
||||
second_turn: Final = [
|
||||
*first_turn,
|
||||
{"type": "function_call", "call_id": "call_2", "name": "get_weather", "arguments": '{"city": "Rome"}'},
|
||||
{"type": "function_call_output", "call_id": "call_2", "output": "Rome: rain"},
|
||||
]
|
||||
session: Final = {"litellm_session_id": f"session-{uuid.uuid4()}"}
|
||||
|
||||
for turn in (first_turn, second_turn):
|
||||
await OpenAIResponsesHandler().process_input_messages(
|
||||
{"model": "gpt-5.4-mini", "input": copy.deepcopy(turn), **session}, guardrail
|
||||
)
|
||||
|
||||
assert sent_texts == [["Weather in Paris and Rome?", "Paris: sunny"], ["Rome: rain"]]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_file_id_screenshot_is_not_scanned_as_text_or_rewritten(self) -> None:
|
||||
guardrail: Final = RecordingMaskingGuardrail(guardrail_name="test")
|
||||
screenshot_output: Final = {
|
||||
"type": "computer_call_output",
|
||||
"call_id": "cu_1",
|
||||
"output": {"type": "computer_screenshot", "file_id": "file-SECRET-screen"},
|
||||
}
|
||||
request: Final = {
|
||||
"model": "gpt-5.4-mini",
|
||||
"input": [
|
||||
{"role": "user", "content": "Open the page"},
|
||||
{"type": "computer_call", "call_id": "cu_1", "id": "cu_1", "action": {"type": "screenshot"}},
|
||||
copy.deepcopy(screenshot_output),
|
||||
],
|
||||
}
|
||||
|
||||
result: Final = await OpenAIResponsesHandler().process_input_messages(request, guardrail)
|
||||
|
||||
assert guardrail.seen_texts == [["Open the page"]]
|
||||
assert result["input"][2] == screenshot_output
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_text_parts_between_file_images_are_rewritten_in_place(self) -> None:
|
||||
guardrail: Final = RecordingMaskingGuardrail(guardrail_name="test")
|
||||
call_item, output_type = _TOOL_TURNS[0]
|
||||
logo_part: Final = {"type": "input_image", "file_id": "file-logo"}
|
||||
image_part: Final = {"type": "input_image", "file_id": "file-map"}
|
||||
parts: Final = [
|
||||
logo_part,
|
||||
{"type": "input_text", "text": "SECRET forecast: "},
|
||||
image_part,
|
||||
{"type": "input_text", "text": "sunny"},
|
||||
]
|
||||
|
||||
result: Final = await OpenAIResponsesHandler().process_input_messages(
|
||||
_tool_result_request(call_item, output_type, parts), guardrail
|
||||
)
|
||||
|
||||
assert guardrail.seen_texts == [["Weather?", "SECRET forecast: ", "sunny"]]
|
||||
assert result["input"][2]["output"] == [
|
||||
logo_part,
|
||||
{"type": "input_text", "text": "SECRET forecast: [GUARDRAILED]"},
|
||||
image_part,
|
||||
{"type": "input_text", "text": "sunny [GUARDRAILED]"},
|
||||
]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_dict_output_is_rewritten_as_the_text_the_guardrail_returns(self) -> None:
|
||||
guardrail: Final = RecordingMaskingGuardrail(guardrail_name="test")
|
||||
call_item, output_type = _TOOL_TURNS[0]
|
||||
|
||||
result: Final = await OpenAIResponsesHandler().process_input_messages(
|
||||
_tool_result_request(call_item, output_type, {"result": "SECRET forecast"}), guardrail
|
||||
)
|
||||
|
||||
assert result["input"][2]["output"] == '{"result": "SECRET forecast"} [GUARDRAILED]'
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_dict_output_the_guardrail_leaves_unchanged_stays_a_dict(self) -> None:
|
||||
guardrail: Final = TextsReplacingGuardrail(guardrail_name="test", texts=("[MASKED]", '{"result": "sunny"}'))
|
||||
call_item, output_type = _TOOL_TURNS[0]
|
||||
|
||||
result: Final = await OpenAIResponsesHandler().process_input_messages(
|
||||
_tool_result_request(call_item, output_type, {"result": "sunny"}), guardrail
|
||||
)
|
||||
|
||||
assert result["input"][0]["content"] == "[MASKED]"
|
||||
assert result["input"][2]["output"] == {"result": "sunny"}
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_tool_result_without_call_id_is_left_out_like_its_structured_row(self) -> None:
|
||||
guardrail: Final = RecordingMaskingGuardrail(guardrail_name="test")
|
||||
call_item, output_type = _TOOL_TURNS[0]
|
||||
orphan: Final = {"type": output_type, "output": "SECRET orphan"}
|
||||
request: Final = {
|
||||
"model": "gpt-5.4-mini",
|
||||
"input": [
|
||||
{"role": "user", "content": "Weather?"},
|
||||
call_item,
|
||||
dict(orphan),
|
||||
{"type": output_type, "call_id": "call_1", "output": "SECRET forecast"},
|
||||
],
|
||||
}
|
||||
|
||||
result: Final = await OpenAIResponsesHandler().process_input_messages(request, guardrail)
|
||||
|
||||
assert guardrail.seen_texts == [["Weather?", "SECRET forecast"]]
|
||||
assert "SECRET orphan" not in guardrail.seen_message_contents[0]
|
||||
assert result["input"][2] == orphan
|
||||
assert result["input"][3]["output"] == "SECRET forecast [GUARDRAILED]"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_textless_output_sharing_a_call_id_keeps_the_text_output_in_structured_messages(self) -> None:
|
||||
guardrail: Final = RecordingMaskingGuardrail(guardrail_name="test")
|
||||
call_item, output_type = _TOOL_TURNS[0]
|
||||
request: Final = _tool_result_request(call_item, output_type, "")
|
||||
request["input"].append({"type": output_type, "call_id": "call_1", "output": "SECRET forecast"})
|
||||
|
||||
result: Final = await OpenAIResponsesHandler().process_input_messages(request, guardrail)
|
||||
|
||||
assert guardrail.seen_texts == [["Weather?", "SECRET forecast"]]
|
||||
assert "SECRET forecast" in guardrail.seen_message_contents[0]
|
||||
assert "" not in guardrail.seen_message_contents[0]
|
||||
assert result["input"][3]["output"] == "SECRET forecast [GUARDRAILED]"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_multi_part_outputs_sharing_a_call_id_each_keep_their_own_text_in_structured_messages(
|
||||
self,
|
||||
) -> None:
|
||||
guardrail: Final = RecordingMaskingGuardrail(guardrail_name="test")
|
||||
call_item, output_type = _TOOL_TURNS[0]
|
||||
request: Final = _tool_result_request(
|
||||
call_item,
|
||||
output_type,
|
||||
[{"type": "input_text", "text": "first "}, {"type": "input_text", "text": "output"}],
|
||||
)
|
||||
request["input"].append(
|
||||
{
|
||||
"type": output_type,
|
||||
"call_id": "call_1",
|
||||
"output": [{"type": "input_text", "text": "second "}, {"type": "input_text", "text": "output"}],
|
||||
}
|
||||
)
|
||||
|
||||
result: Final = await OpenAIResponsesHandler().process_input_messages(request, guardrail)
|
||||
|
||||
assert guardrail.seen_texts == [["Weather?", "first ", "output", "second ", "output"]]
|
||||
assert guardrail.seen_message_contents[0][2:] == [
|
||||
[{"type": "text", "text": "first "}, {"type": "text", "text": "output"}],
|
||||
[{"type": "text", "text": "second "}, {"type": "text", "text": "output"}],
|
||||
]
|
||||
assert [part["text"] for part in result["input"][3]["output"]] == [
|
||||
"second [GUARDRAILED]",
|
||||
"output [GUARDRAILED]",
|
||||
]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize(
|
||||
"textless_output",
|
||||
[
|
||||
{"type": "function_call_output", "call_id": "call_1", "output": ""},
|
||||
{"type": "function_call_output", "call_id": "call_1", "output": None},
|
||||
{
|
||||
"type": "computer_call_output",
|
||||
"call_id": "call_1",
|
||||
"output": {"type": "computer_screenshot", "file_id": "file-screen"},
|
||||
},
|
||||
],
|
||||
ids=["empty_string", "null", "file_id_screenshot"],
|
||||
)
|
||||
async def test_multi_part_output_sharing_a_call_id_with_a_textless_output_lists_its_parts_in_structured_messages(
|
||||
self, textless_output: dict[str, object]
|
||||
) -> None:
|
||||
guardrail: Final = RecordingMaskingGuardrail(guardrail_name="test")
|
||||
call_item, output_type = _TOOL_TURNS[0]
|
||||
request: Final = _tool_result_request(
|
||||
call_item,
|
||||
output_type,
|
||||
[{"type": "input_text", "text": "first "}, {"type": "input_text", "text": "output"}],
|
||||
)
|
||||
request["input"].append(copy.deepcopy(textless_output))
|
||||
|
||||
await OpenAIResponsesHandler().process_input_messages(request, guardrail)
|
||||
|
||||
assert guardrail.seen_texts == [["Weather?", "first ", "output"]]
|
||||
assert guardrail.seen_message_contents[0][2:] == [
|
||||
[{"type": "text", "text": "first "}, {"type": "text", "text": "output"}],
|
||||
]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_tool_result_item_is_scanned_like_function_call_output(self) -> None:
|
||||
guardrail: Final = RecordingMaskingGuardrail(guardrail_name="test")
|
||||
call_item, _ = _TOOL_TURNS[0]
|
||||
|
||||
result: Final = await OpenAIResponsesHandler().process_input_messages(
|
||||
_tool_result_request(call_item, "tool_result", "SECRET forecast"), guardrail
|
||||
)
|
||||
|
||||
assert guardrail.seen_texts == [["Weather?", "SECRET forecast"]]
|
||||
assert guardrail.seen_message_contents[0][2] == "SECRET forecast"
|
||||
assert result["input"][2]["output"] == "SECRET forecast [GUARDRAILED]"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_text_only_parts_the_guardrail_leaves_unchanged_keep_their_parts(self) -> None:
|
||||
call_item, output_type = _TOOL_TURNS[0]
|
||||
parts: Final = [{"type": "input_text", "text": "SECRET forecast: "}, {"type": "input_text", "text": "sunny"}]
|
||||
|
||||
result: Final = await OpenAIResponsesHandler().process_input_messages(
|
||||
_tool_result_request(call_item, output_type, copy.deepcopy(parts)), CustomGuardrail(guardrail_name="test")
|
||||
)
|
||||
|
||||
assert result["input"][2]["output"] == parts
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_text_parts_a_structured_guardrail_returns_unchanged_keep_their_parts(self) -> None:
|
||||
class StructuredEchoGuardrail(CustomGuardrail):
|
||||
async def apply_guardrail(
|
||||
self,
|
||||
inputs: GenericGuardrailAPIInputs,
|
||||
request_data: dict[str, object],
|
||||
input_type: Literal["request", "response"],
|
||||
logging_obj: LiteLLMLoggingObj | None = None,
|
||||
) -> GenericGuardrailAPIInputs:
|
||||
return {**inputs, "structured_messages": copy.deepcopy(inputs.get("structured_messages") or [])}
|
||||
|
||||
call_item, output_type = _TOOL_TURNS[0]
|
||||
parts: Final = [
|
||||
{"type": "output_text", "text": "SECRET forecast: ", "annotations": []},
|
||||
{"type": "output_text", "text": "sunny"},
|
||||
]
|
||||
|
||||
result: Final = await OpenAIResponsesHandler().process_input_messages(
|
||||
_tool_result_request(call_item, output_type, copy.deepcopy(parts)),
|
||||
StructuredEchoGuardrail(guardrail_name="t"),
|
||||
)
|
||||
|
||||
assert result["input"][2]["output"] == parts
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize(
|
||||
("output", "expected_texts"),
|
||||
[
|
||||
pytest.param("SECRET forecast", ["SECRET forecast"], id="string"),
|
||||
pytest.param([{"type": "input_text", "text": "SECRET forecast"}], ["SECRET forecast"], id="one_part"),
|
||||
pytest.param(
|
||||
[{"type": "input_text", "text": "SECRET "}, {"type": "output_text", "text": "forecast"}],
|
||||
["SECRET ", "forecast"],
|
||||
id="parts",
|
||||
),
|
||||
pytest.param(
|
||||
[
|
||||
{"type": "input_text", "text": "SECRET "},
|
||||
{"type": "input_image", "image_url": "https://example.com/map.png"},
|
||||
{"type": "input_text", "text": ""},
|
||||
{"type": "input_text", "text": "forecast"},
|
||||
],
|
||||
["SECRET ", "forecast"],
|
||||
id="parts_with_image",
|
||||
),
|
||||
pytest.param(
|
||||
[
|
||||
{"type": "input_text", "text": "SECRET forecast"},
|
||||
{"type": "summary_text", "text": "not tool output"},
|
||||
],
|
||||
["SECRET forecast"],
|
||||
id="non_text_part_with_text_field",
|
||||
),
|
||||
pytest.param(
|
||||
["SECRET forecast", {"type": "input_text", "text": "visible"}],
|
||||
["visible"],
|
||||
id="non_dict_part",
|
||||
),
|
||||
pytest.param([], ["[]"], id="empty"),
|
||||
pytest.param({"result": "SECRET forecast"}, ['{"result": "SECRET forecast"}'], id="dict"),
|
||||
pytest.param(
|
||||
[{"type": "input_image", "file_id": "file-map"}],
|
||||
['[{"type": "input_image", "file_id": "file-map"}]'],
|
||||
id="no_text_parts",
|
||||
),
|
||||
pytest.param({"type": "input_image", "image_url": "https://example.com/map.png"}, [], id="single_image"),
|
||||
],
|
||||
)
|
||||
async def test_tool_result_texts_match_the_structured_tool_message(
|
||||
self, output: object, expected_texts: list[str]
|
||||
) -> None:
|
||||
guardrail: Final = RecordingMaskingGuardrail(guardrail_name="test")
|
||||
call_item, output_type = _TOOL_TURNS[0]
|
||||
|
||||
await OpenAIResponsesHandler().process_input_messages(
|
||||
_tool_result_request(call_item, output_type, output), guardrail
|
||||
)
|
||||
|
||||
tool_content: Final = guardrail.seen_message_contents[0][2]
|
||||
structured_texts: Final = (
|
||||
[tool_content]
|
||||
if isinstance(tool_content, str)
|
||||
else [part["text"] for part in tool_content if part.get("type") == "text"]
|
||||
)
|
||||
tool_texts: Final = guardrail.seen_texts[0][1:]
|
||||
assert tool_texts == expected_texts
|
||||
assert tool_texts == structured_texts
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize(("call_item", "output_type"), _TOOL_TURNS, ids=["function", "custom"])
|
||||
async def test_skip_tool_message_keeps_the_tool_result_out_of_texts_and_rows(
|
||||
self, call_item: dict[str, str], output_type: str
|
||||
) -> None:
|
||||
guardrail: Final = RecordingMaskingGuardrail(guardrail_name="test")
|
||||
guardrail.skip_tool_message_in_guardrail = True
|
||||
tool_result: Final = {"type": output_type, "call_id": "call_1", "output": "SECRET forecast"}
|
||||
|
||||
result: Final = await OpenAIResponsesHandler().process_input_messages(
|
||||
_tool_result_request(call_item, output_type, "SECRET forecast"), guardrail
|
||||
)
|
||||
|
||||
assert guardrail.seen_texts == [["Weather?"]]
|
||||
assert "SECRET forecast" not in guardrail.seen_message_contents[0]
|
||||
assert result["input"][0] == {"role": "user", "content": "Weather? [GUARDRAILED]"}
|
||||
assert result["input"][2] == tool_result
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize(("call_item", "output_type"), _TOOL_TURNS, ids=["function", "custom"])
|
||||
async def test_scan_only_tool_results_sends_only_the_tool_result(
|
||||
self, call_item: dict[str, str], output_type: str
|
||||
) -> None:
|
||||
guardrail: Final = RecordingMaskingGuardrail(guardrail_name="test")
|
||||
guardrail.scan_only_tool_results = True
|
||||
tool: Final = {"type": "function", "name": "get_weather", "parameters": {"type": "object", "properties": {}}}
|
||||
request: Final = {
|
||||
**_tool_result_request(call_item, output_type, "SECRET forecast"),
|
||||
"instructions": "Answer briefly",
|
||||
"tools": [tool],
|
||||
}
|
||||
|
||||
result: Final = await OpenAIResponsesHandler().process_input_messages(request, guardrail)
|
||||
|
||||
assert guardrail.seen_texts == [["SECRET forecast"]]
|
||||
assert guardrail.seen_message_contents == [["SECRET forecast"]]
|
||||
assert guardrail.seen_tools == [None]
|
||||
assert result["instructions"] == "Answer briefly"
|
||||
assert result["tools"] == [tool]
|
||||
assert result["input"] == [
|
||||
{"role": "user", "content": "Weather?"},
|
||||
call_item,
|
||||
{"type": output_type, "call_id": "call_1", "output": "SECRET forecast [GUARDRAILED]"},
|
||||
]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_scan_only_tool_results_appends_tools_the_guardrail_adds(self) -> None:
|
||||
added_tool: Final = {
|
||||
"type": "function",
|
||||
"function": {"name": "retrieve", "description": "Fetch the full text", "parameters": {"type": "object"}},
|
||||
}
|
||||
renamed_original: Final = {
|
||||
"type": "function",
|
||||
"function": {"name": "get_weather", "description": "rewritten", "parameters": {"type": "object"}},
|
||||
}
|
||||
|
||||
class ToolAddingGuardrail(RecordingMaskingGuardrail):
|
||||
async def apply_guardrail(
|
||||
self,
|
||||
inputs: GenericGuardrailAPIInputs,
|
||||
request_data: dict[str, object],
|
||||
input_type: Literal["request", "response"],
|
||||
logging_obj: LiteLLMLoggingObj | None = None,
|
||||
) -> GenericGuardrailAPIInputs:
|
||||
masked: Final = await super().apply_guardrail(inputs, request_data, input_type, logging_obj)
|
||||
return {**masked, "tools": [renamed_original, added_tool]}
|
||||
|
||||
guardrail: Final = ToolAddingGuardrail(guardrail_name="test")
|
||||
guardrail.scan_only_tool_results = True
|
||||
call_item, output_type = _TOOL_TURNS[0]
|
||||
tool: Final = {"type": "function", "name": "get_weather", "parameters": {"type": "object", "properties": {}}}
|
||||
request: Final = {**_tool_result_request(call_item, output_type, "SECRET forecast"), "tools": [tool]}
|
||||
|
||||
result: Final = await OpenAIResponsesHandler().process_input_messages(request, guardrail)
|
||||
|
||||
assert guardrail.seen_tools == [None]
|
||||
assert result["tools"][0] == tool
|
||||
assert [(t["type"], t["name"]) for t in result["tools"]] == [
|
||||
("function", "get_weather"),
|
||||
("function", "retrieve"),
|
||||
]
|
||||
assert result["tools"][1]["description"] == "Fetch the full text"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_scan_only_tool_results_leaves_a_request_without_tool_results_unscanned(self) -> None:
|
||||
guardrail: Final = RecordingMaskingGuardrail(guardrail_name="test")
|
||||
guardrail.scan_only_tool_results = True
|
||||
request: Final = {"model": "gpt-5.4-mini", "instructions": "Answer briefly", "input": "Weather?"}
|
||||
|
||||
result: Final = await OpenAIResponsesHandler().process_input_messages(dict(request), guardrail)
|
||||
|
||||
assert guardrail.seen_texts == []
|
||||
assert result == request
|
||||
|
||||
|
||||
class TestOpenAIResponsesHandlerOutputProcessing:
|
||||
"""Test output processing functionality"""
|
||||
|
||||
|
|
|
|||
|
|
@ -4834,6 +4834,83 @@ class TestPanwAirsLatestRoleMessageOnlyEveryRequestShape:
|
|||
None,
|
||||
id="function_call_output",
|
||||
),
|
||||
pytest.param(
|
||||
(
|
||||
{"type": "function_call", "call_id": "call_1", "name": "lookup", "arguments": "{}"},
|
||||
{
|
||||
"type": "function_call_output",
|
||||
"call_id": "call_1",
|
||||
"output": [{"type": "input_text", "text": "tool "}, {"type": "input_text", "text": "result"}],
|
||||
},
|
||||
),
|
||||
None,
|
||||
id="function_call_output_text_parts",
|
||||
),
|
||||
pytest.param(
|
||||
(
|
||||
{"type": "function_call", "call_id": "call_1", "name": "lookup", "arguments": "{}"},
|
||||
{
|
||||
"type": "function_call_output",
|
||||
"call_id": "call_1",
|
||||
"output": [
|
||||
{"type": "input_text", "text": "tool "},
|
||||
{"type": "input_image", "image_url": "https://example.com/chart.png"},
|
||||
{"type": "input_text", "text": "result"},
|
||||
],
|
||||
},
|
||||
),
|
||||
None,
|
||||
id="function_call_output_text_and_image_parts",
|
||||
),
|
||||
pytest.param(
|
||||
(
|
||||
{"type": "function_call", "call_id": "call_1", "name": "lookup", "arguments": "{}"},
|
||||
{"type": "function_call_output", "output": "orphan tool result"},
|
||||
{"type": "function_call_output", "call_id": "call_1", "output": "tool result"},
|
||||
),
|
||||
None,
|
||||
id="function_call_output_without_call_id",
|
||||
),
|
||||
pytest.param(
|
||||
(
|
||||
{"type": "function_call", "call_id": "call_1", "name": "lookup", "arguments": "{}"},
|
||||
{"type": "tool_result", "call_id": "call_1", "output": "tool result"},
|
||||
),
|
||||
None,
|
||||
id="tool_result",
|
||||
),
|
||||
pytest.param(
|
||||
(
|
||||
{"type": "computer_call", "call_id": "cu_1", "id": "cu_1", "action": {"type": "screenshot"}},
|
||||
{
|
||||
"type": "computer_call_output",
|
||||
"call_id": "cu_1",
|
||||
"output": {"type": "computer_screenshot", "image_url": "https://example.com/screen.png"},
|
||||
},
|
||||
),
|
||||
None,
|
||||
id="computer_call_output",
|
||||
),
|
||||
pytest.param(
|
||||
(
|
||||
{"type": "computer_call", "call_id": "cu_1", "id": "cu_1", "action": {"type": "screenshot"}},
|
||||
{
|
||||
"type": "computer_call_output",
|
||||
"call_id": "cu_1",
|
||||
"output": {"type": "computer_screenshot", "file_id": "file-screen"},
|
||||
},
|
||||
),
|
||||
None,
|
||||
id="computer_call_output_file_id",
|
||||
),
|
||||
pytest.param(
|
||||
(
|
||||
{"type": "function_call", "call_id": "call_1", "name": "lookup", "arguments": "{}"},
|
||||
{"type": "function_call_output", "call_id": "call_1", "output": ""},
|
||||
),
|
||||
None,
|
||||
id="function_call_output_empty",
|
||||
),
|
||||
pytest.param(
|
||||
({"type": "reasoning", "id": "rs_1", "summary": [{"type": "summary_text", "text": "thinking"}]},),
|
||||
None,
|
||||
|
|
@ -4898,6 +4975,25 @@ class TestPanwAirsLatestRoleMessageOnlyEveryRequestShape:
|
|||
|
||||
assert [call.kwargs["content"] for call in mock_api.call_args_list] == [self.LATEST]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_flag_true_with_skip_tool_still_scans_only_the_latest_turn_on_responses(self) -> None:
|
||||
from litellm.llms.openai.responses.guardrail_translation.handler import (
|
||||
OpenAIResponsesHandler,
|
||||
)
|
||||
|
||||
handler = make_handler(experimental_use_latest_role_message_only=True)
|
||||
handler.skip_tool_message_in_guardrail = True
|
||||
request_data = self._responses_request(
|
||||
{"type": "function_call", "call_id": "call_1", "name": "lookup", "arguments": "{}"},
|
||||
{"type": "function_call_output", "call_id": "call_1", "output": "tool result"},
|
||||
{"role": "user", "content": self.LATEST},
|
||||
)
|
||||
patcher, mock_api = self._scan(handler)
|
||||
with patcher:
|
||||
await OpenAIResponsesHandler().process_input_messages(data=request_data, guardrail_to_apply=handler)
|
||||
|
||||
assert [call.kwargs["content"] for call in mock_api.call_args_list] == [self.LATEST]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_flag_false_responses_scans_instructions_and_full_history(self) -> None:
|
||||
from litellm.llms.openai.responses.guardrail_translation.handler import (
|
||||
|
|
|
|||
|
|
@ -111,6 +111,8 @@ def test_media_normalization_preserves_formats_references_and_metadata(kind: str
|
|||
"keep",
|
||||
),
|
||||
([{"type": "input_image", "file_id": "file-only"}], '[{"type": "input_image", "file_id": "file-only"}]'),
|
||||
({"ids": {7}}, "{'ids': {7}}"),
|
||||
([{"type": "unknown", "ids": {7}}], "[{'type': 'unknown', 'ids': {7}}]"),
|
||||
],
|
||||
)
|
||||
def test_tool_output_without_usable_media_keeps_its_existing_fallback(output: object, expected: str) -> None:
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue