观测 Harness：trace 里该看见什么，不该记录什么

在大模型应用和 Agent 走向生产环境的过程中，可观测性（Observability）往往会和安全隐私发生剧烈冲突。一方面，为了排查模型幻觉、工具调用失败和多轮对话逻辑漂移，我们需要知道模型接收了什么，输出了什么，以及中间调用了哪些外部 API；另一方面，未经处理的原始输入和输出包含了大量敏感的用户数据（PII），甚至是组织的核心商业机密。一旦这些数据被明文记录进集中的 Trace 日志系统，就会直接违反数据合规要求，并显著增加数据泄露的风险。

本指南将详细介绍如何构建一个具备隐私边界的观测 Harness（Observability Harness）。我们将明确 trace 管道中必须捕获的元数据、必须脱敏的敏感边界，以及如何通过这些指标完成闭环的异常排查。

来源说明与时效复核

本单元的工程设计与安全原则基于以下公开技术规范与框架，编写于 2026 年 5 月 28 日：

• OWASP Top 10 for LLM Applications (2025)：特别是有关 LLM01（提示注入）、LLM02（敏感信息泄露）和 LLM08（过度代理）的安全防御机制要求。
• OpenAI Function calling 规范：针对多工具调用（Tool Call）中的参数序列化和执行状态捕获。
• Amazon Bedrock Guardrails 开发者指南：针对输入/输出策略、有害内容过滤及个人身份信息（PII）遮蔽的工程实践。
• Microsoft Azure AI Foundry Risk and safety evaluators：针对跨域提示注入（XPIA）和质量缺陷率（Defect Rate）的定量评估逻辑。
• NIST AI Risk Management Framework (AI RMF)：在“测度（Measure）”与“管理（Manage）”维度上对 AI 系统运行状态的风险量化要求。

由于模型能力和合规监管标准的快速迭代，在遇到新类型的个人信息定义或全新的提示攻击手段时，必须重新复核脱敏正则规则库与安全评估器的拦截基准。

分层观测架构：从 LLM 调用到 Tool Execution 的 Trace 链路

要把大模型应用的执行过程说清楚，不能只把 LLM 当成一个黑盒函数。一个典型的 Agent 运行 trace 应该被视作一个多层嵌套树。标准的 OpenTelemetry（OTel）或类似观测工具需要将其拆解为以下核心 Span 分层：

1. Workflow/Agent Root Span：表示用户发起的一次完整会话请求，包含会话 ID、用户 ID 和最终的业务响应结果。
2. RAG Retrieval Span：表示从向量数据库或知识库检索上下文的阶段，包含查询向量化、检索到的 Document ID 以及召回相似度评分。
3. LLM Chat Completion Span：单次或多次与模型 API 交互的阶段，记录模型型号、推理参数（如 Temperature）、延迟和 Token 消耗明细。
4. Tool Call Execution Span：模型决定调用外部工具的阶段。这里不仅要记录模型期望调用的函数名，还要记录应用端实际执行该工具的起止时间、网络延迟和返回码。

根据 OpenAI 的 Function calling 文档说明，模型通过生成符合特定 JSON Schema 的参数来触发工具调用。因此，在 Tool Call Span 中，必须显式区分两个阶段：

• 模型输出阶段：模型生成的 arguments 字符串（例如 {"location": "Beijing", "unit": "celsius"}）。
• 本地执行阶段：应用层接收此参数，调度对应的本地 Python/JS 函数执行，并将执行结果（或捕获的异常错误）反哺给模型的 tool_response 阶段。

在设计观测 Harness 时，应当在 Trace 管道中对这些不同层级的 Span 注入统一的上下文 ID（Context Propagation），使其能在一个 Trace 图谱中被完美关联。否则，当线上出现由于工具超时导致的 Agent 挂起时，你将无法分辨究竟是模型卡在了幻觉循环中，还是下游 API 发生了硬性超时。

暴露与遮蔽的边界：构建符合隐私合规的 Payload 脱敏 Harness

