跳转至

API reference

Generated from the apogee-ai-guardrails source with mkdocstrings. Every symbol below is exported from apogee_ai_guardrails, so it is part of the supported public surface.

Application

PolicyOutcome dataclass

Python
PolicyOutcome(report: GuardrailReport, redacted_text: str)

report instance-attribute

Python
report: GuardrailReport

redacted_text instance-attribute

Python
redacted_text: str

Application · DTOs

ApplyPolicyDTO dataclass

Python
ApplyPolicyDTO(text: str, direction: str = 'input', policy_name: str = 'default')

text instance-attribute

Python
text: str

direction class-attribute instance-attribute

Python
direction: str = 'input'

policy_name class-attribute instance-attribute

Python
policy_name: str = 'default'

BenchDTO dataclass

Python
BenchDTO(inputs: int = 200)

inputs class-attribute instance-attribute

Python
inputs: int = 200

RedactDTO dataclass

Python
RedactDTO(text: str, detectors: tuple[str, ...] = (), policy_name: str = 'default')

text instance-attribute

Python
text: str

detectors class-attribute instance-attribute

Python
detectors: tuple[str, ...] = ()

policy_name class-attribute instance-attribute

Python
policy_name: str = 'default'

ScanDTO dataclass

Python
ScanDTO(text: str, detectors: tuple[str, ...] = (), direction: str = 'input')

text instance-attribute

Python
text: str

detectors class-attribute instance-attribute

Python
detectors: tuple[str, ...] = ()

direction class-attribute instance-attribute

Python
direction: str = 'input'

Application · Use cases

ApplyPolicyUseCase

Python
ApplyPolicyUseCase(detectors, policy: RedactionPolicy)

Pipeline: scan → block_if_critical → redact → return.

Source code in apogee_ai_guardrails/application/use_cases/apply_policy_use_case.py
Python
def __init__(self, detectors, policy: RedactionPolicy) -> None:
    self._scan = ScanTextUseCase(detectors)
    self._redact = RedactTextUseCase(detectors, policy)
    self._policy = policy

execute async

Python
execute(text: str, direction: ScanDirection = INPUT, raise_on_block: bool = True) -> PolicyOutcome
Source code in apogee_ai_guardrails/application/use_cases/apply_policy_use_case.py
Python
async def execute(
    self,
    text: str,
    direction: ScanDirection = ScanDirection.INPUT,
    raise_on_block: bool = True,
) -> PolicyOutcome:
    report = await self._scan.execute(text, direction=direction)
    blocked = any(
        _ge(v.severity, self._policy.block_severity) for v in report.violations
    )
    if blocked:
        blocked_report = GuardrailReport(
            text=report.text,
            direction=report.direction,
            violations=report.violations,
            blocked=True,
            metadata=report.metadata,
        )
        if raise_on_block:
            raise GuardrailBlockedException(
                reason=f"{len(report.violations)} violation(s)",
                severity=report.severity.value,
            )
        return PolicyOutcome(report=blocked_report, redacted_text="")
    redacted = await self._redact.execute(text)
    return PolicyOutcome(report=report, redacted_text=redacted)

BenchPipelineUseCase

Synthetic load against ApplyPolicy: measure detect/redact rates.

execute async

Python
execute(inputs: int) -> dict[str, float]
Source code in apogee_ai_guardrails/application/use_cases/bench_pipeline_use_case.py
Python
async def execute(self, inputs: int) -> dict[str, float]:
    if inputs <= 0:
        raise ValueError("inputs must be positive")
    rng = random.Random(42)
    registry = DetectorRegistry.default()
    pipeline = ApplyPolicyUseCase(registry.all(), default_redaction_policy())

    detected = 0
    blocked = 0
    start = time.perf_counter()
    for _ in range(inputs):
        r = rng.random()
        if r < 0.4:
            text = rng.choice(_BENIGN)
        elif r < 0.85:
            text = rng.choice(_PII_LACED)
        else:
            text = rng.choice(_INJECTION)
        try:
            outcome = await pipeline.execute(text, raise_on_block=False)
        except Exception:  # pragma: no cover
            continue
        if outcome.report.violations:
            detected += 1
        if outcome.report.blocked:
            blocked += 1
    elapsed = (time.perf_counter() - start) * 1000.0
    return {
        "inputs": float(inputs),
        "detected": float(detected),
        "blocked": float(blocked),
        "detection_rate": detected / inputs,
        "elapsed_ms": elapsed,
        "throughput_per_sec": (inputs / elapsed * 1000.0) if elapsed > 0 else 0.0,
    }

