Skip to content

Detectors & backends

Extension points and the layer internals.

Extension points

wardcat.register_action 

register_action(name, fn)

Register (or override) an anonymization action under name.

Source code in src/wardcat/core/actions.py 
102
103
104
def register_action(name: str, fn: ActionFn) -> None:
    """Register (or override) an anonymization action under *name*."""
    _ACTIONS[name] = fn

wardcat.ActionContext dataclass 

ActionContext(salt='', tokens=TokenAllocator())

Extra context an action may need beyond the span (e.g. the hash salt).

tokens class-attribute instance-attribute 

tokens = field(
    default_factory=TokenAllocator,
    repr=False,
    compare=False,
)

Placeholder vault for reversible actions, scoped to a single scan — its context_id is the one stamped into this scan's tokens. Excluded from repr/equality so two contexts with the same salt still compare equal.

wardcat.TokenAllocator 

TokenAllocator(context_id=None)

Hands out stable, unique placeholders for one scan — the tokenize vault.

A placeholder is [TYPE_index_contextid] — [EMAIL_1_9f3a2c8b71d4]. The index is per entity type in order of first appearance and restarts at 1 for every scan, so it stays short and readable; the context id is drawn once per allocator and shared by every token it hands out, which is what makes one scan's placeholders distinct from another's. An identical value always gets the same token, so a name repeated three times stays one referent for whatever reads the anonymized text.

That distinctness is the safety property behind :meth:~wardcat.ScanResult.restore: two scans running side by side both hold an "EMAIL number 1", and without the context id their placeholders would be the same string — restoring one request's answer against another's result would silently substitute the wrong person's value. With it there is nothing to match, so the mistake becomes a reported non-substitution instead.

Allocation state is per instance and the :class:Anonymizer <wardcat.core.anonymizer.Anonymizer> builds a fresh one for every apply() call, so concurrent scans never share a counter or an id.

Parameters:

Name Type Description Default
context_id str | None

the id to stamp into every token. Defaults to a fresh one; pass "" for bare [TYPE_index] placeholders (no cross-scan protection), or a fixed value to make output reproducible in tests.

.. warning:: Holds the raw values it has seen for the lifetime of the instance.

 None
Source code in src/wardcat/core/actions.py 
66
67
68
69
def __init__(self, context_id: str | None = None) -> None:
    self.context_id = new_context_id() if context_id is None else context_id
    self._tokens: dict[tuple[str, str], str] = {}
    self._counts: dict[str, int] = {}

token_for 

token_for(entity_type, text)

Return the placeholder for text, allocating a new one on first sight.

Source code in src/wardcat/core/actions.py 
71
72
73
74
75
76
77
78
79
80
81
def token_for(self, entity_type: str, text: str) -> str:
    """Return the placeholder for *text*, allocating a new one on first sight."""
    key = (entity_type, text)
    token = self._tokens.get(key)
    if token is None:
        count = self._counts.get(entity_type, 0) + 1
        self._counts[entity_type] = count
        suffix = f"_{self.context_id}" if self.context_id else ""
        token = f"[{entity_type}_{count}{suffix}]"
        self._tokens[key] = token
    return token

Detector interface

wardcat.detectors.base.BaseDetector

Bases: ABC

Interface that all detectors implement.

The engine talks to detectors only through this interface — it never imports a concrete detector. Two optional capabilities are expressed on the base so the engine stays decoupled:

  • can_adjudicate — when True the engine routes the other detectors' spans to this detector via the candidates argument (ensemble mode).
  • detect_async — a default thread-based implementation is provided; I/O-bound detectors (e.g. an LLM backend) override it with native async.

detect abstractmethod 

detect(text, candidates=None)

Scan text and return the spans found.

Parameters:

Name Type Description Default
candidates list[DetectedSpan] | None

spans found by the other detectors. Only meaningful for adjudicating detectors (can_adjudicate=True); others ignore it.

 None
