解析 ‘Trace Masking & Redaction’：在发送监控数据到 LangSmith 前，如何自动移除所有包含隐私信息的 Trace 段？

监控数据隐私保护：LangSmith Traces 的自动脱敏与匿名化实践

在现代软件开发中，可观测性是保障系统稳定性和性能的关键。LangSmith 作为 LangChain 生态系统中的重要组成部分，为大型语言模型（LLM）应用程序提供了强大的追踪、监控和调试能力。然而，随着 LLM 应用的日益普及，其处理的数据量和敏感性也急剧增加。用户输入、LLM 生成内容、工具调用参数以及内部状态等，都可能包含个人身份信息（PII）、受保护健康信息（PHI）或其他敏感数据。将这些未经处理的敏感数据直接发送到外部监控系统，即使是像 LangSmith 这样受信任的平台，也可能构成严重的隐私和合规风险，例如违反 GDPR、CCPA 等数据保护法规。

本讲座将深入探讨如何在将监控数据发送到 LangSmith 之前，自动识别并移除所有包含隐私信息的 Trace 段。我们将从隐私数据的定义出发，分析 LangSmith Trace 的结构特点，进而提出一套基于 LangChain 回调机制的自动化脱敏与匿名化解决方案，并提供详细的代码实现和最佳实践。

第一章：理解隐私数据与追踪系统

1.1 什么是隐私数据 (PII, PHI, PCI)？

在深入技术实现之前，我们首先需要明确“隐私数据”的范畴。通常，隐私数据可以分为以下几类：

• 个人身份信息 (Personally Identifiable Information, PII)：任何可以直接或间接识别个人身份的数据。
  • 直接 PII：姓名、电子邮件地址、电话号码、社会安全号码 (SSN)、护照号码、驾驶执照号码、信用卡号、IP 地址等。
  • 间接 PII：出生日期、性别、种族、地理位置、职业、教育背景等，当与其他信息结合时可能识别个人。
• 受保护健康信息 (Protected Health Information, PHI)：在美国 HIPAA 法规下，任何与个人健康状况、医疗服务或医疗支付相关的可识别个人身份的信息。
• 支付卡行业数据 (Payment Card Industry Data, PCI)：与信用卡和借记卡交易相关的敏感信息，包括卡号、持卡人姓名、有效期、CVV 码等。
• 商业敏感信息 (Commercial Sensitive Information)：API 密钥、数据库连接字符串、内部系统凭证、专有算法细节等，虽然不直接指向个人，但泄露可能导致严重业务损失。

在 LLM 应用中，上述所有类型的数据都可能出现在用户提示、LLM 响应、工具输入/输出、中间步骤日志甚至元数据中。

1.2 为什么 LLM 应用的追踪数据需要脱敏？

1. 合规性要求：GDPR、CCPA、HIPAA 等法规对个人数据的收集、处理、存储和传输有严格规定。未经授权的敏感数据传输可能导致巨额罚款和法律诉讼。
2. 数据泄露风险：即使是内部监控系统，也存在被攻击或内部人员滥用的风险。脱敏可以限制数据泄露时的影响范围。
3. 信任与声誉：用户对数据隐私的关注日益增加。保护用户数据不仅是法律要求，也是建立和维护用户信任的关键。
4. 数据最小化原则：根据隐私设计原则，系统应只收集和处理完成特定目的所需的最少数据。对于监控目的，通常不需要原始敏感数据。
5. 第三方服务风险：LangSmith 虽然是 LangChain 官方服务，但作为外部服务，将原始敏感数据发送出去总是增加了额外的风险点。

1.3 LangSmith 中的追踪数据结构

LangSmith 追踪的核心是 Run 对象。一个完整的 LLM 应用程序执行过程通常是一个 Run，其中包含一系列嵌套的子 Run，形成一个树状结构。这些 Run 可以是：

• LLM Run：表示一个语言模型的调用（例如 llm.invoke(...)）。
• Chain Run：表示一个 LangChain 链的执行（例如 chain.invoke(...)）。
• Tool Run：表示一个工具的调用（例如 agent.run(...) 调用某个工具）。
• Agent Run：表示一个 Agent 的执行循环。

每个 Run 对象都包含以下关键信息，这些信息都可能成为隐私数据的载体：

字段名称	描述	潜在隐私数据类型
id	运行的唯一标识符	通常不是隐私，但如果基于 PII 生成则需注意
name	运行的名称（例如链名、工具名）	通常不是隐私
run_type	运行类型（"llm", "chain", "tool", "agent" 等）	通常不是隐私
inputs	运行的输入数据	高风险：用户提示、工具参数、API 密钥、PII 等
outputs	运行的输出数据	高风险：LLM 响应、工具结果、PII 等
start_time	运行开始时间	通常不是隐私
end_time	运行结束时间	通常不是隐私
extra	额外信息，通常包含 metadata 和其他运行时上下文	高风险：自定义元数据、用户信息、会话 ID 等
error	运行错误信息	可能包含导致错误的敏感输入或系统路径
parent_run_id	父运行的 ID	通常不是隐私
serialized	运行组件的序列化表示（例如 LLM 模型的配置）	可能包含模型配置中的 API 密钥或其他凭证

从上表可以看出，inputs、outputs 和 extra（特别是 metadata）是隐私数据最常出现的地方，也是我们脱敏工作的重点。

第二章：隐私脱敏策略与技术

脱敏的本质是在不影响监控和调试功能的前提下，将敏感数据转换为无害形式。

2.1 常见脱敏方法概述

