22c - 数据隐私与合规

Agent 场景下的 PII 泄露路径

用户输入（含 PII）
       │
       ▼
┌──────────────┐     ┌──────────────┐     ┌──────────────┐
│  Agent 系统   │────▶│  LLM API     │────▶│ 模型提供商    │
│              │     │  (prompt)    │     │ 日志/训练?   │
└──────┬───────┘     └──────────────┘     └──────────────┘
       │
       ├──▶ MCP 工具调用（数据库查询、API 请求）
       │
       ├──▶ 日志系统（可能记录完整 prompt）
       │
       ├──▶ 可观测性平台（LangSmith/Langfuse trace）
       │
       └──▶ Agent 记忆/向量存储（长期保存）

每个节点都是潜在的 PII 泄露点！

# 安装 Presidio
pip install presidio-analyzer presidio-anonymizer
# 下载 spaCy 中文/英文模型
python -m spacy download en_core_web_lg
python -m spacy download zh_core_web_sm

# pii_detector.py — Agent PII 检测与脱敏管线
from presidio_analyzer import AnalyzerEngine, RecognizerRegistry
from presidio_analyzer.nlp_engine import NlpEngineProvider
from presidio_anonymizer import AnonymizerEngine
from presidio_anonymizer.entities import OperatorConfig
from dataclasses import dataclass
from typing import Optional
 
@dataclass
class PIIDetectionResult:
    original_text: str
    anonymized_text: str
    entities_found: list[dict]
    risk_score: float  # 0.0 - 1.0
 
class AgentPIIGuard:
    """Agent PII 防护层 — 在数据发送到 LLM 前检测并脱敏"""
 
    def __init__(self, languages: list[str] = None):
        self.languages = languages or ["en", "zh"]
        self.analyzer = AnalyzerEngine()
        self.anonymizer = AnonymizerEngine()
 
        # 定义需要检测的 PII 实体类型
        self.target_entities = [
            "PERSON",           # 人名
            "EMAIL_ADDRESS",    # 邮箱
            "PHONE_NUMBER",     # 电话号码
            "CREDIT_CARD",      # 信用卡号
            "IBAN_CODE",        # 银行账号
            "IP_ADDRESS",       # IP 地址
            "LOCATION",         # 地理位置
            "US_SSN",           # 美国社保号
            "US_PASSPORT",      # 护照号
            "MEDICAL_LICENSE",  # 医疗执照号
            "URL",              # URL（可能含敏感信息）
        ]
 
    def detect(self, text: str) -> PIIDetectionResult:
        """检测文本中的 PII 实体"""
        # 分析文本
        results = self.analyzer.analyze(
            text=text,
            entities=self.target_entities,
            language="en",  # Presidio 主要支持英文
            score_threshold=0.5,
        )
 
        # 计算风险分数
        risk_score = self._calculate_risk(results)
 
        # 提取检测到的实体信息
        entities = [
            {
                "type": r.entity_type,
                "text": text[r.start:r.end],
                "score": r.score,
                "start": r.start,
                "end": r.end,
            }
            for r in results
        ]
 
        # 执行脱敏
        anonymized = self.anonymizer.anonymize(
            text=text,
            analyzer_results=results,
            operators={
                "PERSON": OperatorConfig("replace",
                    {"new_value": "[姓名已脱敏]"}),
                "EMAIL_ADDRESS": OperatorConfig("replace",
                    {"new_value": "[邮箱已脱敏]"}),
                "PHONE_NUMBER": OperatorConfig("replace",
                    {"new_value": "[电话已脱敏]"}),
                "CREDIT_CARD": OperatorConfig("mask",
                    {"masking_char": "*", "chars_to_mask": 12,
                     "from_end": False}),
                "US_SSN": OperatorConfig("replace",
                    {"new_value": "[SSN已脱敏]"}),
                "IP_ADDRESS": OperatorConfig("replace",
                    {"new_value": "[IP已脱敏]"}),
                "DEFAULT": OperatorConfig("replace",
                    {"new_value": "[已脱敏]"}),
            },
        )
 
        return PIIDetectionResult(
            original_text=text,
            anonymized_text=anonymized.text,
            entities_found=entities,
            risk_score=risk_score,
        )
 
    def _calculate_risk(self, results) -> float:
        """基于检测到的实体计算风险分数"""
        if not results:
            return 0.0
 
        # 高敏感实体权重更高
        weights = {
            "CREDIT_CARD": 1.0,
            "US_SSN": 1.0,
            "MEDICAL_LICENSE": 0.9,
            "IBAN_CODE": 0.9,
            "US_PASSPORT": 0.9,
            "PERSON": 0.5,
            "EMAIL_ADDRESS": 0.6,
            "PHONE_NUMBER": 0.6,
            "IP_ADDRESS": 0.4,
            "LOCATION": 0.3,
            "URL": 0.2,
        }
 
        total_weight = sum(
            weights.get(r.entity_type, 0.3) * r.score
            for r in results
        )
        return min(total_weight / len(results), 1.0)
 
    def guard_prompt(self, prompt: str, threshold: float = 0.3) -> dict:
        """
        Agent prompt 防护 — 在发送到 LLM 前检查并处理 PII
 
        返回:
          - action: "allow" | "anonymize" | "block"
          - text: 处理后的文本
          - warning: 警告信息
        """
        result = self.detect(prompt)
 
        if result.risk_score == 0.0:
            return {
                "action": "allow",
                "text": prompt,
                "warning": None,
            }
        elif result.risk_score < threshold:
            return {
                "action": "allow",
                "text": prompt,
                "warning": f"检测到低风险 PII: "
                    f"{[e['type'] for e in result.entities_found]}",
            }
        else:
            return {
                "action": "anonymize",
                "text": result.anonymized_text,
                "warning": f"已脱敏 {len(result.entities_found)} 个 PII 实体 "
                    f"(风险分数: {result.risk_score:.2f})",
                "entities": result.entities_found,
            }
 
 
