敏感 LLM 数据的脱敏

数据脱敏让你可以精细地控制发送到 Litefuse 服务端的 trace 数据。借助自定义的脱敏函数，你可以控制并清洗被 trace 并发送到服务端的数据。无论是出于 合规需要 还是为了保护 用户隐私，对敏感数据进行脱敏都是负责任的应用开发中的关键一步。它能让你：

对 trace 或 observation 的输入、输出和元数据中的敏感信息进行脱敏。
在事件发送前自定义其内容。
基于自身需求实现细粒度的数据过滤。

关于 Litefuse 对所存储数据的安全和隐私措施，参见我们的安全与合规概览。

工作机制

你定义一个自定义脱敏函数，并把它传给 Langfuse 客户端构造函数。
所有事件的输入、输出和元数据都会经过该函数处理。
脱敏后的数据再发送到 Litefuse 服务端。

通过这种方式，你可以完全掌控应用所记录的事件输入、输出和元数据。

定义一个脱敏函数。无论你使用哪个 Litefuse 维护的集成，该脱敏函数都会作用于所有事件的输入、输出和元数据。

def masking_function(data: any, **kwargs) -> any:
    """Function to mask sensitive data before sending to Litefuse."""
    if isinstance(data, str) and data.startswith("SECRET_"):
        return "REDACTED"
 
    # For more complex data structures
    elif isinstance(data, dict):
        return {k: masking_function(v) for k, v in data.items()}
    elif isinstance(data, list):
        return [masking_function(item) for item in data]
 
    return data

在初始化 Langfuse 客户端时应用脱敏函数：

from langfuse import Langfuse
 
# Initialize with masking function
langfuse = Langfuse(mask=masking_function)
 
# Then get the client
from langfuse import get_client
langfuse = get_client()

配合装饰器：

from langfuse import observe
 
langfuse = Langfuse(mask=masking_function)
 
@observe()
def my_function():
    # This data will be masked before being sent to Litefuse
    return "SECRET_DATA"
 
result = my_function()
print(result)  # Original: "SECRET_DATA"
 
# The trace output in Litefuse will have the output masked as "REDACTED"

使用 context manager：

from langfuse import Langfuse
 
langfuse = Langfuse(mask=masking_function)
 
with langfuse.start_as_current_observation(
    as_type="span",
    name="sensitive-operation",
    input="SECRET_INPUT_DATA"
) as span:
    # ... processing ...
    span.update(output="SECRET_OUTPUT_DATA")
 
# Both input and output will be masked as "REDACTED" in Litefuse

要防止敏感数据被发送到 Litefuse，你可以为 LangfuseSpanProcessor 提供一个 mask 函数。该函数会作用于每个 observation 的 input、output 和 metadata。

该函数接收一个 { data } 对象，其中 data 是属性值序列化为 JSON 的字符串。函数应返回脱敏后的数据。

instrumentation.ts

import { NodeSDK } from "@opentelemetry/sdk-node";
import { LangfuseSpanProcessor } from "@langfuse/otel";
 
const spanProcessor = new LangfuseSpanProcessor({
  mask: ({ data }) => {
    // A simple regex to mask credit card numbers
    const maskedData = data.replace(
      /\b\d{4}[- ]?\d{4}[- ]?\d{4}[- ]?\d{4}\b/g,
      "***MASKED_CREDIT_CARD***"
    );
    return maskedData;
  },
});
 
const sdk = new NodeSDK({
  spanProcessors: [spanProcessor],
});
 
sdk.start();

详见 JS/TS SDK 文档。

使用 CallbackHandler 时，可以通过构造函数传入 mask：

import { CallbackHandler } from "langfuse-langchain";
 
function maskingFunction(params: { data: any }) {
  if (typeof params.data === "string" && params.data.startsWith("SECRET_")) {
    return "REDACTED";
  }
 
  return params.data;
}
 
const handler = new CallbackHandler({
  mask: maskingFunction,
});

示例

下面我们给出几个使用脱敏功能的示例。这里以 Litefuse 装饰器为例，但你也可以类似地使用底层 SDK 或 JS/TS SDK。

示例 1：脱敏信用卡号

本示例演示如何使用正则表达式脱敏字符串中的信用卡号。这有助于满足 PCI DSS 合规要求，确保信用卡号不会被不当传输或存储。

Litefuse 的脱敏功能允许你定义带参数的自定义脱敏函数，并将其传给 Langfuse 客户端构造函数。该函数会作用于 所有事件的输入、输出和元数据，按你的规则处理每条数据以脱敏或抹除敏感信息。通过保证所有事件在发送前都经过你的脱敏函数处理，Litefuse 确保只有脱敏后的数据被传输到 Litefuse 服务端。