脱敏方法	描述	优点	缺点	适用场景
替换 (Substitution)	将敏感数据替换为固定字符串（如 [REDACTED]）或随机生成的值。	简单直接，完全移除敏感信息。	失去原始数据的所有语义和结构，可能影响调试。	绝大多数敏感字符串、字段。
删除 (Deletion)	直接移除包含敏感数据的整个字段或数据段。	完全移除敏感信息。	可能导致数据结构不完整，影响后续分析。	整个字段都被认为是敏感，且无需保留任何信息。
泛化 (Generalization)	将具体值替换为更广泛的类别（如将精确年龄替换为年龄段）。	保留部分数据分布特征，可用于统计分析。	仍然存在逆向工程风险，实现相对复杂。	统计分析场景，LangSmith 监控中较少直接使用。
哈希 (Hashing)	将敏感数据通过哈希算法转换为固定长度的散列值。	保持数据一致性（相同输入得到相同哈希），可用于关联分析。	理论上不可逆，但存在哈希碰撞和彩虹表攻击风险。	需要跟踪特定实体（如用户ID），但不能暴露原始ID。
加密 (Encryption)	使用密钥将敏感数据加密。	提供最高级别的数据保护，可逆。	需要密钥管理，且 LangSmith 可能无法解密。不适合直接发送。	通常用于存储，而非发送到监控系统。

对于 LangSmith 的 Trace 脱敏，我们主要关注替换和哈希这两种方法。替换适用于大多数文本内容，而哈希则适用于需要匿名跟踪但又不能暴露原始值的场景（例如用户 ID）。

2.2 基于模式匹配 (正则表达式) 的脱敏

正则表达式是识别和替换文本中特定模式的强大工具。这是处理非结构化或半结构化字符串内容（如用户提示、LLM 响应）的首选方法。

常见敏感数据模式示例：

• 电子邮件地址：b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+.[A-Z|a-z]{2,}b
• 电话号码：b(?:+?d{1,3}[-.s]?)?(?d{3})?[-.s]?d{3}[-.s]?d{4}b (不同国家地区模式差异大)
• 社会安全号码 (SSN)：bd{3}[-.s]?d{2}[-.s]?d{4}b (美国)
• 信用卡号：b(?:d{4}[-.s]?){3}d{4}b (需要结合 Luhn 算法或前缀匹配进一步验证，单纯正则易误报)
• IP 地址：b(?:[0-9]{1,3}.){3}[0-9]{1,3}b
• API 密钥/Token：通常有特定前缀或格式，例如 sk-[a-zA-Z0-9]{32,} (OpenAI)，Bearer [a-zA-Z0-9-_.]+.[a-zA-Z0-9-_.]+.[a-zA-Z0-9-_.]+ (JWT)。这需要根据实际使用的服务来定义。
• 姓名：难以准确识别，因为普通词语也可能是姓名。除非在特定上下文或已知格式中，否则不建议通过通用正则匹配姓名。

挑战：

• 误报 (False Positives)：正则表达式可能错误地将非敏感数据识别为敏感数据。例如，123-456-7890 可能是电话号码，也可能只是产品型号。
• 漏报 (False Negatives)：无法匹配所有可能的敏感数据变体。
• 性能：复杂的正则表达式在大文本上可能影响性能。

2.3 基于结构/字段的脱敏

当数据以结构化格式（如 JSON、字典）存在时，我们可以根据字段名来直接定位并脱敏。这比正则表达式更精确，且误报率低。

示例：

• 如果已知 user_info 字典中包含 email 和 phone 字段，可以直接访问 data['user_info']['email'] 进行替换。
• 在 LangChain 的 inputs 或 outputs 字典中，我们可能约定某些字段总是包含敏感信息，例如 inputs['user_query']['personal_data']。

挑战：

• 缺乏灵活性：依赖于预定义的结构和字段名。如果数据结构发生变化，脱敏规则也需要更新。
• 深度嵌套：如果敏感数据深层嵌套在复杂的数据结构中，需要递归遍历。

2.4 基于上下文的脱敏 (复杂但有效)

结合模式匹配和结构化脱敏，并利用数据的上下文信息，可以实现更智能的脱敏。

示例：

• 只有当正则表达式匹配到电话号码，并且该电话号码出现在一个名为 shipping_address 的字段中时，才进行脱敏。
• 利用 NLP 技术识别实体（如人名、组织名、位置），然后根据实体类型和周围文本进行判断是否需要脱敏。这超出了本讲座的自动化脱敏范畴，通常需要更复杂的 NLP 模型。

对于 LangSmith Trace 的自动化脱敏，我们将主要聚焦于基于模式匹配和基于结构/字段的组合方法，通过自定义回调处理器在数据发送前进行拦截和处理。

第三章：LangChain 回调系统与脱敏切入点

LangChain 的回调机制是实现自定义行为（如日志记录、度量、脱敏）的强大工具。它允许我们在链、LLM、工具、代理等不同组件的生命周期事件发生时执行自定义逻辑。LangSmith 自身也是通过 LangChain 的回调系统（LangChainTracer）来捕获和发送 Trace 数据的。

我们的目标是在 LangChainTracer 捕获到数据并将其发送出去之前，对原始数据进行脱敏。最直接的方法是创建一个自定义的 BaseCallbackHandler，并将其置于回调链中的 LangChainTracer 之前，或者在它的方法中直接修改传入的数据。

3.1 LangChain Callback 机制简介

LangChain 中的回调是通过 langchain_core.callbacks.BaseCallbackHandler 类及其子类实现的。当 LangChain 组件（如 LLM、Chain、Tool）执行时，它们会触发一系列预定义的方法，这些方法将调用所有注册的 BaseCallbackHandler 实例。

一个典型的回调处理器会继承 BaseCallbackHandler 并覆盖其 on_* 方法。

3.2 BaseCallbackHandler 的核心方法