# 使用示例
guard = AgentPIIGuard()
 
# 模拟用户输入含 PII 的 prompt
user_prompt = """
请帮我分析这个客户的订单问题：
客户姓名：John Smith
邮箱：john.smith@example.com
电话：+1-555-123-4567
信用卡尾号：4532-1234-5678-9012
收货地址：123 Main St, New York, NY 10001
"""
 
result = guard.guard_prompt(user_prompt)
print(f"动作: {result['action']}")
print(f"警告: {result['warning']}")
print(f"处理后文本:\n{result['text']}")

# agent_pii_middleware.py — Agent PII 中间件
from typing import Callable, Any
 
class PIIMiddleware:
    """LLM 调用链 PII 中间件 — 拦截并处理含 PII 的请求"""
 
    def __init__(self, guard: AgentPIIGuard, policy: str = "anonymize"):
        """
        policy:
          - "anonymize": 自动脱敏后继续（默认）
          - "block": 检测到高风险 PII 时阻止请求
          - "warn": 仅警告，不修改内容
          - "log": 静默记录，不干预
        """
        self.guard = guard
        self.policy = policy
        self.audit_log: list[dict] = []
 
    async def process(
        self,
        prompt: str,
        llm_call: Callable,
        **kwargs,
    ) -> dict:
        """处理 Agent 的 LLM 调用请求"""
        # 1. 检测 PII
        detection = self.guard.guard_prompt(prompt)
 
        # 2. 记录审计日志
        self.audit_log.append({
            "action": detection["action"],
            "entities_count": len(detection.get("entities", [])),
            "risk_warning": detection["warning"],
            "policy": self.policy,
        })
 
        # 3. 根据策略处理
        if detection["action"] == "allow":
            # 无 PII 或低风险，直接通过
            return await llm_call(prompt, **kwargs)
 
        if self.policy == "block":
            return {
                "error": "请求被阻止：检测到敏感个人信息",
                "details": detection["warning"],
            }
 
        if self.policy == "anonymize":
            # 使用脱敏后的文本调用 LLM
            response = await llm_call(detection["text"], **kwargs)
            return response
 
        if self.policy == "warn":
            # 添加警告但不修改
            print(f"⚠️ PII 警告: {detection['warning']}")
            return await llm_call(prompt, **kwargs)
 
        # log 模式：静默通过
        return await llm_call(prompt, **kwargs)

# reversible_anonymizer.py — 可逆脱敏器
import hashlib
import secrets
from typing import Optional
 
class ReversibleAnonymizer:
    """
    可逆脱敏器 — 用确定性 token 替换 PII，
    LLM 处理后可恢复原始值。
 
    原理：
    1. 检测 PII → 生成唯一 token（如 <PERSON_a1b2c3>）
    2. 维护 token ↔ 原始值 的映射表（仅存内存/加密存储）
    3. 将 token 化文本发送给 LLM
    4. LLM 响应中的 token 替换回原始值
    """
 
    def __init__(self):
        self._mapping: dict[str, str] = {}  # token → 原始值
        self._reverse: dict[str, str] = {}  # 原始值 → token
        self._session_key = secrets.token_hex(16)
 
    def anonymize(self, text: str, entities: list[dict]) -> str:
        """将检测到的 PII 实体替换为可逆 token"""
        # 按位置从后往前替换（避免偏移问题）
        sorted_entities = sorted(
            entities, key=lambda e: e["start"], reverse=True
        )
 
        result = text
        for entity in sorted_entities:
            original = entity["text"]
            entity_type = entity["type"]
 
            # 检查是否已有映射（同一实体多次出现时保持一致）
            if original in self._reverse:
                token = self._reverse[original]
            else:
                # 生成确定性 token
                hash_input = f"{self._session_key}:{original}"
                short_hash = hashlib.sha256(
                    hash_input.encode()
                ).hexdigest()[:6]
                token = f"<{entity_type}_{short_hash}>"
 
                self._mapping[token] = original
                self._reverse[original] = token
 
            result = (
                result[:entity["start"]]
                + token
                + result[entity["end"]:]
            )
 
        return result
 
    def deanonymize(self, text: str) -> str:
        """将 LLM 响应中的 token 恢复为原始值"""
        result = text
        for token, original in self._mapping.items():
            result = result.replace(token, original)
        return result
 
    def clear_session(self):
        """清除当前会话的映射（安全最佳实践）"""
        self._mapping.clear()
        self._reverse.clear()
        self._session_key = secrets.token_hex(16)
 
 