ListDetectorsUseCase

Python
ListDetectorsUseCase(registry)
Source code in apogee_ai_guardrails/application/use_cases/list_detectors_use_case.py
Python
def __init__(self, registry) -> None:
    self._registry = registry

execute async

Python
execute() -> list[str]
Source code in apogee_ai_guardrails/application/use_cases/list_detectors_use_case.py
Python
async def execute(self) -> list[str]:
    return self._registry.list()

RedactTextUseCase

Python
RedactTextUseCase(detectors: Iterable, policy: RedactionPolicy | None = None)

Applies the policy strategies on a per-violation basis.

Source code in apogee_ai_guardrails/application/use_cases/redact_text_use_case.py
Python
def __init__(self, detectors: Iterable, policy: RedactionPolicy | None = None) -> None:
    self._detectors = list(detectors)
    self._policy = policy or default_redaction_policy()
    self._mask = MaskRedactor(mask_char=self._policy.mask_char)
    self._hash = HashRedactor()
    self._format = FormatPreservingRedactor()

execute async

Python
execute(text: str) -> str
Source code in apogee_ai_guardrails/application/use_cases/redact_text_use_case.py
Python
async def execute(self, text: str) -> str:
    violations: list[GuardrailViolation] = []
    for d in self._detectors:
        violations.extend(await d.detect(text))
    if not violations:
        return text

    grouped: dict[RedactionStrategy, list[GuardrailViolation]] = defaultdict(list)
    for v in violations:
        grouped[self._policy.strategy_for(v.kind)].append(v)

    # Apply right-to-left so spans don't shift. We sort the union once.
    all_with_strategy = sorted(
        ((v, strat) for strat, vs in grouped.items() for v in vs),
        key=lambda item: item[0].span_start,
        reverse=True,
    )

    out = text
    for v, strategy in all_with_strategy:
        replacement = self._render(v, strategy)
        out = out[: v.span_start] + replacement + out[v.span_end :]
    return out

ScanTextUseCase

Python
ScanTextUseCase(detectors: Iterable)

Runs every detector against the text and aggregates violations.

Source code in apogee_ai_guardrails/application/use_cases/scan_text_use_case.py
Python
def __init__(self, detectors: Iterable) -> None:
    self._detectors = list(detectors)
    if not self._detectors:
        raise ValueError("at least one detector is required")

execute async

Python
execute(text: str, direction: ScanDirection = INPUT) -> GuardrailReport
Source code in apogee_ai_guardrails/application/use_cases/scan_text_use_case.py
Python
async def execute(
    self,
    text: str,
    direction: ScanDirection = ScanDirection.INPUT,
) -> GuardrailReport:
    all_violations: list[GuardrailViolation] = []
    for detector in self._detectors:
        all_violations.extend(await detector.detect(text))
    return GuardrailReport(
        text=text,
        direction=direction,
        violations=tuple(all_violations),
        metadata={"detectors": ",".join(d.name for d in self._detectors)},
    )

Domain

GuardrailReport dataclass

Python
GuardrailReport(text: str, direction: ScanDirection = INPUT, violations: tuple[GuardrailViolation, ...] = (), blocked: bool = False, metadata: dict[str, str] = dict())

text instance-attribute

Python
text: str

direction class-attribute instance-attribute

Python
direction: ScanDirection = INPUT

violations class-attribute instance-attribute

Python
violations: tuple[GuardrailViolation, ...] = ()

blocked class-attribute instance-attribute

Python
blocked: bool = False

metadata class-attribute instance-attribute

