Pular para o conteúdo principal
Última atualização em

Ocultar informações de identificação pessoal de rastreamentos antes da exportação

O MLflow Tracing permite mascarar dados confidenciais nas entradas e saídas de span no lado do cliente, antes que o MLflow exporte um rastreamento para o backend, para que as informações de identificação pessoal brutas nunca saiam do seu aplicativo. Isso é diferente de Ocultar informações de identificação pessoal de rastreamentos do OpenTelemetry no Unity Catalog, que oculta spans do OpenTelemetry (OTel) que as tabelas do Unity Catalog já armazenam.

  • Use a redação no lado do cliente quando se deseja garantir que o aplicativo nunca transmita ou persista valores confidenciais.
  • Use a abordagem de pipeline OTel quando você já armazena rastreamentos no Unity Catalog e precisa ocultá-los sem alterar o código do seu aplicativo.
nota

A ocultação do MLflow afeta apenas o que é registrado no rastreamento. O próprio modelo ou chamada de ferramenta ainda recebe e retorna o conteúdo original, não redigido. Use políticas de serviço para bloquear PII de um serviço de AI registrado no Unity Catalog ou para aplicar uma política específica centralmente em toda a sua organização.

Como funciona​

Os processadores de span implementam a ocultação. Esses processadores recebem um intervalo, o alteram in-place e não retornam nada. Registre um ou mais processadores de span com mlflow.tracing.configure, e o MLflow os aplica a cada span antes de exportá-lo.

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])

Tenha em mente o seguinte comportamento ao escrever processadores de span:

  • A filtragem ocorre no lado do cliente, de modo que seu aplicativo nunca transmite ou armazena dados não editados.
  • Múltiplos processadores de span são executados na ordem em que você os registra. Cada um recebe o intervalo depois que o processador anterior o modifica.
  • Processadores de intervalo aplicam-se a cada intervalo em um rastreamento, incluindo intervalos aninhados criados por integrações de estrutura como LangChain e LangGraph.
  • Use span.span_type para aplicar lógica de redação diferente para diferentes tipos de spans, como LLM, TOOL ou AGENT.

Pré-requisitos​

  • MLflow Tracing configurado para sua aplicação GenAI. Essa solução funciona independentemente de você armazenar rastreamentos no Unity Catalog.

  • Instale a biblioteca Python do MLflow com o extra databricks junto com databricks-langchain>=0.19.0 e langgraph>=1.1.0:

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

  • Para seguir o exemplo do Microsoft Presidio, instale também:

    Bash
    pip install presidio_analyzer presidio_anonymizer
    python -m spacy download en_core_web_lg

Ocultar PII com regex​

O exemplo a seguir corresponde a endereços de email nas entradas de um intervalo com uma expressão regular e os substitui por [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")

Aplicar um filtro a tipos de span específicos​

Use span.span_type para ocultar apenas certos tipos de spans, como chamadas de ferramenta. O exemplo a seguir oculta um número de account bancário de spans de chamada de ferramenta de um agente LangGraph. Ele verifica cada mensagem e campo de chamada de ferramenta no payload, então ele oculta números de account onde quer que apareçam na conversa.

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)

Números de account podem aparecer no conteúdo da mensagem, argumentos de chamada de ferramenta e respostas do modelo. O processador lida com cada forma com uma função auxiliar dedicada:

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)

Registre o processador e invoque o agente:

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?"}
        ]
    }
)

Ocultar PII com o Microsoft Presidio​

Para ir além da filtragem baseada em regex, use uma biblioteca dedicada de detecção e anonimização de PII, como Microsoft Presidio. Um AnalyzerEngine detecta entidades como nomes, números de cartão de crédito e endereços de email, e um AnonymizerEngine os reescreve.

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"

Inicialize o analisador e o anonimizador do Presidio e, em seguida, defina o processador de span:

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})

Registre o processador e execute o agente:

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 os processadores de span​

Para parar de ocultar spans, limpe todos os processadores de span registrados.

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

Ou use reset para limpar toda a configuração de rastreamento.

Python
mlflow.tracing.reset()

Recursos adicionais​

Nesta página