Agent 可观测性工程：Trace、Eval 与调试全链路方案

# 传统 APM：一个请求 = 一个 Span
# Agent 可观测性：一个用户请求 = 一棵 Span 树

# 传统方式
@trace_request
def handle_request(request):
    result = process(request)
    return result  # 200 OK = 成功

# Agent 方式 —— 200 OK 不代表成功
@trace_agent_request
def handle_agent_request(request):
    plan = await llm.plan(request)         # Span: planning
    for step in plan.steps:
        tool_result = await execute(step)  # Span: tool_call
        validation = await llm.validate(tool_result)  # Span: validation
        if not validation.is_faithful:
            # HTTP 200，但语义上是失败的
            raise SemanticError("Output not faithful to source")
    return synthesize(results)

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.resources import Resource

# 初始化 Tracer
resource = Resource.create({
    "service.name": "agent-service",
    "service.version": "1.2.0",
    "deployment.environment": "production",
})

provider = TracerProvider(resource=resource)
processor = BatchSpanProcessor(OTLPSpanExporter(endpoint="http://otel-collector:4317"))
provider.add_span_processor(processor)
trace.set_tracer_provider(provider)

tracer = trace.get_tracer("agent.core", "1.0.0")


class AgentTracer:
    """Agent 可观测性追踪器，封装 OpenTelemetry Span 创建逻辑"""

    def __init__(self, tracer):
        self.tracer = tracer

    def trace_llm_call(self, model: str, messages: list, temperature: float = 0.7):
        """追踪单次 LLM 调用"""
        span = self.tracer.start_span(
            name=f"llm.chat.{model}",
            attributes={
                "llm.model": model,
                "llm.temperature": temperature,
                "llm.message_count": len(messages),
                "llm.system_prompt_tokens": self._count_tokens(messages[0]) if messages else 0,
            }
        )
        return span

    def trace_tool_call(self, tool_name: str, parameters: dict):
        """追踪工具调用"""
        span = self.tracer.start_span(
            name=f"tool.execute.{tool_name}",
            attributes={
                "tool.name": tool_name,
                "tool.parameters": str(parameters)[:1024],  # 截断防溢出
            }
        )
        return span

    def trace_agent_step(self, step_type: str, step_index: int):
        """追踪 Agent 推理步骤"""
        span = self.tracer.start_span(
            name=f"agent.step.{step_type}",
            attributes={
                "agent.step.type": step_type,
                "agent.step.index": step_index,
            }
        )
        return span

    def _count_tokens(self, message) -> int:
        return len(str(message)) // 4  # 粗略估算

from langfuse import Langfuse
from langfuse.decorators import observe, langfuse_context

langfuse = Langfuse(
    public_key="pk-xxx",
    secret_key="sk-xxx",
    host="https://your-langfuse-instance.com"
)


@observe(name="agent-pipeline")
async def run_agent_pipeline(user_query: str, session_id: str):
    """完整的 Agent 流水线追踪"""
    langfuse_context.update_current_trace(
        session_id=session_id,
        user_id="user-123",
        metadata={"pipeline_version": "2.1.0"}
    )

    # Step 1: 意图识别
    intent = await classify_intent(user_query)

    # Step 2: 计划生成
    plan = await generate_plan(user_query, intent)

    # Step 3: 执行计划
    results = []
    for step in plan.steps:
        result = await execute_step(step)
        results.append(result)

    # Step 4: 合成输出
    output = await synthesize_output(results, user_query)

    # 记录评估分数
    langfuse_context.score_current_trace(
        name="output_quality",
        value=await evaluate_output(output, user_query),
        comment="Automated quality score"
    )

    return output


@observe(name="classify-intent", capture_input=True, capture_output=True)
async def classify_intent(query: str) -> str:
    """意图分类 - 自动追踪输入输出"""
    response = await llm.chat(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": "Classify user intent into categories..."},
            {"role": "user", "content": query}
        ],
        temperature=0.1
    )
    return response.content


@observe(name="execute-tool")
async def execute_step(step):
    """工具执行步骤 - 追踪工具调用细节"""
    langfuse_context.update_current_observation(
        metadata={"tool": step.tool_name, "retry_count": 0}
    )
    try:
        result = await tool_registry.execute(step.tool_name, step.parameters)
        langfuse_context.update_current_observation(
            level="DEFAULT",
            status_message="success"
        )
        return result
    except Exception as e:
        langfuse_context.update_current_observation(
            level="ERROR",
            status_message=str(e)
        )
        raise

