"""智能采纳：按策略自动处理一批草稿，只对证据足够硬的条目动手，其余留给人。 为什么不是「模型说采纳就采纳」：模型偶尔会把本次任务的具体数据当成可复用知识， 一条错的系统规则会让 Agent 以后一直绕开正确操作。所以要求多路信号一致，且分级： 自动采纳 需要同时满足：类别属于控件技巧 / 入口路径、规则层建议采纳、 模型层（配了才算）也建议采纳、证据够硬（跨校观察或观察次数达标） 自动驳回 只处理明显的垃圾（内容过短、整条基本只剩日期 / 班级 / 人名）， 且标记为 auto —— 再被观察到会退回队列，不像人工驳回那样是终局 其余 一律留给人，包括所有系统规则类 每次都能先干跑（dry_run）看会动哪些，确认了再真跑。单次有条数上限， 避免一次模型抽风批量放行。 """ from __future__ import annotations import asyncio from dataclasses import dataclass from server import review from server.config import settings from server.db import database as db from shared import knowledge_match # 只有这两类可能被自动采纳；系统规则和其他类永远要人看 AUTO_ADOPT_CATEGORIES = knowledge_match.AUTO_VERIFY_CATEGORIES @dataclass class Policy: """自动采纳的门槛，越严越安全。""" min_schools: int = 2 # 至少几所学校各自观察到，才算跨校成立 min_observations: int = 3 # 单校时至少观察多少次 min_trials: int = 2 # Agent 在真实页面上验证相符几次，算实测通过 require_llm: bool = True # 配了模型 Key 时，是否要求模型也同意 auto_reject: bool = True # 是否自动清掉明显的垃圾 max_actions: int = 50 # 单次最多处理多少条 @classmethod def from_settings(cls) -> "Policy": return cls( min_schools=settings.auto_min_schools, min_observations=settings.auto_min_observations, min_trials=settings.auto_min_trials, require_llm=settings.auto_require_llm, auto_reject=settings.auto_reject, max_actions=settings.auto_max_actions, ) def trial_proven(entry: dict, policy: Policy) -> bool: """这条是否已被 Agent 在真实页面上验证过：相符够次数且从没被证伪。 这是最硬的证据。「几所学校推导出同一条」只说明它被反复想到， 「Agent 去页面上对过」才说明它是真的。 """ return int(entry.get("confirmed") or 0) >= policy.min_trials and int(entry.get("contradicted") or 0) == 0 def _evidence_ok(entry: dict, policy: Policy) -> tuple[bool, str]: schools = int(entry.get("school_count") or 1) observations = int(entry.get("observations") or 1) confirmed = int(entry.get("confirmed") or 0) if trial_proven(entry, policy): return True, f"Agent 在真实页面上验证相符 {confirmed} 次" if schools >= policy.min_schools: return True, f"{schools} 所学校各自观察到" if observations >= policy.min_observations: return True, f"单校观察 {observations} 次" extra = f"，实测相符 {confirmed} 次" if confirmed else "" return False, f"证据不足（{schools} 校 / 观察 {observations} 次{extra}）" def decide(entry: dict, rule: tuple[str, str], llm: tuple[str, str] | None, policy: Policy) -> dict: """对一条草稿给出处置。返回 {action, scope, reason}，action 为 verify / reject / hold。 两条路都能走到自动采纳，但都要先过类别硬门槛（只有控件技巧 / 入口路径）： A 证据够硬：跨校观察或观察次数达标。模型配了且要求一致时，模型不能反对。 B 模型背书：证据薄，但模型读过内容、比对过已采纳条目后明确说可复用。 这条路发布范围强制为「仅来源学校」——没有跨校证据就不往别的学校推。 自动驳回同样要两路一致，且标记为 auto（暂缓，不是终局）。 """ rule_verdict, rule_reason = rule llm_verdict, llm_reason = llm if llm else (None, "") llm_active = bool(llm and settings.llm_api_key) category = entry.get("category") or "note" label = knowledge_match.CATEGORY_LABELS.get(category, category) def note(*parts: str) -> str: return "；".join(p for p in parts if p) # ---- 自动驳回：明显的垃圾，两路不能有人说好 ---- junk = rule_verdict == "reject" and llm_verdict in (None, "reject") llm_says_junk = llm_verdict == "reject" and rule_verdict != "adopt" if junk or llm_says_junk: if not policy.auto_reject: return {"action": "hold", "scope": None, "reason": note("疑似无效，但自动驳回已关闭", rule_reason)} return {"action": "reject", "scope": None, "reason": note(rule_reason if junk else "模型判定为一次性数据或重复", llm_reason[:80] if llm_reason else "")} # ---- 类别门槛：系统规则和其他类只有一条路能过，就是 Agent 在真实页面上验过 ---- # 当初禁止自动采纳规则，理由是「读不出真假」。实测不是读：Agent 去页面上对过了， # 这比任何人读一遍都强。所以实测通过就放行，靠猜（规则层启发式、模型意见）的一律不放行。 if category not in AUTO_ADOPT_CATEGORIES: if trial_proven(entry, policy): confirmed = int(entry.get("confirmed") or 0) return {"action": "verify", "scope": "origin", "reason": note(f"{label}类，但已被 Agent 在真实页面上验证相符 {confirmed} 次且从未被证伪")} return {"action": "hold", "scope": None, "reason": f"{label}类影响后续所有执行，必须人工确认或等 Agent 在页面上验证通过。"} if rule_verdict == "reject": return {"action": "hold", "scope": None, "reason": note("规则层与模型层意见相反", rule_reason)} evidence_ok, why = _evidence_ok(entry, policy) llm_adopts = llm_verdict == "adopt" llm_blocks = llm_active and policy.require_llm and not llm_adopts proven = trial_proven(entry, policy) # A 路：证据够硬。实测通过时不再要求规则层也点头——规则层只是启发式， # 「Agent 去页面上对过两次」压过「启发式拿不准」。但规则层明确说 reject 的上面已经拦掉了。 if evidence_ok and (rule_verdict == "adopt" or proven): if llm_blocks: return {"action": "hold", "scope": None, "reason": note("证据够但模型不认可", llm_reason[:80] or "无理由")} schools = int(entry.get("school_count") or 1) scope = "all" if schools >= policy.min_schools else "origin" return {"action": "verify", "scope": scope, "reason": note(why, "规则层与模型层一致" if llm_adopts else "规则层认可")} # 模型认可但同时想改写：原文不够好，自动采纳原文没意义，留给人确认改写 if llm_adopts and review.REWRITE_MARKER in llm_reason: return {"action": "hold", "scope": None, "reason": note("模型认可但建议改写，改写后再采纳", llm_reason[:120])} # B 路：证据薄，但模型明确背书。只发回来源学校 if llm_adopts: return {"action": "verify", "scope": "origin", "reason": note(f"{label}类，模型判定为可复用知识", llm_reason[:80], f"证据尚薄（{why}），先只发回来源学校")} if llm_active: return {"action": "hold", "scope": None, "reason": note(why, "模型也未明确认可")} return {"action": "hold", "scope": None, "reason": note(why, "未配模型 Key，无法进一步判断")} async def _evaluate(entry: dict, policy: Policy, use_llm: bool) -> dict: rule = review.rule_suggestion(entry) llm = None if use_llm and settings.llm_api_key: siblings = [ e for e in db.list_entries(status="verified", knowledge_id=entry["knowledge_id"], business_id=entry["business_id"], limit=20) if e["id"] != entry["id"] ] llm = await review.llm_suggestion(entry, siblings) plan = decide(entry, rule, llm, policy) return { "id": entry["id"], "key": entry["key"], "value": entry["value"][:200], "category": entry["category"], "school_count": entry["school_count"], "observations": entry["observations"], **plan, } async def run(dry_run: bool = True, policy: Policy | None = None, limit: int | None = None, ids: list[int] | None = None, use_llm: bool = True) -> dict: """跑一遍智能采纳。dry_run=True 只给出计划，不改任何状态。""" policy = policy or Policy.from_settings() cap = limit or policy.max_actions entries = ([db.get_entry(i) for i in ids] if ids else db.list_entries(status="draft", limit=cap * 4)) entries = [e for e in entries if e][:cap * 4] plans = await asyncio.gather(*(_evaluate(e, policy, use_llm) for e in entries)) verify = [p for p in plans if p["action"] == "verify"][:cap] reject = [p for p in plans if p["action"] == "reject"][:cap] hold = [p for p in plans if p["action"] == "hold"] if not dry_run: for plan in verify: db.review_entry(plan["id"], "verify", scope=plan["scope"], by="auto") db.save_suggestion(plan["id"], "adopt", f"自动采纳：{plan['reason']}") for plan in reject: db.review_entry(plan["id"], "reject", by="auto") db.save_suggestion(plan["id"], "reject", f"自动驳回（暂缓，再被观察到会退回队列）：{plan['reason']}") for plan in hold: db.save_suggestion(plan["id"], "review", f"留给人工：{plan['reason']}") return { "dry_run": dry_run, "examined": len(plans), "verified": len(verify), "rejected": len(reject), "held": len(hold), "policy": { "min_schools": policy.min_schools, "min_observations": policy.min_observations, "require_llm": policy.require_llm and bool(settings.llm_api_key), "auto_reject": policy.auto_reject, "max_actions": policy.max_actions, }, "plans": {"verify": verify, "reject": reject, "hold": hold[:50]}, }