在享受深度 Trace 带来的排错便利时，必须严格遵守隐私边界。根据 OWASP Top 10 中的 LLM02（敏感信息泄露）漏洞防御指南，直接存储未经脱敏的用户 Prompt 和检索上下文，极易导致下游日志分析人员或外部审计者无意中接触到个人敏感数据。

我们必须明确制定哪些数据可以原样记录，哪些数据必须经过本地拦截和脱敏：

如果系统检测到输入中包含高度敏感的个人识别信息（PII），应该在进入 trace 日志之前执行就地脱敏（Masking），不要依赖后置的安全过滤器，因为一旦未脱敏的 PII 被明文写入持久化日志，就会引发严重的合规性风险，只有在极少数经过隔离且加密的密闭审计环境中才能保留完整原文。

以下是一个用于观测 Harness 的本地脱敏器（Masker）示例代码。它在 Trace 数据发送至远程 Collector 之前拦截并对敏感词、身份证号、邮箱等进行遮蔽：

import re
import json
from typing import Dict, Any, Union

class TraceMasker:
    def __init__(self):
        # 基础 PII 正则表达式（示例，可扩展）
        self.patterns = {
            "email": re.compile(r'[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}'),
            "phone": re.compile(r'\b(?:\+86)?1[3-9]\d{9}\b'), # 简易中国手机号正则
            "id_card": re.compile(r'\b\d{17}[0-9Xx]\b') # 简易身份证正则
        }
        # 敏感工具参数 Key 遮蔽规则
        self.sensitive_keys = {"password", "token", "secret", "credit_card", "api_key"}

    def mask_text(self, text: str) -> str:
        if not isinstance(text, str):
            return text
        
        masked = text
        for pii_type, pattern in self.patterns.items():
            masked = pattern.sub(f"[MASKED_{pii_type.upper()}]", masked)
        return masked

    def mask_json_payload(self, data: Union[Dict, list, str]) -> Any:
        if isinstance(data, dict):
            masked_dict = {}
            for k, v in data.items():
                if k.lower() in self.sensitive_keys:
                    masked_dict[k] = "[MASKED_SENSITIVE_KEY]"
                else:
                    masked_dict[k] = self.mask_json_payload(v)
            return masked_dict
        elif isinstance(data, list):
            return [self.mask_json_payload(item) for item in data]
        elif isinstance(data, str):
            return self.mask_text(data)
        return data

# 验证脱敏逻辑的测试用例
if __name__ == "__main__":
    masker = TraceMasker()
    
    # 测试用例 1：带敏感信息的 Prompt
    raw_prompt = "你好，我的邮箱是 test_user@example.com，电话是 13800138000。请帮我查询。"
    print("Masked Prompt:", masker.mask_text(raw_prompt))
    # 预期输出: 你好，我的邮箱是 [MASKED_EMAIL]，电话是 [MASKED_PHONE]。请帮我查询。
    
    # 测试用例 2：带敏感 Key 的 Function Call 参数
    raw_arguments = {
        "user_id": "usr_9982",
        "api_key": "sk-live-abcdef123456",
        "location": "Beijing",
        "details": {
            "email": "admin@company.com"
        }
    }
    masked_args = masker.mask_json_payload(raw_arguments)
    print("Masked Arguments:", json.dumps(masked_args, indent=2))

安全边界与敏感词拦截：将 Guardrails 遥测深度集成至 Observability 管道

除了被动的文本脱敏，生产环境还需要主动的安全防御与策略拦截。根据 Amazon Bedrock Guardrails 的核心机制，输入和输出策略检测需要在模型推理的边界处，对有害内容（Harmful Content）、非法言论以及注入式攻击（Prompt Attacks）进行拦截。

当 Guardrails 拦截发生时，应用层通常会返回一个通用的替代安全响应（如 "I cannot assist with this request."）。如果此时你的 Trace 仅仅记录了这个通用的替代文本，那么安全运维人员就永远无法还原拦截的根因：究竟是用户的输入触发了“自我伤害（Self-Harm）”防护，还是触发了敏感机构 PII 检测？

因此，观测 Harness 必须支持记录 Guardrail 评估器的遥测详情（Telemetry Details）。当拦截发生时，应该在 Trace Span 的 attributes 中附加以下标准化的评估元数据，而不需要把有害的原始 Payload 写入日志文本中：