# 使用示例
anonymizer = ReversibleAnonymizer()
 
# 原始文本
original = "请联系 John Smith (john@example.com) 确认订单"
 
# 假设 PII 检测结果
entities = [
    {"type": "PERSON", "text": "John Smith", "start": 4, "end": 14, "score": 0.9},
    {"type": "EMAIL_ADDRESS", "text": "john@example.com",
     "start": 16, "end": 32, "score": 0.95},
]
 
# 脱敏
anonymized = anonymizer.anonymize(original, entities)
print(f"脱敏后: {anonymized}")
# 输出: 请联系 <PERSON_a1b2c3> (<EMAIL_ADDRESS_d4e5f6>) 确认订单
 
# LLM 处理后的响应（包含 token）
llm_response = "已向 <PERSON_a1b2c3> 发送确认邮件至 <EMAIL_ADDRESS_d4e5f6>"
 
# 恢复
restored = anonymizer.deanonymize(llm_response)
print(f"恢复后: {restored}")
# 输出: 已向 John Smith 发送确认邮件至 john@example.com
 
# 会话结束后清除映射
anonymizer.clear_session()

你是一位数据隐私工程师。请根据以下 Agent 系统信息，设计 PII 检测与脱敏策略。

## Agent 系统信息
- Agent 类型：[客服Agent / 编码Agent / 数据分析Agent / 医疗Agent]
- 处理的数据类型：[客户对话 / 代码仓库 / 医疗记录 / 财务数据]
- 合规要求：[GDPR / HIPAA / SOC 2 / CCPA / 无特殊要求]
- 部署环境：[云端 / 本地 / 混合]
- 数据量级：[每日处理量]

## 请输出

### 1. PII 实体清单
列出该场景下需要检测的所有 PII 实体类型，按风险等级分类。

### 2. 脱敏策略
为每种实体类型推荐脱敏方法（替换/掩码/哈希/删除/可逆token化）。

### 3. 检测管线架构
设计从用户输入到 LLM 调用的完整 PII 检测管线。

### 4. 误报处理
说明如何处理误报（false positive）以避免影响 Agent 功能。

### 5. 审计与监控
设计 PII 检测的审计日志和监控告警方案。

AI Agent 数据流动的跨境风险

┌─────────────────────────────────────────────────────────┐
│                    你的基础设施                            │
│  ┌──────────┐    ┌──────────┐    ┌──────────┐           │
│  │ 用户数据  │───▶│ Agent    │───▶│ 向量数据库 │           │
│  │ (EU 区域) │    │ 服务器   │    │ (EU 区域) │           │
│  └──────────┘    └────┬─────┘    └──────────┘           │
│                       │                                  │
│                  EU 数据边界                               │
└───────────────────────┼─────────────────────────────────┘
                        │ API 调用（含用户数据）
                        ▼
              ┌──────────────────┐
              │  LLM 提供商       │
              │  推理服务器       │
              │  (美国? 未知?)   │  ← 数据可能跨境！
              └──────────────────┘
                        │
                        ▼
              ┌──────────────────┐
              │  提供商日志/缓存   │
              │  保留多久?        │  ← 数据保留政策不透明
              │  用于训练?        │
              └──────────────────┘

# data_flow_mapper.py — Agent 数据流映射工具
from dataclasses import dataclass, field
from enum import Enum
 
class DataClassification(Enum):
    PUBLIC = "public"
    INTERNAL = "internal"
    CONFIDENTIAL = "confidential"
    RESTRICTED = "restricted"      # PII、PHI、财务数据
    HIGHLY_RESTRICTED = "highly_restricted"  # 国家安全、关键基础设施
 
@dataclass
class DataFlowNode:
    name: str
    location: str          # 地理位置（如 "EU-Frankfurt", "US-East"）
    provider: str          # 服务提供商
    data_retention: str    # 数据保留期限
    encryption: str        # 加密方式
    compliance_certs: list[str] = field(default_factory=list)
 
@dataclass
class DataFlow:
    source: DataFlowNode
    destination: DataFlowNode
    data_types: list[str]
    classification: DataClassification
    crosses_border: bool
    legal_basis: str = ""  # 跨境传输的法律依据
 