以下是 BaseCallbackHandler 中与脱敏密切相关的一些核心方法：

方法签名	触发时机	传入参数	脱敏关注点
on_llm_start(serialized: Dict[str, Any], prompts: List[str], **kwargs: Any) -> Any	LLM 调用开始前	serialized: LLM 配置；prompts: 传递给 LLM 的提示列表；kwargs: 其他参数，如 run_id, parent_run_id 等	prompts (用户输入、系统指令)，serialized (潜在的 API 密钥)
on_llm_end(response: LLMResult, **kwargs: Any) -> Any	LLM 调用结束后	response: LLM 的响应结果 (LLMResult 对象，包含生成的文本)；kwargs: 其他参数	response (LLM 生成的文本可能包含敏感信息)
on_chain_start(serialized: Dict[str, Any], inputs: Dict[str, Any], **kwargs: Any) -> Any	Chain 调用开始前	serialized: Chain 配置；inputs: 传递给 Chain 的输入字典；kwargs: 其他参数	inputs (Chain 的输入通常是用户提供或上一步生成的，包含 PII 风险高)
on_chain_end(outputs: Dict[str, Any], **kwargs: Any) -> Any	Chain 调用结束后	outputs: Chain 的输出字典；kwargs: 其他参数	outputs (Chain 的最终结果，可能包含 PII)
on_tool_start(serialized: Dict[str, Any], input_str: str, **kwargs: Any) -> Any	Tool 调用开始前	serialized: Tool 配置；input_str: 传递给 Tool 的字符串输入；kwargs: 其他参数（如 tool_input 可能是结构化数据）	input_str (工具输入)，kwargs 中的 tool_input (如果工具接受结构化输入)
on_tool_end(output: str, **kwargs: Any) -> Any	Tool 调用结束后	output: Tool 的字符串输出；kwargs: 其他参数	output (工具执行结果，可能包含查询结果、API 响应等敏感信息)
on_agent_action(action: AgentAction, **kwargs: Any) -> Any	Agent 执行一个动作前	action: AgentAction 对象，包含 tool (工具名称) 和 tool_input (工具输入)	action.tool_input
on_agent_finish(finish: AgentFinish, **kwargs: Any) -> Any	Agent 结束执行后	finish: AgentFinish 对象，包含 return_values (Agent 最终返回的值)	finish.return_values
on_text(text: str, **kwargs: Any) -> Any	任何组件打印文本时（例如 Agent 的思考过程），或者 StreamlitCallbackHandler 等用于实时显示中间步骤时。	text: 打印的文本	text (思考过程可能泄露敏感信息)
on_retriever_start(serialized: Dict[str, Any], query: str, **kwargs: Any) -> Any	Retriever 调用开始前	serialized: Retriever 配置；query: 检索查询字符串；kwargs: 其他参数	query (用户查询，可能包含 PII)
on_retriever_end(documents: List[Document], **kwargs: Any) -> Any	Retriever 调用结束后	documents: 检索到的 Document 列表（包含 page_content 和 metadata）；kwargs: 其他参数	documents (文档内容和元数据可能包含敏感信息)
on_chat_model_start(serialized: Dict[str, Any], messages: List[List[BaseMessage]], **kwargs: Any) -> Any	聊天模型调用开始前	serialized: 聊天模型配置；messages: 传递给聊天模型的 BaseMessage 列表（用户消息、系统消息等）	messages (特别是用户消息的内容)

3.3 在 LangSmith Trace 生命周期中识别脱敏点

我们的脱敏策略需要覆盖所有可能出现敏感数据的地方。通过覆盖上述 on_* 方法，我们可以在数据被 LangSmith Tracer 捕获并发送之前，对其进行修改。

• 输入脱敏：
  • on_llm_start 的 prompts：用户向 LLM 发送的原始文本。
  • on_chain_start 的 inputs：链的输入，可能是用户查询或上一步的结果。
  • on_tool_start 的 input_str 或 kwargs['tool_input']：工具接收的参数。
  • on_agent_action 的 action.tool_input：代理决定调用工具时的输入。
  • on_retriever_start 的 query：检索器的查询。
  • on_chat_model_start 的 messages：聊天模型接收的消息列表。
• 输出脱敏：
  • on_llm_end 的 response：LLM 生成的文本。
  • on_chain_end 的 outputs：链的最终输出。
  • on_tool_end 的 output：工具的执行结果。
  • on_agent_finish 的 finish.return_values：代理的最终结果。
  • on_retriever_end 的 documents：检索器返回的文档内容和元数据。
• 中间文本脱敏：
  • on_text 的 text：代理思考过程或其他中间文本。
• 元数据脱敏：
  • 所有 kwargs 中可能包含的 metadata：自定义的运行时附加信息。

值得注意的是，由于回调处理器会修改传入的数据，而 Python 中的字典和列表通常是引用传递，所以对这些数据结构的修改会影响到后续的回调处理器（包括 LangChainTracer）接收到的数据。这正是我们实现脱敏所需要的行为。

第四章：构建自定义隐私脱敏回调处理器

我们将创建一个名为 PrivacyRedactionCallbackHandler 的自定义回调处理器。这个处理器将继承 BaseCallbackHandler，并实现一套通用的脱敏逻辑，应用于所有可能包含敏感信息的 on_* 方法的参数。

4.1 设计 PrivacyRedactionCallbackHandler

我们的处理器需要：

1. 脱敏规则：一个可配置的正则表达式列表，用于匹配常见的敏感数据。
2. 字段黑名单：一个字段名列表，用于直接删除或替换特定字段的值。
3. 递归处理：能够处理深层嵌套的字典和列表结构。
4. 替换占位符：定义敏感数据被替换后的文本，例如 [REDACTED]。
5. 哈希功能：对于某些需要匿名跟踪的字段，提供哈希选项。

