防止在追踪中记录敏感数据

LANGSMITH_HIDE_INPUTS=true
LANGSMITH_HIDE_OUTPUTS=true

from langsmith import Client

# 完全隐藏所有元数据
client = Client(hide_metadata=True)

# 现在当您创建运行时，元数据将为空
client.create_run(
    "my_run",
    inputs={"question": "What is 2+2?"},
    run_type="llm",
    extra={"metadata": {"user_id": "123", "session": "abc"}}
)
# 发送到 LangSmith 的元数据将是 {} 而不是提供的元数据

# 移除敏感键
def hide_sensitive_metadata(metadata: dict) -> dict:
    return {k: v for k, v in metadata.items() if not k.startswith("_private")}

client = Client(hide_metadata=hide_sensitive_metadata)

# 脱敏特定值
def redact_emails(metadata: dict) -> dict:
    import re
    result = {}
    for k, v in metadata.items():
        if isinstance(v, str) and "@" in v:
            result[k] = "[REDACTED_EMAIL]"
        else:
            result[k] = v
    return result

client = Client(hide_metadata=redact_emails)

# 添加转换标记
def add_marker(metadata: dict) -> dict:
    return {**metadata, "transformed": True}

client = Client(hide_metadata=add_marker)

from langsmith import traceable

def process_inputs(inputs: dict) -> dict:
    # inputs 是一个字典，其中键是参数名，值是提供的参数
    # 返回一个包含处理后的输入的新字典
    return {
        "processed_key": inputs.get("my_cool_key", "default"),
        "length": len(inputs.get("my_cool_key", ""))
    }

def process_outputs(output: Any) -> dict:
    # output 是函数的直接返回值
    # 将输出转换为字典
    # 在这种情况下，"output" 将是一个整数
    return {"processed_output": str(output)}

@traceable(process_inputs=process_inputs, process_outputs=process_outputs)
def my_function(my_cool_key: str) -> int:
    # 函数实现
    return len(my_cool_key)

result = my_function("example")

@traceable(process_inputs=process_inputs, process_outputs=process_outputs)
async def async_function(key: str) -> int:
    # 异步实现
    return len(key)

import re
import openai
from langsmith import Client
from langsmith.wrappers import wrap_openai

# 定义各种 PII 的正则表达式模式
SSN_PATTERN = re.compile(r'\b\d{3}-\d{2}-\d{4}\b')
CREDIT_CARD_PATTERN = re.compile(r'\b(?:\d[ -]*?){13,16}\b')
EMAIL_PATTERN = re.compile(r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,7}\b')
PHONE_PATTERN = re.compile(r'\b(?:\+?1[-.\s]?)?\(?\d{3}\)?[-.\s]?\d{3}[-.\s]?\d{4}\b')
FULL_NAME_PATTERN = re.compile(r'\b([A-Z][a-z]*\s[A-Z][a-z]*)\b')

def regex_anonymize(text):
    """
    使用正则表达式模式对文本中的敏感信息进行匿名化。
    参数：
        text (str): 需要匿名化的输入文本。
    返回：
        str: 匿名化后的文本。
    """
    # 使用占位符替换敏感信息
    text = SSN_PATTERN.sub('[REDACTED SSN]', text)
    text = CREDIT_CARD_PATTERN.sub('[REDACTED CREDIT CARD]', text)
    text = EMAIL_PATTERN.sub('[REDACTED EMAIL]', text)
    text = PHONE_PATTERN.sub('[REDACTED PHONE]', text)
    text = FULL_NAME_PATTERN.sub('[REDACTED NAME]', text)
    return text

def recursive_anonymize(data, depth=10):
    """
    递归遍历数据结构并对敏感信息进行匿名化。
    参数：
        data (any): 需要匿名化的输入数据。
        depth (int): 当前递归深度，防止过度递归。
    返回：
        any: 匿名化后的数据。
    """
    if depth == 0:
        return data
    if isinstance(data, dict):
        anonymized_dict = {}
        for k, v in data.items():
            anonymized_value = recursive_anonymize(v, depth - 1)
            anonymized_dict[k] = anonymized_value
        return anonymized_dict
    elif isinstance(data, list):
        anonymized_list = []
        for item in data:
            anonymized_item = recursive_anonymize(item, depth - 1)
            anonymized_list.append(anonymized_item)
        return anonymized_list
    elif isinstance(data, str):
        anonymized_data = regex_anonymize(data)
        return anonymized_data
    else:
        return data

openai_client = wrap_openai(openai.Client())

# 使用匿名化函数初始化 LangSmith @[Client]
langsmith_client = Client(
    hide_inputs=recursive_anonymize, hide_outputs=recursive_anonymize
)