class AgentDataFlowMapper:
    """Agent 数据流映射器 — 识别跨境数据传输风险"""
 
    def __init__(self):
        self.nodes: list[DataFlowNode] = []
        self.flows: list[DataFlow] = []
 
    def add_node(self, node: DataFlowNode):
        self.nodes.append(node)
 
    def add_flow(self, flow: DataFlow):
        self.flows.append(flow)
 
    def identify_risks(self) -> list[dict]:
        """识别数据驻留风险"""
        risks = []
        for flow in self.flows:
            if flow.crosses_border and flow.classification in [
                DataClassification.RESTRICTED,
                DataClassification.HIGHLY_RESTRICTED,
            ]:
                risks.append({
                    "severity": "HIGH",
                    "flow": f"{flow.source.name} → {flow.destination.name}",
                    "issue": (
                        f"受限数据跨境传输: "
                        f"{flow.source.location} → "
                        f"{flow.destination.location}"
                    ),
                    "data_types": flow.data_types,
                    "legal_basis": flow.legal_basis or "未指定",
                    "recommendation": self._get_recommendation(flow),
                })
        return risks
 
    def _get_recommendation(self, flow: DataFlow) -> str:
        if not flow.legal_basis:
            return (
                "需要建立跨境传输法律依据"
                "（标准合同条款/充分性认定/约束性公司规则）"
            )
        return "已有法律依据，建议定期审查有效性"
 
    def generate_report(self) -> str:
        risks = self.identify_risks()
        lines = ["=== Agent 数据流风险报告 ==="]
        lines.append(f"数据节点数: {len(self.nodes)}")
        lines.append(f"数据流数: {len(self.flows)}")
        lines.append(f"跨境风险数: {len(risks)}")
        for r in risks:
            lines.append(f"\n⚠️ [{r['severity']}] {r['flow']}")
            lines.append(f"   问题: {r['issue']}")
            lines.append(f"   数据类型: {r['data_types']}")
            lines.append(f"   法律依据: {r['legal_basis']}")
            lines.append(f"   建议: {r['recommendation']}")
        return "\n".join(lines)
 
 
# 使用示例：映射一个典型 Agent 系统的数据流
mapper = AgentDataFlowMapper()
 
# 定义节点
app_server = DataFlowNode(
    name="Agent 应用服务器",
    location="EU-Frankfurt",
    provider="AWS",
    data_retention="会话期间",
    encryption="AES-256 + TLS 1.3",
    compliance_certs=["SOC 2", "ISO 27001"],
)
 
openai_api = DataFlowNode(
    name="OpenAI API",
    location="US-East (推测)",
    provider="OpenAI",
    data_retention="API: 30天日志 / ZDR 可选",
    encryption="TLS 1.2+",
    compliance_certs=["SOC 2 Type II"],
)
 
vector_db = DataFlowNode(
    name="Pinecone 向量数据库",
    location="EU-Frankfurt (GCP)",
    provider="Pinecone",
    data_retention="持久化",
    encryption="AES-256",
    compliance_certs=["SOC 2 Type II"],
)
 
mapper.add_node(app_server)
mapper.add_node(openai_api)
mapper.add_node(vector_db)
 
# 定义数据流
mapper.add_flow(DataFlow(
    source=app_server,
    destination=openai_api,
    data_types=["用户查询", "客户姓名", "订单信息"],
    classification=DataClassification.RESTRICTED,
    crosses_border=True,  # EU → US
    legal_basis="",  # 缺少法律依据！
))
 
mapper.add_flow(DataFlow(
    source=app_server,
    destination=vector_db,
    data_types=["文档嵌入", "元数据"],
    classification=DataClassification.CONFIDENTIAL,
    crosses_border=False,  # 同区域
))
 
print(mapper.generate_report())

数据驻留策略决策树

你的 Agent 处理的数据是否包含 PII/PHI？
│
├── 否 → 标准部署，无特殊要求
│
└── 是 → 数据受哪些法规管辖？
    │
    ├── 仅 GDPR → 方案 A: EU 区域部署 + SCCs
    │
    ├── GDPR + HIPAA → 方案 B: EU 区域 + BAA + 加密
    │
    ├── 中国《个保法》→ 方案 C: 境内部署 + 国产模型
    │
    └── 多法规交叉 → 方案 D: 本地推理 + 数据不出境

选择 LLM 提供商的数据隐私决策框架

你的数据敏感度？
│
├── 低（公开信息、通用查询）
│   └── 任何提供商均可，优先考虑性价比
│
├── 中（内部文档、业务数据）
│   └── 使用 API 版本（非消费者版）
│       ├── 确认 API 数据不用于训练
│       └── 启用日志最小化
│
├── 高（PII、客户数据、财务数据）
│   └── Enterprise 版本 + DPA
│       ├── 签署数据处理协议 (DPA)
│       ├── 启用零数据保留 (ZDR)
│       ├── 确认数据驻留区域
│       └── 或使用 AWS Bedrock / Azure OpenAI（数据不离开你的云）
│
└── 极高（PHI、国防、关键基础设施）
    └── 自托管模型 或 专用云实例
        ├── 开源模型（Llama 3、Qwen 2.5、Mistral）
        ├── 私有 VPC 部署
        ├── 端到端加密 + CMK
        └── 完整审计日志

# llm-provider-privacy-checklist.yaml
provider_name: ""
review_date: ""
reviewer: ""
 