Source code in src/wardcat/detectors/base.py 
42
43
44
45
46
47
48
49
@abstractmethod
def detect(self, text: str, candidates: list[DetectedSpan] | None = None) -> list[DetectedSpan]:
    """Scan *text* and return the spans found.

    :param candidates: spans found by the other detectors. Only meaningful for
        adjudicating detectors (``can_adjudicate=True``); others ignore it.
    """
    ...

detect_async async 

detect_async(text, candidates=None)

Async variant of :meth:detect.

Default implementation offloads the synchronous :meth:detect to a thread. I/O-bound detectors should override this with native async I/O.

Source code in src/wardcat/detectors/base.py 
51
52
53
54
55
56
57
58
59
async def detect_async(
    self, text: str, candidates: list[DetectedSpan] | None = None
) -> list[DetectedSpan]:
    """Async variant of :meth:`detect`.

    Default implementation offloads the synchronous :meth:`detect` to a
    thread. I/O-bound detectors should override this with native async I/O.
    """
    return await asyncio.to_thread(self.detect, text, candidates)

wardcat.detectors.base.DetectedSpan dataclass 

DetectedSpan(entity_type, text, start, end, confidence=1.0)

A single detected sensitive data span.

entity_type instance-attribute 

entity_type

Entity type — e.g. "EMAIL", "CREDIT_CARD".

text instance-attribute 

text

Full text copied from the original.

start instance-attribute 

start

Start index in the original text (inclusive).

end instance-attribute 

end

End index in the original text (exclusive).

confidence class-attribute instance-attribute 

confidence = 1.0

Detection confidence in [0.0, 1.0]. Regex/checksum detections are 1.0; NER and LLM detections are 0.85 (model-based, not fully deterministic).

LLM backends

The built-in backends are selected with the Backend enum (not user-extensible).

wardcat.Backend

Bases: str, Enum

LLM backend types, as constants — for typo-proof selection.

Pass these to :meth:Wardcat.with_llm instead of bare strings::

from wardcat import Wardcat, Backend

Wardcat(salt="s").with_llm(backend=Backend.OPENAI_COMPATIBLE, model="...")

Each member is its string value (Backend.OLLAMA == "ollama"), so the plain string form is still accepted.

OLLAMA class-attribute instance-attribute 

OLLAMA = 'ollama'

Local Ollama service (supports model download).

OPENAI_COMPATIBLE class-attribute instance-attribute 

OPENAI_COMPATIBLE = 'openai_compatible'

OpenAI-compatible HTTP API — LM Studio, LocalAI, LiteLLM, …

VLLM class-attribute instance-attribute 

VLLM = 'vllm'

vLLM server (OpenAI-compatible API) — native chat, vLLM defaults.

TRANSFORMERS class-attribute instance-attribute 

TRANSFORMERS = 'transformers'

In-process HuggingFace Transformers (no HTTP; loads the model locally).

Engine & anonymizer

wardcat.core.engine.DetectionEngine 

DetectionEngine(config, detectors)

Merges spans from all detectors, resolves overlaps, applies configured actions, and returns a ScanResult.