Python
metadata: dict[str, str] = field(default_factory=dict)

severity property

Python
severity: RuleSeverity

has_violations property

Python
has_violations: bool

GuardrailViolation dataclass

Python
GuardrailViolation(kind: PiiKind, detector: str, span_start: int, span_end: int, matched_text: str, severity: RuleSeverity = MEDIUM, confidence: float = 1.0, metadata: dict[str, str] = dict())

kind instance-attribute

Python
kind: PiiKind

detector instance-attribute

Python
detector: str

span_start instance-attribute

Python
span_start: int

span_end instance-attribute

Python
span_end: int

matched_text instance-attribute

Python
matched_text: str

severity class-attribute instance-attribute

Python
severity: RuleSeverity = MEDIUM

confidence class-attribute instance-attribute

Python
confidence: float = 1.0

metadata class-attribute instance-attribute

Python
metadata: dict[str, str] = field(default_factory=dict)

RedactionPolicy dataclass

Python
RedactionPolicy(name: str = 'default', default_strategy: RedactionStrategy = MASK, per_kind: dict[PiiKind, RedactionStrategy] = dict(), block_severity: RuleSeverity = CRITICAL, mask_char: str = '*')

name class-attribute instance-attribute

Python
name: str = 'default'

default_strategy class-attribute instance-attribute

Python
default_strategy: RedactionStrategy = MASK

per_kind class-attribute instance-attribute

Python
per_kind: dict[PiiKind, RedactionStrategy] = field(default_factory=dict)

block_severity class-attribute instance-attribute

Python
block_severity: RuleSeverity = CRITICAL

mask_char class-attribute instance-attribute

Python
mask_char: str = '*'

strategy_for

Python
strategy_for(kind: PiiKind) -> RedactionStrategy
Source code in apogee_ai_guardrails/domain/value_objects/redaction_policy.py
Python
def strategy_for(self, kind: PiiKind) -> RedactionStrategy:
    return self.per_kind.get(kind, self.default_strategy)

RedactionStrategy

Bases: str, Enum

MASK class-attribute instance-attribute

Python
MASK = 'mask'

HASH class-attribute instance-attribute

Python
HASH = 'hash'

REMOVE class-attribute instance-attribute

Python
REMOVE = 'remove'

FORMAT_PRESERVING class-attribute instance-attribute

Python
FORMAT_PRESERVING = 'format_preserving'

REPLACE class-attribute instance-attribute

Python
REPLACE = 'replace'

RuleSeverity

Bases: str, Enum

INFO class-attribute instance-attribute

Python
INFO = 'info'

LOW class-attribute instance-attribute

Python
LOW = 'low'

MEDIUM class-attribute instance-attribute

Python
MEDIUM = 'medium'

HIGH class-attribute instance-attribute

Python
HIGH = 'high'

CRITICAL class-attribute instance-attribute

Python
CRITICAL = 'critical'

ScanDirection

Bases: str, Enum

INPUT class-attribute instance-attribute

Python
INPUT = 'input'

OUTPUT class-attribute instance-attribute

Python
OUTPUT = 'output'

BOTH class-attribute instance-attribute

Python
BOTH = 'both'

Domain · Enums

PiiKind

Bases: str, Enum

EMAIL class-attribute instance-attribute

Python
EMAIL = 'email'

CPF class-attribute instance-attribute

Python
CPF = 'cpf'

CNPJ class-attribute instance-attribute

Python
CNPJ = 'cnpj'

RG class-attribute instance-attribute

Python
RG = 'rg'

PHONE class-attribute instance-attribute

Python
PHONE = 'phone'

CREDIT_CARD class-attribute instance-attribute

Python
CREDIT_CARD = 'credit_card'

SSN class-attribute instance-attribute

Python
SSN = 'ssn'

IP_V4 class-attribute instance-attribute

Python
IP_V4 = 'ip_v4'

URL class-attribute instance-attribute

Python
URL = 'url'

PROFANITY class-attribute instance-attribute

Python
PROFANITY = 'profanity'

PROMPT_INJECTION class-attribute instance-attribute