data_handling:
  - question: "API 调用数据是否用于模型训练？"
    answer: ""
    evidence: ""  # 链接到官方文档
 
  - question: "数据保留期限是多久？"
    answer: ""
    evidence: ""
 
  - question: "是否支持零数据保留 (ZDR)？"
    answer: ""
    evidence: ""
 
  - question: "是否有人工审查 prompt/响应的可能？"
    answer: ""
    evidence: ""
 
  - question: "数据在传输中和静态时的加密方式？"
    answer: ""
    evidence: ""
 
compliance:
  - question: "是否持有 SOC 2 Type II 认证？"
    answer: ""
    evidence: ""
 
  - question: "是否支持签署 DPA (GDPR)？"
    answer: ""
    evidence: ""
 
  - question: "是否支持签署 BAA (HIPAA)？"
    answer: ""
    evidence: ""
 
  - question: "数据处理的地理位置在哪里？"
    answer: ""
    evidence: ""
 
  - question: "是否支持客户管理的加密密钥 (CMK)？"
    answer: ""
    evidence: ""
 
incident_response:
  - question: "数据泄露通知时限是多久？"
    answer: ""
    evidence: ""
 
  - question: "是否有专门的安全事件响应团队？"
    answer: ""
    evidence: ""
 
  - question: "历史安全事件记录？"
    answer: ""
    evidence: ""

// llm_privacy_config.ts — LLM API 调用隐私配置
interface LLMPrivacyConfig {
  provider: 'openai' | 'anthropic' | 'google' | 'bedrock' | 'azure';
  dataRetention: 'default' | 'minimal' | 'zero';
  region?: string;
  encryptionKeyId?: string;  // 客户管理密钥
  auditLogging: boolean;
  piiPreprocessing: boolean;
}
 
class PrivacyAwareLLMClient {
  private config: LLMPrivacyConfig;
 
  constructor(config: LLMPrivacyConfig) {
    this.config = config;
    this.validateConfig();
  }
 
  private validateConfig(): void {
    // 验证配置是否满足合规要求
    if (this.config.dataRetention === 'zero'
        && !['bedrock', 'azure'].includes(this.config.provider)) {
      console.warn(
        '⚠️ 零数据保留可能需要 Enterprise 合同，'
        + '请确认提供商支持'
      );
    }
 
    if (this.config.region
        && !this.isSupportedRegion(this.config.region)) {
      throw new Error(
        `提供商 ${this.config.provider} `
        + `不支持区域 ${this.config.region}`
      );
    }
  }
 
  private isSupportedRegion(region: string): boolean {
    const regionSupport: Record<string, string[]> = {
      openai: ['us', 'eu'],
      anthropic: ['us'],
      google: ['us', 'eu', 'asia'],
      bedrock: [
        'us-east-1', 'us-west-2', 'eu-west-1',
        'eu-central-1', 'ap-northeast-1',
      ],
      azure: [
        'eastus', 'westeurope', 'swedencentral',
        'japaneast', 'australiaeast',
      ],
    };
    return regionSupport[this.config.provider]?.some(
      r => region.includes(r)
    ) ?? false;
  }
 
  async call(prompt: string): Promise<string> {
    // 1. PII 预处理（如果启用）
    let processedPrompt = prompt;
    if (this.config.piiPreprocessing) {
      processedPrompt = await this.preprocessPII(prompt);
    }
 
    // 2. 添加隐私相关 headers
    const headers = this.getPrivacyHeaders();
 
    // 3. 调用 API
    const response = await this.callProvider(processedPrompt, headers);
 
    // 4. 审计日志
    if (this.config.auditLogging) {
      await this.logAudit(prompt, response);
    }
 
    return response;
  }
 
  private getPrivacyHeaders(): Record<string, string> {
    const headers: Record<string, string> = {};
 
    // OpenAI: 禁止数据用于训练的 header
    if (this.config.provider === 'openai') {
      // OpenAI API 默认不用于训练，但可以显式声明
      headers['X-Data-Retention'] = this.config.dataRetention;
    }
 
    return headers;
  }
 
  private async preprocessPII(text: string): Promise<string> {
    // 调用 PII 检测服务
    // 实际实现中调用 Presidio 或云端 DLP API
    return text; // 占位
  }
 
  private async callProvider(
    prompt: string,
    headers: Record<string, string>,
  ): Promise<string> {
    // 实际 API 调用逻辑
    return ''; // 占位
  }
 
  private async logAudit(
    prompt: string,
    response: string,
  ): Promise<void> {
    // 记录审计日志（不记录实际内容，只记录元数据）
    const auditEntry = {
      timestamp: new Date().toISOString(),
      provider: this.config.provider,
      region: this.config.region,
      promptLength: prompt.length,
      responseLength: response.length,
      piiPreprocessed: this.config.piiPreprocessing,
      dataRetention: this.config.dataRetention,
    };
    console.log('[AUDIT]', JSON.stringify(auditEntry));
  }
}

