Agent 可观测性：日志、Trace 与 Replay 调试

# 关键 metrics
agent_metrics = {
    # 流量
    "agent_request_total": "总请求数（按用户/类型分维度）",
    "agent_active_sessions": "活跃会话数",
    
    # 延迟
    "agent_latency_p50": "中位延迟",
    "agent_latency_p95": "P95 延迟",
    "agent_latency_p99": "P99 延迟",
    "agent_ttft": "首 token 时间",
    
    # 成本
    "agent_token_input_total": "输入 token 总量",
    "agent_token_output_total": "输出 token 总量",
    "agent_cost_usd_total": "总成本",
    "agent_cost_per_request": "单请求成本",
    
    # 异常
    "agent_error_total": "错误数（分类型）",
    "agent_retry_total": "重试次数",
    "agent_timeout_total": "超时次数",
}

import structlog

logger = structlog.get_logger()

# 关键日志点
def log_agent_event(event_type, **kwargs):
    logger.info(
        event_type,
        timestamp=datetime.utcnow().isoformat(),
        request_id=kwargs.get("request_id"),
        user_id=kwargs.get("user_id"),
        session_id=kwargs.get("session_id"),
        agent_name=kwargs.get("agent_name"),
        step=kwargs.get("step"),
        **kwargs
    )

# 使用
log_agent_event("llm_call", request_id=rq.id, agent="researcher", 
                model="gpt-4o", prompt_tokens=1500, completion_tokens=300, cost=0.012)
log_agent_event("tool_call", request_id=rq.id, tool="search_web", args={"q": "..."}, duration_ms=850)
log_agent_event("agent_decision", request_id=rq.id, decision="call_writer", reasoning="...")
log_agent_event("user_feedback", request_id=rq.id, rating=4, comment="...")

quality_metrics = {
    # 自动评估
    "llm_judge_score_avg": "LLM-as-Judge 平均分",
    "faithfulness_avg": "事实一致性",
    "answer_relevancy_avg": "回答相关性",
    
    # 用户反馈
    "thumbs_up_rate": "用户点赞率",
    "user_correction_rate": "用户纠正率",
    "task_completion_rate": "任务完成率",
    
    # 业务信号
    "followup_question_rate": "追问率（高 = 没答清楚）",
    "regeneration_rate": "用户点重新生成率",
    "abandonment_rate": "会话中途放弃率"
}

from traceloop.sdk import Traceloop

Traceloop.init(app_name="my-agent", api_key="...")

# 现有代码不变
response = openai.ChatCompletion.create(
    model="gpt-4o",
    messages=[...]
)
# ↑ 自动产生 trace span，包含：
#   - model name
#   - prompt + completion
#   - token counts
#   - cost
#   - latency

from opentelemetry import trace
from opentelemetry.semconv.ai import SpanAttributes

tracer = trace.get_tracer("agent")

with tracer.start_as_current_span("llm_call") as span:
    span.set_attribute(SpanAttributes.LLM_SYSTEM, "openai")
    span.set_attribute(SpanAttributes.LLM_REQUEST_MODEL, "gpt-4o")
    span.set_attribute(SpanAttributes.LLM_REQUEST_TOKENS, 1500)
    span.set_attribute(SpanAttributes.LLM_RESPONSE_TOKENS, 300)
    span.set_attribute(SpanAttributes.LLM_REQUEST_COST, 0.012)
    
    response = openai.ChatCompletion.create(...)

from opentelemetry import trace

tracer = trace.get_tracer("agent-system")

def run_supervisor(user_input):
    with tracer.start_as_current_span("supervisor.run") as span:
        span.set_attribute("user_input", user_input)
        
        # 决策
        decision = llm_call("决定调用哪个 agent")
        
        # 委托给 worker
        if decision == "researcher":
            with tracer.start_as_current_span("worker.researcher") as sub:
                result = run_researcher(user_input)
                sub.set_attribute("result_size", len(result))
                return result