# 生成的追踪将包含其元数据，但输入和输出将被匿名化
response_with_anonymization = openai_client.chat.completions.create(
    model="gpt-5.4-mini",
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "My name is John Doe, my SSN is 123-45-6789, my credit card number is 4111 1111 1111 1111, my email is john.doe@example.com, and my phone number is (123) 456-7890."},
    ],
    langsmith_extra={"client": langsmith_client},
)

# 生成的追踪不会匿名化输入和输出
response_without_anonymization = openai_client.chat.completions.create(
    model="gpt-5.4-mini",
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "My name is John Doe, my SSN is 123-45-6789, my credit card number is 4111 1111 1111 1111, my email is john.doe@example.com, and my phone number is (123) 456-7890."},
    ],
)

import openai
from langsmith import Client
from langsmith.wrappers import wrap_openai
from presidio_anonymizer import AnonymizerEngine
from presidio_analyzer import AnalyzerEngine

anonymizer = AnonymizerEngine()
analyzer = AnalyzerEngine()

def presidio_anonymize(data):
    """
    对用户发送的或模型返回的敏感信息进行匿名化。
    参数：
        data (any): 需要匿名化的数据。
    返回：
        any: 匿名化后的数据。
    """
    message_list = (
        data.get('messages') or [data.get('choices', [{}])[0].get('message')]
    )
    if not message_list or not all(isinstance(msg, dict) and msg for msg in message_list):
        return data

    for message in message_list:
        content = message.get('content', '')
        if not content.strip():
            print("检测到空内容。跳过匿名化。")
            continue

        results = analyzer.analyze(
            text=content,
            entities=["PERSON", "PHONE_NUMBER", "EMAIL_ADDRESS", "US_SSN"],
            language='en'
        )
        anonymized_result = anonymizer.anonymize(
            text=content,
            analyzer_results=results
        )
        message['content'] = anonymized_result.text

    return data

openai_client = wrap_openai(openai.Client())

# 使用匿名化函数初始化 langsmith @[Client]
langsmith_client = Client(
  hide_inputs=presidio_anonymize, hide_outputs=presidio_anonymize
)

# 生成的追踪将包含其元数据，但输入和输出将被匿名化
response_with_anonymization = openai_client.chat.completions.create(
  model="gpt-5.4-mini",
  messages=[
      {"role": "system", "content": "You are a helpful assistant."},
      {"role": "user", "content": "My name is Slim Shady, call me at 313-666-7440 or email me at real.slim.shady@gmail.com"},
  ],
  langsmith_extra={"client": langsmith_client},
)

# 生成的追踪不会匿名化输入和输出
response_without_anonymization = openai_client.chat.completions.create(
  model="gpt-5.4-mini",
  messages=[
      {"role": "system", "content": "You are a helpful assistant."},
      {"role": "user", "content": "My name is Slim Shady, call me at 313-666-7440 or email me at real.slim.shady@gmail.com"},
  ],
)

import openai
import boto3
from langsmith import Client
from langsmith.wrappers import wrap_openai

comprehend = boto3.client('comprehend', region_name='us-east-1')

def redact_pii_entities(text, entities):
    """
    根据检测到的实体，对文本中的 PII 实体进行脱敏。
    参数：
        text (str): 包含 PII 的原始文本。
        entities (list): 检测到的 PII 实体列表。
    返回：
        str: 脱敏后的文本。
    """
    sorted_entities = sorted(entities, key=lambda x: x['BeginOffset'], reverse=True)
    redacted_text = text
    for entity in sorted_entities:
        begin = entity['BeginOffset']
        end = entity['EndOffset']
        entity_type = entity['Type']
        # 根据实体类型定义脱敏占位符
        placeholder = f"[{entity_type}]"
        # 将文本中的 PII 替换为占位符
        redacted_text = redacted_text[:begin] + placeholder + redacted_text[end:]
    return redacted_text

def detect_pii(text):
    """
    使用 AWS Comprehend 检测给定文本中的 PII 实体。
    参数：
        text (str): 要分析的文本。
    返回：
        list: 检测到的 PII 实体列表。
    """
    try:
        response = comprehend.detect_pii_entities(
            Text=text,
            LanguageCode='en',
        )
        entities = response.get('Entities', [])
        return entities
    except Exception as e:
        print(f"检测 PII 时出错: {e}")
        return []

def comprehend_anonymize(data):
    """
    对用户发送的或模型返回的敏感信息进行匿名化。
    参数：
        data (any): 需要匿名化的输入数据。
    返回：
        any: 匿名化后的数据。
    """
    message_list = (
        data.get('messages') or [data.get('choices', [{}])[0].get('message')]
    )
    if not message_list or not all(isinstance(msg, dict) and msg for msg in message_list):
        return data

    for message in message_list:
        content = message.get('content', '')
        if not content.strip():
            print("检测到空内容。跳过匿名化。")
            continue

        entities = detect_pii(content)
        if entities:
            anonymized_text = redact_pii_entities(content, entities)
            message['content'] = anonymized_text
        else:
            print("未检测到 PII。内容保持不变。")

    return data