4.2 实现核心脱敏逻辑 (正则、字段遍历)

我们将首先实现一个辅助函数，用于递归地遍历任何 Python 数据结构（字符串、列表、字典），并应用脱敏规则。

import re
import hashlib
from typing import Any, Dict, List, Optional, Union
from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.outputs import LLMResult
from langchain_core.messages import BaseMessage, AIMessage, HumanMessage, SystemMessage, FunctionMessage, ToolMessage
from langchain_core.documents import Document
from langchain_core.agents import AgentAction, AgentFinish

# --- 辅助脱敏函数 ---

def _hash_value(value: str) -> str:
    """对字符串进行 SHA256 哈希处理。"""
    return hashlib.sha256(value.encode('utf-8')).hexdigest()

def _redact_text_with_regex(
    text: str, 
    regex_patterns: List[re.Pattern], 
    redaction_placeholder: str = "[REDACTED]"
) -> str:
    """使用正则表达式列表脱敏文本。"""
    if not isinstance(text, str):
        return text # 只处理字符串

    for pattern in regex_patterns:
        text = pattern.sub(redaction_placeholder, text)
    return text

def _redact_data_recursive(
    data: Any,
    regex_patterns: List[re.Pattern],
    field_blacklist: List[str],
    field_hash_list: List[str],
    redaction_placeholder: str = "[REDACTED]",
    max_depth: int = 10 # 避免无限递归
) -> Any:
    """
    递归地遍历数据结构（字典、列表、字符串），并应用脱敏规则。

    Args:
        data: 要脱敏的数据。
        regex_patterns: 用于字符串脱敏的正则表达式列表。
        field_blacklist: 字段名黑名单，这些字段的值将被完全替换。
        field_hash_list: 字段名哈希列表，这些字段的值将被哈希。
        redaction_placeholder: 替换敏感数据的占位符。
        max_depth: 递归的最大深度，防止无限递归。

    Returns:
        脱敏后的数据。
    """
    if max_depth <= 0:
        return redaction_placeholder # 达到最大深度，直接替换

    if isinstance(data, dict):
        redacted_dict = {}
        for key, value in data.items():
            if key in field_blacklist:
                redacted_dict[key] = redaction_placeholder
            elif key in field_hash_list and isinstance(value, str):
                redacted_dict[key] = _hash_value(value)
            else:
                redacted_dict[key] = _redact_data_recursive(
                    value, regex_patterns, field_blacklist, field_hash_list, 
                    redaction_placeholder, max_depth - 1
                )
        return redacted_dict

    elif isinstance(data, list):
        return [
            _redact_data_recursive(
                item, regex_patterns, field_blacklist, field_hash_list, 
                redaction_placeholder, max_depth - 1
            ) 
            for item in data
        ]

    elif isinstance(data, str):
        return _redact_text_with_regex(data, regex_patterns, redaction_placeholder)

    # 对于其他不可变类型（如 int, float, bool, None），直接返回
    return data

# --- 自定义脱敏回调处理器 ---