• guardrail.action: INTERCEPTED | PASSED
• guardrail.rule_set: harmful-content-v2 | pii-mask-v1
• guardrail.triggered_categories: ["HATE_SPEECH", "PROMPT_INJECTION"]
• guardrail.confidence_score: 0.95

通过这种结构化元数据的记录方式，安全团队能够直接在 APM 看板上绘制出“阻断率走势图”与“攻击分类热力图”，同时完全不向日志存储端暴露任何有害或敏感的明文文本。

动态风险量化：在 Trace 管道中埋入安全与质量评测器

要让线上质量退化和合规风险无处遁形，企业需要参考 NIST AI Risk Management Framework（AI RMF）中的“测度（Measure）”准则，将实时评估（Online Evaluators）作为观测 Harness 的核心组件插件化运行。

结合 Microsoft Azure AI Foundry 中提供的 Risk and safety evaluators 的评估设计思路，我们需要在 Trace 收集器（Collector）的处理器中运行轻量级的流式评估。例如，针对以下两类生产环境中的致命风险：

1. XPIA (跨域提示注入攻击 / Cross-Domain Prompt Injection)：当 RAG 检索到不受信任的第三方网页或文档时，这些文档可能包含欺骗模型的恶意指令。通过评估器，我们可以检测召回文本中是否含有特定的高风险命令关键字。
2. 敏感数据泄露缺陷率 (Defect Rate)：线上运行中，模型吐出不应展示的数据的会话比例。

为了将此类评估嵌入 Trace，我们可以设计一个中继处理器（Span Processor），在 trace 刷盘前，自动对其计算安全评分并打上标签：

def evaluate_safety_and_quality(span_data: dict) -> dict:
    """
    对 Trace 中的 llm_response 进行轻量级安全与质量评分评估
    """
    response_text = span_data.get("llm_response", "")
    
    # 简易敏感信息泄露检测 (Defect Evaluator)
    # 依据 Azure AI Foundry 逻辑：检测输出内容中是否包含未屏蔽的敏感格式
    leaked_pii_detected = False
    if re.search(r'\b\d{18}\b', response_text): # 检测到 18 位身份证未被 Mask
        leaked_pii_detected = True
        
    # 计算安全质量属性
    span_data["attributes"]["eval.pii_leak.defect"] = leaked_pii_detected
    span_data["attributes"]["eval.xpia.risk_score"] = estimate_xpia_risk(span_data.get("rag_context", ""))
    
    return span_data

def estimate_xpia_risk(context_text: str) -> float:
    # 检测上下文中是否存在指示性词汇（例如 "ignore previous instructions"）以粗略评估 XPIA 注入风险
    suspicious_patterns = ["ignore previous", "system override", "replace instruction"]
    matches = sum(1 for pattern in suspicious_patterns if pattern in context_text.lower())
    return min(1.0, matches * 0.5)

如果在 Trace 中无法分离系统提示词（System Prompt）、用户提示词（User Prompt）和上下文检索（RAG Retrieve Context），就无法针对性地计算注入攻击或泄露的缺陷率（defect rate）。因此，Trace 必须支持细粒度的字段切分。

延迟与成本归因：Token 消耗与 Tool 执行周期的分段计量

在生产环境中，可观测性不仅关乎安全，还直接决定了系统的成本控制与用户体验优化。一个复杂的 Agent 在完成一次业务请求时，可能在后台多次循环调用大模型和外部工具。如果只观测整体延迟，当接口响应慢时，团队无法判定到底是模型推理本身太慢，还是下游 RAG 检索、三方 API 调用拖了后腿。

因此，观测 Harness 必须严格记录以下两个维度的分段数据：

1. Token 消耗度量

在 LLM Span 的元数据中，必须无条件记录 usage 对象。这是进行成本归因和异常流控的根本依据：

• usage.prompt_tokens：输入提示词消耗的 Token。
• usage.completion_tokens：模型生成的 Token 数量。
• usage.total_tokens：单次总消耗。
• usage.cached_prompt_tokens：（如果模型提供商支持）命中的 Prompt Cache 的 Token 数。这对于 RAG 应用降本至关重要。

