147 lines
7.0 KiB
Python
147 lines
7.0 KiB
Python
"""智能预审：在人看之前先给每条草稿一个建议和理由。
|
||
|
||
两层：
|
||
1. 规则层（不需要模型，永远可用）：多校观察、被大量观察、内容过短或像一次性数据、与已采纳条目重复。
|
||
2. 模型层（配了 Key 才跑）：判断这条是不是可复用的操作知识，还是这次任务的具体数据。
|
||
|
||
建议只是建议，永远不自动改状态——系统规则类错一条会让 Agent 以后一直绕开正确操作。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import re
|
||
|
||
from server.config import settings
|
||
from server.db import database as db
|
||
from shared import knowledge_match
|
||
|
||
# 一次性数据的特征：具体日期、班级、人名。注意可复用的知识里也会拿它们举例
|
||
# （「选项形如「计科2301班（计算机科学与技术）」」是控件技巧，不是一次性数据），
|
||
# 所以只在条目本身很短、基本只剩这些值的时候才判为一次性数据。
|
||
_ONE_OFF_HINTS = (
|
||
r"\d{4}-\d{2}-\d{2}",
|
||
r"\d{4}/\d{1,2}/\d{1,2}",
|
||
r"\d{3,4}班",
|
||
r"[一-龥]{2,3}(老师|同学)",
|
||
)
|
||
# 可复用知识的特征词：出现这些说明在讲「怎么操作」而不是「这次填了什么」
|
||
_HOW_TO_HINTS = ("点击", "点开", "下拉", "选项", "输入框", "按钮", "入口", "路径", "菜单", "先", "再", "不要", "需要", "形如")
|
||
_ONE_OFF_MAX_LEN = 40
|
||
|
||
# 模型给了改写建议时在理由里带这个前缀。智能采纳看到它就不自动采纳原文，留给人确认改写。
|
||
REWRITE_MARKER = "建议改写为："
|
||
|
||
|
||
def _looks_one_off(value: str) -> bool:
|
||
if not any(re.search(p, value) for p in _ONE_OFF_HINTS):
|
||
return False
|
||
if any(w in value for w in _HOW_TO_HINTS):
|
||
return False # 在讲操作，里面的班级 / 日期只是举例
|
||
return len(value.strip()) <= _ONE_OFF_MAX_LEN
|
||
|
||
|
||
def rule_suggestion(entry: dict) -> tuple[str, str]:
|
||
"""只看条目本身和上报情况能得出的建议。返回 (adopt / reject / review, 理由)。
|
||
|
||
顺序即优先级：跨校证据最强，排在启发式正则前面，否则举了例子的控件技巧会被误判成一次性数据。
|
||
"""
|
||
value = str(entry.get("value") or "")
|
||
category = entry.get("category") or "note"
|
||
schools = int(entry.get("school_count") or 1)
|
||
observations = int(entry.get("observations") or 1)
|
||
confirmed = int(entry.get("confirmed") or 0)
|
||
contradicted = int(entry.get("contradicted") or 0)
|
||
|
||
if len(value.strip()) < 8:
|
||
return "reject", "内容太短，几乎没有信息量。"
|
||
# 试用证据最硬：Agent 在真实页面上验过，比「几所学校推导出同一条」强
|
||
if contradicted > 0:
|
||
return "reject", f"已被 Agent 在真实页面上证伪 {contradicted} 次，不该再注入。"
|
||
if confirmed >= 2:
|
||
return "adopt", f"已被 Agent 在真实页面上验证相符 {confirmed} 次。"
|
||
if schools >= 2:
|
||
return "adopt", f"{schools} 所学校各自都观察到了这条，跨校成立的可能性高。"
|
||
if _looks_one_off(value):
|
||
return "reject", "整条基本只是具体日期 / 班级 / 人名，像是本次填报的值而不是可复用知识。"
|
||
if category in knowledge_match.AUTO_VERIFY_CATEGORIES and observations >= 3:
|
||
return "adopt", f"{knowledge_match.CATEGORY_LABELS[category]}类，已被观察 {observations} 次且无冲突。"
|
||
if category == "rule":
|
||
return "review", "系统规则类影响后续所有执行，必须人工确认来源属实。"
|
||
return "review", f"只有 1 所学校、观察 {observations} 次，建议再等一次或人工确认。"
|
||
|
||
|
||
def _prompt(entry: dict, siblings: list[dict]) -> str:
|
||
others = "\n".join(f"- [{s['id']}] {s['key']}：{s['value'][:160]}" for s in siblings[:12]) or "（无）"
|
||
return f"""你在审核一条「智能办事助手」从真实网页操作中自动沉淀的经验，判断它是否值得作为长期知识注入后续任务的提示词。
|
||
|
||
待审条目：
|
||
- 类别：{knowledge_match.CATEGORY_LABELS.get(entry.get('category'), entry.get('category'))}
|
||
- 标题：{entry.get('key')}
|
||
- 内容：{entry.get('value')}
|
||
- 上报学校数：{entry.get('school_count')}，累计观察：{entry.get('observations')}
|
||
|
||
同一业务下已经采纳的条目：
|
||
{others}
|
||
|
||
判断标准：
|
||
1. 可复用的操作知识（控件怎么操作、入口在哪、系统有什么约束）→ adopt
|
||
2. 本次任务的具体数据（某个班级、某个人名、某天的日期、某条记录的编号）→ reject
|
||
3. 与上面已采纳条目重复 → reject，并在 reason 里说明和哪条重复
|
||
4. 看起来有用但表述含糊、或需要人确认来源是否属实 → review
|
||
|
||
只输出 JSON：{{"suggestion": "adopt|reject|review", "reason": "一句话中文理由", "rewrite": "若建议改写则给出改写后的内容，否则空字符串"}}"""
|
||
|
||
|
||
async def llm_suggestion(entry: dict, siblings: list[dict]) -> tuple[str, str] | None:
|
||
"""模型预审。没配 Key 或调用失败时返回 None，由规则层兜底。"""
|
||
if not settings.llm_api_key:
|
||
return None
|
||
try:
|
||
from openai import AsyncOpenAI
|
||
|
||
client = AsyncOpenAI(api_key=settings.llm_api_key, base_url=settings.llm_base_url, timeout=45)
|
||
resp = await client.chat.completions.create(
|
||
model=settings.llm_model,
|
||
messages=[
|
||
{"role": "system", "content": "你只输出 JSON 对象，不输出任何其他文字。"},
|
||
{"role": "user", "content": _prompt(entry, siblings)},
|
||
],
|
||
temperature=0,
|
||
response_format={"type": "json_object"},
|
||
)
|
||
text = (resp.choices[0].message.content or "").strip()
|
||
text = re.sub(r"^```(?:json)?\s*|\s*```$", "", text, flags=re.S)
|
||
data = json.loads(text)
|
||
except Exception as e: # noqa: BLE001 预审失败不能影响审核台可用
|
||
return "review", f"模型预审失败，请人工判断（{type(e).__name__}）"
|
||
|
||
suggestion = str(data.get("suggestion") or "review")
|
||
if suggestion not in ("adopt", "reject", "review"):
|
||
suggestion = "review"
|
||
reason = str(data.get("reason") or "").strip()
|
||
rewrite = str(data.get("rewrite") or "").strip()
|
||
if rewrite and rewrite != entry.get("value"):
|
||
reason = f"{reason}\n{REWRITE_MARKER}{rewrite}"
|
||
return suggestion, reason
|
||
|
||
|
||
async def presuggest(entry_id: int) -> dict | None:
|
||
entry = db.get_entry(entry_id)
|
||
if not entry:
|
||
return None
|
||
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
|
||
]
|
||
rule_verdict, rule_reason = rule_suggestion(entry)
|
||
result = await llm_suggestion(entry, siblings)
|
||
if result is None:
|
||
suggestion, reason = rule_verdict, f"（规则预审）{rule_reason}"
|
||
else:
|
||
suggestion, reason = result
|
||
reason = f"{reason}\n（规则预审：{rule_reason}）"
|
||
db.save_suggestion(entry_id, suggestion, reason)
|
||
return {"id": entry_id, "suggestion": suggestion, "suggestion_note": reason}
|