エクスポート前にトレースからPIIをマスキングする
MLflow Tracingを使用すると、MLflow がトレースをバックエンドにエクスポートする前に、クライアント側でスパンの入力と出力の機密データをマスクできるため、生のPIIがアプリケーションから離れることはありません。これは、Unity CatalogのOpenTelemetryトレースからPIIをマスキングするとは異なります。後者は、Unity Catalogテーブルがすでに保存しているOpenTelemetry (OTel) スパンをマスキングします。
- 機密性の高い値をアプリケーションが送信または保持しないことを保証したい場合は、クライアント側リダクションを使用します。
- Unity Catalog にトレースを既に保存しており、アプリケーションコードを変更せずにトレースを編集する必要がある場合は、OTelパイプラインのアプローチを使用します。
注記
MLflowの秘匿化は、トレースに記録される内容のみに影響します。モデルまたはツール呼び出し自体は、元の編集されていないコンテンツを受け取り、返します。サービスポリシーを使用して、Unity Catalogに登録されたAIサービスからのPIIをブロックしたり、特定のポリシーを組織全体で一元的に適用したりします。
仕組み
スパンプロセッサーはマスキングを実装します。これらのプロセッサーはスパンを受け取り、その場で変更を加え、何も返しません。mlflow.tracing.configureで1つ以上のスパンプロセッサーを登録すると、MLflowはエクスポートする前にそれらをすべてのスパンに適用します。
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])
スパンプロセッサを作成する際は、以下の動作に留意してください：
- フィルタリングはクライアントサイドで行われるため、アプリケーションが編集されていないデータを送信したり保存したりすることはありません。
- 複数のスパンプロセッサーは、登録した順序で実行されます。前のプロセッサがスパンを変更した後、それぞれがスパンを受け取ります。
- スパンプロセッサは、LangChain や LangGraph のようなフレームワーク統合によって作成されたネストされたスパンを含む、トレース内のすべてのスパンに適用されます。
span.span_typeを使用して、LLM、TOOL、またはAGENTのようなさまざまな種類のスパンに異なる編集ロジックを適用します。
前提条件
-
GenAIアプリケーション向けにMLflow Tracingが構成されています。このソリューションは、Unity Catalogにトレースを保存するかどうかにかかわらず機能します。
- トレースを設定するには、MLflow Tracing - GenAIオブザーバビリティを参照してください。
- Unity Catalog にトレースを保存する場合は、このページの例を実行する前に、セットアップ: Unity Catalog トレースロケーションでエクスペリメントを作成するを参照して、Unity Catalog トレースロケーションでエクスペリメントを作成してください。
-
MLflow Pythonライブラリを、
databricks-langchain>=0.19.0とlanggraph>=1.1.0とともにdatabricksエクストラでインストールします：Bashpip install --upgrade "mlflow-skinny[databricks]" databricks-sdk "databricks-langchain>=0.19.0" "langgraph>=1.1.0" -
Microsoft Presidioの例に従うには、以下もインストールします：
Bashpip install presidio_analyzer presidio_anonymizer
python -m spacy download en_core_web_lg
正規表現でPIIをマスキングする
次の例では、スパンの入力にあるEメールアドレスを正規表現で照合し、[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")
特定の スパンタイプ にフィルターを適用します
span.span_type を使用して、特定の種類（たとえばツール呼び出し）のスパンのみをマスキングします。次の例では、LangGraph エージェントのツール呼び出しスパンから銀行口座番号をマスキングします。ペイロード内のすべてのメッセージとツール呼び出しフィールドを確認し、会話のどこにアカウント番号が表示されても消去します。
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)
アカウント番号は、メッセージの内容、ツール呼び出しの引数、およびモデルの応答に表示されることがあります。プロセッサは、専用のヘルパー関数で各シェイプを処理します：
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)
プロセッサーを登録し、エージェントを呼び出す：
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?"}
]
}
)
Microsoft PresidioでPIIをマスキングする
正規表現ベースのフィルタリングを超えて、Microsoft Presidio のような専用の PII 検出および匿名化ライブラリを使用してください。AnalyzerEngine は、名前、クレジットカード番号、Eメールアドレスなどのエンティティを検出し、AnonymizerEngine はそれらを書き換えます。
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"
Presidio のアナライザーと匿名化ツールを初期化し、その後スパンプロセッサを定義します：
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})
プロセッサを登録し、エージェントを実行します：
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 the span processors
スパンの編集を停止するには、登録されているすべてのスパンプロセッサをクリアします。
Python
mlflow.tracing.configure(span_processors=[])
または、resetを使用してすべてのトレース設定をクリアします。
Python
mlflow.tracing.reset()