Python
PROMPT_INJECTION = 'prompt_injection'

JAILBREAK class-attribute instance-attribute

Python
JAILBREAK = 'jailbreak'

SCHEMA_VIOLATION class-attribute instance-attribute

Python
SCHEMA_VIOLATION = 'schema_violation'

BIAS class-attribute instance-attribute

Python
BIAS = 'bias'

HALLUCINATION class-attribute instance-attribute

Python
HALLUCINATION = 'hallucination'

OTHER class-attribute instance-attribute

Python
OTHER = 'other'

Domain · Exceptions

GuardrailBlockedException

Python
GuardrailBlockedException(reason: str, severity: str)

Bases: GuardrailError

Source code in apogee_ai_guardrails/domain/exceptions/guardrail_exceptions.py
Python
def __init__(self, reason: str, severity: str) -> None:
    super().__init__(f"Guardrail blocked content [{severity}]: {reason}")
    self.reason = reason
    self.severity = severity

reason instance-attribute

Python
reason = reason

severity instance-attribute

Python
severity = severity

GuardrailError

Bases: Exception

Base for apogee-ai-guardrails errors.

PolicyViolationException

Python
PolicyViolationException(policy: str, reason: str)

Bases: GuardrailError

Source code in apogee_ai_guardrails/domain/exceptions/guardrail_exceptions.py
Python
def __init__(self, policy: str, reason: str) -> None:
    super().__init__(f"Policy {policy!r} violated: {reason}")
    self.policy = policy
    self.reason = reason

policy instance-attribute

Python
policy = policy

reason instance-attribute

Python
reason = reason

Domain · Protocols (ports)

IDetector

Bases: Protocol

name instance-attribute

Python
name: str

detect async

Python
detect(text: str) -> list[GuardrailViolation]
Source code in apogee_ai_guardrails/domain/services/i_detector.py
Python
async def detect(self, text: str) -> list[GuardrailViolation]: ...

IGuardrailPipeline

Bases: Protocol

scan async

Python
scan(text: str) -> GuardrailReport
Source code in apogee_ai_guardrails/domain/services/i_guardrail_pipeline.py
Python
async def scan(self, text: str) -> GuardrailReport: ...

IRedactor

Bases: Protocol

name instance-attribute

Python
name: str

redact

Python
redact(text: str, violations: Iterable[GuardrailViolation], policy: RedactionPolicy | None = None) -> str
Source code in apogee_ai_guardrails/domain/services/i_redactor.py
Python
def redact(
    self,
    text: str,
    violations: Iterable[GuardrailViolation],
    policy: RedactionPolicy | None = None,
) -> str: ...

Infrastructure

DetectorRegistry

Python
DetectorRegistry(detectors: Iterable | None = None)
Source code in apogee_ai_guardrails/infrastructure/registries/detector_registry.py
Python
def __init__(self, detectors: Iterable | None = None) -> None:
    self._detectors: dict[str, object] = {}
    for d in detectors or []:
        self._detectors[d.name] = d

default classmethod

Python
default() -> 'DetectorRegistry'
Source code in apogee_ai_guardrails/infrastructure/registries/detector_registry.py
Python
@classmethod
def default(cls) -> "DetectorRegistry":
    return cls([
        PiiDetector(),
        ProfanityDetector(),
        PromptInjectionDetector(),
        JailbreakDetector(),
    ])

register

Python
register(detector) -> None
Source code in apogee_ai_guardrails/infrastructure/registries/detector_registry.py
Python
def register(self, detector) -> None:
    self._detectors[detector.name] = detector

get

Python
get(name: str)
Source code in apogee_ai_guardrails/infrastructure/registries/detector_registry.py
Python
def get(self, name: str):
    if name not in self._detectors:
        raise KeyError(f"unknown detector: {name}")
    return self._detectors[name]

list

Python
list() -> list[str]
Source code in apogee_ai_guardrails/infrastructure/registries/detector_registry.py
Python
def list(self) -> list[str]:
    return list(self._detectors.keys())

all