class PrivacyRedactionCallbackHandler(BaseCallbackHandler):
    """
    一个自定义的 LangChain 回调处理器，用于在数据发送到 LangSmith 前自动脱敏。
    它支持正则表达式匹配、字段黑名单替换和字段值哈希。
    """

    def __init__(
        self,
        regex_patterns: Optional[List[str]] = None,
        field_blacklist: Optional[List[str]] = None,
        field_hash_list: Optional[List[str]] = None,
        redaction_placeholder: str = "[REDACTED]",
        max_redaction_depth: int = 10,
        **kwargs: Any
    ) -> None:
        super().__init__(**kwargs)
        self.compiled_regex_patterns = [re.compile(p) for p in (regex_patterns or [])]
        self.field_blacklist = [f.lower() for f in (field_blacklist or [])] # 统一为小写，方便匹配
        self.field_hash_list = [f.lower() for f in (field_hash_list or [])]
        self.redaction_placeholder = redaction_placeholder
        self.max_redaction_depth = max_redaction_depth

        # 默认的常见敏感信息正则表达式，可以根据需要扩展或修改
        if not regex_patterns:
            self.compiled_regex_patterns.extend([
                re.compile(r'b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+.[A-Z|a-z]{2,}b'), # 邮箱
                re.compile(r'b(?:+?d{1,3}[-.s]?)?(?d{3})?[-.s]?d{3}[-.s]?d{4}b'), # 电话号码 (通用)
                re.compile(r'bd{3}[-.s]?d{2}[-.s]?d{4}b'), # 美国 SSN 格式
                re.compile(r'b(?:4d{12}(?:d{3})?|5[1-5]d{14}|6(?:011|5d{2})d{12}|3[47]d{13}|3(?:0[0-5]|[68]d)d{11}|(?:2131|1800|35d{3})d{11})b'), # 信用卡号 (宽泛匹配，易误报)
                re.compile(r'b(?:[0-9]{1,3}.){3}[0-9]{1,3}b'), # IP 地址
                re.compile(r'sk-[a-zA-Z0-9]{32,}'), # OpenAI API 密钥
                re.compile(r'Bearers[a-zA-Z0-9-_.]+.[a-zA-Z0-9-_.]+.[a-zA-Z0-9-_.]+'), # JWT Token
            ])

        # 默认的字段黑名单，可以根据应用场景扩展
        if not field_blacklist:
            self.field_blacklist.extend([
                "password", "secret", "api_key", "token", "private_key", 
                "ssn", "credit_card_number", "email", "phone_number", "address",
                "user_id_raw", "customer_id_raw", # 如果有原始ID字段
            ])

        # 默认的哈希字段，用于需要匿名跟踪但不能暴露原始值的字段
        if not field_hash_list:
            self.field_hash_list.extend([
                "user_id", "customer_id", "session_id", "device_id"
            ])

    def _redact_common_data(self, data: Any) -> Any:
        """应用通用的递归脱敏逻辑。"""
        return _redact_data_recursive(
            data, 
            self.compiled_regex_patterns, 
            self.field_blacklist,
            self.field_hash_list,
            self.redaction_placeholder,
            self.max_redaction_depth
        )

    # --- 覆盖 BaseCallbackHandler 的方法以实现脱敏 ---

    def on_llm_start(
        self, serialized: Dict[str, Any], prompts: List[str], **kwargs: Any
    ) -> Any:
        # 脱敏 prompts
        for i in range(len(prompts)):
            prompts[i] = self._redact_common_data(prompts[i])

        # 脱敏 serialized 中的潜在敏感信息（例如模型配置中的 API 密钥）
        if "kwargs" in serialized and isinstance(serialized["kwargs"], dict):
            if "openai_api_key" in serialized["kwargs"]:
                serialized["kwargs"]["openai_api_key"] = self.redaction_placeholder
            # 也可以在这里检查其他模型的 API 密钥字段

        # 脱敏 kwargs 中的元数据
        if "metadata" in kwargs and isinstance(kwargs["metadata"], dict):
            kwargs["metadata"] = self._redact_common_data(kwargs["metadata"])

    def on_llm_end(self, response: LLMResult, **kwargs: Any) -> Any:
        # 脱敏 LLM 响应
        for gen_list in response.generations:
            for i in range(len(gen_list)):
                gen_list[i].text = self._redact_common_data(gen_list[i].text)

        # 脱敏 kwargs 中的元数据
        if "metadata" in kwargs and isinstance(kwargs["metadata"], dict):
            kwargs["metadata"] = self._redact_common_data(kwargs["metadata"])

    def on_chain_start(
        self, serialized: Dict[str, Any], inputs: Dict[str, Any], **kwargs: Any
    ) -> Any:
        # 脱敏 Chain 输入
        for key, value in inputs.items():
            inputs[key] = self._redact_common_data(value)

        # 脱敏 kwargs 中的元数据
        if "metadata" in kwargs and isinstance(kwargs["metadata"], dict):
            kwargs["metadata"] = self._redact_common_data(kwargs["metadata"])

    def on_chain_end(self, outputs: Dict[str, Any], **kwargs: Any) -> Any:
        # 脱敏 Chain 输出
        for key, value in outputs.items():
            outputs[key] = self._redact_common_data(value)

        # 脱敏 kwargs 中的元数据
        if "metadata" in kwargs and isinstance(kwargs["metadata"], dict):
            kwargs["metadata"] = self._redact_common_data(kwargs["metadata"])

    def on_tool_start(
        self, serialized: Dict[str, Any], input_str: str, **kwargs: Any
    ) -> Any:
        # 脱敏工具输入字符串
        input_str = self._redact_common_data(input_str)

        # 脱敏 kwargs 中的结构化 tool_input (如果存在)
        if "tool_input" in kwargs and isinstance(kwargs["tool_input"], (dict, list, str)):
            kwargs["tool_input"] = self._redact_common_data(kwargs["tool_input"])

        # 脱敏 kwargs 中的元数据
        if "metadata" in kwargs and isinstance(kwargs["metadata"], dict):
            kwargs["metadata"] = self._redact_common_data(kwargs["metadata"])

    def on_tool_end(self, output: str, **kwargs: Any) -> Any:
        # 脱敏工具输出
        output = self._redact_common_data(output)

        # 脱敏 kwargs 中的元数据
        if "metadata" in kwargs and isinstance(kwargs["metadata"], dict):
            kwargs["metadata"] = self._redact_common_data(kwargs["metadata"])

    def on_agent_action(self, action: AgentAction, **kwargs: Any) -> Any:
        # 脱敏 Agent 动作的工具输入
        action.tool_input = self._redact_common_data(action.tool_input)

        # 脱敏 kwargs 中的元数据
        if "metadata" in kwargs and isinstance(kwargs["metadata"], dict):
            kwargs["metadata"] = self._redact_common_data(kwargs["metadata"])

    def on_agent_finish(self, finish: AgentFinish, **kwargs: Any) -> Any:
        # 脱敏 Agent 最终返回的值
        finish.return_values = self._redact_common_data(finish.return_values)

        # 脱敏 kwargs 中的元数据
        if "metadata" in kwargs and isinstance(kwargs["metadata"], dict):
            kwargs["metadata"] = self._redact_common_data(kwargs["metadata"])

    def on_text(self, text: str, **kwargs: Any) -> Any:
        # 脱敏中间文本，如 Agent 的思考过程
        text = self._redact_common_data(text)

        # 脱敏 kwargs 中的元数据
        if "metadata" in kwargs and isinstance(kwargs["metadata"], dict):
            kwargs["metadata"] = self._redact_common_data(kwargs["metadata"])

    def on_retriever_start(
        self, serialized: Dict[str, Any], query: str, **kwargs: Any
    ) -> Any:
        query = self._redact_common_data(query)
        if "metadata" in kwargs and isinstance(kwargs["metadata"], dict):
            kwargs["metadata"] = self._redact_common_data(kwargs["metadata"])

    def on_retriever_end(self, documents: List[Document], **kwargs: Any) -> Any:
        for doc in documents:
            doc.page_content = self._redact_common_data(doc.page_content)
            doc.metadata = self._redact_common_data(doc.metadata)
        if "metadata" in kwargs and isinstance(kwargs["metadata"], dict):
            kwargs["metadata"] = self._redact_common_data(kwargs["metadata"])

    def on_chat_model_start(
        self, serialized: Dict[str, Any], messages: List[List[BaseMessage]], **kwargs: Any
    ) -> Any:
        for message_list in messages: # messages 是 List[List[BaseMessage]]
            for i, msg in enumerate(message_list):
                if isinstance(msg, (HumanMessage, AIMessage, SystemMessage, FunctionMessage, ToolMessage)):
                    if hasattr(msg, 'content') and isinstance(msg.content, str):
                        msg.content = self._redact_common_data(msg.content)
                    elif hasattr(msg, 'content') and isinstance(msg.content, list): # 针对 Vision 模型等多模态输入
                         # 递归处理 list of dicts/strings
                        msg.content = self._redact_common_data(msg.content)
                    # For FunctionMessage/ToolMessage, also check name/tool_call/tool_returns
                    if isinstance(msg, (FunctionMessage, ToolMessage)):
                        # These usually contain content as string or a dict.
                        # The _redact_common_data handles both
                        pass # Already handled content above
        if "metadata" in kwargs and isinstance(kwargs["metadata"], dict):
            kwargs["metadata"] = self._redact_common_data(kwargs["metadata"])