def run_researcher(query):
    with tracer.start_as_current_span("llm_call") as span:
        span.set_attribute("gen_ai.system", "anthropic")
        span.set_attribute("gen_ai.request.model", "claude-sonnet-4-5")
        return claude.messages.create(...)

class TraceSampler:
    def __init__(self):
        self.full_sample_rate = 0.05  # 5% 完整记录
        self.summary_sample_rate = 0.50  # 50% 摘要记录
    
    def should_record_full(self, request_id):
        return random.random() < self.full_sample_rate
    
    def should_record_summary(self, request_id):
        return random.random() < self.summary_sample_rate

# 完整记录：所有 prompt + completion
# 摘要记录：长度、关键 token、首尾片段
# 都不记录：聚合 metric

trace_strategy = {
    "full": {
        "rate": 0.05,
        "what": "完整 prompt + completion + tool args + tool result",
        "cost": "高（5% 流量）"
    },
    "summary": {
        "rate": 0.50,
        "what": "token 数 + cost + 工具名 + 首尾 100 char",
        "cost": "低"
    },
    "metric_only": {
        "rate": 1.00,
        "what": "counter / histogram",
        "cost": "极低"
    }
}

def should_record_full(request):
    # 异常请求一定完整记录
    if request.error:
        return True
    if request.latency > p99_threshold:
        return True
    if request.user_thumbs_down:
        return True
    if request.cost > cost_threshold:
        return True
    
    # 正常请求按概率采样
    return random.random() < 0.05

class AgentReplayer:
    def __init__(self, trace_storage):
        self.storage = trace_storage
    
    def replay(self, request_id: str, mode: str = "actual"):
        """
        mode: 
          - "actual": 重放时使用记录的 LLM 响应（确定性重放）
          - "live": 重放时重新调用 LLM（验证修复）
          - "step": 单步调试
        """
        trace = self.storage.get_trace(request_id)
        
        if mode == "actual":
            # 用 trace 中记录的 LLM 响应，不重新调用
            return self._replay_with_recorded(trace)
        elif mode == "live":
            # 重新调用 LLM，验证修复是否有效
            return self._replay_live(trace)
    
    def _replay_with_recorded(self, trace):
        """模拟每一步，用记录的响应而不是真实调用"""
        for span in trace.spans:
            if span.kind == "llm_call":
                # 直接返回记录的 completion
                span.output = span.recorded_completion
            elif span.kind == "tool_call":
                span.output = span.recorded_tool_result
        return trace.reconstruct_final_output()

from langfuse import Langfuse

# 生产环境自动收集 trace
langfuse = Langfuse()

@langfuse.observe()
def run_agent(user_input):
    response = llm.complete(user_input)
    return response

# 出问题时，从 Langfuse UI 找到对应的 trace
# 点击 "Replay" → 用记录的 LLM 响应完整重放
# 或选择 "Live" → 重新调用 LLM 验证修复

# 每个 span 都打上成本
span.set_attribute("gen_ai.request.cost_usd", cost)
span.set_attribute("gen_ai.request.tokens_input", input_tokens)
span.set_attribute("gen_ai.request.tokens_output", output_tokens)

# 在监控里聚合
cost_metrics = {
    "total_cost_usd_per_minute": "每分钟总成本",
    "cost_p50_per_request": "单请求 P50 成本",
    "cost_p99_per_request": "单请求 P99 成本",
    "cost_per_user_today": "单用户日成本",
    "cost_by_agent": "按 agent 拆解",
    "cost_by_model": "按模型拆解"
}

# 监控异常飙升
def detect_cost_anomaly(current_cost_per_minute):
    baseline = get_baseline_cost_per_minute(window="7d")
    
    if current_cost_per_minute > baseline * 2:
        alert(f"💰 成本异常: {current_cost_per_minute} > baseline*2 ({baseline*2})")
    
    # 单用户成本
    user_costs = get_top_users_by_cost(limit=10)
    for user_id, cost in user_costs:
        if cost > baseline_per_user * 5:
            alert(f"⚠️ 用户 {user_id} 成本异常: ${cost}")