skill-creator 源码深度拆解:LLM Skill 触发率、防过拟合与三 Agent 评审完整指南(05)

def run_single_query(query, skill_name, skill_description, timeout, project_root, model):    # 1. 创建临时 Skill 命令文件    command_file = project_commands_dir / f"{clean_name}.md"    command_file.write_text(f"---\ndescription: |\n  {indented_desc}\n---\n...")    # 2. 执行 claude -p，使用 SSE 流式输出    cmd = ["claude", "-p", query,           "--output-format", "stream-json",           "--verbose",           "--include-partial-messages"]  # 关键：提前检测触发    # 3. 实时监听流事件    while time.time() - start_time < timeout:        event = json.loads(line)        if event["type"] == "stream_event":            se_type = event["event"]["type"]            if se_type == "content_block_start":                tool_name = event["event"]["content_block"]["name"]                if tool_name in ("Skill", "Read"):                    pending_tool_name = tool_name  # 可能触发                else:                    return False  # 调用了其他工具，说明没用 Skill            elif se_type == "content_block_delta" and pending_tool_name:                accumulated_json += delta["partial_json"]                if clean_name in accumulated_json:                    return True  # 确认触发！立即返回    # 4. 清理临时文件    command_file.unlink()

• --include-partial-messages：不等完整输出，从 SSE 流中实时检测触发。这让单次检测从分钟级降到秒级。

• 临时命令文件：每次测试用不同的唯一 ID（uuid4().hex[:8]），测完立即删除，不污染环境。

• 10 worker 并行（run_eval.py:198）：ProcessPoolExecutor(max_workers=10)，10 个 query 同时跑。

• 触发判断逻辑：看工具调用的 name 字段，Skill 或 Read + 包含 skill 名字 = 触发；调用了其他工具 = 未触发。

def run_eval(eval_set, skill_name, description, runs_per_query=1):    with ProcessPoolExecutor(max_workers=num_workers) as executor:        # 每个 query 跑 runs_per_query 次（默认3次）        for item in eval_set:            for run_idx in range(runs_per_query):                future = executor.submit(run_single_query, ...)    # 计算每个 query 的触发率均值    trigger_rate = sum(triggers) / len(triggers)    # 按 should_trigger 判断 pass/fail    did_pass = trigger_rate >= trigger_threshold  # 默认阈值 0.5

def split_eval_set(eval_set, holdout=0.4, seed=42):    # 按 should_trigger 分层采样，保证 train/test 比例一致    trigger = [e for e in eval_set if e["should_trigger"]]    no_trigger = [e for e in eval_set if not e["should_trigger"]]    # 各取 40% 放入 test set    n_trigger_test = max(1, int(len(trigger) * holdout))    test_set = trigger[:n_trigger_test] + no_trigger[:n_no_trigger_test]    train_set = trigger[n_trigger_test:] + no_trigger[n_no_trigger_test:]    return train_set, test_set

for iteration in range(1, max_iterations + 1):    # 1. 跑所有 query（train + test 合批，节省时间）    all_results = run_eval(eval_set=train_set + test_set, description=current_description)    # 2. 按 query 归属分回 train/test    train_result_list = [r for r in all_results if r["query"] in train_queries_set]    # 3. 如果 train 全通过，停止    if train_summary["failed"] == 0:        break    # 4. 关键：对改进模型「盲化」test 成绩    blinded_history = [{k: v for k, v in h.items() if not k.startswith("test_")} ...]    # 5. 调用 improve_description，只看 train 结果来改进    new_description = improve_description(        history=blinded_history,  # test 成绩被抹掉        eval_results=train_results    )    current_description = new_description# 最终按 test 成绩选 best（不是 train）best = max(history, key=lambda h: h["test_passed"] or 0)

def calculate_stats(values):    n = len(values)    mean = sum(values) / n    # 使用样本标准差（n-1），不是总体标准差（n）    variance = sum((x - mean) ** 2 for x in values) / (n - 1)    stddev = math.sqrt(variance)    return {"mean": round(mean, 4), "stddev": round(stddev, 4), ...}

{  "pass_rate": {    "mean": 0.917,     ← 三次平均通过率 91.7%    "stddev": 0.047,   ← 标准差 4.7%，稳定（< 0.05 为稳定）    "min": 0.875,    "max": 0.958  }}

def aggregate_results(results):    # 计算 with_skill / without_skill 各自的统计量    for config in ["with_skill", "without_skill"]:        pass_rates = [r["pass_rate"] for r in results[config]]        run_summary[config] = {"pass_rate": calculate_stats(pass_rates), ...}    # 计算 delta    delta_pass_rate = primary["pass_rate"]["mean"] - baseline["pass_rate"]["mean"]    run_summary["delta"] = {"pass_rate": f"{delta_pass_rate:+.2f}", ...}

Step 1: 读 transcript → 注意工具调用、返回值、报错Step 2: 读 output 文件 → 注意结构、内容、质量Step 3: 对每条 expectation：    - PASS：有明确、具体、真实的证据    - FAIL：无证据 / 证据矛盾 / 技术上满足但实质错误Step 4: 提取隐含 claims 并验证（factual/process/quality 三类）Step 5: 读 user_notes.md（执行者留下的疑问）Step 6: 批评断言质量（eval_feedback）

「You receive two outputs labeled A and B, but you do NOT know which skill produced which. This prevents bias toward a particular skill or approach.」

内容维度（Content）：  正确性（Correctness）  1-5 分  完整性（Completeness） 1-5 分  准确性（Accuracy）     1-5 分结构维度（Structure）：  组织性（Organization） 1-5 分  格式规范（Formatting） 1-5 分  可用性（Usability）    1-5 分综合得分 = (内容均值 + 结构均值) / 2 × 2（换算 1-10 分）

{  "improvement_suggestions": [    {      "priority": "high",      "category": "instructions",      "suggestion": "将'process the document appropriately'替换为明确的5步流程",      "expected_impact": "消除歧义，防止 Agent 自行发挥"    }  ]}

断言总是双方都通过 → 可能无判别力，不能区分 Skill 价值断言总是双方都失败 → 可能超出模型能力范围断言 with_skill 总通过而 without 总失败 → Skill 明确有价值断言 with_skill 总失败而 without 总通过 → Skill 可能帮了倒忙

1. 流式检测触发：--include-partial-messages + SSE 流，秒级完成触发判断

2. 防过拟合的 60/40 分割：对改进模型盲化 test，按 test 选 best

3. 样本标准差（n-1）：比总体标准差更准确估计真实波动盲测消除偏见：Comparator 不知道哪个是「好版本」，纯质量评分

4. evidence 必须引用原文：Grader 的核心约束，消除感觉判断