4.3 代码示例：一个基础的正则匹配脱敏器

让我们创建一个简单的 LangChain 链，并通过我们的脱敏回调处理器运行它。

首先，确保安装 LangChain 和 LangSmith 相关的库：
pip install langchain langchain-openai langsmith

import os
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.callbacks import LangChainTracer
from langchain.agents import AgentExecutor, create_react_agent, Tool
from langchain import hub

# 设置 LangSmith 环境变量
# os.environ["LANGCHAIN_TRACING_V2"] = "true"
# os.environ["LANGCHAIN_ENDPOINT"] = "https://api.smith.langchain.com"
# os.environ["LANGCHAIN_API_KEY"] = "YOUR_LANGSMITH_API_KEY" # 替换为你的 LangSmith API 密钥
# os.environ["OPENAI_API_KEY"] = "YOUR_OPENAI_API_KEY" # 替换为你的 OpenAI API 密钥

# 实例化脱敏处理器
# 我们可以自定义正则表达式和黑名单
custom_regex_patterns = [
    r"user_id_d+", # 匹配如 "user_id_123"
    r"confidential_doc_d{4}", # 匹配如 "confidential_doc_2023"
]
custom_field_blacklist = ["secret_data", "sensitive_param"]
custom_field_hash_list = ["user_identifier"]

privacy_handler = PrivacyRedactionCallbackHandler(
    regex_patterns=custom_regex_patterns,
    field_blacklist=custom_field_blacklist,
    field_hash_list=custom_field_hash_list,
    redaction_placeholder="[PRIVATE_INFO]"
)

# LangSmith Tracer 必须是最后一个，以确保它接收到的是已脱敏的数据
# 如果不设置 LANGCHAIN_TRACING_V2 环境变量，可以手动实例化 LangChainTracer
# langsmith_tracer = LangChainTracer(project_name="redaction-demo")

# 示例 1: LLM 调用和 Chain 的输入/输出脱敏

llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个乐于助人的 AI 助手。"),
    ("human", "我的电子邮件是 {email}，我的电话号码是 {phone_number}。请问我如何重置我的密码？我的 user_id_raw 是 user_id_123。我的 user_identifier 是 test_user_456.")
])

chain = (
    prompt 
    | llm 
    | StrOutputParser()
)

print("--- 运行 Chain (带敏感信息) ---")
# 在这里传递回调处理器列表
# privacy_handler 应该在 LangChainTracer 之前，确保 LangChainTracer 收到脱敏后的数据
# 如果 LANGCHAIN_TRACING_V2 环境变量已设置，LangChainTracer 会自动被添加，
# 我们的 handler 也会被添加到其之前。
# 如果没有设置环境变量，我们可以手动构建回调列表。
# callbacks = [privacy_handler, langsmith_tracer] if 'LANGCHAIN_TRACING_V2' not in os.environ else [privacy_handler]
# 为了演示，我们假设 LangChainTracer 已经通过环境变量激活，或者我们手动将其添加到回调列表的末尾
# 在实际生产中，你通常会依赖环境变量来激活 LangSmith。
# 确保 privacy_handler 在 LangSmithTracer 之前被调用。
# LangChain Tracing v2 默认将回调添加到列表的开头，所以我们的自定义回调会先执行。
# 参见：https://python.langchain.com/docs/modules/callbacks/custom_callbacks/
output = chain.invoke(
    {
        "email": "[email protected]", 
        "phone_number": "123-456-7890",
        "user_id_raw": "user_id_123", # 这是一个黑名单字段
        "user_identifier": "test_user_456", # 这是一个哈希字段
        "secret_data": "my_super_secret_password", # 另一个黑名单字段
        "some_other_info": "This is a normal piece of info.",
    },
    config={"callbacks": [privacy_handler]} # 直接传入回调处理器
)
print("Chain 输出:", output)
print("n请检查 LangSmith UI 中对应的 Trace，查看 email, phone_number, user_id_raw, user_identifier 和 secret_data 是否被脱敏。")
print("注意：LLM 响应中的敏感信息也会被脱敏。")

# 示例 2: Tool 调用和 Agent 的输入/输出脱敏

# 模拟一个可能返回敏感信息的工具
def get_user_profile(user_id: str, sensitive_param: str) -> str:
    """获取用户档案，可能包含敏感信息。"""
    if user_id == "test_user_456":
        return f"User profile for {user_id}: Name: Jane Doe, Email: [email protected], Address: 123 Main St. API Key: sk-test-key-123. Sensitive Param: {sensitive_param}"
    return f"User profile for {user_id}: No sensitive data. Sensitive Param: {sensitive_param}"

