LLM Guard is archived: here's another open-source alternative to protect personal data

$ uv pip install --dry-run "llm-guard==0.3.16" "cryptography>=46.0.5"
error: No solution found when resolving dependencies
  cause: Because presidio-anonymizer==2.2.358 depends on cryptography<44.1 and llm-guard>=0.3.16 depends on presidio-anonymizer==2.2.358, we can conclude that llm-guard>=0.3.16 depends on cryptography<44.1.
         And because you require llm-guard==0.3.16 and cryptography>=46.0.5, we can conclude that your requirements are unsatisfiable.

from llm_guard.input_scanners import Anonymize
from llm_guard.output_scanners import Deanonymize
from llm_guard.input_scanners.anonymize_helpers import BERT_LARGE_NER_CONF
from llm_guard.vault import Vault

vault = Vault()
anonymize = Anonymize(vault, recognizer_conf=BERT_LARGE_NER_CONF, language="en")
deanonymize = Deanonymize(vault)


def fake_llm(prompt: str) -> str:
    """Stands in for the model: answers with the placeholders it was given."""
    return "Done. I wrote to [REDACTED_PERSON_1] at [REDACTED_EMAIL_ADDRESS_1]."


prompt = "Email John Doe at john.doe@example.com about the invoice."
sanitized, is_valid, risk = anonymize.scan(prompt)
print("sent to LLM :", sanitized)

answer = deanonymize.scan(sanitized, fake_llm(sanitized))[0]
print("user reads  :", answer)
print("vault       :", vault.get())

# Second turn, same vault: is John Doe still [REDACTED_PERSON_1]?
sanitized_2, _, _ = anonymize.scan("Also tell John Doe that Jane Smith joins the call.")
print("turn 2      :", sanitized_2)

# A sentence the model gets wrong, on a fresh vault.
miss, _, _ = Anonymize(Vault(), recognizer_conf=BERT_LARGE_NER_CONF, language="en").scan(
    "Email John Doe about the invoice."
)
print("miss        :", miss)

sent to LLM : Email [REDACTED_PERSON_1] at [REDACTED_EMAIL_ADDRESS_1] about the invoice.
user reads  : Done. I wrote to John Doe at john.doe@example.com.
vault       : [('[REDACTED_EMAIL_ADDRESS_1]', 'john.doe@example.com'), ('[REDACTED_PERSON_1]', 'John Doe')]
turn 2      : Also tell [REDACTED_PERSON_1] that [REDACTED_PERSON_2] joins the call.
miss        : Email John [REDACTED_PERSON_1]e about the invoice.

import asyncio

from piighost.components.detector import CompositeDetector, RegexDetector
from piighost.components.detector.ner import TransformersDetector
from piighost.pipeline import ThreadAnonymizationPipeline

names = TransformersDetector(
    "dslim/bert-large-NER",
    labels={"PERSON": "PER", "LOCATION": "LOC", "ORGANIZATION": "ORG"},
)
formats = RegexDetector.from_catalog("catalog:piighost/generic")
pipeline = ThreadAnonymizationPipeline(CompositeDetector([names, formats]))


def fake_llm(prompt: str) -> str:
    """Stands in for the model: answers with the placeholders it was given."""
    return "Done. I wrote to <<PERSON:1>> at <<EMAIL:1>>."


async def main() -> None:
    prompt = "Email John Doe at john.doe@example.com about the invoice."
    safe = await pipeline.anonymize(prompt, thread_id="thread-42")
    print("sent to LLM :", safe.text)

    answer = await pipeline.deanonymize(fake_llm(safe.text), thread_id="thread-42")
    print("user reads  :", answer)

    # Second turn, same thread: is John Doe still <<PERSON:1>>?
    turn_2 = await pipeline.anonymize(
        "Also tell John Doe that Jane Smith joins the call.", thread_id="thread-42"
    )
    print("turn 2      :", turn_2.text)

    # Another conversation starts its own numbering.
    other = await pipeline.anonymize("Call Jane Smith.", thread_id="thread-43")
    print("thread-43   :", other.text)

    # A sentence the model gets wrong, on a fresh thread.
    miss = await pipeline.anonymize("Email John Doe about the invoice.", thread_id="thread-44")
    print("miss        :", miss.text)


asyncio.run(main())

sent to LLM : Email <<PERSON:1>> at <<EMAIL:1>> about the invoice.
user reads  : Done. I wrote to John Doe at john.doe@example.com.
turn 2      : Also tell <<PERSON:1>> that <<PERSON:2>> joins the call.
thread-43   : Call <<PERSON:1>>.
miss        : Email <<ORGANIZATION:1>> <<PERSON:1>><<ORGANIZATION:2>> about the invoice.

from llm_guard.input_scanners import Anonymize
from llm_guard.input_scanners.anonymize_helpers import BERT_LARGE_NER_CONF
from llm_guard.vault import Vault

text = "Écrivez à Marie Dubois au 06 12 34 56 78, IBAN FR76 3000 6000 0112 3456 7890 189."
default = Anonymize(Vault())
bert_large = Anonymize(Vault(), recognizer_conf=BERT_LARGE_NER_CONF, language="en")
print("default    :", default.scan(text)[0])
print("bert-large :", bert_large.scan(text)[0])