步骤：

导入所需模块。
定义一个脱敏函数，使用正则表达式检测并替换信用卡号。
在 Litefuse 中配置该脱敏函数。
创建一个示例函数 来模拟处理敏感数据。
观察 trace 看到脱敏后的输出。

import re
from langfuse import Langfuse, observe, get_client
 
# Step 2: Define the masking function
def masking_function(data, **kwargs):
    if isinstance(data, str):
        # Regular expression to match credit card numbers (Visa, MasterCard, AmEx, etc.)
        pattern = r'\b(?:\d[ -]*?){13,19}\b'
        data = re.sub(pattern, '[REDACTED CREDIT CARD]', data)
    return data
 
# Step 3: Configure the masking function
langfuse = Langfuse(mask=masking_function)
 
# Step 4: Create a sample function with sensitive data
@observe()
def process_payment():
    # Simulated sensitive data containing a credit card number
    transaction_info = "Customer paid with card number 4111 1111 1111 1111."
    return transaction_info
 
# Step 5: Observe the trace
result = process_payment()
 
print(result)
# Output: Customer paid with card number [REDACTED CREDIT CARD].
 
# Flush events in short-lived applications
langfuse.flush()

Litefuse 中脱敏后的 trace 1

Litefuse 中的 trace 链接

示例 2：使用 `llm-guard` 库

本示例使用 llm-guard 中的 Anonymize 扫描器从数据中移除人名等其他 PII。这对匿名化用户数据、保护隐私非常有用。

更多关于 llm-guard 库的信息见其文档。

步骤：

安装 llm-guard 库。
导入所需模块。
初始化 Vault 并配置 Anonymize 扫描器。
定义一个脱敏函数，使用 Anonymize 扫描器。
在 Litefuse 中配置该脱敏函数。
创建一个示例函数 来模拟处理含 PII 的数据。
观察 trace 看到脱敏后的输出。

pip install llm-guard

from langfuse import Langfuse, observe, get_client
from llm_guard.vault import Vault
from llm_guard.input_scanners import Anonymize
from llm_guard.input_scanners.anonymize_helpers import BERT_LARGE_NER_CONF
 
# Step 3: Initialize the Vault and configure the Anonymize scanner
vault = Vault()
 
def create_anonymize_scanner():
    scanner = Anonymize(
        vault,
        recognizer_conf=BERT_LARGE_NER_CONF,
        language="en"
    )
    return scanner
 
# Step 4: Define the masking function
def masking_function(data, **kwargs):
    if isinstance(data, str):
        scanner = create_anonymize_scanner()
        # Scan and redact the data
        sanitized_data, is_valid, risk_score = scanner.scan(data)
        return sanitized_data
    return data
 
# Step 5: Configure the masking function
langfuse = Langfuse(mask=masking_function)
 
# Step 6: Create a sample function with PII
@observe()
def generate_report():
    # Simulated data containing personal names
    report = "John Doe met with Jane Smith to discuss the project."
    return report
 
# Step 7: Observe the trace
result = generate_report()
 
print(result)
# Output: [REDACTED_PERSON] met with [REDACTED_PERSON] to discuss the project.
 
# Flush events in short-lived applications
langfuse.flush()

Litefuse 中脱敏后的 trace

Litefuse 2 中的 trace 链接

示例 3：脱敏邮箱与电话号码

你可以扩展脱敏函数，使用正则表达式脱敏其他类型的 PII，如邮箱地址和电话号码。

import re
from langfuse import Langfuse, observe, get_client
 
def masking_function(data, **kwargs):
    if isinstance(data, str):
        # Mask email addresses
        data = re.sub(r'\b[\w.-]+?@\w+?\.\w+?\b', '[REDACTED EMAIL]', data)
        # Mask phone numbers
        data = re.sub(r'\b\d{3}[-. ]?\d{3}[-. ]?\d{4}\b', '[REDACTED PHONE]', data)
    return data
 
langfuse = Langfuse(mask=masking_function)
 
@observe()
def contact_customer():
    info = "Please contact John at john.doe@example.com or call 555-123-4567."
    return info
 
result = contact_customer()
 
print(result)
# Output: Please contact John at [REDACTED EMAIL] or call [REDACTED PHONE].
 
# Flush events in short-lived applications
langfuse.flush()

Litefuse 中脱敏后的 trace 3