Source code in src/wardcat/core/engine.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
def __init__(self, config: dict[str, Any], detectors: list[BaseDetector]) -> None:
    self.config = config
    self.detectors = detectors
    # Ensemble adjudication: when enabled and an LLM detector is present,
    # regex/NER spans are passed to the LLM as candidates to verify/relabel/
    # drop (one combined detection + adjudication call). Every regex span
    # (confidence >= _ADJUDICATION_KEEP_CONFIDENCE) is always kept regardless
    # of the LLM — a weak adjudicator that fails to re-detect a real match
    # would otherwise leak it. Only the model layers (NER/LLM) are
    # candidates the LLM may drop/relabel.
    # Detectors are addressed only through BaseDetector — the engine never
    # imports a concrete detector. Adjudicators advertise themselves via the
    # can_adjudicate flag.
    self._adjudicators = [d for d in detectors if d.can_adjudicate]
    self._other_detectors = [d for d in detectors if not d.can_adjudicate]
    self._use_adjudication = bool(
        config.get("llm_detector", {}).get("adjudicate", False)
    ) and bool(self._adjudicators)
    self.salt: str = config.get("salt", "")
    self.entity_config: dict[str, Any] = config.get("entities", {})
    self._max_text_bytes: int = config.get("max_text_bytes", _MAX_TEXT_BYTES)
    self._allowlist: set[str] = set(config.get("allowlist", []))
    # Confidence floor. A span scoring below it is dropped before any action
    # is applied. The default sits above the "uncued" tier the regex layer
    # gives a checksum match with no supporting keyword, and below every
    # other tier, so lowering it trades precision for recall and nothing
    # else changes.
    self._min_confidence: float = float(config.get("min_confidence", 0.8))
    self._denylist: list[dict[str, str]] = config.get("denylist", [])
    # Value propagation: once any layer detects a value, redact every other
    # whole-token occurrence of that exact value too. Closes the gap where a
    # model-based layer (NER/LLM) reports a repeated value only once.
    # Opt-in — it can over-redact, so short values are skipped and matches
    # must be token-bounded.
    self._propagate: bool = bool(config.get("propagate_matches", False))
    self._propagate_min_len: int = config.get("propagate_min_length", 3)
    # Detection (this class) is kept separate from anonymization (applying the
    # configured action to each span); the Anonymizer owns that stage.
    self._anonymizer = Anonymizer(self.entity_config, self.salt)

    if not self.salt:
        logger.debug(
            "Hash salt is empty — identical values will produce the same hash. "
            "Pass salt=... to Wardcat(...) in production."
        )

scan 

scan(text)

Run all detectors, apply actions, and return the result.

Source code in src/wardcat/core/engine.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
def scan(self, text: str) -> ScanResult:
    """Run all detectors, apply actions, and return the result."""
    t_start = time.perf_counter()
    self._check_size(text)

    warnings: list[str] = []
    if self._use_adjudication:
        candidate_spans: list[DetectedSpan] = []
        for detector in self._other_detectors:
            candidate_spans.extend(self._safe_detect(detector, text, warnings))
        raw_spans = [
            s for s in candidate_spans if s.confidence >= _ADJUDICATION_KEEP_CONFIDENCE
        ]
        for adjudicator in self._adjudicators:
            raw_spans.extend(
                self._safe_detect(adjudicator, text, warnings, candidates=candidate_spans)
            )
    else:
        raw_spans = []
        for detector in self.detectors:
            raw_spans.extend(self._safe_detect(detector, text, warnings))

    raw_spans.extend(self._collect_denylist_spans(text))
    spans = self._filter_spans(raw_spans, text)
    # One context id per scan — it is what keeps this scan's reversible
    # placeholders distinct from every other scan's (see TokenAllocator).
    context_id = new_context_id()
    sanitized, violations = self._anonymizer.apply(text, spans, context_id=context_id)

    elapsed_ms = (time.perf_counter() - t_start) * 1000
    logger.info(
        "scan completed: %d violation(s), %d character(s), %.1f ms",
        len(violations),
        len(text),
        elapsed_ms,
    )
    return ScanResult(
        original_text=text,
        sanitized_text=sanitized,
        violations=violations,
        warnings=warnings,
        context_id=context_id,
        _salt=self.salt,
    )

scan_async async 

scan_async(text)

Async variant — uses native async for I/O-bound detectors (LLM backend).

CPU-bound detectors (regex, NER) run via asyncio.to_thread; the LLM detector uses its own detect_async() method with a native httpx.AsyncClient.