✅ GDPR AI Agent 合规清单

□ 法律基础
  □ 确定数据处理的法律依据（同意/合法利益/合同必要）
  □ 如使用同意，实现可撤回的同意机制
  □ 记录每项数据处理活动的法律依据

□ 透明性
  □ 隐私政策中披露 AI/Agent 的使用
  □ 告知用户数据将发送给哪些第三方（LLM 提供商）
  □ 说明自动化决策的逻辑（GDPR 第 22 条）
  □ 提供人工审查自动化决策的选项

□ 数据最小化
  □ Agent prompt 中只包含必要数据
  □ 实施 PII 检测和脱敏管线
  □ 定期审查 Agent 收集的数据范围

□ 数据主体权利
  □ 实现数据访问请求 (SAR) 处理流程
  □ 实现数据删除请求（"被遗忘权"）处理
  □ 实现数据可携带性导出
  □ Agent 记忆/向量存储中的数据也必须可删除

□ 跨境传输
  □ 确认 LLM 提供商的数据处理位置
  □ 签署标准合同条款 (SCCs) 或确认充分性认定
  □ 签署数据处理协议 (DPA)
  □ 进行传输影响评估 (TIA)

□ 数据保护影响评估 (DPIA)
  □ 对高风险 Agent 应用执行 DPIA
  □ 评估 Agent 自动化决策对个人的影响
  □ 记录风险缓解措施

□ 数据泄露响应
  □ 建立 72 小时内通知监管机构的流程
  □ 建立通知受影响数据主体的流程
  □ 定期演练数据泄露响应

✅ SOC 2 AI Agent 合规清单

□ 安全性 (Security) — 必选
  □ Agent API 端点实施认证和授权
  □ Agent 与 LLM 提供商之间的通信加密 (TLS 1.2+)
  □ 静态数据加密 (AES-256)
  □ Agent 执行环境沙箱隔离
  □ 定期漏洞扫描和渗透测试
  □ 安全事件检测和响应流程
  □ 密钥轮换策略（API keys、加密密钥）

□ 可用性 (Availability)
  □ Agent 服务 SLA 定义和监控
  □ LLM 提供商故障时的降级策略
  □ 自动故障转移和恢复机制
  □ 容量规划和自动扩缩容
  □ 灾难恢复计划和定期演练

□ 处理完整性 (Processing Integrity)
  □ Agent 输出验证机制（格式、范围、合理性）
  □ 幻觉检测和事实核查流程
  □ 完整的请求-响应审计追踪
  □ 数据处理错误检测和纠正
  □ 变更管理流程（Agent 配置、prompt 模板）

□ 保密性 (Confidentiality)
  □ 数据分类策略（公开/内部/机密/受限）
  □ 基于分类的访问控制
  □ Agent 日志中的敏感数据脱敏
  □ 可观测性平台中的数据保护
  □ 数据保留和销毁策略

□ 隐私性 (Privacy)
  □ 隐私声明覆盖 Agent 数据处理
  □ PII 检测和脱敏管线
  □ 数据主体权利请求处理流程
  □ 第三方数据处理协议（LLM 提供商 DPA）
  □ 隐私影响评估

✅ HIPAA AI Agent 合规清单

□ 业务伙伴协议 (BAA)
  □ 与 LLM 提供商签署 BAA
  □ 与所有处理 PHI 的第三方签署 BAA
  □ BAA 中明确 AI 特定条款（数据训练禁止、保留期限）
  □ 定期审查 BAA 合规性

□ 技术保障措施
  □ Agent 系统实施访问控制（唯一用户 ID + MFA）
  □ ePHI 传输加密 (TLS 1.2+)
  □ ePHI 静态加密 (AES-256)
  □ 审计日志记录所有 PHI 访问
  □ 自动会话超时和注销
  □ Agent 记忆中的 PHI 自动过期清理

□ PHI 数据处理
  □ 实施 PHI 检测管线（姓名、MRN、诊断、处方等）
  □ 发送到 LLM 前对 PHI 进行脱敏/去标识化
  □ 使用 HIPAA Safe Harbor 或 Expert Determination 去标识化方法
  □ Agent 不在日志中记录完整 PHI
  □ 可观测性平台中的 PHI 脱敏

□ 管理保障措施
  □ 指定 HIPAA 隐私官和安全官
  □ Agent 系统风险评估（年度）
  □ 员工 HIPAA 培训（含 AI 使用专项）
  □ Agent 使用政策和程序文档化
  □ 事件响应计划（含 AI 特定场景）

□ 物理保障措施
  □ Agent 服务器物理访问控制
  □ 工作站安全（开发/运维人员）
  □ 设备和介质控制

□ 违规通知
  □ 60 天内通知 HHS（影响 500+ 人）
  □ 通知受影响个人
  □ 通知当地媒体（影响 500+ 人的州级违规）
  □ 维护违规日志

