文档 / 遮蔽敏感的 LLM 数据

遮蔽敏感的 LLM 数据

遮蔽(Masking)让你能够控制发送到 Langfuse 的追踪数据。使用遮蔽函数在追踪数据离开你的应用之前编辑敏感信息,例如用于:

从 trace 或 observation 的输入、输出和元数据中编辑敏感信息。
在导出前转换 OpenTelemetry span 属性。
为合规或隐私要求实现细粒度的数据过滤。

关于 Langfuse 针对所存储数据的数据安全和隐私措施,在我们的安全与合规概览中了解更多。

配置遮蔽

Python SDK 支持两个遮蔽钩子。对于新的 Python SDK 设置,优先使用 mask_otel_spans。

方法	状态	何时运行	覆盖范围
`mask_otel_spans`	推荐	在导出阶段,Langfuse 决定该客户端应导出哪些 OpenTelemetry spans 之后,以及 Langfuse 媒体处理之后	来自 Langfuse SDK spans 和由此 Langfuse 客户端导出的第三方埋点的原始 OpenTelemetry span 属性
`mask`	旧版	在 Langfuse SDK 属性创建时同步运行	通过 Langfuse SDK API(如 `start_observation()`、`update()` 和 `set_trace_io()`)设置的数据

使用 mask_otel_spans 在 Langfuse 客户端导出之前修补 OpenTelemetry span 属性。该钩子接收一个 OpenTelemetry 导出批次的只读快照,并为应更改的 spans 返回稀疏补丁。

python

from typing import Optional

from langfuse import Langfuse
from langfuse.types import (
    MaskOtelSpansParams,
    MaskOtelSpansResult,
    OtelSpanPatch,
)


def mask_otel_spans(
    *, params: MaskOtelSpansParams
) -> Optional[MaskOtelSpansResult]:
    patches = {}

    for identifier, span in params.spans.items():
        if span.instrumentation_scope_name == "openai":
            patches[identifier] = OtelSpanPatch(
                delete_attributes=(
                    "gen_ai.prompt.0.content",
                    "gen_ai.completion.0.content",
                ),
                set_attributes={"masking.applied": True},
            )

    return MaskOtelSpansResult(span_patches=patches)


langfuse = Langfuse(mask_otel_spans=mask_otel_spans)

`mask_otel_spans` 行为

mask_otel_spans 遵循公共 Python SDK 类型契约:

它接收一个 OpenTelemetry 导出批次作为 params.spans。一个批次不保证包含完整的 trace、请求或 Langfuse observation 树。
每个键是一个 OtelSpanIdentifier(trace_id, span_id)。返回补丁时复用 params.spans 中的标识符对象。
每个值是经过 should_export_span 过滤和导出阶段媒体处理后的 OtelSpanData 快照。其 attributes 和 resource_attributes 映射是只读的。
返回 None 以让整个批次保持不变。
返回 MaskOtelSpansResult(span_patches=...) 以删除或替换所选 spans 上的属性。
补丁是稀疏的。省略不需要更改的 spans。
OtelSpanPatch 先删除 delete_attributes,然后应用 set_attributes,因此当同一键同时出现时 set_attributes 胜出。
set_attributes 的值必须是有效的 OpenTelemetry 属性值:字符串、布尔值、整数、浮点数,或这些标量类型的同构序列。
该钩子只能更改 span 属性。它不能更改 span 名称、ID、父关系、resource 属性、events、links 或 instrumentation scope。
该钩子仅影响由此 Langfuse 客户端导出的 spans。如果相同的 OpenTelemetry spans 被发送到另一个导出器,该导出器会收到其自己未修改的副本。

mask_otel_spans 仅影响通过 Langfuse Python SDK span processor 并由此 Langfuse 客户端导出的 spans。如果你还通过单独的 OpenTelemetry span processor 或导出器将遥测数据发送到另一个可观测性后端,该后端会收到其自己未修改的 spans 副本。为任何非 Langfuse 导出器单独配置遮蔽。

mask_otel_spans 是同步的。它通常运行在 OpenTelemetry 批量 span processor 工作线程上,因此不应阻塞应用的主调用线程。在 flush() 和关闭期间,它可能在调用线程上运行。

保持函数确定且快速。网络调用是可能的,但缓慢的遮蔽会使 OpenTelemetry 导出队列积压并延迟 span 导出。避免在遮蔽函数内进行长时间运行的工作、无界重试、请求局部状态、当前活动 span 和异步 I/O。

如果 mask_otel_spans 抛出异常或返回无效的 MaskOtelSpansResult,Langfuse 会丢弃整个导出批次。如果单个 OtelSpanPatch 无效,Langfuse 仅从 Langfuse 导出中丢弃该 span。无效的返回属性值仅删除受影响的属性。

旧版 `mask` 行为

mask 参数是旧版 Python SDK 遮蔽钩子。它在 Langfuse SDK 属性创建时同步运行,仅适用于通过 Langfuse SDK API(如 start_observation()、update() 和 set_trace_io())设置的数据。它不检查来自第三方埋点的最终原始 OpenTelemetry span 属性。

