Event Callbacks
React to detection, anonymization, and deanonymization events
Rdakt AI fires callbacks at key points in the processing pipeline. Use them for logging, monitoring, or custom logic.
Available Callbacks
| Callback | When it fires | Parameters |
|---|---|---|
on_entities_detected | After detection, before anonymization | (entities: list[Entity], text: str) |
on_anonymized | After anonymization | (original: str, anonymized: str, mapping: dict[str, str]) |
on_deanonymized | After deanonymization of a response | (anonymized: str, restored: str) |
on_error | When detection or anonymization fails | (error: Exception, stage: str) |
Usage
from rdakt_ai import RdaktMiddleware, Entity
def log_entities(entities: list[Entity], text: str) -> None:
for entity in entities:
print(f"Found {entity.type}: {entity.value}")
def log_anonymization(original: str, anonymized: str, mapping: dict[str, str]) -> None:
print(f"Anonymized {len(mapping)} entities")
for token, value in mapping.items():
print(f" {value} → {token}")
def log_deanonymization(anonymized: str, restored: str) -> None:
print(f"Deanonymized response ({len(restored)} chars)")
def handle_error(error: Exception, stage: str) -> None:
print(f"Error in {stage}: {error}")
middleware = RdaktMiddleware(
on_entities_detected=log_entities,
on_anonymized=log_anonymization,
on_deanonymized=log_deanonymization,
on_error=handle_error,
)Monitoring Example
Send metrics to your observability platform:
import time
def track_metrics(entities: list[Entity], text: str) -> None:
# Count entity types
type_counts: dict[str, int] = {}
for entity in entities:
type_counts[entity.type] = type_counts.get(entity.type, 0) + 1
# Send to your metrics system
for entity_type, count in type_counts.items():
metrics.increment("rdakt.entities_detected", count, tags=[f"type:{entity_type}"])
middleware = RdaktMiddleware(
on_entities_detected=track_metrics,
)Audit Logging Example
Create a compliance audit trail:
import json
import logging
audit_logger = logging.getLogger("rdakt.audit")
def audit_log(original: str, anonymized: str, mapping: dict[str, str]) -> None:
audit_logger.info(json.dumps({
"event": "anonymization",
"entities_count": len(mapping),
"entity_types": list({v.split("_")[0].strip("<") for v in mapping}),
}))
middleware = RdaktMiddleware(
on_anonymized=audit_log,
)