Python
all() -> list
Source code in apogee_ai_guardrails/infrastructure/registries/detector_registry.py
Python
def all(self) -> list:
    return list(self._detectors.values())

FormatPreservingRedactor

Replaces digits with X / letters with x, keeping punctuation intact.

Useful when downstream parsing depends on the original layout (e.g. CPF 123.456.789-09 becomes XXX.XXX.XXX-XX).

name class-attribute instance-attribute

Python
name = 'format_preserving'

redact

Python
redact(text: str, violations: Iterable[GuardrailViolation], policy: RedactionPolicy | None = None) -> str
Source code in apogee_ai_guardrails/infrastructure/redactors/format_preserving_redactor.py
Python
def redact(
    self,
    text: str,
    violations: Iterable[GuardrailViolation],
    policy: RedactionPolicy | None = None,
) -> str:
    return apply_replacements(
        text, violations, lambda v: self._redact_value(v.matched_text)
    )

HashRedactor

Python
HashRedactor(salt: str = 'apogee', algo: str = 'sha256')

Replaces matched spans with <KIND:hash8> for deterministic correlation.

Source code in apogee_ai_guardrails/infrastructure/redactors/hash_redactor.py
Python
def __init__(self, salt: str = "apogee", algo: str = "sha256") -> None:
    self._salt = salt
    self._algo = algo

name class-attribute instance-attribute

Python
name = 'hash'

redact

Python
redact(text: str, violations: Iterable[GuardrailViolation], policy: RedactionPolicy | None = None) -> str
Source code in apogee_ai_guardrails/infrastructure/redactors/hash_redactor.py
Python
def redact(
    self,
    text: str,
    violations: Iterable[GuardrailViolation],
    policy: RedactionPolicy | None = None,
) -> str:
    return apply_replacements(
        text,
        violations,
        lambda v: f"<{v.kind.value.upper()}:{self._hash(v.matched_text)}>",
    )

JailbreakDetector

Python
JailbreakDetector(severity: RuleSeverity = CRITICAL)
Source code in apogee_ai_guardrails/infrastructure/detectors/jailbreak_detector.py
Python
def __init__(self, severity: RuleSeverity = RuleSeverity.CRITICAL) -> None:
    self._severity = severity

name class-attribute instance-attribute

Python
name = 'jailbreak'

detect async

Python
detect(text: str) -> list[GuardrailViolation]
Source code in apogee_ai_guardrails/infrastructure/detectors/jailbreak_detector.py
Python
async def detect(self, text: str) -> list[GuardrailViolation]:
    out: list[GuardrailViolation] = []
    for pat in _PATTERNS:
        for m in pat.finditer(text):
            out.append(
                GuardrailViolation(
                    kind=PiiKind.JAILBREAK,
                    detector=self.name,
                    span_start=m.start(),
                    span_end=m.end(),
                    matched_text=m.group(0),
                    severity=self._severity,
                    confidence=0.9,
                )
            )
    return out

MaskRedactor

Python
MaskRedactor(mask_char: str = '*')

Replaces matched text with a single repeated mask char of equal length.

Source code in apogee_ai_guardrails/infrastructure/redactors/mask_redactor.py
Python
def __init__(self, mask_char: str = "*") -> None:
    if len(mask_char) != 1:
        raise ValueError("mask_char must be a single character")
    self._mask_char = mask_char

name class-attribute instance-attribute

Python
name = 'mask'

redact

Python
redact(text: str, violations: Iterable[GuardrailViolation], policy: RedactionPolicy | None = None) -> str
Source code in apogee_ai_guardrails/infrastructure/redactors/mask_redactor.py
Python
def redact(
    self,
    text: str,
    violations: Iterable[GuardrailViolation],
    policy: RedactionPolicy | None = None,
) -> str:
    char = (policy.mask_char if policy is not None else self._mask_char) or "*"
    return apply_replacements(
        text, violations, lambda v: char * (v.span_end - v.span_start)
    )

PiiDetector

Python
PiiDetector(kinds: tuple[PiiKind, ...] | None = None)

Regex-based PII scanner for PT-BR + EN patterns. Zero external deps.