EU AI Act 风险金字塔

          ┌───────────┐
          │  禁止的    │  社会评分、实时远程生物识别（执法例外）
          │ AI 系统    │  操纵性 AI、利用弱势群体
          └─────┬─────┘  → 2025.2 起禁止
                │
        ┌───────┴───────┐
        │   高风险       │  信用评分、招聘筛选、医疗诊断
        │   AI 系统      │  关键基础设施、执法、教育
        └───────┬───────┘  → 2026.8 前必须合规
                │
    ┌───────────┴───────────┐
    │     有限风险            │  聊天机器人、深度伪造
    │     AI 系统            │  情感识别、生物分类
    └───────────┬───────────┘  → 透明性义务
                │
┌───────────────┴───────────────┐
│          最小风险               │  垃圾邮件过滤、游戏 AI
│          AI 系统               │  推荐系统（非高风险）
└───────────────────────────────┘  → 自愿行为准则

✅ EU AI Act AI Agent 合规清单

□ 风险分类
  □ 确定 Agent 系统的风险等级（禁止/高/有限/最小）
  □ 如果是高风险，确认属于 Annex III 的哪个类别
  □ 记录风险分类的依据和理由

□ 通用 AI 模型 (GPAI) 义务（如使用第三方基础模型）
  □ 确认 LLM 提供商是否已履行 GPAI 义务
  □ 获取提供商的技术文档和模型卡
  □ 确认提供商的版权合规声明
  □ 评估系统性风险（如使用 >10^25 FLOP 训练的模型）

□ 高风险系统要求（如适用）
  □ 建立风险管理系统（全生命周期）
  □ 数据治理：训练/验证/测试数据的质量管理
  □ 技术文档：系统设计、开发、测试的完整记录
  □ 日志记录：自动记录系统运行日志
  □ 透明性：向部署者提供使用说明
  □ 人工监督：确保人类可以有效监督系统
  □ 准确性、鲁棒性、网络安全措施
  □ 合格评定（自评或第三方）
  □ CE 标志
  □ EU 数据库注册

□ 有限风险系统要求（如适用）
  □ 告知用户正在与 AI 系统交互（聊天机器人标识）
  □ 标记 AI 生成的内容（深度伪造标识）
  □ 情感识别/生物分类系统的告知义务

□ 透明性义务（所有 Agent）
  □ 用户知道自己在与 AI Agent 交互
  □ AI 生成的内容有适当标识
  □ 提供 Agent 能力和限制的说明

□ 持续合规
  □ 建立上市后监控系统
  □ 严重事件报告机制
  □ 定期审查和更新合规措施

你的 Agent 需要遵守哪些合规框架？

你的用户/客户在哪里？
│
├── 欧盟/EEA → GDPR（必须）+ EU AI Act（必须）
│
├── 美国 → SOC 2（B2B 几乎必须）
│   │
│   ├── 处理健康数据？ → + HIPAA
│   ├── 处理金融数据？ → + SOX / PCI DSS
│   └── 加州用户？ → + CCPA/CPRA
│
├── 中国 → 《个保法》+《数据安全法》+《AI 管理办法》
│
├── 全球 → GDPR + SOC 2 + 目标市场法规
│
└── 不确定 → 至少实施 SOC 2 + GDPR（覆盖面最广）

合规客服 Agent 架构

┌─────────────────────────────────────────────────────────┐
│                    用户交互层                              │
│  ┌──────────────────────────────────────────────────┐   │
│  │  💬 聊天界面                                      │   │
│  │  • AI 交互标识："您正在与 AI 助手对话"              │   │
│  │  • 同意横幅：数据处理同意                          │   │
│  │  • 人工转接按钮                                   │   │
│  └──────────────────────────────────────────────────┘   │
└─────────────────────────┬───────────────────────────────┘
                          │
┌─────────────────────────▼───────────────────────────────┐
│                    PII 防护层                              │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐              │
│  │ PII 检测  │→│ 风险评估  │→│ 脱敏处理  │              │
│  │ Presidio │  │ 分级判断  │  │ 可逆token │              │
│  └──────────┘  └──────────┘  └──────────┘              │
│  • 检测：姓名、邮箱、电话、地址、支付信息                   │
│  • 策略：高风险 PII 脱敏，低风险 PII 标记                  │
└─────────────────────────┬───────────────────────────────┘
                          │
┌─────────────────────────▼───────────────────────────────┐
│                    Agent 核心层                            │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐              │
│  │ LLM 调用  │  │ 工具调用  │  │ 记忆管理  │              │
│  │ (脱敏后)  │  │ (权限控制)│  │ (加密+TTL)│              │
│  └──────────┘  └──────────┘  └──────────┘              │
│  • LLM: Azure OpenAI (EU 区域, ZDR)                     │
│  • 工具: 只读数据库访问 + 参数化查询                       │
│  • 记忆: 会话记忆 24h TTL, 加密存储                       │
└─────────────────────────┬───────────────────────────────┘
                          │
