Trace、Replay 与线上闭环

import json, time, hashlib
from dataclasses import dataclass, field, asdict
@dataclass
class ToolCall:
    turn: int
    name: str
    args: dict
    status: str                     # ok | empty | bad_args | timeout | ...
    duration_ms: int
    result_digest: str              # 结果指纹，而不是全文（全文另存）
@dataclass
class Trace:
    trace_id: str
    started_at: float
    model: str
    request_body: dict              # 脱敏后的完整请求体 —— 重放的依据
    tool_calls: list[ToolCall] = field(default_factory=list)
    context_sizes: dict = field(default_factory=dict)   # {"instructions":1200,...}
    ttfb_ms: int | None = None
    total_ms: int = 0
    retries: list[dict] = field(default_factory=list)
    outcome: str = "unknown"        # done | budget_exhausted | failed | need_human
    user_signal: str = "none"       # none | retried | aborted | corrected
    def to_json(self) -> str:
        return json.dumps(asdict(self), ensure_ascii=False)
def trace_id_for(request_body: dict, model: str) -> str:
    """
    trace id 用「请求内容 + 模型」派生。
    好处：同一个输入重复出现时能自动聚类，便于统计高频问题。
    """
    key = json.dumps(request_body, sort_keys=True, ensure_ascii=False) + model
    return hashlib.sha256(key.encode()).hexdigest()[:16]
def replay(trace: Trace, run_once) -> dict:
    """
    重放：用当时的请求体重新跑一遍，对比结果差异。
    这是「改了代码到底有没有改善」的最直接验证方式。
    """
    result = run_once(trace.request_body)          # 用同一份请求体
    diffs = {
        "outcome_changed": result.get("outcome") != trace.outcome,
        "tool_call_count": (len(result.get("tool_calls", [])), len(trace.tool_calls)),
        "retry_count": (len(result.get("retries", [])), len(trace.retries)),
        "latency_ms": (result.get("total_ms"), trace.total_ms),
    }
    return {"trace_id": trace.trace_id, "improved": not diffs["outcome_changed"]
            or result.get("outcome") == "done", "diffs": diffs}
def classify_failure(trace: Trace) -> str:
    """
    归因：把一次失败定位到具体环节。
    这一段逻辑是「评测报告」与「评测工具」的分界线 ——
    有归因，报告才能直接变成改动清单。
    """
    if trace.outcome == "budget_exhausted":
        if trace.context_sizes.get("state", 0) > 0.6 * sum(trace.context_sizes.values()):
            return "上下文膨胀：状态占比过高，需要加强压缩或状态外置"
        return "轮数/预算上限不足，或陷入无进展循环（检查调用签名重复度）"
    bad = [c for c in trace.tool_calls if c.status == "bad_args"]
    if bad:
        return f"参数错误（{len(bad)} 次）：检查工具描述与参数约束"
    to = [c for c in trace.tool_calls if c.status == "timeout"]
    if to:
        return f"工具超时（{len(to)} 次）：检查超时设置与退避策略"
    if trace.retries:
        return f"重试 {len(trace.retries)} 次后成功：属隐患，应查清根因"
    if trace.ttfb_ms and trace.ttfb_ms > 15000:
        return "首字节延迟过高：可能需要换 provider 或降级流式"
    return "未归因：需要人工查看 trace"
def attribution_report(traces: list[Trace]) -> dict:
    """归因分布：这份报告比「失败率 12%」有用得多"""
    from collections import Counter
    failed = [t for t in traces if t.outcome != "done"]
    reasons = Counter(classify_failure(t) for t in failed)
    return {
        "total": len(traces),
        "failed": len(failed),
        "failure_rate": len(failed) / len(traces) if traces else 0,
        "by_reason": reasons.most_common(),
        "user_retry_rate": sum(1 for t in traces if t.user_signal == "retried") / len(traces)
                           if traces else 0,
    }