2. 精确的延迟分段归因

我们必须通过父子 Span 机制将一次完整请求的耗时（例如 5200ms）分摊到具体环节：

• 模型首字延迟 (Time to First Token - TTFT)：在流式（Streaming）输出时，首个 Token 返回的延迟直接决定了用户感知的流畅度。此指标必须单独记录在 LLM Span 的元数据中。
• 工具执行时间 (Tool Duration)：工具在本地或微服务中运行的持续时间。通过比对 llm_span.duration 与 tool_span.duration，分析模型等待外部 IO 响应的时间占比。

闭环异常排查：基于 Trace 判定 System Failures 与 Agent Drift

线上发生异常时，排查路径需要区分系统级错误（System Failures）和智能体能力漂移/幻觉（Agent Drift）。通过结构化的 Trace，我们可以根据清晰的链路逻辑进行定位：

异常场景一：工具调用参数不满足 Schema 要求

• 现象：模型生成的 Function call 在本地应用层执行时抛出 ValidationError。
• 定位路径：
  1. 检查 LLM Span 的 tool_calls[0].arguments 原始输出。如果调用 OpenAI 的 Function calling 时工具执行返回了空值或格式错误的错误信息，不应该仅记录“API调用失败”，而必须在 trace span 的元数据中完整捕获工具返回的原始报错和当前大模型的 retry 次数，否则在后期追溯 Agent 执行逻辑时，开发人员将无法分清是模型幻觉（未按 schema 生成）还是工具服务本身发生了硬性崩溃。
  2. 比对该 arguments 是否符合工具注册时定义的 JSON Schema（在 attributes.tool_schema 中记录）。
  3. 如果 Schema 本身无误，则判定为模型幻觉，需要对当前模型进行 Prompt 微调，或升级至对 Schema 遵循度更好的模型版本。

异常场景二：Agent 进入死循环（Looping Drill）

• 现象：用户请求超时未返回，系统后端持续处于计算状态，直到强制熔断。
• 定位路径：
  1. 在 Root Span 下观察，若发现同一层级下嵌套了 5 次以上完全相同的 LLM Chat Completion Span -> Tool Call Span 循环序列，则意味着 Agent 陷入了死循环。
  2. 查看每一次循环中 tool_response 传回给模型的报错信息。如果是工具调用报错（例如 403 Unauthorized），而模型在缺乏异常处理指令的情况下反复用相同参数重新尝试调用，这属于典型的 Agent 决策链断裂。系统需要优化 System Prompt，指导模型在工具连续失败时直接向用户报告错误，而不是无休止地盲目重试。

异常场景三：召回无关上下文导致的输出质量下降（RAG Hallucination）

• 现象：用户投诉系统胡言乱语，回答与提问完全不相关。
• 定位路径：
  1. 找到对应的 Root Trace，定位到 RAG Retrieval Span。
  2. 检查 attributes.retrieved_docs 的召回列表。如果所有召回文档的相似度得分（Similarity Score）均低于设定的置信度阈值（例如 score < 0.6），则说明是检索阶段未找到正确答案，而非模型生成能力问题。
  3. 此时应该优化 Embedding 模型、切片策略或重新清洗知识库，而非盲目去调整 LLM 的 Prompt。

实践交付物：可落地的 Observability 字段清单与脱敏验证练习

1. 核心 Trace 字段 Schema 清单

请确保你的 Observability Harness 输出的 JSON 格式 Trace 包含了以下标准的字段结构：

