Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/policy_engine/pipeline_executor.py: 18%
294 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 12:01 +0000
1"""
2Pipeline Executor - Executes guardrail pipelines with conditional step logic.
4Runs guardrails sequentially per pipeline step definitions, handling
5pass/fail actions (allow, block, next, modify_response) and data forwarding.
6"""
8import copy
9import time
10from collections.abc import Callable, Mapping, Sequence
11from typing import TYPE_CHECKING, Final, Literal, TypeVar
13from pydantic import BaseModel
15import litellm
16from litellm._logging import verbose_proxy_logger
17from litellm.constants import LOGS_GUARDRAIL_INFORMATION_MARKER
18from litellm.integrations.custom_guardrail import (
19 CustomGuardrail,
20 ModifyResponseException,
21)
22from litellm.integrations.custom_logger import CustomLogger
23from litellm.litellm_core_utils.core_helpers import (
24 get_metadata_variable_name_from_kwargs,
25 get_or_create_metadata_bucket,
26 independent_snapshot,
27)
28from litellm.proxy.common_utils.callback_utils import add_guardrail_to_applied_guardrails_header
29from litellm.proxy.guardrails.guardrail_hooks.unified_guardrail.unified_guardrail import (
30 UnifiedLLMGuardrails,
31)
32from litellm.types.proxy.policy_engine.pipeline_types import (
33 PipelineExecutionResult,
34 PipelineStep,
35 PipelineStepResult,
36)
37from litellm.types.utils import GenericGuardrailAPIInputs, StandardLoggingGuardrailInformation
39if TYPE_CHECKING: 39 ↛ 40line 39 didn't jump to line 40 because the condition on line 39 was never true
40 from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
41 from litellm.llms.base_llm.guardrail_translation.base_translation import (
42 BaseTranslation,
43 )
44 from litellm.proxy._types import UserAPIKeyAuth
46try:
47 from fastapi.exceptions import HTTPException
48except ImportError:
49 HTTPException = None
52class UndeliverableStreamRewrite(Exception):
53 def __init__(self, guardrail_name: str, reason: str) -> None:
54 super().__init__(guardrail_name, reason)
55 self.guardrail_name: Final = guardrail_name
56 self.reason: Final = reason
58 def __str__(self) -> str:
59 return (
60 f"Guardrail '{self.guardrail_name}' rewrote the streamed response but the rewrite cannot be written "
61 f"back to the stream: {self.reason}"
62 )
65def _tool_call_shape(tool_call: object) -> tuple[object, object]:
66 plain: Final = tool_call.model_dump() if isinstance(tool_call, BaseModel) else tool_call
67 function: Final = plain.get("function") if isinstance(plain, Mapping) else None
68 if not isinstance(function, Mapping):
69 return (None, None)
70 return (function.get("name"), function.get("arguments"))
73def _text_snapshot(texts: Sequence[str] | None) -> tuple[str, ...] | None:
74 return None if texts is None else tuple(texts)
77def _scanned_texts(texts: Sequence[str] | None) -> tuple[str, ...]:
78 return tuple(texts or ())
81def _tool_call_shapes(tool_calls: Sequence[object] | None) -> tuple[tuple[object, object], ...] | None:
82 return None if tool_calls is None else tuple(_tool_call_shape(tool_call) for tool_call in tool_calls)
85def _rewrote(sent: tuple[object, ...] | None, returned: tuple[object, ...] | None) -> bool:
86 return sent is not None and returned is not None and returned != sent
89def _count_change(sent: tuple[object, ...] | None, returned: tuple[object, ...] | None) -> tuple[int, int] | None:
90 if sent is None or returned is None or len(returned) == len(sent):
91 return None
92 return (len(sent), len(returned))
95def _tool_call_mismatch_reason(
96 sent: tuple[tuple[object, object], ...] | None, returned: tuple[tuple[object, object], ...] | None
97) -> str | None:
98 if sent == returned:
99 return None
100 sent_count: Final = len(sent or ())
101 returned_count: Final = len(returned or ())
102 if sent_count == returned_count:
103 return "the legacy hook changed a tool call's name or arguments, which this path cannot write back"
104 return f"the legacy hook returned {returned_count} tool calls for a stream that carried {sent_count}"
107_GuardrailMethodT = TypeVar("_GuardrailMethodT", bound=Callable[..., object])
110def _logged_by_inner_guardrail(method: _GuardrailMethodT) -> _GuardrailMethodT:
111 vars(method)[LOGS_GUARDRAIL_INFORMATION_MARKER] = True
112 return method
115class _StreamRewriteObserver(CustomGuardrail):
116 """Stand-in handed to the endpoint translation in place of a streaming pipeline step's
117 guardrail. It records whether the guardrail returned different output than it was given,
118 which for guardrails like Bedrock's ANONYMIZED action is only known at runtime. Text and
119 tool-call rewrites are deliverable on translations that write them back across the
120 buffered chunks (``delivers_ended_stream_rewrites``); rewrites on any other translation,
121 and a rewrite that drops or adds a tool call on any translation, are discarded by the
122 executor, which releases the original chunks.
123 The inner guardrail's ``apply_guardrail`` already records the guardrail information
124 and span, so the observer's stays out of ``log_guardrail_information``."""
126 def __init__(self, inner: CustomGuardrail) -> None:
127 super().__init__(guardrail_name=inner.guardrail_name)
128 self.inner: Final = inner
129 self.rewrote_texts = False
130 self.rewrote_tool_calls = False
131 self.tool_call_count_change: tuple[int, int] | None = None
133 def structured_messages_cover_full_request(self) -> bool:
134 return self.inner.structured_messages_cover_full_request()
136 @_logged_by_inner_guardrail
137 async def apply_guardrail(
138 self,
139 inputs: GenericGuardrailAPIInputs,
140 request_data: dict, # mutable-ok: matches CustomGuardrail.apply_guardrail
141 input_type: Literal["request", "response"],
142 logging_obj: "LiteLLMLoggingObj | None" = None,
143 ) -> GenericGuardrailAPIInputs:
144 sent_texts: Final = _text_snapshot(inputs.get("texts"))
145 sent_tool_shapes: Final = _tool_call_shapes(inputs.get("tool_calls"))
146 outputs: Final = await self.inner.apply_guardrail(
147 inputs=inputs, request_data=request_data, input_type=input_type, logging_obj=logging_obj
148 )
149 returned_tool_shapes: Final = _tool_call_shapes(outputs.get("tool_calls"))
150 self.rewrote_texts = self.rewrote_texts or _rewrote(sent_texts, _text_snapshot(outputs.get("texts")))
151 self.rewrote_tool_calls = self.rewrote_tool_calls or _rewrote(sent_tool_shapes, returned_tool_shapes)
152 self.tool_call_count_change = self.tool_call_count_change or _count_change(
153 sent_tool_shapes, returned_tool_shapes
154 )
155 return outputs
157 def discard_reason(self, deliver_rewrites: bool) -> str | None:
158 if self.tool_call_count_change is not None:
159 sent, returned = self.tool_call_count_change
160 return (
161 f"the guardrail returned {returned} tool calls for a stream that carried {sent}, and a rewrite "
162 "that drops or adds a tool call cannot be written back"
163 )
164 if not deliver_rewrites and (self.rewrote_texts or self.rewrote_tool_calls):
165 return "this endpoint's streaming pipeline does not write ended-stream rewrites back yet"
166 return None
169class _ScannedTextRecorder(CustomGuardrail):
170 def __init__(self, guardrail_name: str) -> None:
171 super().__init__(guardrail_name=guardrail_name)
172 self.inputs: GenericGuardrailAPIInputs | None = None
174 @_logged_by_inner_guardrail
175 async def apply_guardrail(
176 self,
177 inputs: GenericGuardrailAPIInputs,
178 request_data: dict, # mutable-ok: matches CustomGuardrail.apply_guardrail
179 input_type: Literal["request", "response"],
180 logging_obj: "LiteLLMLoggingObj | None" = None,
181 ) -> GenericGuardrailAPIInputs:
182 self.inputs = inputs
183 return inputs
186class _LegacyHookStreamAdapter(CustomGuardrail):
187 """Runs a guardrail that only implements the legacy post-call hook (no unified
188 ``apply_guardrail``, or ``use_native_lifecycle_hooks``) as a streaming pipeline step. The
189 endpoint translation hands it the texts it scanned plus the assembled response under
190 ``request_data["response"]``; the hook gets that response in the shape its route gives
191 non-streaming hooks, an exception it raises ends the stream through the executor's
192 fail/error classification, and the response it hands back, or the one it changed in place
193 and returned ``None`` for, is re-scanned by the same translation so its texts reach the
194 client through the translation's ended-stream write-back. A
195 replacement whose scanned texts do not line up with the originals, or whose tool calls
196 differ from them, is undeliverable, so the executor releases the original chunks. A stream
197 that carried no text to scan, such as a tool-only Anthropic message, stays deliverable as
198 long as the hook left the tool calls alone."""
200 def __init__(
201 self,
202 inner: CustomGuardrail,
203 endpoint_translation: "BaseTranslation",
204 user_api_key_dict: "UserAPIKeyAuth",
205 ) -> None:
206 super().__init__(guardrail_name=inner.guardrail_name)
207 self.inner: Final = inner
208 self.endpoint_translation: Final = endpoint_translation
209 self.user_api_key_dict: Final = user_api_key_dict
211 def structured_messages_cover_full_request(self) -> bool:
212 return self.inner.structured_messages_cover_full_request()
214 @_logged_by_inner_guardrail
215 async def apply_guardrail(
216 self,
217 inputs: GenericGuardrailAPIInputs,
218 request_data: dict, # mutable-ok: matches CustomGuardrail.apply_guardrail
219 input_type: Literal["request", "response"],
220 logging_obj: "LiteLLMLoggingObj | None" = None,
221 ) -> GenericGuardrailAPIInputs:
222 hooked: Final = self.endpoint_translation.post_call_hook_response(request_data.get("response"))
223 replacement: Final = await self.inner.async_post_call_success_hook(
224 data=request_data,
225 user_api_key_dict=self.user_api_key_dict,
226 response=hooked,
227 )
228 rewrite: Final = hooked if replacement is None else replacement
229 if rewrite is None:
230 return inputs
231 rescanned: Final = await self._rescan(rewrite, logging_obj)
232 guardrail_name: Final = self.guardrail_name or "unknown"
233 if rescanned is None:
234 raise UndeliverableStreamRewrite(
235 guardrail_name, "the legacy hook's response could not be rescanned by this endpoint's translation"
236 )
237 rewritten: Final = rescanned.get("texts")
238 returned_text_count: Final = len(_scanned_texts(rewritten))
239 sent_text_count: Final = len(_scanned_texts(inputs.get("texts")))
240 if returned_text_count != sent_text_count:
241 raise UndeliverableStreamRewrite(
242 guardrail_name,
243 f"the legacy hook returned {returned_text_count} texts for a stream that carried {sent_text_count}",
244 )
245 tool_call_mismatch: Final = _tool_call_mismatch_reason(
246 _tool_call_shapes(inputs.get("tool_calls")), _tool_call_shapes(rescanned.get("tool_calls"))
247 )
248 if tool_call_mismatch is not None:
249 raise UndeliverableStreamRewrite(guardrail_name, tool_call_mismatch)
250 if not rewritten:
251 return inputs
252 rewritten_inputs: Final[GenericGuardrailAPIInputs] = {**inputs, "texts": rewritten}
253 return rewritten_inputs
255 async def _rescan(
256 self, response: object, logging_obj: "LiteLLMLoggingObj | None"
257 ) -> GenericGuardrailAPIInputs | None:
258 recorder: Final = _ScannedTextRecorder(self.guardrail_name or "unknown")
259 await self.endpoint_translation.process_output_response(
260 response=response,
261 guardrail_to_apply=recorder,
262 litellm_logging_obj=logging_obj,
263 user_api_key_dict=self.user_api_key_dict,
264 )
265 return recorder.inputs
268def _prepare_hook_input(
269 step: PipelineStep,
270 callback: CustomGuardrail,
271 data: dict, # mutable-ok: same request-payload shape the hooks mutate
272 raw_request_snapshot: dict | None, # mutable-ok: same request-payload shape as data
273) -> tuple[dict, bool]: # mutable-ok: returns that same request-payload dict
274 """Inject the step's guardrail name into metadata so should_run_guardrail() allows it,
275 and pick the payload the step scans: a scan_raw_request step evaluates the pristine
276 pre-pipeline snapshot instead of `data` (which earlier pass_data steps in this same
277 pipeline may have already rewritten), same reason the normal sequential/parallel
278 guardrail loops do this."""
279 if "metadata" not in data:
280 data["metadata"] = {} # mutable-ok: request metadata bucket, hooks mutate it
281 data["metadata"]["guardrails"] = [step.guardrail]
283 scans_raw_request: Final = callback.scan_raw_request
284 hook_input: Final[dict] = ( # mutable-ok: same request-payload shape as data
285 independent_snapshot(raw_request_snapshot) if scans_raw_request and raw_request_snapshot is not None else data
286 )
287 if hook_input is not data:
288 hook_input.setdefault("metadata", {})["guardrails"] = [step.guardrail] # mutable-ok: request metadata shape
289 return hook_input, scans_raw_request
292def _release_original_chunks(
293 guardrail_name: str,
294 reason: str,
295 streaming_chunks: list[object], # mutable-ok: shared buffered-stream chunks, restored in place
296 originals: Sequence[object],
297) -> None:
298 streaming_chunks[:] = originals # rebind-ok: the caller's buffer is the stream the client receives
299 verbose_proxy_logger.warning(
300 "Pipeline: guardrail '%s' rewrote the streamed response but the rewrite could not be written back to "
301 "the stream: %s. The whole rewrite, text rewrites included, was discarded and the original stream released",
302 guardrail_name,
303 reason,
304 )
307class PipelineExecutor:
308 """Executes guardrail pipelines with ordered, conditional step logic."""
310 @staticmethod
311 async def execute_steps(
312 steps: list[PipelineStep],
313 mode: str,
314 data: dict,
315 user_api_key_dict: "UserAPIKeyAuth",
316 call_type: str,
317 policy_name: str,
318 raw_request_snapshot: dict | None = None, # mutable-ok: same request-payload shape as data
319 streaming_chunks: list[object] | None = None, # mutable-ok: shared buffered-stream chunks, read per step
320 endpoint_translation: "BaseTranslation | None" = None,
321 ) -> PipelineExecutionResult:
322 """
323 Execute pipeline steps sequentially with conditional actions.
325 Args:
326 steps: Ordered list of pipeline steps
327 mode: Event hook mode (pre_call, post_call)
328 data: Request data dict
329 user_api_key_dict: User API key auth
330 call_type: Type of call (completion, etc.)
331 policy_name: Name of the owning policy (for logging)
332 raw_request_snapshot: pristine pre-pipeline, pre-guardrail request
333 (taken by the caller before any guardrail or pipeline ran), so a
334 step whose guardrail opted into ``scan_raw_request`` evaluates
335 the original request instead of whatever an earlier
336 ``pass_data`` step in this same pipeline already rewrote.
337 streaming_chunks: buffered chunks of a completed stream. When set
338 (with ``endpoint_translation``), post_call steps scan the
339 assembled streamed output through the endpoint translation
340 instead of calling ``async_post_call_success_hook``.
341 endpoint_translation: the guardrail translation for the streamed
342 endpoint, resolved by the caller.
344 Returns:
345 PipelineExecutionResult with terminal action and step results
346 """
347 step_results: Final[list[PipelineStepResult]] = []
348 working_data = data.copy()
349 if "metadata" in working_data:
350 working_data["metadata"] = working_data["metadata"].copy()
352 for i, step in enumerate(steps):
353 start_time = time.perf_counter()
355 (
356 outcome,
357 modified_data,
358 error_detail,
359 original_exception,
360 ) = await PipelineExecutor._run_step(
361 step=step,
362 mode=mode,
363 data=working_data,
364 user_api_key_dict=user_api_key_dict,
365 call_type=call_type,
366 raw_request_snapshot=raw_request_snapshot,
367 streaming_chunks=streaming_chunks,
368 endpoint_translation=endpoint_translation,
369 )
371 duration = time.perf_counter() - start_time
373 action = _pipeline_action_for_outcome(step, outcome)
375 step_result = PipelineStepResult(
376 guardrail_name=step.guardrail,
377 outcome=outcome,
378 action_taken=action,
379 modified_data=modified_data,
380 error_detail=error_detail,
381 duration_seconds=round(duration, 4),
382 )
383 step_results.append(step_result)
385 verbose_proxy_logger.debug(
386 "Pipeline '%s' step %s: guardrail=%s, outcome=%s, action=%s",
387 policy_name,
388 i,
389 step.guardrail,
390 outcome,
391 action,
392 )
394 # Forward modified data to the next step if pass_data is True;
395 # post_call response replacements always chain, matching the flat
396 # callback loop where each hook sees the previous hook's response
397 if modified_data is not None and (step.pass_data or mode == "post_call"):
398 working_data = {**working_data, **modified_data}
400 # Handle terminal actions
401 if action == "allow":
402 return _allow_result(step_results=step_results, working_data=working_data, request_data=data)
404 if action == "block":
405 _carry_working_guardrail_information(working_data=working_data, request_data=data)
406 return PipelineExecutionResult(
407 terminal_action="block",
408 step_results=step_results,
409 error_message=error_detail,
410 original_exception=original_exception,
411 modified_data=working_data if working_data != data else None,
412 )
414 if action == "modify_response":
415 _carry_working_guardrail_information(working_data=working_data, request_data=data)
416 return PipelineExecutionResult(
417 terminal_action="modify_response",
418 step_results=step_results,
419 modify_response_message=step.modify_response_message or error_detail,
420 modified_data=working_data if working_data != data else None,
421 )
423 # action == "next" → continue to next step
425 # Ran out of steps without a terminal action → default allow
426 return _allow_result(step_results=step_results, working_data=working_data, request_data=data)
428 @staticmethod
429 async def _run_streaming_step(
430 step: PipelineStep,
431 callback: CustomGuardrail,
432 endpoint_translation: "BaseTranslation",
433 streaming_chunks: list[object], # mutable-ok: shared buffered-stream chunks the translation rewrites in place
434 hook_input: dict[str, object], # mutable-ok: same request-payload shape as data
435 user_api_key_dict: "UserAPIKeyAuth",
436 litellm_logging_obj: "LiteLLMLoggingObj | None",
437 ) -> None:
438 """Run one streaming post_call step through the endpoint translation, delivering
439 text and tool-call rewrites on translations that support ended-stream write-back. A
440 guardrail without the unified interface runs its legacy post-call hook against the
441 assembled response through ``_LegacyHookStreamAdapter``. A rewrite that cannot reach the
442 client yet (one on a translation without write-back, one that drops or adds a tool call,
443 or one the translation or adapter refused with ``UndeliverableStreamRewrite``) is
444 discarded: the buffered chunks go back to the originals and the step passes, so the
445 client gets the stream the merge base sent, and the guardrail stays out of the
446 applied-guardrails header since its output never reached the client. The response an
447 earlier step's translation stored under ``request_data["response"]`` is dropped first,
448 so this step's hook sees the stream as the steps before it left it."""
449 scanner: Final = (
450 callback
451 if PipelineExecutor.supports_unified_execution(callback)
452 else _LegacyHookStreamAdapter(callback, endpoint_translation, user_api_key_dict)
453 )
454 observer: Final = _StreamRewriteObserver(scanner)
455 deliver_rewrites: Final = type(endpoint_translation).delivers_ended_stream_rewrites
456 originals: Final = copy.deepcopy(streaming_chunks)
457 hook_input.pop("response", None)
458 try:
459 if deliver_rewrites:
460 await endpoint_translation.process_output_streaming_response(
461 responses_so_far=streaming_chunks,
462 guardrail_to_apply=observer,
463 litellm_logging_obj=litellm_logging_obj,
464 user_api_key_dict=user_api_key_dict,
465 request_data=hook_input,
466 deliver_ended_stream_rewrites=True,
467 )
468 else:
469 await endpoint_translation.process_output_streaming_response(
470 responses_so_far=streaming_chunks,
471 guardrail_to_apply=observer,
472 litellm_logging_obj=litellm_logging_obj,
473 user_api_key_dict=user_api_key_dict,
474 request_data=hook_input,
475 )
476 except UndeliverableStreamRewrite as undeliverable:
477 _release_original_chunks(step.guardrail, undeliverable.reason, streaming_chunks, originals)
478 return
479 discard_reason: Final = observer.discard_reason(deliver_rewrites)
480 if discard_reason is not None:
481 _release_original_chunks(step.guardrail, discard_reason, streaming_chunks, originals)
482 return
483 if not callback.records_own_guardrail_information:
484 add_guardrail_to_applied_guardrails_header(request_data=hook_input, guardrail_name=step.guardrail)
486 @staticmethod
487 async def _run_step(
488 step: PipelineStep,
489 mode: str,
490 data: dict,
491 user_api_key_dict: "UserAPIKeyAuth",
492 call_type: str,
493 raw_request_snapshot: dict | None = None, # mutable-ok: same request-payload shape as data
494 streaming_chunks: list[object] | None = None, # mutable-ok: shared buffered-stream chunks, read per step
495 endpoint_translation: "BaseTranslation | None" = None,
496 ) -> tuple[
497 Literal["pass", "fail", "error"],
498 dict | None,
499 str | None,
500 Exception | None,
501 ]:
502 """
503 Run a single pipeline step's guardrail.
505 Returns:
506 Tuple of (outcome, modified_data, error_detail, original_exception):
507 - outcome: "pass", "fail", or "error"
508 - modified_data: dict if guardrail returned modified data, else None
509 - error_detail: error message string if fail/error, else None
510 - original_exception: the exception the guardrail raised, so the
511 pipeline can re-raise it verbatim and match the direct-attachment
512 response/trace, else None
513 """
514 callback: Final = PipelineExecutor.find_guardrail_callback(step.guardrail)
515 if callback is None:
516 verbose_proxy_logger.warning("Pipeline: guardrail '%s' not found in callbacks", step.guardrail)
517 return ("error", None, f"Guardrail '{step.guardrail}' not found", None)
519 hook_input, scans_raw_request = _prepare_hook_input(step, callback, data, raw_request_snapshot)
520 snapshot_entries_before: Final = len(_recorded_guardrail_information(hook_input))
522 # Use unified_guardrail path if callback implements apply_guardrail
523 target: CustomLogger = callback
524 use_unified: Final = PipelineExecutor.supports_unified_execution(callback)
525 if use_unified and streaming_chunks is None:
526 hook_input["guardrail_to_apply"] = callback
527 target = UnifiedLLMGuardrails()
529 try:
530 if mode == "pre_call":
531 response = await target.async_pre_call_hook(
532 user_api_key_dict=user_api_key_dict,
533 cache=None,
534 data=hook_input,
535 call_type=call_type,
536 )
537 if isinstance(callback, CustomGuardrail):
538 callback.mark_pre_call_hook_ran(data)
539 if isinstance(response, dict):
540 callback.mark_pre_call_hook_ran(response)
541 elif mode == "post_call" and streaming_chunks is not None:
542 if endpoint_translation is None:
543 return (
544 "error",
545 None,
546 f"Guardrail '{step.guardrail}' cannot run on a stream without an endpoint translation",
547 None,
548 )
549 await PipelineExecutor._run_streaming_step(
550 step=step,
551 callback=callback,
552 endpoint_translation=endpoint_translation,
553 streaming_chunks=streaming_chunks,
554 hook_input=hook_input,
555 user_api_key_dict=user_api_key_dict,
556 litellm_logging_obj=data.get("litellm_logging_obj"),
557 )
558 response = None
559 elif mode == "post_call":
560 response = await target.async_post_call_success_hook(
561 user_api_key_dict=user_api_key_dict,
562 data=data,
563 response=data.get("response"),
564 )
565 else:
566 return ("error", None, f"Unsupported pipeline mode: {mode}", None)
568 # Normal return means pass. A scan_raw_request step is block-only,
569 # same contract as run_in_parallel/scan_raw_request elsewhere: any
570 # data it returned is discarded, since applying it on top of the
571 # raw snapshot would silently undo whatever an earlier step in
572 # this pipeline already did. A post_call hook's non-None return is
573 # a replacement response (the flat callback-loop contract), carried
574 # under the same "response" key the step input uses.
575 if response is None or scans_raw_request:
576 return ("pass", None, None, None)
577 if mode == "post_call":
578 return (
579 "pass",
580 {"response": response},
581 None,
582 None,
583 )
584 return ("pass", response if isinstance(response, dict) else None, None, None)
586 except Exception as e:
587 if CustomGuardrail._is_guardrail_intervention(e):
588 error_msg: Final = _extract_error_message(e)
589 return ("fail", None, error_msg, e)
590 else:
591 verbose_proxy_logger.error("Pipeline: unexpected error from guardrail '%s': %s", step.guardrail, e)
592 return ("error", None, str(e), e)
593 finally:
594 if hook_input is not data:
595 _append_guardrail_information(
596 request_data=data,
597 entries=_recorded_guardrail_information(hook_input)[snapshot_entries_before:],
598 )
600 @staticmethod
601 def supports_unified_execution(callback: CustomGuardrail) -> bool:
602 """Whether this guardrail runs through the unified apply_guardrail path."""
603 return "apply_guardrail" in type(callback).__dict__ and not callback.use_native_lifecycle_hooks
605 @staticmethod
606 def supports_streaming_execution(callback: CustomGuardrail) -> bool:
607 """Whether a streaming pipeline step can run this guardrail against the buffered
608 stream: through the unified path, or through its post-call hook on the assembled
609 response when that hook is its only streaming path. A guardrail with its own
610 streaming iterator hook, or with neither hook, keeps running on its own."""
611 callback_type: Final = type(callback)
612 return PipelineExecutor.supports_unified_execution(callback) or (
613 callback_type.async_post_call_success_hook is not CustomLogger.async_post_call_success_hook
614 and callback_type.async_post_call_streaming_iterator_hook
615 is CustomLogger.async_post_call_streaming_iterator_hook
616 )
618 @staticmethod
619 def find_guardrail_callback(guardrail_name: str) -> CustomGuardrail | None:
620 """Look up an initialized guardrail callback by name from litellm.callbacks."""
621 for callback in litellm.callbacks:
622 if isinstance(callback, CustomGuardrail):
623 if callback.guardrail_name == guardrail_name:
624 return callback
625 return None
628def _allow_result(
629 step_results: Sequence[PipelineStepResult],
630 working_data: dict, # mutable-ok: same request-payload shape as execute_steps' data
631 request_data: dict, # mutable-ok: same request-payload shape as execute_steps' data
632) -> PipelineExecutionResult:
633 """Build the terminal-allow result, propagating pipeline modifications without the per-step guardrail override."""
634 restored: Final = _restore_request_guardrails(working_data, request_data)
635 return PipelineExecutionResult(
636 terminal_action="allow",
637 step_results=list(step_results), # mutable-ok: PipelineExecutionResult field is a list
638 modified_data=restored if restored != request_data else None,
639 )
642def _restore_request_guardrails(
643 working_data: dict, # mutable-ok: same request-payload shape as execute_steps' data
644 request_data: dict, # mutable-ok: same request-payload shape as execute_steps' data
645) -> dict: # mutable-ok: merged back into the request dict, which downstream code mutates
646 """
647 Restore the request's own metadata["guardrails"] activation list.
649 _run_step overrides it to [step.guardrail] so should_run_guardrail() allows each
650 step; letting that override escape via modified_data permanently drops every
651 independently activated guardrail from later lifecycle stages (post_call, etc.).
652 """
653 working_metadata: Final = working_data.get("metadata")
654 if not isinstance(working_metadata, dict):
655 return working_data
656 request_metadata: Final = request_data.get("metadata")
657 original_guardrails: Final = request_metadata.get("guardrails") if isinstance(request_metadata, dict) else None
658 stripped: Final = {k: v for k, v in working_metadata.items() if k != "guardrails"} # mutable-ok: request dict
659 if original_guardrails is not None:
660 restored: Final = {**stripped, "guardrails": original_guardrails} # mutable-ok: request dict
661 return {**working_data, "metadata": restored} # mutable-ok: request dict
662 if not stripped and not isinstance(request_metadata, dict):
663 return {k: v for k, v in working_data.items() if k != "metadata"} # mutable-ok: request dict
664 return {**working_data, "metadata": stripped} # mutable-ok: request dict
667_GUARDRAIL_INFORMATION_KEY: Final = "standard_logging_guardrail_information"
670def _recorded_guardrail_information(source: Mapping[str, object]) -> list[StandardLoggingGuardrailInformation]:
671 bucket: Final = source.get(get_metadata_variable_name_from_kwargs(source))
672 recorded: Final = bucket.get(_GUARDRAIL_INFORMATION_KEY) if isinstance(bucket, dict) else None
673 return recorded if isinstance(recorded, list) else []
676def _append_guardrail_information(
677 request_data: dict[str, object], # mutable-ok: same request-payload shape as execute_steps' data
678 entries: Sequence[StandardLoggingGuardrailInformation],
679) -> None:
680 if not entries:
681 return
682 _, request_bucket = get_or_create_metadata_bucket(request_data)
683 existing: Final = request_bucket.get(_GUARDRAIL_INFORMATION_KEY)
684 if isinstance(existing, list):
685 existing.extend(entries)
686 return
687 request_bucket[_GUARDRAIL_INFORMATION_KEY] = list(entries)
690def _carry_working_guardrail_information(
691 working_data: Mapping[str, object],
692 request_data: dict[str, object], # mutable-ok: same request-payload shape as execute_steps' data
693) -> None:
694 recorded: Final = _recorded_guardrail_information(working_data)
695 existing: Final = _recorded_guardrail_information(request_data)
696 if recorded is existing:
697 return
698 _append_guardrail_information(request_data=request_data, entries=[e for e in recorded if e not in existing])
701def _pipeline_action_for_outcome(step: PipelineStep, outcome: str) -> str:
702 """
703 Map pipeline step outcome to the configured action.
705 - pass -> on_pass
706 - fail -> on_fail (content/policy intervention)
707 - error -> on_error if set, else on_fail (backward compatible)
708 """
709 if outcome == "pass":
710 return step.on_pass
711 if outcome == "fail":
712 return step.on_fail
713 if step.on_error is not None:
714 return step.on_error
715 return step.on_fail
718def _extract_error_message(e: Exception) -> str:
719 """Extract a human-readable error message from a guardrail exception."""
720 if isinstance(e, ModifyResponseException):
721 return str(e)
722 if HTTPException is not None and isinstance(e, HTTPException):
723 detail: Final[object] = getattr(e, "detail", None)
724 if detail:
725 return str(detail)
726 return str(e)