refactor: move tone detection into litellm_content_filter as sub-module

Move tone detection from standalone tone_detector/ guardrail into
litellm_content_filter/ as an optional sub-feature (like competitor_intent).

- New sub-module: litellm_content_filter/tone_detection/ with ToneChecker
- Wire into ContentFilterGuardrail.__init__ via tone_detection_config param
- Add ToneDetection type to ContentFilterDetection union
- Add tone_detection_config to LitellmContentFilterGuardrailConfigModel
- Remove standalone tone_detector/ directory and TONE_DETECTOR enum
- Update tests to use ContentFilterGuardrail with tone_detection_config

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Krrish Dholakia 2026-03-11 20:31:37 -07:00
parent 1ec567a4dd
commit c60ea50baf
9 changed files with 238 additions and 266 deletions

View file

@ -47,6 +47,9 @@ def initialize_guardrail(
competitor_intent_config=getattr(
litellm_params, "competitor_intent_config", None
),
tone_detection_config=getattr(
litellm_params, "tone_detection_config", None
),
end_session_after_n_fails=getattr(litellm_params, "end_session_after_n_fails", None),
on_violation=getattr(litellm_params, "on_violation", None),
realtime_violation_message=getattr(litellm_params, "realtime_violation_message", None),

View file

@ -56,6 +56,7 @@ from litellm.types.proxy.guardrails.guardrail_hooks.litellm_content_filter impor
ContentFilterCategoryConfig,
ContentFilterDetection,
PatternDetection,
ToneDetection,
)
from .competitor_intent import (
@ -63,6 +64,7 @@ from .competitor_intent import (
BaseCompetitorIntentChecker,
)
from .patterns import PATTERN_EXTRA_CONFIG, get_compiled_pattern
from .tone_detection import ToneChecker
MAX_KEYWORD_VALUE_GAP_WORDS = 1
GAP_WORD_TOKENIZER = re.compile(r"\b\w+\b")
@ -169,6 +171,7 @@ class ContentFilterGuardrail(CustomGuardrail):
llm_router: Optional[Router] = None,
image_model: Optional[str] = None,
competitor_intent_config: Optional[Dict[str, Any]] = None,
tone_detection_config: Optional[Dict[str, Any]] = None,
**kwargs,
):
"""
@ -229,6 +232,11 @@ class ContentFilterGuardrail(CustomGuardrail):
if competitor_intent_config and isinstance(competitor_intent_config, dict):
self._init_competitor_intent_checker(competitor_intent_config)
# Tone checker (optional; CPU-only regex detection of inappropriate chatbot tone)
self._tone_checker: Optional[ToneChecker] = None
if tone_detection_config is not None and isinstance(tone_detection_config, dict):
self._init_tone_checker(tone_detection_config)
# Load categories if provided
if categories:
self._load_categories(categories)
@ -301,6 +309,46 @@ class ContentFilterGuardrail(CustomGuardrail):
e,
)
def _init_tone_checker(self, tone_detection_config: Dict[str, Any]) -> None:
try:
self._tone_checker = ToneChecker(tone_detection_config)
verbose_proxy_logger.debug(
"ContentFilterGuardrail: tone checker enabled"
)
except Exception as e:
verbose_proxy_logger.warning(
"ContentFilterGuardrail: failed to init tone checker: %s",
e,
)
def _apply_tone_detection_policy(
self,
tone_result: Dict[str, str],
detections: List[ContentFilterDetection],
) -> None:
"""Raise HTTPException(400) when a tone violation is detected."""
category = tone_result["category"]
matched = tone_result["matched_text"]
detection: ToneDetection = {
"type": "tone",
"category": category,
"matched_text": matched,
}
detections.append(detection)
verbose_proxy_logger.warning(
"ContentFilterGuardrail: tone violation (%s): '%s'",
category,
matched,
)
raise HTTPException(
status_code=400,
detail={
"error": f"Tone violation detected: {category}",
"category": category,
"matched_text": matched,
},
)
@staticmethod
def _normalize_patterns(
patterns: Optional[List[ContentFilterPattern]],
@ -1821,6 +1869,13 @@ class ContentFilterGuardrail(CustomGuardrail):
self._apply_competitor_intent_policy(
intent_result, request_data, detections
)
# Tone detection (optional; raises on violation)
if self._tone_checker and text:
tone_result = self._tone_checker.run(text)
if tone_result is not None:
self._apply_tone_detection_policy(
tone_result, detections
)
filtered_text = self._filter_single_text(text, detections=detections)
processed_texts.append(filtered_text)

View file

@ -0,0 +1,14 @@
"""
Tone detection: CPU-only regex/keyword checks for inappropriate chatbot tone.
Detects dismissive, condescending, blaming, unprofessional language while
allowing domain-specific safe phrases.
"""
from litellm.proxy.guardrails.guardrail_hooks.litellm_content_filter.tone_detection.base import (
ToneChecker,
)
__all__ = [
"ToneChecker",
]

View file

@ -0,0 +1,123 @@
"""
Tone checker: CPU-only regex detection of inappropriate chatbot tone.
Detects 6 categories: dismissive, blaming, refusal, condescension,
impatience, unprofessional. Supports user-supplied blocked_phrases
(additional patterns) and safe_phrases (exemptions for domain jargon).
"""
import re
from typing import Dict, List, Optional, Pattern, Tuple
from litellm._logging import verbose_proxy_logger
# ---------------------------------------------------------------------------
# Built-in tone patterns — each is (raw_regex, category_label)
# ---------------------------------------------------------------------------
_TONE_PATTERNS: List[Tuple[str, str]] = [
# Dismissive
(r"\bthat(?:'s| is) not (?:really )?my (?:problem|concern|issue)\b", "dismissive"),
(r"\bi (?:don't|do not) see what the big deal is\b", "dismissive"),
(r"\byou(?:'re| are) overthinking\b", "dismissive"),
(r"\bjust read the (?:FAQ|docs|documentation|manual)\b", "dismissive"),
# Blaming the customer
# "you should have" is blame only when followed by a blame verb
# (e.g. "you should have received" is informational, not blame)
(r"\byou should have (?:read|known|checked|done|thought|realized|paid|looked|noticed|seen to)\b", "blaming"),
(r"\bthat(?:'s| is) your (?:fault(?! tolerance| tolerant)|problem|mistake)\b", "blaming"),
(r"\bif you had (?:followed|read|done)\b", "blaming"),
(r"\byou clearly (?:didn't|did not)\b", "blaming"),
(r"\bthis (?:issue|problem) is on your end\b", "blaming"),
# Refusal to help (without offering alternatives)
# "I can't help you" but NOT "I can't help but notice" (which is polite)
(r"\bi (?:can't|cannot|can not) help you\b", "refusal"),
# "nothing I can do" is refusal only when NOT followed by "but" / "to"
(r"\bthere(?:'s| is) nothing (?:i|we) can do(?! (?:but|to))\b", "refusal"),
(r"\byou(?:'ll| will) (?:just )?have to figure it out\b", "refusal"),
(r"\btry somewhere else\b", "refusal"),
# Sarcasm / condescension
(r"\bif you(?:'d| had| would have) been paying attention\b", "condescension"),
(r"\bhow (?:to )?make this any simpler\b", "condescension"),
(r"\blet me spell it out for you\b", "condescension"),
(r"\bsince you (?:don't|do not) (?:seem to )?get it\b", "condescension"),
(r"\bdo your (?:job|work) for you\b", "condescension"),
# Impatience / frustration — "told you" specifically, not "told our team"
(r"\bi(?:'ve| have) already told you\b", "impatience"),
(r"\bhow many times do i have to\b", "impatience"),
(r"\bi (?:don't|do not) have time to\b", "impatience"),
(r"\bare you even listening\b", "impatience"),
(r"\bjust do what i said\b", "impatience"),
# Unprofessional casual language
(r"\b(?:bruh|lol|idk|smh|lmao|wtf)\b", "unprofessional"),
(r"\bmy bad dude\b", "unprofessional"),
(r"\bwhatever,? just deal with it\b", "unprofessional"),
(r"\bsounds like a you problem\b", "unprofessional"),
]
def _compile_patterns(
patterns: List[Tuple[str, str]],
) -> List[Tuple[Pattern[str], str]]:
return [(re.compile(p, re.IGNORECASE), cat) for p, cat in patterns]
# Pre-compiled at module load
_COMPILED_TONE_PATTERNS = _compile_patterns(_TONE_PATTERNS)
class ToneChecker:
"""
CPU-only tone checker.
Config keys (all optional):
blocked_phrases: list of additional regex strings to block
safe_phrases: list of regex strings that exempt text from blocking
"""
def __init__(self, config: Dict) -> None:
self._extra_blocked: List[Tuple[Pattern[str], str]] = []
for phrase in config.get("blocked_phrases") or []:
self._extra_blocked.append(
(re.compile(phrase, re.IGNORECASE), "custom_blocked")
)
self._safe_patterns: List[Pattern[str]] = []
for phrase in config.get("safe_phrases") or []:
self._safe_patterns.append(re.compile(phrase, re.IGNORECASE))
verbose_proxy_logger.debug(
"ToneChecker: initialized with %d extra blocked, %d safe phrases",
len(self._extra_blocked),
len(self._safe_patterns),
)
def _is_safe(self, text: str) -> bool:
"""Return True if text matches any user-defined safe phrase."""
return any(p.search(text) for p in self._safe_patterns)
def run(self, text: str) -> Optional[Dict]:
"""
Check text for tone violations.
Returns a dict with {matched_text, category} on first match, or None.
"""
if self._is_safe(text):
return None
for pattern, category in _COMPILED_TONE_PATTERNS:
m = pattern.search(text)
if m:
return {"matched_text": m.group(0), "category": category}
for pattern, category in self._extra_blocked:
m = pattern.search(text)
if m:
return {"matched_text": m.group(0), "category": category}
return None

View file

@ -1,45 +0,0 @@
from typing import TYPE_CHECKING, Optional
import litellm
from litellm.proxy.guardrails.guardrail_hooks.tone_detector.tone_detector import (
ToneDetectorGuardrail,
)
from litellm.types.guardrails import SupportedGuardrailIntegrations
if TYPE_CHECKING:
from litellm import Router
from litellm.types.guardrails import Guardrail, LitellmParams
def initialize_guardrail(
litellm_params: "LitellmParams",
guardrail: "Guardrail",
llm_router: Optional["Router"] = None,
):
"""Initialize the Tone Detector Guardrail."""
guardrail_name = guardrail.get("guardrail_name")
if not guardrail_name:
raise ValueError("Tone Detector: guardrail_name is required")
tone_guardrail = ToneDetectorGuardrail(
guardrail_name=guardrail_name,
blocked_phrases=getattr(litellm_params, "blocked_phrases", None),
safe_phrases=getattr(litellm_params, "safe_phrases", None),
event_hook=litellm_params.mode, # type: ignore
default_on=litellm_params.default_on or False,
violation_message_template=getattr(
litellm_params, "violation_message_template", None
),
)
litellm.logging_callback_manager.add_litellm_callback(tone_guardrail)
return tone_guardrail
guardrail_initializer_registry = {
SupportedGuardrailIntegrations.TONE_DETECTOR.value: initialize_guardrail,
}
guardrail_class_registry = {
SupportedGuardrailIntegrations.TONE_DETECTOR.value: ToneDetectorGuardrail,
}

View file

@ -1,186 +0,0 @@
"""
Tone Detector Guardrail for LiteLLM.
CPU-only guardrail that detects inappropriate tone in customer-facing chatbot
responses (dismissive, condescending, blaming, unprofessional language) while
allowing domain-specific terms that might otherwise trigger false positives.
"""
import re
from typing import TYPE_CHECKING, List, Literal, Optional, Pattern, Tuple
from fastapi import HTTPException
from litellm._logging import verbose_proxy_logger
from litellm.integrations.custom_guardrail import CustomGuardrail
from litellm.types.guardrails import GuardrailEventHooks
from litellm.types.utils import GenericGuardrailAPIInputs
if TYPE_CHECKING:
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
# ---------------------------------------------------------------------------
# Built-in tone patterns — each is (compiled_regex, category_label)
# ---------------------------------------------------------------------------
_TONE_PATTERNS: List[Tuple[str, str]] = [
# Dismissive
(r"\bthat(?:'s| is) not (?:really )?my (?:problem|concern|issue)\b", "dismissive"),
(r"\bi (?:don't|do not) see what the big deal is\b", "dismissive"),
(r"\byou(?:'re| are) overthinking\b", "dismissive"),
(r"\bjust read the (?:FAQ|docs|documentation|manual)\b", "dismissive"),
# Blaming the customer
# "you should have" is blame only when NOT followed by a past participle
# describing something the customer already received/seen
# (e.g. "you should have received" is informational, not blame)
(r"\byou should have (?:read|known|checked|done|thought|realized|paid|looked|noticed|seen to)\b", "blaming"),
(r"\bthat(?:'s| is) your (?:fault(?! tolerance| tolerant)|problem|mistake)\b", "blaming"),
(r"\bif you had (?:followed|read|done)\b", "blaming"),
(r"\byou clearly (?:didn't|did not)\b", "blaming"),
(r"\bthis (?:issue|problem) is on your end\b", "blaming"),
# Refusal to help (without offering alternatives)
# "I can't help you" but NOT "I can't help but notice" (which is polite)
(r"\bi (?:can't|cannot|can not) help you\b", "refusal"),
# "nothing I can do" is refusal only when NOT followed by "but" / "to" + helpful verb
(r"\bthere(?:'s| is) nothing (?:i|we) can do(?! (?:but|to))\b", "refusal"),
(r"\byou(?:'ll| will) (?:just )?have to figure it out\b", "refusal"),
(r"\btry somewhere else\b", "refusal"),
# Sarcasm / condescension
(r"\bif you(?:'d| had| would have) been paying attention\b", "condescension"),
(r"\bhow (?:to )?make this any simpler\b", "condescension"),
(r"\blet me spell it out for you\b", "condescension"),
(r"\bsince you (?:don't|do not) (?:seem to )?get it\b", "condescension"),
(r"\bdo your (?:job|work) for you\b", "condescension"),
# Impatience / frustration — "told you" specifically, not "told our team"
(r"\bi(?:'ve| have) already told you\b", "impatience"),
(r"\bhow many times do i have to\b", "impatience"),
(r"\bi (?:don't|do not) have time to\b", "impatience"),
(r"\bare you even listening\b", "impatience"),
(r"\bjust do what i said\b", "impatience"),
# Unprofessional casual language
(r"\b(?:bruh|lol|idk|smh|lmao|wtf)\b", "unprofessional"),
(r"\bmy bad dude\b", "unprofessional"),
(r"\bwhatever,? just deal with it\b", "unprofessional"),
(r"\bsounds like a you problem\b", "unprofessional"),
]
def _compile_patterns(
patterns: List[Tuple[str, str]],
) -> List[Tuple[Pattern[str], str]]:
return [(re.compile(p, re.IGNORECASE), cat) for p, cat in patterns]
# Pre-compiled at module load
_COMPILED_TONE_PATTERNS = _compile_patterns(_TONE_PATTERNS)
class ToneDetectorGuardrail(CustomGuardrail):
"""
Detects inappropriate tone in chatbot responses.
Configuration accepts:
blocked_phrases: additional regex patterns to block
safe_phrases: regex patterns that exempt a text from blocking
(e.g. domain-specific jargon)
"""
def __init__(
self,
guardrail_name: Optional[str] = None,
blocked_phrases: Optional[List[str]] = None,
safe_phrases: Optional[List[str]] = None,
**kwargs,
):
super().__init__(
guardrail_name=guardrail_name,
supported_event_hooks=[
GuardrailEventHooks.pre_call,
GuardrailEventHooks.post_call,
GuardrailEventHooks.during_call,
],
**kwargs,
)
# User-supplied additional blocked patterns
self._extra_blocked: List[Tuple[Pattern[str], str]] = []
for phrase in blocked_phrases or []:
self._extra_blocked.append(
(re.compile(phrase, re.IGNORECASE), "custom_blocked")
)
# User-supplied safe-phrase patterns — if ANY safe phrase matches
# the text, that text is allowed through even if a tone pattern fires.
self._safe_patterns: List[Pattern[str]] = []
for phrase in safe_phrases or []:
self._safe_patterns.append(re.compile(phrase, re.IGNORECASE))
# ------------------------------------------------------------------
# Core detection
# ------------------------------------------------------------------
def _is_safe(self, text: str) -> bool:
"""Return True if text matches any user-defined safe phrase."""
return any(p.search(text) for p in self._safe_patterns)
def _check_tone(self, text: str) -> Optional[Tuple[str, str]]:
"""
Check a single text for tone violations.
Returns (matched_text, category) on first match, or None.
"""
if self._is_safe(text):
return None
for pattern, category in _COMPILED_TONE_PATTERNS:
m = pattern.search(text)
if m:
return (m.group(0), category)
for pattern, category in self._extra_blocked:
m = pattern.search(text)
if m:
return (m.group(0), category)
return None
# ------------------------------------------------------------------
# CustomGuardrail hook
# ------------------------------------------------------------------
async def apply_guardrail(
self,
inputs: GenericGuardrailAPIInputs,
request_data: dict,
input_type: Literal["request", "response"],
logging_obj: Optional["LiteLLMLoggingObj"] = None,
) -> GenericGuardrailAPIInputs:
texts = inputs.get("texts") or []
for text in texts:
if not text:
continue
result = self._check_tone(text)
if result is not None:
matched, category = result
verbose_proxy_logger.warning(
"ToneDetector blocked (%s): '%s'",
category,
matched,
)
raise HTTPException(
status_code=400,
detail={
"error": f"Tone violation detected: {category}",
"category": category,
"matched_text": matched,
},
)
return inputs

View file

@ -78,7 +78,6 @@ class SupportedGuardrailIntegrations(Enum):
SEMANTIC_GUARD = "semantic_guard"
MCP_END_USER_PERMISSION = "mcp_end_user_permission"
BLOCK_CODE_EXECUTION = "block_code_execution"
TONE_DETECTOR = "tone_detector"
class Role(Enum):

View file

@ -4,8 +4,7 @@ from typing import Any, Dict, List, Literal, Optional, TypedDict, Union
from pydantic import Field
from litellm.types.llms.base import BaseLiteLLMOpenAIResponseObject
from litellm.types.proxy.guardrails.guardrail_hooks.base import \
GuardrailConfigModel
from litellm.types.proxy.guardrails.guardrail_hooks.base import GuardrailConfigModel
# --- Competitor intent blocker (generic, industry-agnostic) ---
@ -81,11 +80,20 @@ class CompetitorIntentDetection(TypedDict):
evidence: List[Dict[str, Any]]
class ToneDetection(TypedDict):
"""Detection from tone checker (category + matched text)."""
type: Literal["tone"]
category: str
matched_text: str
ContentFilterDetection = Union[
PatternDetection,
BlockedWordDetection,
CategoryKeywordDetection,
CompetitorIntentDetection,
ToneDetection,
]
@ -173,6 +181,14 @@ class LitellmContentFilterGuardrailConfigModel(GuardrailConfigModel):
"reframe_message_template, refuse_message_template.",
)
# Tone detection (customer-facing chatbot tone checks)
tone_detection_config: Optional[Dict[str, Any]] = Field(
default=None,
description="Optional config for CPU-only tone detection. "
"Keys: blocked_phrases (list of regex strings, optional), "
"safe_phrases (list of regex strings that exempt text from blocking, optional).",
)
@staticmethod
def ui_friendly_name() -> str:
return "LiteLLM Content Filter"

View file

@ -1,5 +1,5 @@
"""
Tests for the Tone Detector Guardrail.
Tests for the Tone Detection feature of ContentFilterGuardrail.
Covers:
- True positives: inappropriate tone is blocked
@ -19,16 +19,28 @@ sys.path.insert(0, os.path.abspath("../../"))
from fastapi import HTTPException
from litellm.proxy.guardrails.guardrail_hooks.tone_detector.tone_detector import (
ToneDetectorGuardrail,
from litellm.proxy.guardrails.guardrail_hooks.litellm_content_filter.content_filter import (
ContentFilterGuardrail,
)
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _make_guardrail(**kwargs) -> ToneDetectorGuardrail:
return ToneDetectorGuardrail(guardrail_name="test-tone", **kwargs)
def _make_guardrail(
blocked_phrases=None,
safe_phrases=None,
) -> ContentFilterGuardrail:
"""Create a ContentFilterGuardrail with only tone detection enabled."""
config = {}
if blocked_phrases is not None:
config["blocked_phrases"] = blocked_phrases
if safe_phrases is not None:
config["safe_phrases"] = safe_phrases
return ContentFilterGuardrail(
guardrail_name="test-tone",
tone_detection_config=config,
)
def _inputs(text: str) -> dict:
@ -211,42 +223,34 @@ class TestFalsePositiveResistance:
@pytest.mark.parametrize(
"text,description",
[
# "you should have" + informational continuation
(
"You should have received a confirmation email within 5 minutes.",
"informational 'should have received'",
),
# "problem is on your end" with helpful framing
(
"If the problem is on your end, here are some steps to troubleshoot.",
"'the problem' not 'this problem/issue'",
),
# "I can't help but" (positive usage)
(
"I can't help but notice you've been a loyal customer — thank you!",
"'can't help but notice' is a compliment",
),
# "spell out" without "for you"
(
"Let me spell out the steps clearly so nothing is missed.",
"'spell out the steps' is helpful, not condescending",
),
# "I've already told" + someone other than the customer
(
"I've already told our engineering team about this, and they're working on a fix.",
"'told our team' is reassuring, not impatient",
),
# "nothing I can do" + "but" (offers alternative)
(
"There's nothing I can do to speed up the shipment, but I can offer a discount on your next order.",
"'nothing I can do to X, but Y' offers an alternative",
),
# "FAQ" in a polite context (no "just read the")
(
"Please check the FAQ for a list of supported file formats — it's very comprehensive.",
"polite FAQ reference without 'just read the'",
),
# "your fault" as a technical term
(
"I understand that's your fault tolerance threshold — let me adjust it for you.",
"'fault tolerance' is a technical term",
@ -371,17 +375,11 @@ class TestEdgeCases:
result = await g.apply_guardrail({"texts": [""]}, {}, "response")
assert result["texts"] == [""]
@pytest.mark.asyncio
async def test_none_in_texts_passes(self):
g = _make_guardrail()
result = await g.apply_guardrail({"texts": [None]}, {}, "response")
assert result["texts"] == [None]
@pytest.mark.asyncio
async def test_no_texts_key(self):
g = _make_guardrail()
result = await g.apply_guardrail({}, {}, "response")
assert result == {}
assert result["texts"] == []
@pytest.mark.asyncio
async def test_multiple_texts_blocks_on_first_violation(self):
@ -409,24 +407,19 @@ class TestEdgeCases:
# ---------------------------------------------------------------------------
# INIT & REGISTRATION
# INIT
# ---------------------------------------------------------------------------
class TestRegistration:
class TestInit:
def test_guardrail_name_set(self):
g = _make_guardrail()
assert g.guardrail_name == "test-tone"
def test_init_with_no_extras(self):
def test_tone_checker_enabled(self):
g = _make_guardrail()
assert g._extra_blocked == []
assert g._safe_patterns == []
assert g._tone_checker is not None
def test_init_with_blocked_and_safe(self):
g = _make_guardrail(
blocked_phrases=[r"foo", r"bar"],
safe_phrases=[r"baz"],
)
assert len(g._extra_blocked) == 2
assert len(g._safe_patterns) == 1
def test_tone_checker_disabled_when_no_config(self):
g = ContentFilterGuardrail(guardrail_name="test-no-tone")
assert g._tone_checker is None