Aller au contenu principal
Dernière mise à jour le

Anonymiser les PII des traces avant l'exportation

MLflow Tracing vous permet de masquer les données sensibles dans les entrées et sorties d'étendue côté client, avant que MLflow n'exporte une trace vers le backend, afin que les données PII brutes ne quittent jamais votre application. Ceci est différent de Anonymiser les PII des traces OpenTelemetry dans Unity Catalog, qui anonymise les étendues OpenTelemetry (OTel) que les tables Unity Catalog stockent déjà.

  • Utilisez la rédaction côté client lorsque vous voulez garantir que votre application ne transmette ou ne persiste jamais de valeurs sensibles.
  • Utilisez l'approche de pipeline OTel si vous stockez déjà des traces dans Unity Catalog et que vous devez les expurger sans modifier le code de votre application.
remarque

La rédaction MLflow n'affecte que ce qui est enregistré dans la trace. L'appel de modèle ou d'outil lui-même reçoit et renvoie toujours le contenu original et non expurgé. Utilisez les stratégies de service pour bloquer les informations personnelles identifiables (PII) d'un service d'IA enregistré dans Unity Catalog ou pour appliquer une stratégie spécifique de manière centralisée au sein de votre organisation.

Comment cela fonctionne​

Les processeurs de spans assurent la rédaction. Ces processeurs reçoivent une étendue, la mutent sur place et ne renvoient rien. Enregistrez un ou plusieurs processeurs d’étendue avec mlflow.tracing.configure, et MLflow les applique à chaque étendue avant de l’exporter.

Python
from mlflow.entities.span import Span

def filter_function(span: Span) -> None:
    # Read span.inputs / span.outputs, redact, then write back.
    span.set_inputs(...)
    span.set_outputs(...)

mlflow.tracing.configure(span_processors=[filter_function])

Gardez à l’esprit le comportement suivant lorsque vous écrivez des processeurs d’étendue :

  • Le filtrage s'effectue côté client, de sorte que votre application ne transmet ni ne stocke jamais de données non expurgées.
  • Plusieurs processeurs d'étendue s'exécutent dans l'ordre où vous les enregistrez. Chacun reçoit l'étendue après que le processeur précédent l'a mutée.
  • Les processeurs de spans s'appliquent à chaque span dans une trace, y compris les spans imbriqués créés par des intégrations de framework telles que LangChain et LangGraph.
  • Utilisez span.span_type pour appliquer une logique de rédaction différente à différents types d'étendues, tels que LLM, TOOL, ou AGENT.

Prérequis​

  • MLflow Tracing configuré pour votre application d'IA. Cette solution fonctionne que vous stockiez ou non des traces dans Unity Catalog.

  • Installez la bibliothèque Python MLflow avec l’extra databricks ainsi que databricks-langchain>=0.19.0 et langgraph>=1.1.0:

    Bash
    pip install --upgrade "mlflow-skinny[databricks]" databricks-sdk "databricks-langchain>=0.19.0" "langgraph>=1.1.0"

  • Pour suivre l’exemple de Microsoft Presidio, installez également :

    Bash
    pip install presidio_analyzer presidio_anonymizer
    python -m spacy download en_core_web_lg

Anonymiser les PII avec une regex​

L'exemple suivant correspond aux adresses e-mail dans les entrées d'une étendue avec une expression régulière et les remplace par [REDACTED].

Python
import re
import mlflow
from mlflow.entities.span import Span

# mlflow.set_experiment(experiment_id=experiment_id)


# Your application code (simplified).
@mlflow.trace
def predict(text: str):
    return "Answer"


# Regex pattern to match e-mail addresses.
EMAIL_PATTERN = r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}"


# Define a filter function that takes a span as input and mutates it in place.
def redact_email(span: Span) -> None:
    raw_input = span.inputs.get("text")
    redacted_input = re.sub(EMAIL_PATTERN, "[REDACTED]", raw_input)
    span.set_inputs({"text": redacted_input})


# Register the filter function.
mlflow.tracing.configure(span_processors=[redact_email])

# Run the application.
predict("My e-mail address is test@example.com")