Source code in src/wardcat/core/engine.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
async def scan_async(self, text: str) -> ScanResult:
    """Async variant — uses native async for I/O-bound detectors (LLM backend).

    CPU-bound detectors (regex, NER) run via ``asyncio.to_thread``;
    the LLM detector uses its own ``detect_async()`` method with a
    native ``httpx.AsyncClient``.
    """
    t_start = time.perf_counter()
    self._check_size(text)

    warnings: list[str] = []
    if self._use_adjudication:
        cand_results = await asyncio.gather(
            *(self._safe_detect_async(d, text) for d in self._other_detectors)
        )
        candidate_spans: list[DetectedSpan] = [s for spans, _ in cand_results for s in spans]
        warnings.extend(w for _, w in cand_results if w)
        raw_spans = [
            s for s in candidate_spans if s.confidence >= _ADJUDICATION_KEEP_CONFIDENCE
        ]
        adj_results = await asyncio.gather(
            *(
                self._safe_detect_async(d, text, candidates=candidate_spans)
                for d in self._adjudicators
            )
        )
        for spans, w in adj_results:
            raw_spans.extend(spans)
            if w:
                warnings.append(w)
    else:
        results = await asyncio.gather(
            *(self._safe_detect_async(d, text) for d in self.detectors)
        )
        raw_spans = [s for spans, _ in results for s in spans]
        warnings.extend(w for _, w in results if w)
    raw_spans.extend(self._collect_denylist_spans(text))
    spans = self._filter_spans(raw_spans, text)
    # One context id per scan — it is what keeps this scan's reversible
    # placeholders distinct from every other scan's (see TokenAllocator).
    context_id = new_context_id()
    sanitized, violations = self._anonymizer.apply(text, spans, context_id=context_id)

    elapsed_ms = (time.perf_counter() - t_start) * 1000
    logger.info(
        "scan_async completed: %d violation(s), %d character(s), %.1f ms",
        len(violations),
        len(text),
        elapsed_ms,
    )
    return ScanResult(
        original_text=text,
        sanitized_text=sanitized,
        violations=violations,
        warnings=warnings,
        context_id=context_id,
        _salt=self.salt,
    )

wardcat.core.anonymizer.Anonymizer 

Anonymizer(entity_config, salt='')

Applies configured actions to detected spans and rebuilds the text.

Source code in src/wardcat/core/anonymizer.py 
22
23
24
def __init__(self, entity_config: dict[str, Any], salt: str = "") -> None:
    self._entity_config = entity_config
    self._salt = salt

apply 

apply(text, spans, *, context_id=None)

Return (sanitized_text, violations) for spans (already filtered).

spans must be sorted/non-overlapping (the engine guarantees this).

Parameters:

Name Type Description Default
context_id str | None

stamped into reversible (tokenize) placeholders so this call's tokens cannot collide with another scan's. Defaults to a fresh id; pass one to record it alongside the result, or "" for bare [TYPE_index] placeholders.

 None
Source code in src/wardcat/core/anonymizer.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def apply(
    self,
    text: str,
    spans: list[DetectedSpan],
    *,
    context_id: str | None = None,
) -> tuple[str, list[Violation]]:
    """Return ``(sanitized_text, violations)`` for *spans* (already filtered).

    *spans* must be sorted/non-overlapping (the engine guarantees this).

    :param context_id: stamped into reversible (``tokenize``) placeholders so
        this call's tokens cannot collide with another scan's. Defaults to a
        fresh id; pass one to record it alongside the result, or ``""`` for
        bare ``[TYPE_index]`` placeholders.
    """
    # A fresh context per call: reversible actions allocate placeholders in it,
    # so state must not leak between scans — one Anonymizer instance is shared
    # by every scan on a guard, including the concurrent ones in scan_batch.
    ctx = ActionContext(salt=self._salt, tokens=TokenAllocator(context_id))
    violations: list[Violation] = []
    sanitized = text
    offset = 0

    for span in spans:
        action_name = self._entity_config.get(span.entity_type, {}).get("action", "warn")
        replacement = get_action(action_name)(span, ctx)

        if replacement is not None:
            adj_start = span.start + offset
            adj_end = span.end + offset
            sanitized = sanitized[:adj_start] + replacement + sanitized[adj_end:]
            offset += len(replacement) - (span.end - span.start)

        violations.append(
            Violation(
                entity_type=span.entity_type,
                original=span.text,
                start=span.start,
                end=span.end,
                action=action_name,
                replacement=replacement,
                confidence=span.confidence,
            )
        )

    return sanitized, violations