Detection Pipeline

The detection pipeline is a composable chain of stages that analyze text and return detected entities. Each stage runs a detector and passes results to the next stage.

Pipeline Stages

A pipeline is a list of RdaktStage objects. The built-in stages are:

Stage	Detector	What it finds
`DetectorStage(RegexDetector())`	Regex	Emails, SSNs, credit cards, phone numbers, IPs, IBANs, JWTs, API keys
`DetectorStage(SpacyDetector())`	NER (spaCy)	Person names, organizations, locations, dates

Configuration

In YAML, specify detectors by name:

pipeline:
  - regex
  - ner:
      model: en_core_web_sm

In code, build the pipeline directly:

from rdakt_ai import RdaktPipeline, DetectorStage, RegexDetector

pipeline = RdaktPipeline(stages=[
    DetectorStage(RegexDetector()),
])

entities = pipeline.detect_sync("Email me at john@example.com")
# [Entity(value='john@example.com', type='EMAIL', start=12, end=28)]

Built-in Regex Patterns

The RegexDetector ships with 8 built-in patterns:

Entity Type	Example Match
`EMAIL`	`john@example.com`
`SSN`	`123-45-6789`
`CREDIT_CARD`	`4111 1111 1111 1111`
`PHONE`	`(555) 123-4567`
`IP_ADDRESS`	`192.168.1.1`
`IBAN`	`DE89370400440532013000`
`JWT`	`eyJhbG...`
`API_KEY`	`sk-proj-...`, `AKIA...`, `ghp_...`, `glpat-...`

Custom Patterns

Add custom regex patterns for domain-specific PII:

entities:
  EMPLOYEE_ID:
    pattern: "EMP-\\d{6}"
    strategy: token
  ACCOUNT_NUMBER:
    pattern: "\\d{4}-\\d{4}-\\d{4}"
    strategy: token

Or in code:

detector = RegexDetector(custom_patterns={
    "EMPLOYEE_ID": r"EMP-\d{6}",
    "ACCOUNT_NUMBER": r"\d{4}-\d{4}-\d{4}",
})

Overlap Resolution

When multiple detectors find overlapping entities, the pipeline resolves conflicts:

Longest match wins — if one entity spans a larger portion of text, it takes priority
Priority tiebreaker — if two entities have the same span, the one from the higher-priority detector wins

This means you can safely run both regex and NER detectors — they won't produce duplicate or conflicting results.

Chain of Responsibility

The pipeline uses a chain-of-responsibility pattern. Each stage receives:

The input text
A RdaktContext (mutable list of entities)
A call_next function to invoke the next stage

This allows stages to modify the context or even short-circuit the pipeline:

from rdakt_ai import RdaktStage, RdaktContext

class AllowlistStage(RdaktStage):
    """Remove entities that match an allowlist."""

    def __init__(self, allowed: set[str]):
        self.allowed = allowed

    async def process(self, text, context, call_next):
        await call_next(text, context)
        context.entities = [
            e for e in context.entities if e.value not in self.allowed
        ]

    def process_sync(self, text, context, call_next_sync):
        call_next_sync(text, context)
        context.entities = [
            e for e in context.entities if e.value not in self.allowed
        ]