aa48145557
体检修复:启动日志不再打印 AUTH_TOKEN、密钥可清空、推送失败不占整段冷却、目录名与文案统一成 plugins/。
123 lines
5.3 KiB
Python
123 lines
5.3 KiB
Python
"""题库匹配 + 大模型兜底,与平台无关。
|
||
|
||
插件负责两头:把平台的题目转成统一格式、把答案填回平台;中间「这题选什么」由这里决定:
|
||
1. 题库命中(题干归一化后:精确 -> 互相包含 -> 相似度)就直接用;
|
||
2. 没命中的题交给设置里配置的大模型;没配大模型就留空,由插件决定怎么处理。
|
||
|
||
题目格式:{"no": 题号, "type": 题型名(单选/多选/判断),"title": 题干, "options": [{"key": "A", "text": "..."}]}
|
||
题库格式(插件目录里的 *.json,可放多个):[{"q": 题干, "a": "B"}],多选写 "ABC"。
|
||
返回:{题号: {"answer": ["A", "C"], "source": "bank" | "llm"}},没答出的题不在结果里。
|
||
"""
|
||
import difflib
|
||
import json
|
||
import re
|
||
from pathlib import Path
|
||
|
||
from study_tool import runtime
|
||
from study_tool.config.settings import load_settings
|
||
from study_tool.llm import llm_chat
|
||
|
||
BATCH = 5 # 大模型每批问几题,单批失败只影响这一批
|
||
FUZZY = 0.9 # 相似度命中的下限
|
||
MIN_CONTAIN = 8 # 互相包含匹配时,两边都至少要有这么多字,免得短题干乱命中
|
||
SYSTEM = (
|
||
"你是考试答题助手。根据题干和选项判断正确答案。\n"
|
||
"单选题返回一个选项字母;多选题返回全部正确的选项字母连写(如 AC);判断题返回 A(正确)或 B(错误)。\n"
|
||
'只输出 JSON:{"answers":{"题号":"答案"}},不要输出解释或多余文字。'
|
||
)
|
||
|
||
|
||
def norm(s):
|
||
"""去掉空白和中英文标点、统一大小写,题干才能跨页面格式比较。"""
|
||
return re.sub(r"[\W_]+", "", str(s or "")).lower()
|
||
|
||
|
||
def letters(raw, options=None):
|
||
"""把 "A#C" / "AC" / ["A", "C"] 收敛成选项字母列表;有选项时只留合法字母,不是字母答案返回空。"""
|
||
keys = [o.get("key") for o in options or [] if o.get("key")]
|
||
if isinstance(raw, (list, tuple)):
|
||
raw = "".join(map(str, raw))
|
||
s = re.sub(r"[#,,、/\s]+", "", str(raw or "")).upper()
|
||
if not re.fullmatch(r"[A-F]+", s):
|
||
return []
|
||
return [c for c in dict.fromkeys(s) if not keys or c in keys]
|
||
|
||
|
||
class Bank:
|
||
def __init__(self, items):
|
||
self.items = [(norm(i["q"]), i["a"]) for i in items if i.get("q") and i.get("a")]
|
||
self.exact = dict(self.items)
|
||
|
||
def find(self, title):
|
||
"""返回题库里的答案原文,没命中返回空串。"""
|
||
t = norm(title)
|
||
if not t:
|
||
return ""
|
||
if t in self.exact:
|
||
return self.exact[t]
|
||
for q, a in self.items:
|
||
if min(len(q), len(t)) >= MIN_CONTAIN and (q in t or t in q):
|
||
return a
|
||
q, a = max(self.items, key=lambda x: difflib.SequenceMatcher(None, t, x[0]).ratio(), default=("", ""))
|
||
return a if q and difflib.SequenceMatcher(None, t, q).ratio() >= FUZZY else ""
|
||
|
||
|
||
def load(folder):
|
||
"""读目录下全部 *.json 合并成一个题库;目录不存在就是空题库。"""
|
||
items = []
|
||
for f in sorted(Path(folder).glob("*.json")):
|
||
items += json.loads(f.read_text(encoding="utf-8"))
|
||
return Bank(items)
|
||
|
||
|
||
def ask(part, cfg):
|
||
"""问大模型一批题,返回 {题号: 原始答案}。"""
|
||
lines = []
|
||
for q in part:
|
||
opts = ";".join("%s. %s" % (o.get("key"), o.get("text")) for o in q.get("options") or [])
|
||
lines.append("第%s题(%s)%s\n选项:%s" % (q.get("no"), q.get("type") or "单选", q.get("title") or "", opts or "无"))
|
||
content = llm_chat(
|
||
[{"role": "system", "content": cfg.get("prompt") or SYSTEM}, {"role": "user", "content": "\n\n".join(lines)}],
|
||
cfg,
|
||
max_tokens=1600,
|
||
)
|
||
m = re.search(r"\{.*\}", content.replace("```json", "").replace("```", ""), re.S)
|
||
data = json.loads(m.group(0)) if m else {}
|
||
data = data.get("answers") if isinstance(data.get("answers"), dict) else data
|
||
return {str(k): v for k, v in data.items()}
|
||
|
||
|
||
def llm_fill(key, miss, out):
|
||
cfg = load_settings()["llm"]
|
||
if not (cfg.get("api_key") and cfg.get("base_url")):
|
||
runtime.log(key, "题库没命中 %d 题,且还没配置大模型(右上角「设置」),这些题留空" % len(miss), level="warn")
|
||
return
|
||
before = len(out)
|
||
for i in range(0, len(miss), BATCH):
|
||
part = miss[i : i + BATCH]
|
||
try:
|
||
got = ask(part, cfg)
|
||
except Exception as e:
|
||
runtime.log(key, "AI 答题第 %d 批失败:%s" % (i // BATCH + 1, str(e)[:160]), level="warn")
|
||
continue
|
||
for q in part:
|
||
a = letters(got.get(str(q.get("no"))), q.get("options"))
|
||
if a:
|
||
out[str(q.get("no"))] = {"answer": a, "source": "llm"}
|
||
runtime.log(key, "AI 补答 %d/%d 题(答案仅供参考,未核对)" % (len(out) - before, len(miss)))
|
||
|
||
|
||
def answer(key, questions, bank=None, use_llm=True):
|
||
"""题库优先,没命中的走大模型。日志统一走 runtime.log(key, ...)。"""
|
||
out, miss = {}, []
|
||
for q in questions:
|
||
a = letters(bank.find(q.get("title")), q.get("options")) if bank else []
|
||
if a:
|
||
out[str(q.get("no"))] = {"answer": a, "source": "bank"}
|
||
else:
|
||
miss.append(q)
|
||
runtime.log(key, "题库命中 %d/%d 题" % (len(out), len(questions)))
|
||
if miss and use_llm:
|
||
llm_fill(key, miss, out)
|
||
return out
|