Codex 额度消耗太快怎么排查：用 NDJSON 定位上下文膨胀、重试与工具失败

{"ts":"2026-08-25T08:00:00Z","task_id":"t-001","auth_mode":"chatgpt","duration_s":860,"input_chars":184000,"retries":2,"tool_failures":1,"result":"ok"}
{"ts":"2026-08-25T08:28:00Z","task_id":"t-002","auth_mode":"api","duration_s":92,"input_chars":24000,"retries":3,"tool_failures":3,"result":"rate_limit"}

#!/usr/bin/env python3
import argparse
import json
from collections import Counter
from pathlib import Path

def pct(n, total):
    return f"{(n / total):.1%}" if total else "0.0%"

def main():
    p = argparse.ArgumentParser()
    p.add_argument("file", type=Path)
    p.add_argument("--long-seconds", type=int, default=600)
    p.add_argument("--large-chars", type=int, default=120000)
    args = p.parse_args()

    rows, bad = [], 0
    with args.file.open(encoding="utf-8") as f:
        for line in f:
            if not line.strip():
                continue
            try:
                r = json.loads(line)
                rows.append({
                    "task_id": str(r.get("task_id", "unknown")),
                    "auth_mode": str(r.get("auth_mode", "unknown")),
                    "duration_s": max(0, float(r.get("duration_s", 0))),
                    "input_chars": max(0, int(r.get("input_chars", 0))),
                    "retries": max(0, int(r.get("retries", 0))),
                    "tool_failures": max(0, int(r.get("tool_failures", 0))),
                    "result": str(r.get("result", "unknown")),
                })
            except (ValueError, TypeError, json.JSONDecodeError):
                bad += 1

    total = len(rows)
    long_tasks = [r for r in rows if r["duration_s"] >= args.long_seconds]
    large_tasks = [r for r in rows if r["input_chars"] >= args.large_chars]
    retried = [r for r in rows if r["retries"] > 0]
    tool_failed = [r for r in rows if r["tool_failures"] > 0]
    quota_hits = [r for r in rows if r["result"] in {"quota", "rate_limit"}]
    auth = Counter(r["auth_mode"] for r in rows)

    print(f"有效任务: {total}，坏行: {bad}")
    print("认证方式:", dict(auth))
    print(f"长任务率: {pct(len(long_tasks), total)}")
    print(f"大上下文率: {pct(len(large_tasks), total)}")
    print(f"发生重试率: {pct(len(retried), total)}")
    print(f"工具失败率: {pct(len(tool_failed), total)}")
    print(f"quota/rate_limit 命中: {len(quota_hits)}")

    score = lambda r: (
        r["duration_s"] / max(args.long_seconds, 1)
        + r["input_chars"] / max(args.large_chars, 1)
        + r["retries"]
        + r["tool_failures"]
        + (2 if r["result"] in {"quota", "rate_limit"} else 0)
    )
    print("\n优先排查:")
    for r in sorted(rows, key=score, reverse=True)[:10]:
        print(f"{r['task_id']}: auth={r['auth_mode']}, "
              f"sec={r['duration_s']:.0f}, chars={r['input_chars']}, "
              f"retry={r['retries']}, tool_fail={r['tool_failures']}, "
              f"result={r['result']}")

if __name__ == "__main__":
    main()

python3 analyze_codex_usage.py codex_tasks.ndjson

python3 analyze_codex_usage.py codex_tasks.ndjson \
  --long-seconds 900 --large-chars 200000