Wire Presidio as a LangGraph node

Presidio is a Microsoft open-source library that detects PII with a combination of NER models and pattern matchers. It handles names, emails, phone numbers, credit cards, IBANs, and dozens more. We wrap it in a lightweight class that produces placeholders like PERSON_1 and EMAIL_2, and we keep a mapping from each placeholder back to the original value.

anonymizer.py
python
class PIIAnonymizer:
    """Reversible PII masking using Presidio."""

    def anonymize(self, text: str) -> AnonymizationResult:
        self._ensure_engines()
        if not self._analyzer:
            return AnonymizationResult(anonymized_text=text)

        results = self._analyzer.analyze(text=text, language=self.language)
        # Sort by start position descending so we can splice in place.
        results = sorted(results, key=lambda r: r.start, reverse=True)

        anonymized = text
        mapping: dict[str, str] = {}
        counters: dict[str, int] = {}

        for r in results:
            counters[r.entity_type] = counters.get(r.entity_type, 0) + 1
            placeholder = f"<{r.entity_type}_{counters[r.entity_type]}>"
            original = anonymized[r.start:r.end]
            anonymized = anonymized[:r.start] + placeholder + anonymized[r.end:]
            mapping[placeholder] = original

        return AnonymizationResult(anonymized_text=anonymized, mapping=mapping)

Iterate matches in reverse so splicing earlier positions does not shift later offsets. The counter per entity type produces stable, unique placeholders.

anonymizer.py
python
def deanonymize(self, text: str, mapping: dict[str, str]) -> str:
    """Swap placeholders back to original values."""
    if not mapping:
        return text
    for placeholder, original in mapping.items():
        text = text.replace(placeholder, original)
    return text

De-anonymization is just a set of string replacements. Simple, predictable, and easy to audit.

rag_graph.py
python
async def _anonymize_node(state: GraphState) -> GraphState:
    """Mask PII in the user question before any LLM call."""
    if not state.get("anonymize", True):
        return {"anonymized_question": state["question"], "pii_mapping": {}}
    result = _anonymizer.anonymize(state["question"])
    return {
        "anonymized_question": result.anonymized_text,
        "pii_mapping": result.mapping,
    }

The node is small and toggleable. When anonymize is false, the scrubbed question is just the original. That keeps downstream nodes identical either way.

rag_graph.py
python
# Tail end of _synthesize_node:
# De-anonymize the answer so the caller gets real entities back.
mapping = state.get("pii_mapping") or {}
if mapping:
    answer = _anonymizer.deanonymize(answer, mapping)

return {"answer": answer}

Restoration happens inside the synthesizer after generation completes. The LLM never sees the mapping, and the caller never sees placeholders.