238 lines
9.2 KiB
Python
238 lines
9.2 KiB
Python
"""意图识别与字段抽取。
|
||
|
||
一次 OpenAI 兼容接口调用（DeepSeek / 千问 DashScope 都兼容），要求模型只输出 JSON：
|
||
{"business_id": "...", "confidence": 0.9, "inputs": {"字段名": "值"}, "reply": "缺字段时的追问"}
|
||
接口不可用时退回关键词匹配 + 正则抽取，保证流程不中断。
|
||
"""
|
||
|
||
import json
|
||
import re
|
||
from datetime import date, timedelta
|
||
from typing import Any
|
||
|
||
from backend.browser.llm_factory import get_llm_config
|
||
|
||
WEEKDAYS = "一二三四五六日"
|
||
|
||
|
||
# ---------- prompt ----------
|
||
|
||
|
||
def _field_desc(f: dict) -> str:
|
||
parts = [f"{f['name']}（{f['label']}"]
|
||
if f.get("required"):
|
||
parts.append("，必填")
|
||
if f.get("type") == "date":
|
||
parts.append("，格式 YYYY-MM-DD")
|
||
if f.get("hint"):
|
||
parts.append(f"，{f['hint']}")
|
||
return "".join(parts) + "）"
|
||
|
||
|
||
def _catalog(businesses: list[dict]) -> str:
|
||
lines = []
|
||
for b in businesses:
|
||
fields = "、".join(_field_desc(f) for f in b.get("required_inputs", [])) or "无"
|
||
lines.append(f"- id: {b['id']} | 名称: {b['name']} | 适用: {b.get('intent', '')} | 字段: {fields}")
|
||
return "\n".join(lines)
|
||
|
||
|
||
def _build_prompt(message: str, businesses: list[dict], existing: dict, fixed_business_id: str | None) -> str:
|
||
today = date.today()
|
||
lines = [
|
||
"你是学校办事助手的意图识别与资料抽取模块。",
|
||
f"今天是 {today.isoformat()}（星期{WEEKDAYS[today.weekday()]}）。",
|
||
"",
|
||
"可选业务：",
|
||
_catalog(businesses),
|
||
"",
|
||
]
|
||
if fixed_business_id:
|
||
lines.append(f"业务已经确定为 {fixed_business_id}，不要更改，只需从用户这句话里补充抽取字段。")
|
||
if existing:
|
||
shown = {k: v for k, v in existing.items() if k not in ("goal", "business_id", "user_notes") and v}
|
||
if shown:
|
||
lines.append(f"已有资料（无需重复抽取，除非用户明确修改）：{json.dumps(shown, ensure_ascii=False)}")
|
||
lines += [
|
||
f"用户说：{message}",
|
||
"",
|
||
"输出一个 JSON 对象，字段：",
|
||
' business_id：最匹配的业务 id；都不匹配时用 "generic_explore"',
|
||
" confidence：0 到 1 的小数",
|
||
" inputs：从用户话里抽到的字段，键用字段名。日期统一 YYYY-MM-DD，「今天/明天/昨天/下周一」按今天换算；没提到的字段不要编造，也不要输出空字符串",
|
||
" reply：如果该业务还有必填字段没抽到，用一句自然的中文向用户追问，把缺的项列出来；否则为空字符串",
|
||
"只输出 JSON，不要解释。",
|
||
]
|
||
return "\n".join(lines)
|
||
|
||
|
||
# ---------- LLM ----------
|
||
|
||
|
||
def _parse_json(text: str) -> dict | None:
|
||
text = (text or "").strip()
|
||
if text.startswith("```"):
|
||
text = re.sub(r"^```(?:json)?\s*|\s*```$", "", text, flags=re.S)
|
||
try:
|
||
obj = json.loads(text)
|
||
return obj if isinstance(obj, dict) else None
|
||
except ValueError:
|
||
pass
|
||
m = re.search(r"\{.*\}", text, flags=re.S)
|
||
if m:
|
||
try:
|
||
obj = json.loads(m.group(0))
|
||
return obj if isinstance(obj, dict) else None
|
||
except ValueError:
|
||
return None
|
||
return None
|
||
|
||
|
||
async def _call_llm(prompt: str, cfg: dict) -> dict | None:
|
||
"""一次 JSON 调用。主 Key 402 余额不足时切兜底 Key 重发（cfg 来自 llm_factory.get_llm_config）。"""
|
||
from openai import AsyncOpenAI
|
||
|
||
from backend.browser import llm_keys
|
||
|
||
messages = [
|
||
{"role": "system", "content": "你只输出 JSON 对象，不输出任何其他文字。"},
|
||
{"role": "user", "content": prompt},
|
||
]
|
||
|
||
async def ask(api_key: str):
|
||
client = AsyncOpenAI(api_key=api_key, base_url=cfg["base_url"], timeout=45)
|
||
try:
|
||
return await client.chat.completions.create(
|
||
model=cfg["model"], messages=messages, temperature=0, response_format={"type": "json_object"}
|
||
)
|
||
except Exception as error: # noqa: BLE001 千问部分模型不支持 response_format，去掉再试一次；余额不足直接抛
|
||
if llm_keys.is_balance_error(error):
|
||
raise
|
||
return await client.chat.completions.create(model=cfg["model"], messages=messages, temperature=0)
|
||
|
||
key = cfg["api_key"]
|
||
try:
|
||
resp = await ask(key)
|
||
if key and key == cfg.get("primary_api_key"):
|
||
llm_keys.clear_primary_exhausted()
|
||
except Exception as error: # noqa: BLE001
|
||
fallback = cfg.get("fallback_api_key") or ""
|
||
if fallback and key != fallback and llm_keys.is_balance_error(error):
|
||
llm_keys.mark_primary_exhausted(str(error))
|
||
resp = await ask(fallback)
|
||
else:
|
||
raise
|
||
return _parse_json(resp.choices[0].message.content or "")
|
||
|
||
|
||
# ---------- 关键词退路 ----------
|
||
|
||
|
||
def _resolve_relative_date(text: str) -> str | None:
|
||
today = date.today()
|
||
table = {"今天": 0, "今日": 0, "明天": 1, "明日": 1, "后天": 2, "昨天": -1, "昨日": -1, "前天": -2}
|
||
for k, d in table.items():
|
||
if k in text:
|
||
return (today + timedelta(days=d)).isoformat()
|
||
m = re.search(r"(\d{4})[-/年.](\d{1,2})[-/月.](\d{1,2})", text)
|
||
if m:
|
||
return f"{m.group(1)}-{int(m.group(2)):02d}-{int(m.group(3)):02d}"
|
||
m = re.search(r"(\d{1,2})月(\d{1,2})[日号]", text)
|
||
if m:
|
||
return f"{today.year}-{int(m.group(1)):02d}-{int(m.group(2)):02d}"
|
||
return None
|
||
|
||
|
||
def _phrase_score(phrase: str, message: str) -> int:
|
||
"""短语与用户原话的最长公共子串长度（少于 2 个字不算）。
|
||
|
||
不能要求整段短语原样出现：知识里写的是「每月查课日志」，用户说的是「帮我填今天的查课日志」，
|
||
整段匹配为 0，只有最长公共子串「查课日志」能把这条业务认出来。
|
||
"""
|
||
import difflib
|
||
|
||
if len(phrase) < 2:
|
||
return 0
|
||
match = difflib.SequenceMatcher(None, phrase, message, autojunk=False).find_longest_match(0, len(phrase), 0, len(message))
|
||
return match.size if match.size >= 2 else 0
|
||
|
||
|
||
def _fallback(message: str, businesses: list[dict], fixed_business_id: str | None) -> dict:
|
||
"""模型不可用时的保底路由：按业务名 / intent 与用户原话的字面重合度打分。"""
|
||
best, best_score = None, 0
|
||
if fixed_business_id:
|
||
best = next((b for b in businesses if b["id"] == fixed_business_id), None)
|
||
else:
|
||
for b in businesses:
|
||
if b.get("id") == "generic_explore":
|
||
continue # 兜底业务不参与打分，避免它的 intent（「查找某个入口」等）抢走具体业务
|
||
phrases = [p for p in re.split(r"[、，,/ \s]+", f"{b.get('name', '')} {b.get('intent', '')}") if p]
|
||
score = sum(_phrase_score(p, message) for p in phrases)
|
||
if score > best_score:
|
||
best, best_score = b, score
|
||
if best is None:
|
||
best = next((b for b in businesses if b["id"] == "generic_explore"), businesses[0])
|
||
|
||
inputs: dict[str, str] = {}
|
||
for f in best.get("required_inputs", []):
|
||
if f.get("type") == "date":
|
||
d = _resolve_relative_date(message)
|
||
if d:
|
||
inputs[f["name"]] = d
|
||
elif f["name"] in ("class_name", "class"):
|
||
cleaned = re.sub(r"班级[是为:：]?", " ", message)
|
||
m = re.search(r"([一-龥A-Za-z]{1,6}\d{3,4}班?)", cleaned)
|
||
if m:
|
||
inputs[f["name"]] = m.group(1)
|
||
return {"business_id": best["id"], "confidence": 0.4 if best_score else 0.2, "inputs": inputs, "reply": ""}
|
||
|
||
|
||
# ---------- 对外 ----------
|
||
|
||
|
||
async def classify_and_extract(
|
||
message: str,
|
||
businesses: list[dict],
|
||
existing: dict | None = None,
|
||
fixed_business_id: str | None = None,
|
||
) -> dict:
|
||
"""返回 {business_id, confidence, inputs, reply, source, error}。inputs 只含该业务声明过的字段。"""
|
||
existing = existing or {}
|
||
cfg = get_llm_config()
|
||
result: dict | None = None
|
||
source = "llm"
|
||
error = ""
|
||
if cfg.get("api_key"):
|
||
try:
|
||
result = await _call_llm(_build_prompt(message, businesses, existing, fixed_business_id), cfg)
|
||
except Exception as e: # noqa: BLE001
|
||
error = f"{type(e).__name__}: {e}"
|
||
if not result:
|
||
result = _fallback(message, businesses, fixed_business_id)
|
||
source = "fallback"
|
||
|
||
ids = {b["id"] for b in businesses}
|
||
bid = fixed_business_id or str(result.get("business_id") or "")
|
||
if bid not in ids:
|
||
bid = "generic_explore" if "generic_explore" in ids else businesses[0]["id"]
|
||
business = next(b for b in businesses if b["id"] == bid)
|
||
declared = {f["name"] for f in business.get("required_inputs", [])}
|
||
raw_inputs = result.get("inputs") or {}
|
||
inputs = {
|
||
k: str(v).strip()
|
||
for k, v in raw_inputs.items()
|
||
if k in declared and v is not None and str(v).strip() and str(v).strip().lower() not in ("null", "none", "未提及")
|
||
}
|
||
try:
|
||
confidence = float(result.get("confidence", 0.5))
|
||
except (TypeError, ValueError):
|
||
confidence = 0.5
|
||
return {
|
||
"business_id": bid,
|
||
"confidence": max(0.0, min(1.0, confidence)),
|
||
"inputs": inputs,
|
||
"reply": str(result.get("reply") or "").strip(),
|
||
"source": source,
|
||
"error": error,
|
||
}
|