┌─────────────────────────▼───────────────────────────────┐
│                    审计与监控层                             │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐              │
│  │ 审计日志  │  │ 合规监控  │  │ 告警系统  │              │
│  │ (脱敏)   │  │ (自动检查)│  │ (实时)   │              │
│  └──────────┘  └──────────┘  └──────────┘              │
│  • 日志: 不记录原始 PII，只记录脱敏后的元数据               │
│  • 监控: PII 泄露检测、异常访问模式                        │
│  • 告警: 高风险操作实时通知                                │
└─────────────────────────────────────────────────────────┘

# compliant_agent.py — 合规客服 Agent 核心实现
from dataclasses import dataclass
from typing import Optional
import time
 
@dataclass
class ComplianceConfig:
    gdpr_enabled: bool = True
    soc2_audit_logging: bool = True
    pii_anonymization: bool = True
    data_retention_hours: int = 24
    llm_provider: str = "azure_openai"
    llm_region: str = "swedencentral"  # EU 区域
    zero_data_retention: bool = True
 
class CompliantCustomerAgent:
    """合规客服 Agent — 满足 GDPR + SOC 2 要求"""
 
    def __init__(self, config: ComplianceConfig):
        self.config = config
        self.pii_guard = AgentPIIGuard()
        self.anonymizer = ReversibleAnonymizer()
        self.audit_logger = AuditLogger(enabled=config.soc2_audit_logging)
 
    async def handle_query(
        self,
        user_id: str,
        query: str,
        consent_given: bool,
    ) -> dict:
        """处理客户查询（合规流程）"""
 
        # 1. GDPR: 检查同意
        if self.config.gdpr_enabled and not consent_given:
            return {
                "response": "在处理您的请求前，"
                    "我们需要您同意我们的数据处理政策。",
                "requires_consent": True,
            }
 
        # 2. PII 检测与脱敏
        pii_result = self.pii_guard.detect(query)
        processed_query = query
 
        if pii_result.entities_found and self.config.pii_anonymization:
            processed_query = self.anonymizer.anonymize(
                query, pii_result.entities_found
            )
            self.audit_logger.log(
                "pii_detected",
                user_id=user_id,
                entity_count=len(pii_result.entities_found),
                entity_types=[
                    e["type"] for e in pii_result.entities_found
                ],
                risk_score=pii_result.risk_score,
            )
 
        # 3. 调用 LLM（使用脱敏后的查询）
        llm_response = await self._call_llm(processed_query)
 
        # 4. 恢复脱敏内容（如果需要在回复中包含客户信息）
        final_response = self.anonymizer.deanonymize(llm_response)
 
        # 5. SOC 2: 审计日志
        self.audit_logger.log(
            "query_processed",
            user_id=user_id,
            query_length=len(query),
            pii_anonymized=bool(pii_result.entities_found),
            llm_provider=self.config.llm_provider,
            llm_region=self.config.llm_region,
        )
 
        # 6. 清理会话数据
        self.anonymizer.clear_session()
 
        return {
            "response": final_response,
            "metadata": {
                "ai_generated": True,  # EU AI Act 透明性
                "pii_processed": bool(pii_result.entities_found),
                "data_region": self.config.llm_region,
            },
        }
 
    async def handle_data_deletion_request(
        self, user_id: str
    ) -> dict:
        """GDPR 第 17 条：处理数据删除请求（被遗忘权）"""
        deleted_items = []
 
        # 1. 删除会话历史
        await self._delete_session_history(user_id)
        deleted_items.append("会话历史")
 
        # 2. 删除向量存储中的用户数据
        await self._delete_vector_embeddings(user_id)
        deleted_items.append("向量嵌入")
 
        # 3. 删除审计日志中的可识别数据
        # 注意：审计日志可能有法律保留要求
        await self._anonymize_audit_logs(user_id)
        deleted_items.append("审计日志（已匿名化）")
 
        # 4. 记录删除操作
        self.audit_logger.log(
            "data_deletion_request",
            user_id="[已删除]",  # 不再记录用户 ID
            deleted_items=deleted_items,
        )
 
        return {
            "status": "completed",
            "deleted_items": deleted_items,
            "confirmation_id": f"DEL-{int(time.time())}",
        }
 
    async def _call_llm(self, query: str) -> str:
        """调用 LLM（占位实现）"""
        return f"[LLM 响应: {query[:50]}...]"
 
    async def _delete_session_history(self, user_id: str):
        pass  # 实际实现：删除数据库中的会话记录
 
    async def _delete_vector_embeddings(self, user_id: str):
        pass  # 实际实现：从向量数据库中删除用户相关嵌入
 
    async def _anonymize_audit_logs(self, user_id: str):
        pass  # 实际实现：将审计日志中的用户 ID 替换为哈希
 
 
class AuditLogger:
    """SOC 2 合规审计日志器"""
 
    def __init__(self, enabled: bool = True):
        self.enabled = enabled
 
    def log(self, event_type: str, **kwargs):
        if not self.enabled:
            return
        entry = {
            "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
            "event_type": event_type,
            **kwargs,
        }
        # 实际实现：写入不可变审计日志存储
        print(f"[AUDIT] {entry}")