{
  "trace_id": "t_8f89bc72110c492b",
  "span_id": "s_4492aef3189",
  "parent_id": "s_root_001",
  "name": "llm_completion_call",
  "status": {
    "code": "ERROR",
    "message": "ToolCallValidationError: 'price' is a required property"
  },
  "attributes": {
    "gen_ai.model": "gpt-4o",
    "gen_ai.temperature": 0.2,
    "gen_ai.usage.prompt_tokens": 1204,
    "gen_ai.usage.completion_tokens": 145,
    "gen_ai.usage.total_tokens": 1349,
    "gen_ai.usage.cached_prompt_tokens": 512,
    "latency.ttft_ms": 240,
    "latency.total_ms": 1850,
    "security.guardrail.triggered": false,
    "security.masking.applied": true,
    "security.masking.masked_keys": ["user_phone", "user_email"]
  },
  "events": [
    {
      "time": "2026-05-28T14:30:01.120Z",
      "name": "tool_call_triggered",
      "attributes": {
        "tool.name": "get_user_balance",
        "tool.arguments_masked": "{\"user_id\": \"[MASKED_SENSITIVE_KEY]\"}"
      }
    }
  ]
}

2. 实践自测练习

如果你要将外部检索数据（RAG Context）加入到 context 字段中进行观测，必须确保限制单次 trace 记录的字节大小，不要无限制地转储大文本块，因为这不仅会使日志链路产生高昂的存储成本，还会增加 OWASP LLM06 提到的敏感信息泄露概率，除非该 trace 管道配置了专用的自动化 PII 清洗插件。

为了检验你对本单元概念的掌握，请在你的开发或测试环境中完成以下挑战：

• 任务目标：编写一个 Python 装饰器（Decorator）或中间件，包裹大模型的 API 调用函数。该中间件需要在模型调用前对 Prompt 进行脱敏处理，统计 Prompt Token 消耗，并在模型抛出异常或返回有害内容时，能够向你的 Trace 系统输出类似上述 Schema 定义的 JSON 数据结构。
• 验收基准：
  1. 输入含有明文邮箱和手机号的 Prompt，确保本地打印的日志输出中这些信息已被成功替换为 [MASKED_EMAIL] 和 [MASKED_PHONE]。
  2. 制造一个工具参数校验失败（例如返回错误的 JSON 格式），检查生成的 Trace JSON 中是否包含了正确的错误分类（ValidationError）以及工具执行时长的精确分段度量。
  3. 确保你的日志或 trace 数据中没有包含完整的、未加密的企业系统提示词（System Prompt）。

观测Harness：trace里该看见什么，不该记录什么：把判断写进 Harness 证据链

观测数据既要帮助排障，也要控制隐私和成本边界。本课交付物是 一份 GenAI observability 字段清单、脱敏策略和异常排查路径，它必须能被复跑、复核、追踪和复盘。

如果 观测Harness：trace里 还没有对应的输入样例，先不要讨论自动化覆盖率，因为没有样例就无法判断 harness 是否真的抓住问题。 如果一次评测失败会影响发布，应该保留失败输入、模型输出、工具轨迹和判分理由，否则团队只能凭记忆争论。 如果你准备把某个结果自动放行，必须先写清人工复核的例外条件，只有例外条件明确时，门禁才不会变成新的风险源。

CI gate 误报时要查阈值、样例稳定性和模型随机性，而不是直接关掉门禁。围绕 观测Harness：trace里该看 做排查时，记录步骤、样例、指标、风险、修复动作和复测结果。

OpenAI 的《Function calling》说明：支撑工具调用、应用层执行、参数 schema 和工具 harness 设计。；这意味着 观测Harness：trace里 要把来源转成可执行断言。OWASP 的《OWASP Top 10 for LLM Applications》提醒：支撑提示注入、敏感信息泄露、过度代理、供应链、输出处理和红队 harness。；因此本课必须写清自动判断和人工判断的边界。NIST 的《NIST AI Risk Management Framework》提供的证据是：支撑 AI 风险治理、Govern/Map/Measure/Manage 和组织级 harness。；所以当前结论按 2026-05-28 的来源状态使用。

复核触发条件要具体：如果模型 API/SDK、评测工具版本、Agent 框架、RAG 指标、安全规范、合规政策、平台 release/changelog 或关键来源页面变化，需要重新运行本课样例并更新 harness。

练习验收：把 观测Harness：trace里该看 接入一个最小 AI 应用，提交包含输入样例、评测结果、失败诊断、来源证据、人工复核结论和下一步修复动作的记录。缺少任何一项，都标记为 needs_review。