Appliquer un filtre à des types d’étendue spécifiques​

Utilisez span.span_type pour masquer uniquement certains types de spans, tels que les appels d'outils. L'exemple suivant masque un numéro de compte bancaire des spans d'appel d'outil d'un agent LangGraph. Il vérifie chaque message et chaque champ d'appel d'outil dans la charge utile, de sorte qu'il masque les numéros de compte partout où ils apparaissent dans la conversation.

Python
import mlflow
from langchain_core.tools import tool
from databricks_langchain import ChatDatabricks
from langchain.agents import create_agent

# autolog() registers a LangChain callback so MLflow automatically captures spans
# for every LLM call, tool invocation, and agent step. Without this call,
# LangGraph operations would not appear in the trace.
mlflow.langchain.autolog()


@tool
def get_bank_account_number(user_name: str):
    """Return the bank account number for the given user name."""
    return "1234567890"


llm = ChatDatabricks(
    model="databricks-claude-opus-4-6",
    use_ai_gateway=True,
)
tools = [get_bank_account_number]
graph = create_agent(llm, tools)

Les numéros de compte peuvent apparaître dans le contenu du message, les arguments des appels d'outils et les réponses du modèle. Le processeur traite chaque forme avec une fonction d'assistance dédiée :

Python
import re
from mlflow.entities.span import Span, SpanType

ACCOUNT_NUMBER_PATTERN = re.compile(r"\d{10}")


def _redact_str(s):
    """Replace every pattern match in s with [REDACTED]. Non-strings pass through unchanged."""
    return ACCOUNT_NUMBER_PATTERN.sub("[REDACTED]", s) if isinstance(s, str) else s


def _redact_content(content):
    """Redact pattern matches from a message content field.

    content is either a plain string or a multimodal list of content parts
    (e.g., [{"type": "text", "text": "..."}, {"type": "image_url", ...}]).
    Only text fields are redacted; non-text parts pass through unchanged."""
    if isinstance(content, str):
        return _redact_str(content)
    if isinstance(content, list):
        out = []
        for part in content:
            if isinstance(part, dict) and "text" in part:
                out.append({**part, "text": _redact_str(part.get("text"))})
            elif isinstance(part, str):
                out.append(_redact_str(part))
            else:
                out.append(part)
        return out
    return content


def _redact_tool_calls(tool_calls):
    """Redact pattern matches from an OpenAI-format tool call list.

    In the OpenAI schema, each tool call's `function.arguments` value is a
    JSON-encoded string, not a parsed dict. This function redacts it as raw
    text to avoid parsing and re-serializing the JSON."""
    if not isinstance(tool_calls, list):
        return tool_calls
    out = []
    for tc in tool_calls:
        if not isinstance(tc, dict):
            out.append(tc)
            continue
        fn = tc.get("function")
        if isinstance(fn, dict) and isinstance(fn.get("arguments"), str):
            # arguments is a JSON string — redact the raw text.
            tc = {**tc, "function": {**fn, "arguments": _redact_str(fn["arguments"])}}
        out.append(tc)
    return out


def _redact_message(msg):
    """Redact pattern matches from a single chat message dict.

    A message is a dict like {"role": "user", "content": "..."}.
    Assistant messages may also include a `tool_calls` key, which is
    also redacted."""
    if not isinstance(msg, dict):
        return msg
    new = {**msg}
    if "content" in new:
        new["content"] = _redact_content(new.get("content"))
    if "tool_calls" in new:
        new["tool_calls"] = _redact_tool_calls(new.get("tool_calls"))
    return new


def redact_messages(messages):
    """Works for both input and output message lists."""
    if isinstance(messages, dict):  # {"messages": [...]}
        messages = messages.get("messages")
    if not isinstance(messages, list):
        return messages
    return [_redact_message(m) for m in messages]


