refactor(guardrails): type the request scan context helpers as read-only mappings

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
yucheng 2026-09-17 06:54:17 +00:00
parent 85444b56d9
commit b536289233
5 changed files with 34 additions and 14 deletions

View file

@ -965,10 +965,7 @@ class CustomGuardrail(CustomLogger):
translation: "BaseTranslation",
) -> dict[str, object]: # mutable-ok: BaseTranslation.process_output_response contract
"""The logged request in OpenAI chat shape, for an output scan whose translation differs from the input's."""
context: Final = translation.request_scan_context(
dict(scratch_request), # mutable-ok: BaseTranslation.request_scan_context requires a dict
self,
)
context: Final = translation.request_scan_context(scratch_request, self)
return {
**scratch_request,
"messages": list(context.structured_messages),

View file

@ -528,7 +528,9 @@ class AnthropicMessagesHandler(BaseTranslation):
)
return result if result else None
def request_scan_context(self, data: dict, guardrail_to_apply: "CustomGuardrail") -> RequestScanContext:
def request_scan_context(
self, data: Mapping[str, object], guardrail_to_apply: "CustomGuardrail"
) -> RequestScanContext:
if data.get("messages") is None:
return RequestScanContext()
translated: Final = self._translate_to_openai(
@ -715,9 +717,7 @@ class AnthropicMessagesHandler(BaseTranslation):
return data
def _hoisted_top_level_system_message(
self, data: dict
) -> AllMessageValues | None: # mutable-ok: API message payload
def _hoisted_top_level_system_message(self, data: Mapping[str, object]) -> AllMessageValues | None:
"""Return the system message produced by translating the top-level prompt."""
system: Final = data.get("system")
if not system:

View file

@ -1,5 +1,5 @@
from abc import ABC, abstractmethod
from collections.abc import Sequence
from collections.abc import Mapping, Sequence
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Any, ClassVar, Final, Optional
@ -7,6 +7,7 @@ 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,
request_tools,
response_assistant_turn,
scoped_structured_message_indices,
)
@ -301,16 +302,21 @@ class BaseTranslation(ABC):
"""
return None
def request_scan_context(self, data: dict, guardrail_to_apply: "CustomGuardrail") -> RequestScanContext:
def request_scan_context(
self, data: Mapping[str, object], guardrail_to_apply: "CustomGuardrail"
) -> RequestScanContext:
"""Override wherever ``process_input_messages`` scopes or translates the request differently."""
structured_messages: Final = self.get_structured_messages(
dict(data) # mutable-ok: get_structured_messages takes the request as a dict
)
return RequestScanContext.scoped(
self.get_structured_messages(data) or (), data.get("tools") or (), guardrail_to_apply
structured_messages or (), request_tools(data.get("tools")), guardrail_to_apply
)
def with_response_context(
self,
inputs: "GenericGuardrailAPIInputs",
request_data: dict | None,
request_data: Mapping[str, object] | None,
guardrail_to_apply: "CustomGuardrail",
) -> "GenericGuardrailAPIInputs":
"""``inputs`` plus the scoped request conversation, closed by the scanned reply, and the request tools."""

View file

@ -14,6 +14,7 @@ from litellm.types.llms.openai import (
ChatCompletionTextObject,
ChatCompletionToolCallChunk,
ChatCompletionToolCallFunctionChunk,
ChatCompletionToolParam,
ResponseAPIUsage,
)
@ -331,6 +332,15 @@ def response_assistant_turn(
ToolT = TypeVar("ToolT")
def request_tools(raw_tools: object) -> tuple[ChatCompletionToolParam, ...]:
"""The request's ``tools`` list, as the chat completion request model already validated it upstream."""
if not isinstance(raw_tools, list):
return ()
return tuple(
cast(Sequence[ChatCompletionToolParam], raw_tools) # cast-ok: the request model validated tools upstream
)
def openai_tool_name(tool: object) -> str | None:
if not isinstance(tool, dict):
return None

View file

@ -452,9 +452,16 @@ class OpenAIResponsesHandler(BaseTranslation):
)
return cast(list[AllMessageValues], messages) if messages else None
def request_scan_context(self, data: dict, guardrail_to_apply: "CustomGuardrail") -> RequestScanContext:
def request_scan_context(
self, data: Mapping[str, object], guardrail_to_apply: "CustomGuardrail"
) -> RequestScanContext:
raw_tools: Final = data.get("tools")
structured_messages: Final = tuple(self.get_structured_messages(data) or ())
structured_messages: Final = tuple(
self.get_structured_messages(
dict(data) # mutable-ok: get_structured_messages takes the request as a dict
)
or ()
)
return RequestScanContext(
structured_messages=structured_messages,
tools=tuple(