Source code in apogee_ai_guardrails/infrastructure/detectors/pii_detector.py
Python
def __init__(self, kinds: tuple[PiiKind, ...] | None = None) -> None:
    if kinds is None:
        self._kinds = tuple(_PATTERNS.keys())
    else:
        self._kinds = tuple(k for k in kinds if k in _PATTERNS)

name class-attribute instance-attribute

Python
name = 'pii'

detect async

Python
detect(text: str) -> list[GuardrailViolation]
Source code in apogee_ai_guardrails/infrastructure/detectors/pii_detector.py
Python
async def detect(self, text: str) -> list[GuardrailViolation]:
    out: list[GuardrailViolation] = []
    for kind in self._kinds:
        pattern, severity = _PATTERNS[kind]
        for match in pattern.finditer(text):
            # Skip IPs that are fully zero (likely not real PII).
            if kind is PiiKind.IP_V4 and match.group(0) == "0.0.0.0":
                continue
            out.append(
                GuardrailViolation(
                    kind=kind,
                    detector=self.name,
                    span_start=match.start(),
                    span_end=match.end(),
                    matched_text=match.group(0),
                    severity=severity,
                    confidence=0.95,
                )
            )
    return out

PresidioPiiDetector

Python
PresidioPiiDetector(language: str = 'en')

Lazy adapter for Microsoft Presidio. install via extras=presidio.

Source code in apogee_ai_guardrails/infrastructure/detectors/presidio_pii_detector.py
Python
def __init__(self, language: str = "en") -> None:
    self._language = language
    self._analyzer = None

name class-attribute instance-attribute

Python
name = 'presidio_pii'

detect async

Python
detect(text: str) -> list[GuardrailViolation]
Source code in apogee_ai_guardrails/infrastructure/detectors/presidio_pii_detector.py
Python
async def detect(self, text: str) -> list[GuardrailViolation]:
    self._ensure()
    results = self._analyzer.analyze(text=text, language=self._language)  # type: ignore[union-attr]
    out: list[GuardrailViolation] = []
    for r in results:
        out.append(
            GuardrailViolation(
                kind=PiiKind.OTHER,
                detector=self.name,
                span_start=int(r.start),
                span_end=int(r.end),
                matched_text=text[r.start : r.end],
                severity=RuleSeverity.HIGH,
                confidence=float(r.score),
                metadata={"entity_type": str(r.entity_type)},
            )
        )
    return out

ProfanityDetector

Python
ProfanityDetector(terms: tuple[str, ...] = _DEFAULT_TERMS, severity: RuleSeverity = LOW)
Source code in apogee_ai_guardrails/infrastructure/detectors/profanity_detector.py
Python
def __init__(
    self,
    terms: tuple[str, ...] = _DEFAULT_TERMS,
    severity: RuleSeverity = RuleSeverity.LOW,
) -> None:
    if not terms:
        raise ValueError("terms list cannot be empty")
    self._terms = tuple(t.lower() for t in terms)
    self._severity = severity
    joined = "|".join(re.escape(t) for t in self._terms)
    self._pattern = re.compile(rf"\b({joined})\b", re.IGNORECASE)

name class-attribute instance-attribute

Python
name = 'profanity'

detect async

Python
detect(text: str) -> list[GuardrailViolation]
Source code in apogee_ai_guardrails/infrastructure/detectors/profanity_detector.py
Python
async def detect(self, text: str) -> list[GuardrailViolation]:
    return [
        GuardrailViolation(
            kind=PiiKind.PROFANITY,
            detector=self.name,
            span_start=m.start(),
            span_end=m.end(),
            matched_text=m.group(0),
            severity=self._severity,
            confidence=0.85,
        )
        for m in self._pattern.finditer(text)
    ]

PromptInjectionDetector

Python
PromptInjectionDetector(severity: RuleSeverity = HIGH)
Source code in apogee_ai_guardrails/infrastructure/detectors/prompt_injection_detector.py
Python
def __init__(self, severity: RuleSeverity = RuleSeverity.HIGH) -> None:
    self._severity = severity