default    : Écrivez à [REDACTED_PERSON_1] au [REDACTED_PHONE_NUMBER_1], IBAN [REDACTED_IBAN_CODE_1].
bert-large : Écrivez à Marie [REDACTED_PERSON_1]ois au 06 12 34 56 78, IBAN [REDACTED_IBAN_CODE_1].

import asyncio

from piighost.catalog import pull
from piighost.components.detector import CompositeDetector, RegexDetector
from piighost.components.detector.ner import Gliner2Detector
from piighost.pipeline import ThreadAnonymizationPipeline

french = ThreadAnonymizationPipeline(
    CompositeDetector(
        [
            Gliner2Detector(model="fastino/gliner2-multi-v1", labels=["PERSON"]),
            RegexDetector(pull("catalog:piighost/fr") | pull("catalog:piighost/eu")),
        ]
    )
)
TEXT = "Écrivez à Marie Dubois au 06 12 34 56 78, IBAN FR76 3000 6000 0112 3456 7890 189."


async def compare() -> None:
    print("bert-large :", (await pipeline.anonymize(TEXT, thread_id="fr-1")).text)
    print("gliner2    :", (await french.anonymize(TEXT, thread_id="fr-2")).text)


asyncio.run(compare())

bert-large : <<ORGANIZATION:1>>crivez <<ORGANIZATION:2>> <<ORGANIZATION:3>> <<PERSON:1>><<ORGANIZATION:4>> au 06 12 34 56 78, <<ORGANIZATION:5>>BAN FR76 <<CREDIT_CARD:1>> 7890 189.
gliner2    : Écrivez à <<PERSON:1>> au <<FR_PHONE:1>>, IBAN <<FR_IBAN:1>>.

# Mirrors llm_guard_api/app/app.py: one Vault() created at startup, shared by all requests.
from llm_guard.input_scanners import Anonymize
from llm_guard.input_scanners.anonymize_helpers import BERT_LARGE_NER_CONF
from llm_guard.output_scanners import Deanonymize
from llm_guard.vault import Vault

vault = Vault()
anonymize = Anonymize(vault, recognizer_conf=BERT_LARGE_NER_CONF, language="en")
deanonymize = Deanonymize(vault)

# Request from user A
print("user A sends :", anonymize.scan("Email John Doe at john.doe@example.com about the invoice.")[0])
# Request from user B: their model output happens to contain the same placeholder
print("user B reads :", deanonymize.scan("Hi", "Hello [REDACTED_PERSON_1], how can I help?")[0])

user A sends : Email [REDACTED_PERSON_1] at [REDACTED_EMAIL_ADDRESS_1] about the invoice.
user B reads : Hello John Doe, how can I help?

import asyncio


async def two_users() -> None:
    sent = await pipeline.anonymize(
        "Email John Doe at john.doe@example.com about the invoice.", thread_id="user-a"
    )
    print("user A sends :", sent.text)
    # User B's model happens to write the same placeholder.
    reply = await pipeline.deanonymize("Hello <<PERSON:1>>, how can I help?", thread_id="user-b")
    print("user B reads :", reply)


asyncio.run(two_users())

user A sends : Email <<PERSON:1>> at <<EMAIL:1>> about the invoice.
user B reads : Hello <<PERSON:1>>, how can I help?

import asyncio

from langchain.agents import create_agent
from langchain.tools import tool
from langchain_core.messages import AIMessage

from piighost.integrations.langchain import PIIAnonymizationMiddleware

middleware = PIIAnonymizationMiddleware(pipeline=pipeline)
PROMPT = "Send the invoice to john.doe@example.com for John Doe."


@tool
def send_invoice(to: str) -> str:
    """Send the invoice to an email address."""
    print("tool receives:", to)
    return f"Invoice sent to {to}."


agent = create_agent(
    model="openai:gpt-5.6-terra",
    tools=[send_invoice],
    middleware=[middleware],
)


async def model_text():
    async for chunk, _meta in agent.astream(
        {"messages": [{"role": "user", "content": PROMPT}]},
        {"configurable": {"thread_id": "thread-42"}},
        stream_mode="messages",
    ):
        if isinstance(chunk, AIMessage) and chunk.tool_call_chunks:
            print("LLM writes   :", chunk.tool_call_chunks[0]["args"])
        elif isinstance(chunk, AIMessage) and isinstance(chunk.content, str):
            yield chunk.content


async def main() -> None:
    async for piece in middleware.deanonymize_stream(model_text(), "thread-42"):
        print("user reads   :", repr(piece))


asyncio.run(main())

LLM writes   : {"to": "<<EMAIL:1>>"}
tool receives: john.doe@example.com
user reads   : 'Done'
user reads   : '. '
user reads   : 'John Doe g'
user reads   : 'ets '
user reads   : 'the '
user reads   : 'invo'
user reads   : 'ice '
user reads   : 'at '
user reads   : 'john.doe@example.com.'