tools = [
    Tool(
        name="get_user_profile",
        func=get_user_profile,
        description="用于获取用户档案信息，需要 user_id 作为输入。"
    )
]

# 提示模板
prompt_agent = ChatPromptTemplate.from_messages([
    ("system", "你是一个有用的助手。当需要查询用户档案时，使用 get_user_profile 工具。"),
    ("human", "{input}")
])

# 创建 ReAct 代理
agent = create_react_agent(llm, tools, prompt_agent)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

print("n--- 运行 Agent (带敏感信息) ---")
agent_input = {
    "input": "请帮我查找用户 'test_user_456' 的档案。我的 secret_data 是 my_agent_secret。",
    "user_identifier": "test_user_456", # 这个字段会被哈希
    "sensitive_param": "agent_sensitive_value" # 这个字段会在 tool input 中被脱敏
}
output_agent = agent_executor.invoke(
    agent_input,
    config={"callbacks": [privacy_handler]}
)
print("Agent 输出:", output_agent)
print("n请检查 LangSmith UI 中对应的 Trace，查看 Agent 的输入、工具的输入/输出是否被脱敏。")
print("特别是 tool input 'test_user_456' 应该被哈希，'[email protected]' 和 'sk-test-key-123' 应该被替换。")

运行上述代码并观察 LangSmith UI

当你运行上述代码后，如果 LangSmith 环境变量正确配置，你会在 LangSmith UI 中看到两个新的 Trace。

• 对于第一个 Chain Trace：
  • inputs 中 email 和 phone_number 会被 [PRIVATE_INFO] 替换。
  • user_id_raw 和 secret_data 字段的值也会被 [PRIVATE_INFO] 替换。
  • user_identifier 的值 test_user_456 会被替换成其 SHA256 哈希值。
  • LLM 的 prompts 和 response 中匹配到电话号码、邮箱或 user_id_ 模式的文本都会被脱敏。
• 对于第二个 Agent Trace：
  • Agent 的 input 会被脱敏。
  • 当 Agent 调用 get_user_profile 工具时，工具的 input_str 和 tool_input 参数中匹配到的敏感信息（例如 sensitive_param 的值）会被脱敏。
  • 工具的 output 中包含的 [email protected] 和 sk-test-key-123 会被 [PRIVATE_INFO] 替换。
  • Agent 最终的 return_values 也会是脱敏后的结果。

通过这种方式，我们确保了在数据离开应用程序并发送到 LangSmith 之前，所有指定的敏感信息都已经被有效地处理。

第五章：高级脱敏技巧与最佳实践

5.1 配置化脱敏规则：灵活性与可维护性

将脱敏规则硬编码到代码中是不灵活的。最佳实践是将正则表达式、字段黑名单、哈希字段列表和占位符等配置外部化。

• 环境变量：适用于少量、全局性的规则。例如，通过 PII_REGEX_PATTERNS 环境变量传递逗号分隔的正则表达式字符串。
• 配置文件：使用 YAML、JSON 或 TOML 文件来定义更复杂、分层的规则。这允许按环境（开发、测试、生产）或按应用模块定义不同的规则集。
• 集中式配置服务：对于微服务架构，可以使用 HashiCorp Vault、AWS Secrets Manager 或 Kubernetes Secrets 等服务来安全地存储和动态加载脱敏规则。

示例：从环境变量加载配置

import os
import json

# ... (Previous _redact_data_recursive and _redact_text_with_regex functions) ...

class ConfigurablePrivacyRedactionCallbackHandler(PrivacyRedactionCallbackHandler):
    def __init__(self, **kwargs: Any) -> None:
        # 从环境变量加载自定义规则
        env_regex_str = os.getenv("REDACTION_REGEX_PATTERNS")
        env_field_blacklist_str = os.getenv("REDACTION_FIELD_BLACKLIST")
        env_field_hash_list_str = os.getenv("REDACTION_FIELD_HASH_LIST")
        env_placeholder = os.getenv("REDACTION_PLACEHOLDER", "[REDACTED_ENV]")

        # 优先级：传入参数 > 环境变量 > 默认值
        regex_patterns = kwargs.pop("regex_patterns", None)
        if regex_patterns is None and env_regex_str:
            regex_patterns = json.loads(env_regex_str) # 假设是 JSON 数组字符串

        field_blacklist = kwargs.pop("field_blacklist", None)
        if field_blacklist is None and env_field_blacklist_str:
            field_blacklist = [f.strip() for f in env_field_blacklist_str.split(',')]

        field_hash_list = kwargs.pop("field_hash_list", None)
        if field_hash_list is None and env_field_hash_list_str:
            field_hash_list = [f.strip() for f in env_field_hash_list_str.split(',')]

        redaction_placeholder = kwargs.pop("redaction_placeholder", env_placeholder)

        super().__init__(
            regex_patterns=regex_patterns,
            field_blacklist=field_blacklist,
            field_hash_list=field_hash_list,
            redaction_placeholder=redaction_placeholder,
            **kwargs
        )

# 使用示例
# os.environ["REDACTION_REGEX_PATTERNS"] = '["user_id_\\d+", "credit_card_\\d{4}"]'
# os.environ["REDACTION_FIELD_BLACKLIST"] = "password,secret_key"
# os.environ["REDACTION_PLACEHOLDER"] = "[CENSORED]"
#
# configurable_handler = ConfigurablePrivacyRedactionCallbackHandler()
#
# # ... (然后将 configurable_handler 传递给 chain.invoke)

5.2 性能考量：异步脱敏与效率优化

递归遍历复杂的数据结构和应用多个正则表达式可能会引入性能开销，尤其是在高并发或处理大量数据时。

