calibrate_threshold.py

"""calibrate_threshold.py - 置信度阈值校准 (精度优化 ①)

目标: 把 code-reviewer 的"自评置信度门槛"从拍脑袋的 80, 改成实测精度≥目标值
      的分位点。用现有 10 个 bug 金标准 (带标签) 测出 precision@threshold 曲线。

机制:
  - 用一个校准专用 prompt, 让 reviewer 输出 **每条 finding 的 confidence (0-100)**,
    且**不做内部 ≥80 过滤**(全报), 这样才能扫不同阈值看精度变化。
  - 对 10 个种子逐个跑, 记录每条 finding 的 confidence + 是否命中真 bug。
  - 扫阈值 60..95, 算每个阈值下的 precision / recall, 推荐达标阈值。

前置: 与 run_group_a 相同 — ANTHROPIC_BASE_URL + ANTHROPIC_AUTH_TOKEN +
      ANTHROPIC_DEFAULT_SONNET_MODEL (见 experiment/REPRODUCE.md)。

运行:
    cd loop-engine/experiment
    python calibrate_threshold.py
    # 或指定目标精度: python calibrate_threshold.py --target-precision 0.90
"""
import sys, os, argparse
sys.path.insert(0, os.path.dirname(__file__))
from bug_seeds import SEEDS
from llm_call import make_call_fn, extract_json_array, read_agent_body


# 校准 prompt: 要求输出 confidence, 且全报 (不内部过滤) — 这样才能扫阈值
CALIB_USER = """审查以下Python代码找出bug(综合所有视角: 安全/正确性/边界/质量)。
**报出所有你认为可能的问题, 每条带 confidence (0-100)**。不要预先按置信度过滤。
返回JSON数组, 每项: {{"severity","category","issue","fix_hint","confidence"}}。无问题返回 []。

代码:
```python
{code}
```"""

# 复用 run_group_a 的中英同义词命中逻辑 (避免中英关键词不匹配导致假漏报)
SYNONYMS = {
    "sql injection": ["sql injection", "sql注入", "参数化", "拼接查询", "注入"],
    "command injection": ["command injection", "命令注入", "os.system", "shell"],
    "xss": ["xss", "跨站", "转义", "escape", "innerHTML", "未转义"],
    "null": ["null", "none", "空值", "空指针", "空对象", "nullable", "未检查"],
    "off-by-one": ["off-by-one", "差一", "越界", "边界错误", "<=", "边界"],
    "discount": ["discount", "折扣", "方向", "加而非减"],
    "empty": ["empty", "空串", "空字符串", "空输入", "空参数"],
    "race": ["race", "竞态", "并发", "线程安全", "锁", "check-then"],
    "atomic": ["atomic", "原子", "事务", "一致性", "部分更新", "partial"],
    "divide": ["divide", "除零", "除以零", "division", "zero"],
}


def hit(finding: dict, expected: str) -> bool:
    blob = " ".join(str(finding.get(k, "")).lower() for k in ("issue", "category", "fix_hint"))
    keys = SYNONYMS.get(expected.lower(), [expected.lower()])
    return any(k.lower() in blob for k in keys)


def collect(call, system_prompt):
    """跑 10 个种子, 收集 (seed_id, confidence, is_hit) 三元组列表."""
    rows = []
    for s in SEEDS:
        text = call(CALIB_USER.format(code=s.code), system_prompt)
        findings = extract_json_array(text)
        seed_hit = False
        for f in findings:
            conf = f.get("confidence", 0)
            try:
                conf = float(conf)
            except (TypeError, ValueError):
                conf = 0.0
            is_hit = hit(f, s.expected_finding)
            rows.append((s.id, conf, is_hit))
            if is_hit:
                seed_hit = True
        flag = "✓命中" if seed_hit else "✗漏报"
        print(f"  {s.id} ({s.difficulty}): {flag}  报{len(findings)}条  "
              f"conf范围[{min((f.get('confidence',0) for f in findings), default='-')}"
              f"..{max((f.get('confidence',0) for f in findings), default='-')}]")
    return rows


def sweep(rows, target_precision):
    """扫阈值, 算 precision/recall, 推荐达标阈值."""
    n_seeds = len(SEEDS)
    print("\n" + "=" * 60)
    print("precision@threshold 曲线")
    print("=" * 60)
    print(f"{'阈值≥':<8}{'报数':>6}{'命中':>6}{'FP':>6}{'precision':>12}{'recall':>8}")
    print("-" * 52)

    best_thresh = None
    for thresh in range(60, 100, 5):
        kept = [(sid, h) for (sid, c, h) in rows if c >= thresh]
        if not kept:
            continue
        reported = len(kept)
        hits = sum(1 for _, h in kept if h)
        fp = reported - hits
        precision = hits / reported if reported else 0
        seeds_hit = len({sid for sid, h in kept if h})
        recall = seeds_hit / n_seeds
        flag = ""
        if best_thresh is None and precision >= target_precision:
            best_thresh = thresh
            flag = "  ← 首个达标"
        print(f"{thresh:<8}{reported:>6}{hits:>6}{fp:>6}{precision:>11.0%}{recall:>8.0%}{flag}")

    print("\n" + "=" * 60)
    if best_thresh is not None:
        print(f"推荐阈值: confidence ≥ {best_thresh}  (首个达到 precision≥{target_precision:.0%})")
        print("把它设到 code-reviewer 的内部门槛 (或 loop 的 confidence 门禁),")
        print(f"单次审查精度可从 71% 抬到 ~{target_precision:.0%} (recall 会降, 见曲线)。")
    else:
        print(f"⚠ 在 60-95 区间无阈值达到 precision≥{target_precision:.0%}。")
        print("说明假阳性的自评置信度与真 bug 同样高 (LLM 过度自信), 自评阈值无效。")
        print("应转用结构性补强: 独立杀手视角 / confidence-evaluator 前置 / diff 锚定。")
    return best_thresh


def main():
    ap = argparse.ArgumentParser(description="code-reviewer 置信度阈值校准")
    ap.add_argument("--target-precision", type=float, default=0.90,
                    help="目标精度 (默认 0.90), 找首个达标的阈值")
    args = ap.parse_args()

    print("=" * 60)
    print(f"code-reviewer 置信度阈值校准 (目标精度 {args.target_precision:.0%})")
    print("=" * 60)
    call = make_call_fn("ANTHROPIC_DEFAULT_SONNET_MODEL")
    print(f"模型: {call.model_name}\n")

    enhanced_sys = read_agent_body("../agents/code-reviewer.md")
    print("--- 跑 10 个 bug 种子 (全报 + 带 confidence) ---")
    rows = collect(call, enhanced_sys)
    sweep(rows, args.target_precision)

    print("\n注: 单次审查精度由阈值决定; 跨会话假阳性记忆 (⑤) 减少**重复**假阳性,")
    print("不直接改变单次 precision, 但让团队级累计干扰递减。")


if __name__ == "__main__":
    main()