diff --git a/litellm/proxy/guardrails/guardrail_hooks/litellm_content_filter/content_filter.py b/litellm/proxy/guardrails/guardrail_hooks/litellm_content_filter/content_filter.py index 16c46a76ccd..0e75d8e73c6 100644 --- a/litellm/proxy/guardrails/guardrail_hooks/litellm_content_filter/content_filter.py +++ b/litellm/proxy/guardrails/guardrail_hooks/litellm_content_filter/content_filter.py @@ -10,8 +10,19 @@ import json import os import re from datetime import datetime -from typing import (TYPE_CHECKING, Any, AsyncGenerator, Dict, List, Literal, - Optional, Pattern, Tuple, Union, cast) +from typing import ( + TYPE_CHECKING, + Any, + AsyncGenerator, + Dict, + List, + Literal, + Optional, + Pattern, + Tuple, + Union, + cast, +) import yaml from fastapi import HTTPException @@ -20,22 +31,37 @@ from litellm import Router from litellm._logging import verbose_proxy_logger from litellm.integrations.custom_guardrail import CustomGuardrail from litellm.proxy._types import UserAPIKeyAuth -from litellm.types.utils import (GenericGuardrailAPIInputs, GuardrailStatus, - GuardrailTracingDetail, ModelResponseStream) +from litellm.types.utils import ( + GenericGuardrailAPIInputs, + GuardrailStatus, + GuardrailTracingDetail, + ModelResponseStream, +) if TYPE_CHECKING: from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj -from litellm.types.guardrails import (BlockedWord, ContentFilterAction, - ContentFilterPattern, - GuardrailEventHooks, Mode) +from litellm.types.guardrails import ( + BlockedWord, + ContentFilterAction, + ContentFilterPattern, + GuardrailEventHooks, + Mode, +) from litellm.types.proxy.guardrails.guardrail_hooks.litellm_content_filter import ( - BlockedWordDetection, CategoryKeywordDetection, CompetitorIntentDetection, - CompetitorIntentResult, ContentFilterCategoryConfig, - ContentFilterDetection, PatternDetection) + BlockedWordDetection, + CategoryKeywordDetection, + CompetitorIntentDetection, + CompetitorIntentResult, + ContentFilterCategoryConfig, + ContentFilterDetection, + PatternDetection, +) -from .competitor_intent import (AirlineCompetitorIntentChecker, - BaseCompetitorIntentChecker) +from .competitor_intent import ( + AirlineCompetitorIntentChecker, + BaseCompetitorIntentChecker, +) from .patterns import PATTERN_EXTRA_CONFIG, get_compiled_pattern MAX_KEYWORD_VALUE_GAP_WORDS = 1 @@ -196,48 +222,15 @@ class ContentFilterGuardrail(CustomGuardrail): # Competitor intent checker (optional; airline uses major_airlines.json, generic requires competitors) self._competitor_intent_checker: Optional[BaseCompetitorIntentChecker] = None if competitor_intent_config and isinstance(competitor_intent_config, dict): - try: - competitor_intent_type = competitor_intent_config.get( - "competitor_intent_type", "airline" - ) - if competitor_intent_type == "generic": - self._competitor_intent_checker = BaseCompetitorIntentChecker( - competitor_intent_config - ) - else: - self._competitor_intent_checker = AirlineCompetitorIntentChecker( - competitor_intent_config - ) - verbose_proxy_logger.debug( - "ContentFilterGuardrail: competitor intent checker enabled (%s)", - competitor_intent_type, - ) - except Exception as e: - verbose_proxy_logger.warning( - "ContentFilterGuardrail: failed to init competitor intent checker: %s", - e, - ) + self._init_competitor_intent_checker(competitor_intent_config) # Load categories if provided if categories: self._load_categories(categories) # Normalize inputs: convert dicts to Pydantic models for consistent handling - normalized_patterns: List[ContentFilterPattern] = [] - if patterns: - for pattern_config in patterns: - if isinstance(pattern_config, dict): - normalized_patterns.append(ContentFilterPattern(**pattern_config)) - else: - normalized_patterns.append(pattern_config) - - normalized_blocked_words: List[BlockedWord] = [] - if blocked_words: - for word in blocked_words: - if isinstance(word, dict): - normalized_blocked_words.append(BlockedWord(**word)) - else: - normalized_blocked_words.append(word) + normalized_patterns = self._normalize_patterns(patterns) + normalized_blocked_words = self._normalize_blocked_words(blocked_words) # Compile regex patterns self.compiled_patterns: List[Dict[str, Any]] = [] @@ -278,6 +271,57 @@ class ContentFilterGuardrail(CustomGuardrail): f"{len(self.category_keywords)} keywords" ) + def _init_competitor_intent_checker( + self, competitor_intent_config: Dict[str, Any] + ) -> None: + try: + competitor_intent_type = competitor_intent_config.get( + "competitor_intent_type", "airline" + ) + if competitor_intent_type == "generic": + self._competitor_intent_checker = BaseCompetitorIntentChecker( + competitor_intent_config + ) + else: + self._competitor_intent_checker = AirlineCompetitorIntentChecker( + competitor_intent_config + ) + verbose_proxy_logger.debug( + "ContentFilterGuardrail: competitor intent checker enabled (%s)", + competitor_intent_type, + ) + except Exception as e: + verbose_proxy_logger.warning( + "ContentFilterGuardrail: failed to init competitor intent checker: %s", + e, + ) + + @staticmethod + def _normalize_patterns( + patterns: Optional[List[ContentFilterPattern]], + ) -> List[ContentFilterPattern]: + result: List[ContentFilterPattern] = [] + if patterns: + for pattern_config in patterns: + if isinstance(pattern_config, dict): + result.append(ContentFilterPattern(**pattern_config)) + else: + result.append(pattern_config) + return result + + @staticmethod + def _normalize_blocked_words( + blocked_words: Optional[List[BlockedWord]], + ) -> List[BlockedWord]: + result: List[BlockedWord] = [] + if blocked_words: + for word in blocked_words: + if isinstance(word, dict): + result.append(BlockedWord(**word)) + else: + result.append(word) + return result + @staticmethod def _resolve_category_file_path(file_path: str) -> str: """ @@ -1878,7 +1922,8 @@ class ContentFilterGuardrail(CustomGuardrail): @staticmethod def get_config_model(): - from litellm.types.proxy.guardrails.guardrail_hooks.litellm_content_filter import \ - LitellmContentFilterGuardrailConfigModel + from litellm.types.proxy.guardrails.guardrail_hooks.litellm_content_filter import ( + LitellmContentFilterGuardrailConfigModel, + ) return LitellmContentFilterGuardrailConfigModel diff --git a/litellm/proxy/guardrails/guardrail_hooks/litellm_content_filter/guardrail_benchmarks/test_eval.py b/litellm/proxy/guardrails/guardrail_hooks/litellm_content_filter/guardrail_benchmarks/test_eval.py index 61c724b1836..e6f07b08adb 100644 --- a/litellm/proxy/guardrails/guardrail_hooks/litellm_content_filter/guardrail_benchmarks/test_eval.py +++ b/litellm/proxy/guardrails/guardrail_hooks/litellm_content_filter/guardrail_benchmarks/test_eval.py @@ -69,6 +69,67 @@ def _run(checker, text: str) -> dict: raise +def _print_confusion_report(label: str, metrics: dict, wrong: list) -> None: + """Print the confusion matrix report to stdout.""" + print("\n") # noqa: T201 + print("=" * 70) # noqa: T201 + print(f" {label}") # noqa: T201 + print("=" * 70) # noqa: T201 + print(f" Total cases: {metrics['total']}") # noqa: T201 + print(f" Correct: {metrics['tp'] + metrics['tn']}") # noqa: T201 + print(f" Wrong: {metrics['fp'] + metrics['fn']}") # noqa: T201 + print() # noqa: T201 + print(f" TP (correctly blocked): {metrics['tp']}") # noqa: T201 + print(f" TN (correctly allowed): {metrics['tn']}") # noqa: T201 + print(f" FP (wrongly blocked): {metrics['fp']}") # noqa: T201 + print(f" FN (wrongly allowed): {metrics['fn']}") # noqa: T201 + print() # noqa: T201 + print(f" Precision: {metrics['precision']:.1%}") # noqa: T201 + print(f" Recall: {metrics['recall']:.1%}") # noqa: T201 + print(f" F1: {metrics['f1']:.1%}") # noqa: T201 + print(f" Accuracy: {metrics['accuracy']:.1%}") # noqa: T201 + print() # noqa: T201 + print(f" Latency p50: {metrics['p50']:.1f}ms") # noqa: T201 + print(f" Latency p95: {metrics['p95']:.1f}ms") # noqa: T201 + print(f" Latency avg: {metrics['avg_lat']:.1f}ms") # noqa: T201 + print() # noqa: T201 + if wrong: + print("WRONG ANSWERS:") # noqa: T201 + for line in wrong: + print(line) # noqa: T201 + else: + print("ALL CASES CORRECT") # noqa: T201 + print("=" * 70) # noqa: T201 + + +def _save_confusion_results(label: str, metrics: dict, wrong: list, rows: list) -> dict: + """Save confusion matrix results to a JSON file and return the result dict.""" + os.makedirs(RESULTS_DIR, exist_ok=True) + safe_label = label.lower().replace(" ", "_").replace("—", "-") + result = { + "label": label, + "timestamp": datetime.now(timezone.utc).isoformat(), + "total": metrics["total"], + "tp": metrics["tp"], + "tn": metrics["tn"], + "fp": metrics["fp"], + "fn": metrics["fn"], + "precision": round(metrics["precision"], 4), + "recall": round(metrics["recall"], 4), + "f1": round(metrics["f1"], 4), + "accuracy": round(metrics["accuracy"], 4), + "latency_p50_ms": round(metrics["p50"], 3), + "latency_p95_ms": round(metrics["p95"], 3), + "latency_avg_ms": round(metrics["avg_lat"], 3), + "wrong": wrong, + "rows": rows, + } + result_path = os.path.join(RESULTS_DIR, f"{safe_label}.json") + with open(result_path, "w") as f: + json.dump(result, f, indent=2) + return result + + def _confusion_matrix(checker, cases: List[dict], label: str): """Run all cases, print confusion matrix, save results JSON.""" tp = fp = tn = fn = 0 @@ -131,62 +192,13 @@ def _confusion_matrix(checker, cases: List[dict], label: str): p95 = sorted_lat[int(len(sorted_lat) * 0.95)] if sorted_lat else 0 avg_lat = sum(latencies) / len(latencies) if latencies else 0 - # Print confusion matrix (noqa: T201 — intentional eval output) - print("\n") # noqa: T201 - print("=" * 70) # noqa: T201 - print(f" {label}") # noqa: T201 - print("=" * 70) # noqa: T201 - print(f" Total cases: {total}") # noqa: T201 - print(f" Correct: {tp + tn}") # noqa: T201 - print(f" Wrong: {fp + fn}") # noqa: T201 - print() # noqa: T201 - print(f" TP (correctly blocked): {tp}") # noqa: T201 - print(f" TN (correctly allowed): {tn}") # noqa: T201 - print(f" FP (wrongly blocked): {fp}") # noqa: T201 - print(f" FN (wrongly allowed): {fn}") # noqa: T201 - print() # noqa: T201 - print(f" Precision: {precision:.1%}") # noqa: T201 - print(f" Recall: {recall:.1%}") # noqa: T201 - print(f" F1: {f1:.1%}") # noqa: T201 - print(f" Accuracy: {accuracy:.1%}") # noqa: T201 - print() # noqa: T201 - print(f" Latency p50: {p50:.1f}ms") # noqa: T201 - print(f" Latency p95: {p95:.1f}ms") # noqa: T201 - print(f" Latency avg: {avg_lat:.1f}ms") # noqa: T201 - print() # noqa: T201 - if wrong: - print("WRONG ANSWERS:") # noqa: T201 - for line in wrong: - print(line) # noqa: T201 - else: - print("ALL CASES CORRECT") # noqa: T201 - print("=" * 70) # noqa: T201 - - # Save results - os.makedirs(RESULTS_DIR, exist_ok=True) - safe_label = label.lower().replace(" ", "_").replace("—", "-") - result = { - "label": label, - "timestamp": datetime.now(timezone.utc).isoformat(), - "total": total, - "tp": tp, - "tn": tn, - "fp": fp, - "fn": fn, - "precision": round(precision, 4), - "recall": round(recall, 4), - "f1": round(f1, 4), - "accuracy": round(accuracy, 4), - "latency_p50_ms": round(p50, 3), - "latency_p95_ms": round(p95, 3), - "latency_avg_ms": round(avg_lat, 3), - "wrong": wrong, - "rows": rows, + metrics = { + "total": total, "tp": tp, "tn": tn, "fp": fp, "fn": fn, + "precision": precision, "recall": recall, "f1": f1, "accuracy": accuracy, + "p50": p50, "p95": p95, "avg_lat": avg_lat, } - result_path = os.path.join(RESULTS_DIR, f"{safe_label}.json") - with open(result_path, "w") as f: - json.dump(result, f, indent=2) - + _print_confusion_report(label, metrics, wrong) + result = _save_confusion_results(label, metrics, wrong, rows) return result diff --git a/litellm/proxy/utils.py b/litellm/proxy/utils.py index 0591d5b04d5..06297ec1b35 100644 --- a/litellm/proxy/utils.py +++ b/litellm/proxy/utils.py @@ -14,6 +14,8 @@ from email.mime.text import MIMEText from typing import ( TYPE_CHECKING, Any, + Callable, + Coroutine, Dict, List, Literal, @@ -4009,6 +4011,19 @@ class PrismaClient: async with self._db_reconnect_lock: return await _attempt_reconnect_inside_lock() + return await self._attempt_reconnect_with_lock_timeout( + _attempt_reconnect_inside_lock, + reason=reason, + lock_timeout_seconds=lock_timeout_seconds, + ) + + async def _attempt_reconnect_with_lock_timeout( + self, + reconnect_fn: Callable[[], Coroutine[Any, Any, bool]], + reason: str, + lock_timeout_seconds: float, + ) -> bool: + """Acquire the reconnect lock with a timeout, then run reconnect_fn.""" lock_acquired_by_timeout_task = False async def _acquire_reconnect_lock() -> bool: @@ -4056,7 +4071,7 @@ class PrismaClient: return False try: - return await _attempt_reconnect_inside_lock() + return await reconnect_fn() finally: self._db_reconnect_lock.release()