• 优化正则表达式：确保正则表达式高效且尽可能精确，避免使用贪婪匹配（.*）在长字符串上。
• 缓存编译后的正则：re.compile() 应该只执行一次，我们的实现已经考虑到了这一点。
• 限制递归深度：设置 max_redaction_depth 可以防止无限递归并限制处理复杂度的上限。
• 异步处理：对于非常大的数据或复杂的脱敏规则，可以考虑在单独的线程或进程中进行脱敏，避免阻塞主应用程序线程。然而，由于 LangChain 回调是同步的，这需要更复杂的设计，例如在回调中将数据发送到队列，然后由另一个异步服务处理并发送到 LangSmith。对于大多数 LLM 应用，同步脱敏通常足够快。
• 选择性脱敏：只对已知可能包含敏感信息的字段进行深度扫描，对其他字段进行浅层检查。

5.3 误报与漏报：平衡安全与可用性

• 误报 (False Positives)：将非敏感数据错误地标记为敏感。这可能导致有用的调试信息丢失。
  • 缓解：精确的正则表达式、字段白名单（只允许已知安全字段）、基于上下文的规则。
• 漏报 (False Negatives)：未能识别出真正的敏感数据。这是最危险的情况，可能导致合规问题。
  • 缓解：全面的正则表达式库、定期更新规则、结合结构化字段检查、安全审计。

在实际应用中，通常需要在“过于激进的脱敏”（高误报，低可用性）和“过于宽松的脱敏”（高漏报，高风险）之间找到平衡点。建议从稍微激进的策略开始，然后根据实际调试需求和安全审计结果逐步放宽。

5.4 数据一致性与可追溯性：脱敏后的数据分析

脱敏后的数据虽然失去了原始的敏感信息，但仍应尽可能保留其结构和非敏感部分的语义，以便进行后续的分析和调试。

• 哈希技术：对于用户 ID 等需要跟踪但不能暴露原始值的字段，哈希是一个很好的选择。相同的原始 ID 总是生成相同的哈希值，从而允许我们在 LangSmith 中跟踪特定用户的会话，而无需知道其真实身份。
• 保留数据结构：替换敏感值而非删除整个字段，有助于保持数据的完整性，使得 LangSmith 上的 Trace 结构仍然可读。
• 审计日志：记录哪些数据被脱敏、使用了哪些规则。这对于合规性审计和调试脱敏逻辑本身至关重要。

5.5 动态脱敏与用户权限

在某些高级场景中，可能需要根据访问 LangSmith UI 的用户权限来动态决定脱敏级别。例如，安全团队成员可能被允许查看更多原始数据，而开发人员只能看到脱敏后的数据。这种“按需解密”或“分级脱敏”需要 LangSmith 本身的支持或在数据发送前进行更复杂的权限检查。目前，我们的解决方案是在数据发送前进行永久脱敏，即一旦脱敏，原始数据就不可恢复，这是最安全也最符合数据最小化原则的做法。

第六章：实际案例与部署考量

6.1 在生产环境中集成脱敏回调

将 PrivacyRedactionCallbackHandler 集成到生产环境的 LangChain 应用程序中非常直接。

1. 全局注册：在应用程序启动时，将 privacy_handler 添加到 LangChainGlobalCallbackManager.set_handlers([privacy_handler, langsmith_tracer]) 中，确保它对所有 LangChain 操作生效。
2. 局部注册：通过 chain.invoke(..., config={"callbacks": [privacy_handler]}) 或 llm.invoke(..., callbacks=[privacy_handler]) 等方式，将其应用于特定的 Chain、LLM 或 Tool 调用。
3. 环境变量激活：确保 LangSmith Tracer 是通过环境变量 LANGCHAIN_TRACING_V2=true 激活的，而不是手动实例化 LangChainTracer。当 LANGCHAIN_TRACING_V2 被设置为 true 时，LangChain 会自动创建一个 LangChainTracer 实例并将其添加到回调列表中。我们的自定义回调应该在 Tracer 之前注册，以确保数据在被 Tracer 处理之前就已经脱敏。LangChain 默认会将用户提供的回调添加到全局回调管理器的前面，所以通常 privacy_handler 会先于 LangChainTracer 执行。

6.2 监控与审计脱敏效果

• 定期检查 LangSmith Traces：随机抽样检查 LangSmith 中的 Trace，以验证敏感数据是否被正确脱敏。这有助于发现漏报或新的敏感数据模式。
• 日志记录：在脱敏处理器内部添加日志，记录哪些数据被匹配、被替换或被哈希。这有助于审计和调试脱敏规则。
• 单元测试和集成测试：编写测试用例，包含各种敏感数据模式，并验证脱敏处理器是否按预期工作。

6.3 与其他安全策略的协同

脱敏不是唯一的安全措施。它应该与其他安全策略协同工作：

• 数据传输加密：确保数据在传输到 LangSmith 时使用 HTTPS 等加密协议。
• 访问控制：严格控制谁可以访问 LangSmith 账户和其中的 Trace 数据。
• 数据保留策略：根据合规性要求，在 LangSmith 中设置合理的数据保留期限，并定期清理旧数据。
• 安全编码实践：从源头减少在应用程序中处理和存储敏感数据的机会。

通过上述方法，我们可以构建一个既能利用 LangSmith 强大的监控能力，又能严格遵守数据隐私和合规性要求的高效、安全的 LLM 应用程序。

---

通过在 LangChain 回调机制中注入自定义脱敏逻辑，我们可以在监控数据发送至 LangSmith 之前，对敏感信息进行自动、细粒度的处理。这不仅满足了合规性要求，降低了数据泄露风险，还为 LLM 应用程序提供了更安全、更负责任的观测能力。