scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-25 01:02:15 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 81

refactor(guardrails): satisfy the type-discipline gate in the video handler

Browse source 
Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
yucheng 2026-09-21 22:55:01 +00:00
parent dfc5ef70b4
commit 652bddfdc6
2 changed files with 11 additions and 9 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
litellm/llms/openai/videos/guardrail_translation/__init__.py
View file

@ -7,7 +7,7 @@ from litellm.llms.openai.videos.guardrail_translation.handler import (
)
from litellm.types.utils import CallTypes
guardrail_translation_mappings: Final = {
guardrail_translation_mappings: Final = { # mutable-ok: discover_guardrail_translation_mappings only accepts isinstance(mappings, dict)
CallTypes.video_generation: OpenAIVideoGenerationHandler,
CallTypes.avideo_generation: OpenAIVideoGenerationHandler,
CallTypes.create_video: OpenAIVideoGenerationHandler,
@ -20,4 +20,4 @@ guardrail_translation_mappings: Final = {
CallTypes.avideo_extension: OpenAIVideoGenerationHandler,
}
__all__ = ["OpenAIVideoGenerationHandler", "guardrail_translation_mappings"]
__all__ = ("OpenAIVideoGenerationHandler", "guardrail_translation_mappings")

16
litellm/llms/openai/videos/guardrail_translation/handler.py
View file

@ -14,19 +14,20 @@ class OpenAIVideoGenerationHandler(BaseTranslation):
async def process_input_messages(
self,
data: dict[str, object],
data: dict[str, object], # mutable-ok: BaseTranslation contract passes the proxy's request dict through
guardrail_to_apply: "CustomGuardrail",
litellm_logging_obj: "LiteLLMLoggingObj | None" = None,
) -> dict[str, object]:
) -> dict[str, object]: # mutable-ok: BaseTranslation contract returns the proxy's request dict
prompt: Final = data.get("prompt")
if not isinstance(prompt, str):
return data
model: Final = data.get("model")
texts: Final = [prompt] # mutable-ok: GenericGuardrailAPIInputs.texts is declared list[str]
inputs: Final = (
GenericGuardrailAPIInputs(texts=[prompt], model=model)
GenericGuardrailAPIInputs(texts=texts, model=model)
if isinstance(model, str)
else GenericGuardrailAPIInputs(texts=[prompt])
else GenericGuardrailAPIInputs(texts=texts)
)
guardrailed_inputs: Final = await guardrail_to_apply.apply_guardrail( # pyright: ignore[reportUnknownMemberType] # request_data is a bare dict
inputs=inputs,
@ -34,8 +35,9 @@ class OpenAIVideoGenerationHandler(BaseTranslation):
input_type="request",
logging_obj=litellm_logging_obj,
)
guardrailed_texts: Final = guardrailed_inputs.get("texts", [])
return {**data, "prompt": guardrailed_texts[0] if guardrailed_texts else prompt}
guardrailed_texts: Final = guardrailed_inputs.get("texts")
guardrailed_prompt: Final = guardrailed_texts[0] if guardrailed_texts else prompt
return {**data, "prompt": guardrailed_prompt} # mutable-ok: BaseTranslation contract returns a dict
async def process_output_response(
self,
@ -43,6 +45,6 @@ class OpenAIVideoGenerationHandler(BaseTranslation):
guardrail_to_apply: "CustomGuardrail",
litellm_logging_obj: "LiteLLMLoggingObj | None" = None,
user_api_key_dict: "UserAPIKeyAuth | None" = None,
request_data: dict[str, object] | None = None,
request_data: dict[str, object] | None = None, # mutable-ok: BaseTranslation contract
) -> object:
return response