仅当你特别需要在 Langfuse SDK 属性创建时转换数据时才使用 mask。

python

from typing import Any

from langfuse import Langfuse


def masking_function(*, data: Any, **kwargs: Any) -> Any:
    if isinstance(data, str) and data.startswith("SECRET_"):
        return "REDACTED"

    if isinstance(data, dict):
        return {key: masking_function(data=value) for key, value in data.items()}

    if isinstance(data, list):
        return [masking_function(data=item) for item in data]

    return data


langfuse = Langfuse(mask=masking_function)

对于新的 Python SDK 遮蔽设置,优先使用 mask_otel_spans。

示例

用 `mask_otel_spans` 编辑信用卡号

本示例扫描导出的 OpenTelemetry 字符串属性,查找类似信用卡的模式,并在 span 导出到 Langfuse 之前替换匹配项。

python

import re
from typing import Optional

from langfuse import Langfuse, observe
from langfuse.types import (
    MaskOtelSpansParams,
    MaskOtelSpansResult,
    OtelSpanPatch,
)

credit_card_pattern = re.compile(r"\b(?:\d[ -]*?){13,19}\b")


def mask_otel_spans(
    *, params: MaskOtelSpansParams
) -> Optional[MaskOtelSpansResult]:
    patches = {}

    for identifier, span in params.spans.items():
        replacements = {}

        for key, value in span.attributes.items():
            if isinstance(value, str):
                masked_value = credit_card_pattern.sub(
                    "[REDACTED CREDIT CARD]", value
                )

                if masked_value != value:
                    replacements[key] = masked_value

        if replacements:
            patches[identifier] = OtelSpanPatch(set_attributes=replacements)

    return MaskOtelSpansResult(span_patches=patches)


langfuse = Langfuse(mask_otel_spans=mask_otel_spans)


@observe()
def process_payment():
    return "Customer paid with card number 4111 1111 1111 1111."


result = process_payment()

print(result)
# Output: Customer paid with card number 4111 1111 1111 1111.

# Flush spans in short-lived applications.
langfuse.flush()

在你的应用中打印的函数结果不变。发送到 Langfuse 的导出 span 属性包含编辑后的值。

编辑电子邮件地址和电话号码

python

import re
from typing import Optional

from langfuse import Langfuse
from langfuse.types import (
    MaskOtelSpansParams,
    MaskOtelSpansResult,
    OtelSpanPatch,
)

email_pattern = re.compile(r"\b[\w.-]+?@[\w.-]+?\.\w+?\b")
phone_pattern = re.compile(r"\b\d{3}[-. ]?\d{3}[-. ]?\d{4}\b")


def mask_otel_spans(
    *, params: MaskOtelSpansParams
) -> Optional[MaskOtelSpansResult]:
    patches = {}

    for identifier, span in params.spans.items():
        replacements = {}

        for key, value in span.attributes.items():
            if isinstance(value, str):
                masked_value = email_pattern.sub("[REDACTED EMAIL]", value)
                masked_value = phone_pattern.sub("[REDACTED PHONE]", masked_value)

                if masked_value != value:
                    replacements[key] = masked_value

        if replacements:
            patches[identifier] = OtelSpanPatch(set_attributes=replacements)

    return MaskOtelSpansResult(span_patches=patches)


langfuse = Langfuse(mask_otel_spans=mask_otel_spans)

为防止敏感数据被发送到 Langfuse,你可以向 LangfuseSpanProcessor 提供一个 mask 函数。该函数将应用于每个 observation 的 input、output 和 metadata。

该函数接收一个对象 ``,其中 data 是属性值的字符串化 JSON。它应返回遮蔽后的数据。

tsinstrumentation.ts

import { NodeSDK } from "@opentelemetry/sdk-node";
import { LangfuseSpanProcessor } from "@langfuse/otel";

const spanProcessor = new LangfuseSpanProcessor({
  mask: ({ data }) => {
    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 文档。

遮蔽配置在 LangfuseSpanProcessor 上——由 LangChain CallbackHandler 创建的 spans 会自动流经它,因此不需要 LangChain 专属的遮蔽设置:

typescript

import { NodeSDK } from "@opentelemetry/sdk-node";
import { LangfuseSpanProcessor } from "@langfuse/otel";
import { CallbackHandler } from "@langfuse/langchain";

const spanProcessor = new LangfuseSpanProcessor({
  mask: ({ data }) => {
    if (typeof data === "string" && data.startsWith("SECRET_")) {
      return "REDACTED";
    }

    return data;
  },
});

const sdk = new NodeSDK({ spanProcessors: [spanProcessor] });
sdk.start();

const handler = new CallbackHandler();

遮蔽敏感的 LLM 数据

配置遮蔽

mask_otel_spans 行为

旧版 mask 行为

示例

用 mask_otel_spans 编辑信用卡号

编辑电子邮件地址和电话号码

相关资源

`mask_otel_spans` 行为

旧版 `mask` 行为

用 `mask_otel_spans` 编辑信用卡号