def _redact_choices(choices):
    """OpenAI Chat Completions: choices[i].message / .delta / .text"""
    if not isinstance(choices, list):
        return choices
    out = []
    for ch in choices:
        if not isinstance(ch, dict):
            out.append(ch)
            continue
        new = dict(ch)
        for key in ("message", "delta"):
            if isinstance(new.get(key), dict):
                new[key] = _redact_message(new[key])
        if isinstance(new.get("text"), str):  # legacy completions
            new["text"] = _redact_str(new["text"])
        out.append(new)
    return out


def _redact_payload(payload):
    """Apply to either span.inputs or span.outputs uniformly."""
    if isinstance(payload, str):
        return _redact_str(payload)
    if isinstance(payload, list):  # bare list of messages
        return redact_messages(payload)
    if not isinstance(payload, dict):
        return payload
    new = dict(payload)
    if isinstance(new.get("messages"), list):
        new["messages"] = redact_messages(new["messages"])
    if "choices" in new:  # OpenAI chat / completions
        new["choices"] = _redact_choices(new["choices"])
    if isinstance(new.get("content"), (str, list)):  # Anthropic Messages
        new["content"] = _redact_content(new["content"])
    if isinstance(new.get("output_text"), str):  # OpenAI Responses API
        new["output_text"] = _redact_str(new["output_text"])
    if isinstance(new.get("prompt"), str):  # legacy completion input
        new["prompt"] = _redact_str(new["prompt"])
    return new


def filter_bank_account_number(span: Span) -> None:
    if span.span_type == SpanType.TOOL:
        span.set_outputs("[REDACTED]")
        return

    if span.inputs is not None:
        new_in = _redact_payload(span.inputs)
        if new_in != span.inputs:
            span.set_inputs(new_in)

    if span.outputs is not None:
        new_out = _redact_payload(span.outputs)
        if new_out != span.outputs:
            span.set_outputs(new_out)

Enregistrez le processeur et invoquez l'agent :

Python
# Register the filter function.
mlflow.tracing.configure(span_processors=[filter_bank_account_number])

# Run the application.
result = graph.invoke(
    {
        "messages": [
            {"role": "user", "content": "What is the bank account number for John Doe?"}
        ]
    }
)

Masquer les PII avec Microsoft Presidio​

Pour aller au-delà du filtrage basé sur des expressions régulières, utilisez une bibliothèque dédiée de détection et d'anonymisation des PII, telle que Microsoft Presidio. Un AnalyzerEngine détecte les entités telles que les noms, les numéros de carte de crédit et les adresses e-mail, et un AnonymizerEngine les réécrit.

Python
import mlflow
from mlflow.entities.span import Span, SpanType


# Dummy application code for a customer support agent.
@mlflow.trace(span_type=SpanType.AGENT)
def customer_support_agent(request: str):
    return "Yes"

Initialisez l'analyseur et l'anonymiseur de Presidio, puis définissez le processeur d'étendue :

Python
from presidio_analyzer import AnalyzerEngine
from presidio_anonymizer import AnonymizerEngine

# Initialize the analyzer and anonymizer.
analyzer = AnalyzerEngine()
anonymizer = AnonymizerEngine()


# Define a filter function.
def filter_pii(span: Span) -> None:
    text = span.inputs.get("request")

    results = analyzer.analyze(
        text=text,
        entities=["PERSON", "CREDIT_CARD", "EMAIL_ADDRESS", "LOCATION", "DATE_TIME"],
        language="en",
    )
    anonymized_text = anonymizer.anonymize(text=text, analyzer_results=results)

    span.set_inputs({"request": anonymized_text.text})

Enregistrez le processeur et exécutez l'agent :

Python
# Register the filter function.
mlflow.tracing.configure(span_processors=[filter_pii])

# Run the application.
customer_support_agent(
    "Please cancel my credit card effective September 19th. My name is John Doe and my credit "
    "card number is 4095-2609-9393-4932. My email is john.doe@example.com and I live in Amsterdam."
)

Reset les processeurs de spans​

Pour arrêter de masquer les étendues, effacez tous les processeurs d'étendue enregistrés.

Python
mlflow.tracing.configure(span_processors=[])

Ou utilisez reset pour effacer toute la configuration de traçage.

Python
mlflow.tracing.reset()

Ressources supplémentaires​

Sur cette page