// TypeScript: 结构化 LLM-as-Judge 评估器
import { OpenAI } from "openai";
import { z } from "zod";

// 定义评估维度的结构化输出 Schema
const EvalResultSchema = z.object({
  faithfulness: z.object({
    score: z.number().min(0).max(1),
    reasoning: z.string(),
    evidence: z.array(z.string()),
  }),
  relevance: z.object({
    score: z.number().min(0).max(1),
    reasoning: z.string(),
  }),
  completeness: z.object({
    score: z.number().min(0).max(1),
    missing_aspects: z.array(z.string()),
  }),
  hallucination: z.object({
    detected: z.boolean(),
    hallucinated_claims: z.array(z.string()),
    severity: z.enum(["none", "minor", "major", "critical"]),
  }),
});

type EvalResult = z.infer<typeof EvalResultSchema>;

interface EvalInput {
  query: string;
  context: string[];
  response: string;
  groundTruth?: string;
}

class LLMJudgeEvaluator {
  private client: OpenAI;
  private model: string;

  constructor(apiKey: string, model = "gpt-4o") {
    this.client = new OpenAI({ apiKey });
    this.model = model;
  }

  async evaluate(input: EvalInput): Promise<EvalResult> {
    const rubric = this.buildRubric(input);

    const response = await this.client.chat.completions.create({
      model: this.model,
      temperature: 0.1, // 低温度提高评估一致性
      response_format: { type: "json_object" },
      messages: [
        {
          role: "system",
          content: `You are an expert evaluator for AI agent outputs.
Evaluate the response strictly according to the rubric provided.
Return ONLY a JSON object matching the specified schema.
Be critical and precise - do not inflate scores.`,
        },
        {
          role: "user",
          content: rubric,
        },
      ],
    });

    const parsed = JSON.parse(response.choices[0].message.content!);
    return EvalResultSchema.parse(parsed);
  }

