#!/usr/bin/env python3 """ quality_tuner.py — 质量阈值自动调优器 调优结果写入 evolution/thresholds.json，不修改 SKILL.md。 """ import sys, os, re, json, math from pathlib import Path from typing import Any sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from utils import ( find_project_root, get_evolution_dir, load_evolution, save_evolution, parse_quality_log, extract_quality_features, log_evolution_event, ) DEFAULT_THRESHOLDS = { "pass": 80, "revise_low": 60, "rewrite_low": 40, "max_revision_rounds": 5, "weights": {"P1": 0.35, "P2": 0.25, "P3": 0.15, "P4": 0.15, "P5": 0.10}, "stale_hook_limit": 10, "burst_hook_limit": 5, "no_advance_limit": 5, "payoff_interval_min": 300, "payoff_interval_max": 500, "weak_hook_consecutive_max": 3, } class QualityTuner: def __init__(self, thresholds: dict | None = None): self.thresholds = thresholds or load_evolution("thresholds.json") or DEFAULT_THRESHOLDS.copy() self.history: list[dict] = [] self.adjustments: list[dict] = [] def load_history(self, project: Path) -> None: self.history = parse_quality_log(project) def analyze(self) -> dict: if not self.history: return {"status": "no_data"} features = extract_quality_features(self.history) suggestions = [] recent = [r for r in self.history[-10:] if r.get("score")] recent_scores = [r["score"] for r in recent] pass_rate = sum(1 for s in recent_scores if s >= self.thresholds["pass"]) / max(len(recent_scores), 1) if pass_rate > 0.9 and len(recent_scores) >= 5: suggestions.append({ "type": "pass_threshold_raise", "current": self.thresholds["pass"], "suggested": min(88, self.thresholds["pass"] + 3), "reason": f"近期通过率 {pass_rate:.0%}，阈值可能偏低", "confidence": 0.7, }) elif pass_rate < 0.3 and len(recent_scores) >= 5: suggestions.append({ "type": "pass_threshold_lower", "current": self.thresholds["pass"], "suggested": max(70, self.thresholds["pass"] - 3), "reason": f"近期通过率 {pass_rate:.0%}，阈值可能偏高", "confidence": 0.6, }) pass_weights = ["P1", "P2", "P3", "P4", "P5"] pass_scores = {} for p in pass_weights: scores = [r.get(f"score_{p.lower()}", 0) for r in self.history if f"score_{p.lower()}" in r] if scores: pass_scores[p] = sum(scores) / len(scores) if pass_scores: weakest = min(pass_scores, key=pass_scores.get) strongest = max(pass_scores, key=pass_scores.get) if pass_scores[strongest] - pass_scores[weakest] > 15: current_weight = self.thresholds["weights"][weakest] new_weight = min(0.40, current_weight + 0.05) suggestions.append({ "type": "weight_adjust", "target_pass": weakest, "current_weight": current_weight, "suggested_weight": new_weight, "reason": f"{weakest} 最弱 ({pass_scores[weakest]:.0f})，增权", "confidence": 0.75, "redistribute_from": strongest, }) rounds = [r.get("revision_rounds", 1) for r in self.history if r.get("revision_rounds")] avg_rounds = sum(rounds) / len(rounds) if rounds else 1.0 if avg_rounds > 3: suggestions.append({ "type": "root_cause_warning", "reason": f"平均修订 {avg_rounds:.1f} 轮，建议向上追溯大纲/设定问题", "confidence": 0.8, }) s1_recent = sum(r.get("s1_count", 0) for r in recent) if s1_recent > 0: suggestions.append({ "type": "quality_alert", "level": "S1", "count": s1_recent, "reason": f"近10章有 {s1_recent} 个S1致命问题", "confidence": 0.9, }) return { "status": "ok", "features": features, "pass_rate": pass_rate, "recent_scores": recent_scores, "pass_scores": pass_scores, "suggestions": suggestions, "avg_revision_rounds": avg_rounds, } def apply(self, suggestions: list[dict], auto_apply: bool = False) -> list[dict]: applied = [] for s in suggestions: if s.get("type") == "quality_alert": applied.append({**s, "applied": False}) continue if not auto_apply and s.get("confidence", 0) < 0.7: applied.append({**s, "applied": False}) continue if s["type"] in ("pass_threshold_raise", "pass_threshold_lower"): old = self.thresholds["pass"] self.thresholds["pass"] = s["suggested"] self.adjustments.append({"what": "pass", "from": old, "to": s["suggested"]}) applied.append({**s, "applied": True}) elif s["type"] == "weight_adjust": target = s["target_pass"] old_w = self.thresholds["weights"][target] delta = s["suggested_weight"] - old_w self.thresholds["weights"][target] = s["suggested_weight"] # 从其他维度扣减 delta，每维度不低于 floor=0.05 floor = 0.05 donors = sorted( [(dim, w) for dim, w in self.thresholds["weights"].items() if dim != target and w > floor], key=lambda x: x[1], reverse=True ) remaining = delta for dim, w in donors: if remaining <= 0: break available = w - floor take = min(available, remaining) self.thresholds["weights"][dim] -= take remaining -= take # 归一化到 1.0，避免浮点累积 total = sum(self.thresholds["weights"].values()) for dim in self.thresholds["weights"]: self.thresholds["weights"][dim] = round( self.thresholds["weights"][dim] / total, 10 ) self.adjustments.append({"what": f"weight_{target}", "from": old_w, "to": self.thresholds["weights"][target]}) applied.append({**s, "applied": True}) return applied def sync_to_evolution(self) -> None: """将当前阈值写入 evolution/thresholds.json""" save_evolution("thresholds.json", self.thresholds) def sync_to_genre_evolution(self, genre_id: str) -> None: """将当前阈值写入 evolution/genres/{genre_id}/thresholds.json""" from utils import save_genre_evolution save_genre_evolution(genre_id, "thresholds.json", self.thresholds) # ── CLI ───────────────────────────────────────────────── def analyze_and_tune(project: Path, auto_apply: bool = False) -> dict: print(f"🔧 质量阈值调优: {project.name}") tuner = QualityTuner() tuner.load_history(project) analysis = tuner.analyze() if analysis["status"] == "no_data": print(" ⚠ 没有质检历史数据，使用当前阈值") return analysis features = analysis.get("features", {}) avg = features.get('avg_score', 0) print(f" 平均得分: {avg:.1f}" if isinstance(avg, (int, float)) else f" 平均得分: N/A") print(f" 通过率: {analysis.get('pass_rate', 0):.0%}") print(f" S1/S2: {features.get('total_s1', 0)}/{features.get('total_s2', 0)}") suggestions = analysis.get("suggestions", []) if suggestions: print(f"\n {len(suggestions)} 条优化建议：") for s in suggestions: print(f" [{s['type']}] {s.get('reason', '')}") applied = tuner.apply(suggestions, auto_apply=auto_apply) applied_count = sum(1 for a in applied if a.get("applied")) print(f"\n 应用了 {applied_count} 条，跳过 {len(applied) - applied_count} 条") else: print(" ✓ 阈值配置合理，无需调整") applied = [] from utils import find_project_genre, map_genre_to_inkos_id genre_id = map_genre_to_inkos_id(find_project_genre(project)) tuner.sync_to_evolution() tuner.sync_to_genre_evolution(genre_id) print(f" ✓ 阈值已同步到 skill 级和题材级 evolution/genres/{genre_id}/") return {"analysis": analysis, "applied": applied} if __name__ == "__main__": import argparse parser = argparse.ArgumentParser(description="质量阈值调优器") parser.add_argument("action", choices=["analyze", "tune", "show"], help="analyze=分析 | tune=分析并应用 | show=查看当前阈值") parser.add_argument("--project", "-p", default=".") parser.add_argument("--auto", action="store_true") args = parser.parse_args() if args.action == "show": thresholds = load_evolution("thresholds.json") or DEFAULT_THRESHOLDS print(json.dumps(thresholds, ensure_ascii=False, indent=2)) elif args.action in ("analyze", "tune"): root = find_project_root(Path(args.project).resolve()) if not root: print("错误: 未找到小说项目根目录") sys.exit(1) auto = args.auto or (args.action == "tune") result = analyze_and_tune(root, auto_apply=auto) if args.action == "tune" and result.get("applied"): log_evolution_event(root, "质量阈值调优", { "应用": len([a for a in result["applied"] if a.get("applied")]), })