name class-attribute instance-attribute

Python
name = 'prompt_injection'

detect async

Python
detect(text: str) -> list[GuardrailViolation]
Source code in apogee_ai_guardrails/infrastructure/detectors/prompt_injection_detector.py
Python
async def detect(self, text: str) -> list[GuardrailViolation]:
    out: list[GuardrailViolation] = []
    for pat in _PATTERNS:
        for m in pat.finditer(text):
            out.append(
                GuardrailViolation(
                    kind=PiiKind.PROMPT_INJECTION,
                    detector=self.name,
                    span_start=m.start(),
                    span_end=m.end(),
                    matched_text=m.group(0),
                    severity=self._severity,
                    confidence=0.85,
                    metadata={"pattern": pat.pattern[:60]},
                )
            )
    return out

SchemaDetector

Python
SchemaDetector(required_keys: tuple[str, ...] = (), severity: RuleSeverity = HIGH)

Validates that the text parses as JSON and contains required keys.

Lightweight — no jsonschema dependency. Required keys are matched against the top-level object only.

Source code in apogee_ai_guardrails/infrastructure/detectors/schema_detector.py
Python
def __init__(
    self,
    required_keys: tuple[str, ...] = (),
    severity: RuleSeverity = RuleSeverity.HIGH,
) -> None:
    self._required = required_keys
    self._severity = severity

name class-attribute instance-attribute

Python
name = 'schema'

detect async

Python
detect(text: str) -> list[GuardrailViolation]
Source code in apogee_ai_guardrails/infrastructure/detectors/schema_detector.py
Python
async def detect(self, text: str) -> list[GuardrailViolation]:
    try:
        payload = json.loads(text)
    except json.JSONDecodeError as exc:
        return [
            GuardrailViolation(
                kind=PiiKind.SCHEMA_VIOLATION,
                detector=self.name,
                span_start=0,
                span_end=max(len(text), 1),
                matched_text=text[:100],
                severity=self._severity,
                confidence=1.0,
                metadata={"reason": f"invalid json: {exc.msg}"},
            )
        ]
    if not isinstance(payload, dict):
        return [
            GuardrailViolation(
                kind=PiiKind.SCHEMA_VIOLATION,
                detector=self.name,
                span_start=0,
                span_end=max(len(text), 1),
                matched_text=text[:100],
                severity=self._severity,
                confidence=1.0,
                metadata={"reason": "expected json object at top level"},
            )
        ]
    missing = [k for k in self._required if k not in payload]
    if missing:
        return [
            GuardrailViolation(
                kind=PiiKind.SCHEMA_VIOLATION,
                detector=self.name,
                span_start=0,
                span_end=max(len(text), 1),
                matched_text=", ".join(missing),
                severity=self._severity,
                confidence=1.0,
                metadata={"reason": "missing required keys"},
            )
        ]
    return []

default_redaction_policy

Python
default_redaction_policy() -> RedactionPolicy

Sensible defaults: format-preserving for IDs, mask for everything else.

Source code in apogee_ai_guardrails/infrastructure/policies/default_policy.py
Python
def default_redaction_policy() -> RedactionPolicy:
    """Sensible defaults: format-preserving for IDs, mask for everything else."""

    return RedactionPolicy(
        name="default",
        default_strategy=RedactionStrategy.MASK,
        per_kind={
            PiiKind.CPF: RedactionStrategy.FORMAT_PRESERVING,
            PiiKind.CNPJ: RedactionStrategy.FORMAT_PRESERVING,
            PiiKind.RG: RedactionStrategy.FORMAT_PRESERVING,
            PiiKind.CREDIT_CARD: RedactionStrategy.FORMAT_PRESERVING,
            PiiKind.SSN: RedactionStrategy.FORMAT_PRESERVING,
            PiiKind.EMAIL: RedactionStrategy.HASH,
            PiiKind.JAILBREAK: RedactionStrategy.REMOVE,
            PiiKind.PROMPT_INJECTION: RedactionStrategy.REMOVE,
        },
        block_severity=RuleSeverity.CRITICAL,
    )