  private buildRubric(input: EvalInput): string {
    return `## Evaluation Task

### User Query
${input.query}

### Retrieved Context
${input.context.map((c, i) => `[${i + 1}] ${c}`).join("\n")}

### Agent Response
${input.response}

${input.groundTruth ? `### Ground Truth\n${input.groundTruth}` : ""}

### Scoring Rubric

**Faithfulness** (0-1): Does the response ONLY contain claims supported by the context?
- 1.0: Every claim is directly supported by context
- 0.7: Minor unsupported claims that don't affect correctness
- 0.3: Contains speculative claims without evidence
- 0.0: Fabricates information contradicting context

**Relevance** (0-1): Does the response address the user's actual question?
- 1.0: Directly and completely answers the query
- 0.5: Partially addresses the query with some tangential content
- 0.0: Completely off-topic

**Completeness** (0-1): Does the response cover all aspects of the query?
- 1.0: Addresses all sub-questions and aspects
- 0.5: Covers main points but misses important details
- 0.0: Only superficially touches the topic

**Hallucination Detection**: Identify any claims NOT supported by context.

Return your evaluation as a JSON object.`;
  }
}

from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall,
)
from datasets import Dataset

# 构建评估数据集
eval_dataset = Dataset.from_dict({
    "question": [
        "如何配置 OpenTelemetry 进行 LLM 追踪？",
        "LangSmith 和 LangFuse 哪个更适合私有化部署？",
    ],
    "answer": [
        agent_responses[0],  # Agent 实际输出
        agent_responses[1],
    ],
    "contexts": [
        [retrieved_context_1],  # 检索到的上下文
        [retrieved_context_2],
    ],
    "ground_truth": [
        "使用 OpenTelemetry SDK 创建自定义 Span...",
        "LangFuse 支持自托管部署，适合私有化需求...",
    ],
})

# 执行评估
results = evaluate(
    dataset=eval_dataset,
    metrics=[
        faithfulness,       # 忠实度：输出是否忠于上下文
        answer_relevancy,   # 相关性：输出是否回答了问题
        context_precision,  # 上下文精度：检索内容是否相关
        context_recall,     # 上下文召回：是否检索到足够信息
    ],
)

print(results.to_pandas())
# Output:
#   faithfulness  answer_relevancy  context_precision  context_recall
# 0         0.92              0.88               0.85            0.78
# 1         0.95              0.91               0.90            0.82

import json
from dataclasses import dataclass, field
from typing import Any
from datetime import datetime


@dataclass
class AgentSnapshot:
    """Agent 运行时快照，支持 Time-Travel 调试"""
    timestamp: datetime
    step_index: int
    step_type: str
    input_state: dict
    output_state: dict
    llm_messages: list
    llm_response: str
    tool_calls: list = field(default_factory=list)
    metadata: dict = field(default_factory=dict)


class TimeTravelDebugger:
    """Time-Travel 调试器：记录每一步快照，支持回放和断点"""

    def __init__(self):
        self.snapshots: list[AgentSnapshot] = []
        self.breakpoints: dict[str, callable] = {}

    def capture(self, step_index: int, step_type: str, **kwargs) -> AgentSnapshot:
        """捕获当前步骤的完整快照"""
        snapshot = AgentSnapshot(
            timestamp=datetime.utcnow(),
            step_index=step_index,
            step_type=step_type,
            input_state=kwargs.get("input_state", {}),
            output_state=kwargs.get("output_state", {}),
            llm_messages=kwargs.get("messages", []),
            llm_response=kwargs.get("response", ""),
            tool_calls=kwargs.get("tool_calls", []),
            metadata=kwargs.get("metadata", {}),
        )
        self.snapshots.append(snapshot)
        self._check_breakpoints(snapshot)
        return snapshot

    def replay_from(self, step_index: int) -> list[AgentSnapshot]:
        """从指定步骤开始回放"""
        return [s for s in self.snapshots if s.step_index >= step_index]

    def detect_loop(self, window_size: int = 5) -> bool:
        """检测循环模式：连续 N 步的工具调用序列是否重复"""
        if len(self.snapshots) < window_size * 2:
            return False

        recent = self.snapshots[-window_size:]
        previous = self.snapshots[-window_size * 2:-window_size]

        recent_pattern = [(s.step_type, tuple(s.tool_calls)) for s in recent]
        previous_pattern = [(s.step_type, tuple(s.tool_calls)) for s in previous]

        return recent_pattern == previous_pattern

    def find_divergence_point(self, expected_trace: list[dict]) -> int:
        """对比实际执行和预期路径，找到分叉点"""
        for i, (actual, expected) in enumerate(zip(self.snapshots, expected_trace)):
            if actual.step_type != expected.get("step_type"):
                return i
            if actual.output_state != expected.get("expected_output"):
                return i
        return -1

    def _check_breakpoints(self, snapshot: AgentSnapshot):
        """检查是否触发调试断点"""
        for name, condition in self.breakpoints.items():
            if condition(snapshot):
                print(f"[BREAKPOINT] {name} triggered at step {snapshot.step_index}")
                self._dump_snapshot(snapshot)

    def _dump_snapshot(self, snapshot: AgentSnapshot):
        """输出快照详情"""
        print(json.dumps({
            "step": snapshot.step_index,
            "type": snapshot.step_type,
            "input": snapshot.input_state,
            "output": snapshot.output_state,
            "tools": snapshot.tool_calls,
        }, indent=2, ensure_ascii=False))


# 使用示例
debugger = TimeTravelDebugger()

# 设置断点：当检测到循环时暂停
debugger.breakpoints["loop_detected"] = lambda s: debugger.detect_loop()

# 设置断点：Token 用量超过阈值
debugger.breakpoints["token_overflow"] = lambda s: s.metadata.get("total_tokens", 0) > 100000

// TypeScript: 结构化日志与 Trace 关联
import { SpanContext, trace } from "@opentelemetry/api";
import pino from "pino";

interface AgentLogEntry {
  traceId: string;
  spanId: string;
  level: "info" | "warn" | "error" | "debug";
  event: string;
  agentId: string;
  stepIndex: number;
  data: Record<string, unknown>;
}

class AgentLogger {
  private logger: pino.Logger;

  constructor(serviceName: string) {
    this.logger = pino({
      name: serviceName,
      formatters: {
        log: (obj) => {
          // 自动注入当前 Trace 上下文
          const span = trace.getActiveSpan();
          if (span) {
            const ctx: SpanContext = span.spanContext();
            return {
              ...obj,
              traceId: ctx.traceId,
              spanId: ctx.spanId,
              traceFlags: ctx.traceFlags,
            };
          }
          return obj;
        },
      },
    });
  }

  logAgentStep(entry: Omit<AgentLogEntry, "traceId" | "spanId">): void {
    const span = trace.getActiveSpan();
    const ctx = span?.spanContext();

    this.logger[entry.level]({
      traceId: ctx?.traceId ?? "unknown",
      spanId: ctx?.spanId ?? "unknown",
      event: entry.event,
      agentId: entry.agentId,
      stepIndex: entry.stepIndex,
      ...entry.data,
    });
  }

  logToolResult(toolName: string, success: boolean, latencyMs: number): void {
    this.logAgentStep({
      level: success ? "info" : "error",
      event: "tool_execution_complete",
      agentId: "current",
      stepIndex: -1,
      data: { toolName, success, latencyMs },
    });
  }
}

from enum import Enum
from typing import Optional
import random
import hashlib


class ImportanceLevel(Enum):
    CRITICAL = "critical"   # 100% 采样 - 付费操作、高风险决策
    HIGH = "high"           # 50% 采样 - 工具执行失败、长链推理
    MEDIUM = "medium"       # 10% 采样 - 正常请求
    LOW = "low"             # 1% 采样 - 健康检查、内部调用


class AdaptiveSampler:
    """基于重要性的自适应采样器"""

    SAMPLE_RATES = {
        ImportanceLevel.CRITICAL: 1.0,
        ImportanceLevel.HIGH: 0.5,
        ImportanceLevel.MEDIUM: 0.1,
        ImportanceLevel.LOW: 0.01,
    }

    def __init__(self, error_boost_multiplier: float = 5.0):
        self.error_boost = error_boost_multiplier
        self.error_rate_window: list[bool] = []

    def should_sample(
        self,
        trace_id: str,
        importance: ImportanceLevel,
        is_error: bool = False,
    ) -> bool:
        """决定是否采样当前 Trace"""
        base_rate = self.SAMPLE_RATES[importance]

        # 错误请求提升采样率
        if is_error:
            base_rate = min(1.0, base_rate * self.error_boost)

        # 基于 trace_id 的确定性采样（同一 trace 始终一致）
        hash_value = int(hashlib.md5(trace_id.encode()).hexdigest()[:8], 16)
        threshold = hash_value / 0xFFFFFFFF

        return threshold < base_rate

    def classify_importance(self, request_metadata: dict) -> ImportanceLevel:
        """根据请求元数据分类重要性"""
        if request_metadata.get("involves_payment"):
            return ImportanceLevel.CRITICAL
        if request_metadata.get("tool_count", 0) > 5:
            return ImportanceLevel.HIGH
        if request_metadata.get("is_internal"):
            return ImportanceLevel.LOW
        return ImportanceLevel.MEDIUM

# 告警规则配置示例
ALERT_RULES = {
    "agent_loop_detected": {
        "condition": "span_repeat_count > 3 within 30s",
        "severity": "critical",
        "action": "auto_terminate + page_oncall",
    },
    "hallucination_rate_spike": {
        "condition": "hallucination_rate > 0.15 for 5min",
        "severity": "high",
        "action": "alert_oncall + increase_sampling",
    },
    "latency_p99_breach": {
        "condition": "agent_latency_p99 > 30s for 3min",
        "severity": "medium",
        "action": "alert_channel",
    },
    "token_cost_anomaly": {
        "condition": "hourly_token_cost > 2x daily_average",
        "severity": "high",
        "action": "alert_oncall + enable_rate_limit",
    },
    "eval_score_degradation": {
        "condition": "faithfulness_score_avg < 0.7 for 15min",
        "severity": "high",
        "action": "rollback_prompt_version",
    },
}

class PIIScrubber:
    """在 Trace 存储前清洗 PII 信息"""

    PATTERNS = {
        "email": r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}",
        "phone": r"\b\d{3}[-.]?\d{4}[-.]?\d{4}\b",
        "credit_card": r"\b\d{4}[-\s]?\d{4}[-\s]?\d{4}[-\s]?\d{4}\b",
    }

    def scrub(self, text: str) -> str:
        import re
        for pii_type, pattern in self.PATTERNS.items():
            text = re.sub(pattern, f"[REDACTED_{pii_type.upper()}]", text)
        return text