openai_client = wrap_openai(openai.Client())

# 使用匿名化函数初始化 langsmith @[Client]
langsmith_client = Client(
  hide_inputs=comprehend_anonymize, hide_outputs=comprehend_anonymize
)

# 生成的追踪将包含其元数据，但输入和输出将被匿名化
response_with_anonymization = openai_client.chat.completions.create(
  model="gpt-5.4-mini",
  messages=[
      {"role": "system", "content": "You are a helpful assistant."},
      {"role": "user", "content": "My name is Slim Shady, call me at 313-666-7440 or email me at real.slim.shady@gmail.com"},
  ],
  langsmith_extra={"client": langsmith_client},
)

# 生成的追踪不会匿名化输入和输出
response_without_anonymization = openai_client.chat.completions.create(
  model="gpt-5.4-mini",
  messages=[
      {"role": "system", "content": "You are a helpful assistant."},
      {"role": "user", "content": "My name is Slim Shady, call me at 313-666-7440 or email me at real.slim.shady@gmail.com"},
  ],
)

# 创建操作 — 在运行开始时发送
{
    "id": "018f1b2c-...",
    "name": "my_llm_call",
    "run_type": "llm",
    "inputs": {"messages": [{"role": "user", "content": "My name is Jane Smith..."}]},
    "start_time": "2024-01-01T00:00:00.000Z",
    "trace_id": "018f1b2c-...",
    "dotted_order": "20240101T000000000000Z018f1b2c-...",
    "extra": {"metadata": {}, "runtime": {...}},
    "session_name": "default",
}

# 更新操作 — 在运行结束时发送（相同的 id，添加了 outputs）
{
    "id": "018f1b2c-...",
    "outputs": {"choices": [{"message": {"role": "assistant", "content": "Hello Jane..."}}]},
    "end_time": "2024-01-01T00:00:01.000Z",
    "trace_id": "018f1b2c-...",
    "dotted_order": "20240101T000000000000Z018f1b2c-...",
}

import boto3
from langsmith import Client, traceable

comprehend = boto3.client("comprehend", region_name="us-east-1")

def redact_entities(text: str, entities: list) -> str:
    for entity in sorted(entities, key=lambda e: e["BeginOffset"], reverse=True):
        placeholder = f"[{entity['Type']}]"
        text = text[:entity["BeginOffset"]] + placeholder + text[entity["EndOffset"]:]
    return text

def comprehend_anonymize_batch(runs: list[dict]) -> list[dict]:
    # 收集所有消息文本并记住它们来自哪里。
    # 注意：同一个运行 ID 可能会出现两次 — 一次作为创建（带有 inputs），
    # 另一次作为更新（带有 outputs）。
    locations = []  # (run_idx, field, msg_idx)
    texts = []
    for run_idx, run in enumerate(runs):
        for field in ("inputs", "outputs"):
            data = run.get(field)
            if not isinstance(data, dict):
                continue
            for msg_idx, message in enumerate(data.get("messages") or []):
                content = message.get("content", "")
                if content.strip():
                    locations.append((run_idx, field, msg_idx))
                    texts.append(content)

    # 将所有文本按 25 个一组（API 限制）发送给 Comprehend。
    # 对于 1000 个操作（约 500 次运行），每个有 2 条消息：进行 40 次 API 调用，而不是 1000 次。
    redacted_texts = []
    for i in range(0, len(texts), 25):
        chunk = texts[i : i + 25]
        response = comprehend.batch_detect_entities(
            TextList=chunk, LanguageCode="en"
        )
        for text, result in zip(chunk, response["ResultList"]):
            redacted_texts.append(redact_entities(text, result.get("Entities", [])))

    # 将脱敏后的文本写回运行字典
    for (run_idx, field, msg_idx), redacted in zip(locations, redacted_texts):
        runs[run_idx][field]["messages"][msg_idx]["content"] = redacted

    return runs

client = Client(
    process_buffered_run_ops=comprehend_anonymize_batch,
    run_ops_buffer_size=1000,        # 约 500 次被追踪的调用（每次 2 个操作：创建 + 更新）
    run_ops_buffer_timeout_ms=3000,  # 或者 3 秒后，以先到者为准
)

@traceable(client=client)
def my_llm_call(messages: list) -> dict:
    # ... 您的 LLM 调用 ...
    pass

try:
    my_llm_call([{"role": "user", "content": "My name is Jane Smith, call me at 555-867-5309"}])
finally:
    client.flush()  # 退出前务必刷新