Overall Class Structure
Rules inherit fromRuleBase and must implement the process method. They detect specific content and return a RuleOutput with confidence score and detected keywords.
from upsonic.safety_engine.base import RuleBase
from upsonic.safety_engine.models import PolicyInput, RuleOutput
from typing import Optional, Dict, Any
class MyCustomRule(RuleBase):
name = "My Custom Rule"
description = "Detects specific content in text"
language = "en"
def __init__(self, options: Optional[Dict[str, Any]] = None):
super().__init__(options)
self.keywords = ["keyword1", "keyword2"]
def process(self, policy_input: PolicyInput) -> RuleOutput:
combined_text = " ".join(policy_input.input_texts or []).lower()
triggered = []
for keyword in self.keywords:
if keyword in combined_text:
triggered.append(keyword)
if not triggered:
return RuleOutput(
confidence=0.0,
content_type="SAFE",
details="No issues detected"
)
return RuleOutput(
confidence=1.0,
content_type="DETECTED",
details=f"Found {len(triggered)} matches",
triggered_keywords=triggered
)
RuleOutput Fields
| Field | Type | Description |
|---|---|---|
confidence | float | Detection confidence between 0.0 and 1.0 |
content_type | str | Category label (e.g., "SAFE", "PII", "CONFIDENTIAL") |
details | str | Human-readable description of what was detected |
triggered_keywords | Optional[List[str]] | List of matched keywords/values to be passed to the action |
triggered_keywords list is important — it tells the action (e.g., replace_triggered_keywords or anonymize_triggered_keywords) which exact strings to transform.
Example Rule
Here’s a complete example of a custom rule that detects company-specific confidential terms:import re
from upsonic.safety_engine.base import RuleBase
from upsonic.safety_engine.models import PolicyInput, RuleOutput
from typing import Optional, Dict, Any
class CompanySecretRule(RuleBase):
name = "Company Secret Rule"
description = "Detects confidential company terms and code names"
language = "en"
def __init__(self, options: Optional[Dict[str, Any]] = None):
super().__init__(options)
self.secret_keywords = [
"project zeus", "alpha build", "confidential",
"internal only", "trade secret", "proprietary"
]
self.secret_patterns = [
r'\b(?:project|operation)\s+[A-Z][a-z]+\b',
r'\b[A-Z]{3}-\d{4}\b',
]
if options and "keywords" in options:
self.secret_keywords.extend(options["keywords"])
def process(self, policy_input: PolicyInput) -> RuleOutput:
combined_text = " ".join(policy_input.input_texts or [])
triggered_keywords = []
for keyword in self.secret_keywords:
pattern = r'\b' + re.escape(keyword) + r'\b'
if re.search(pattern, combined_text, re.IGNORECASE):
triggered_keywords.append(keyword)
for pattern in self.secret_patterns:
matches = re.findall(pattern, combined_text)
triggered_keywords.extend(matches)
if not triggered_keywords:
return RuleOutput(
confidence=0.0,
content_type="SAFE",
details="No confidential content detected"
)
confidence = min(1.0, len(triggered_keywords) * 0.5)
return RuleOutput(
confidence=confidence,
content_type="CONFIDENTIAL",
details=f"Detected {len(triggered_keywords)} confidential terms",
triggered_keywords=triggered_keywords
)
When using regex patterns with capturing groups
(...), use non-capturing groups (?:...) instead. Capturing groups cause re.findall to return only the captured group content (which may be empty), rather than the full match. This can lead to empty strings being passed to the action as triggered keywords.Using LLM in Rules
For more intelligent detection, you can use LLM-powered content finding:class SmartConfidentialRule(RuleBase):
name = "Smart Confidential Rule"
description = "Uses LLM to detect confidential content with context"
language = "en"
def __init__(self, options: Optional[Dict[str, Any]] = None, text_finder_llm=None):
super().__init__(options, text_finder_llm)
def process(self, policy_input: PolicyInput) -> RuleOutput:
if not self.text_finder_llm:
return RuleOutput(
confidence=0.0,
content_type="SAFE",
details="LLM not available"
)
triggered = self._llm_find_keywords_with_input(
"confidential company information",
policy_input
)
if not triggered:
return RuleOutput(
confidence=0.0,
content_type="SAFE",
details="No confidential content detected"
)
return RuleOutput(
confidence=1.0,
content_type="CONFIDENTIAL",
details=f"LLM detected {len(triggered)} confidential items",
triggered_keywords=triggered
)