251 lines
12 KiB
Python
251 lines
12 KiB
Python
"""意图识别：JSON 解析、相对日期、关键词退路、以及模型不可用时 classify_and_extract 仍然给出可用结果。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
from datetime import date, timedelta
|
||
|
||
import pytest
|
||
|
||
from backend import nlu
|
||
|
||
|
||
class TestParseJson:
|
||
def test_plain_object(self):
|
||
assert nlu._parse_json('{"business_id": "a", "confidence": 0.9}') == {"business_id": "a", "confidence": 0.9}
|
||
|
||
def test_fenced_block(self):
|
||
assert nlu._parse_json('```json\n{"business_id": "a"}\n```') == {"business_id": "a"}
|
||
|
||
def test_object_embedded_in_prose(self):
|
||
# 模型偶尔会在 JSON 前后带一句话，取第一个到最后一个花括号之间的内容
|
||
assert nlu._parse_json('好的，结果如下：\n{"business_id": "a"}\n以上。') == {"business_id": "a"}
|
||
|
||
@pytest.mark.parametrize("text", ["", " ", "根本没有 JSON", "{不是合法 JSON}", "[1, 2, 3]"])
|
||
def test_unusable_returns_none(self, text):
|
||
assert nlu._parse_json(text) is None
|
||
|
||
def test_top_level_array_rejected(self):
|
||
# 只接受对象；数组不是调用方期望的形状
|
||
assert nlu._parse_json('[{"business_id": "a"}]') is None
|
||
|
||
|
||
class TestRelativeDate:
|
||
@pytest.mark.parametrize("word,offset", [("今天", 0), ("今日", 0), ("明天", 1), ("后天", 2), ("昨天", -1), ("前天", -2)])
|
||
def test_relative_words(self, word, offset):
|
||
assert nlu._resolve_relative_date(f"帮我填{word}的查课日志") == (date.today() + timedelta(days=offset)).isoformat()
|
||
|
||
@pytest.mark.parametrize("text,expected", [
|
||
("听课时间 2026-09-17", "2026-09-17"),
|
||
("听课时间 2026/9/7", "2026-09-07"),
|
||
("听课时间 2026年9月7日", "2026-09-07"),
|
||
])
|
||
def test_absolute_dates_normalised(self, text, expected):
|
||
assert nlu._resolve_relative_date(text) == expected
|
||
|
||
def test_month_day_uses_current_year(self):
|
||
assert nlu._resolve_relative_date("9月7日那节课") == f"{date.today().year}-09-07"
|
||
|
||
def test_no_date_returns_none(self):
|
||
assert nlu._resolve_relative_date("帮我看看有什么能办的") is None
|
||
|
||
|
||
class TestPhraseScore:
|
||
"""退路打分用最长公共子串，不能要求整段短语原样出现。"""
|
||
|
||
def test_partial_overlap_counts(self):
|
||
# 知识里写「每月查课日志」，用户说「帮我填今天的查课日志」：整段不匹配，公共子串「查课日志」4 个字
|
||
assert nlu._phrase_score("每月查课日志", "帮我填今天的查课日志") == 4
|
||
|
||
def test_single_character_overlap_ignored(self):
|
||
assert nlu._phrase_score("辅导员填写谈心谈话记录", "帮我填查课日志") == 0
|
||
|
||
def test_too_short_phrase_ignored(self):
|
||
assert nlu._phrase_score("的", "帮我填今天的查课日志") == 0
|
||
|
||
def test_no_overlap(self):
|
||
assert nlu._phrase_score("请假销假", "今天天气怎么样") == 0
|
||
|
||
|
||
class TestKeywordFallback:
|
||
@pytest.mark.parametrize("message", [
|
||
"帮我填查课日志",
|
||
"帮我填今天的查课日志，班级计科2301班",
|
||
"我要填每月查课要求的日志",
|
||
"听课记录还好还好",
|
||
])
|
||
def test_recognises_business_from_varied_phrasings(self, knowledge, message):
|
||
# 回归：早先按整段短语匹配，这几句全都落到 generic_explore，模型不可用时等于没有保底
|
||
assert nlu._fallback(message, knowledge["businesses"], None)["business_id"] == "counselor_worklog"
|
||
|
||
def test_matched_business_gets_higher_confidence(self, knowledge):
|
||
assert nlu._fallback("帮我填查课日志", knowledge["businesses"], None)["confidence"] == 0.4
|
||
|
||
def test_unmatched_message_falls_back_to_generic_explore(self, knowledge):
|
||
result = nlu._fallback("今天天气怎么样", knowledge["businesses"], None)
|
||
assert result["business_id"] == "generic_explore"
|
||
assert result["confidence"] == 0.2
|
||
|
||
def test_generic_explore_does_not_outscore_a_specific_business(self, knowledge):
|
||
# generic_explore 的 intent 里有「查找某个入口」「有哪些字段」，不排除它就会抢走具体业务
|
||
assert nlu._fallback("帮我查一下课的日志入口", knowledge["businesses"], None)["business_id"] == "counselor_worklog"
|
||
|
||
def test_fixed_business_id_wins_over_keywords(self, knowledge):
|
||
result = nlu._fallback("帮我填查课日志", knowledge["businesses"], "generic_explore")
|
||
assert result["business_id"] == "generic_explore"
|
||
|
||
def test_extracts_class_and_date_without_model(self, knowledge):
|
||
result = nlu._fallback("帮我填今天的查课日志，班级计科2301班", knowledge["businesses"], None)
|
||
assert result["inputs"]["class_name"] == "计科2301班"
|
||
assert result["inputs"]["observe_date"] == date.today().isoformat()
|
||
|
||
def test_class_label_prefix_stripped(self, knowledge):
|
||
# 「班级是计科2301班」里的「班级是」不能被当成班级名的一部分
|
||
result = nlu._fallback("班级是计科2301班", knowledge["businesses"], "counselor_worklog")
|
||
assert result["inputs"]["class_name"] == "计科2301班"
|
||
|
||
def test_only_declared_fields_extracted(self, knowledge):
|
||
# generic_explore 没声明任何字段，即使原话里有班级也不抽
|
||
result = nlu._fallback("帮我看看计科2301班", knowledge["businesses"], "generic_explore")
|
||
assert result["inputs"] == {}
|
||
|
||
|
||
class TestClassifyAndExtract:
|
||
async def test_uses_model_result_and_filters_undeclared_fields(self, knowledge, monkeypatch):
|
||
async def fake_llm(prompt, cfg):
|
||
return {
|
||
"business_id": "counselor_worklog",
|
||
"confidence": 0.95,
|
||
"inputs": {"class_name": "计科2301班", "course_name": "高等数学", "不存在的字段": "x"},
|
||
}
|
||
|
||
monkeypatch.setattr(nlu, "get_llm_config", lambda: {"api_key": "k", "model": "m", "base_url": "u"})
|
||
monkeypatch.setattr(nlu, "_call_llm", fake_llm)
|
||
result = await nlu.classify_and_extract("帮我填查课日志", knowledge["businesses"])
|
||
assert result["source"] == "llm"
|
||
assert result["business_id"] == "counselor_worklog"
|
||
assert result["inputs"] == {"class_name": "计科2301班", "course_name": "高等数学"}
|
||
|
||
async def test_model_failure_falls_back_to_keywords(self, knowledge, monkeypatch):
|
||
async def boom(prompt, cfg):
|
||
raise RuntimeError("Error code: 402 - Insufficient Balance")
|
||
|
||
monkeypatch.setattr(nlu, "get_llm_config", lambda: {"api_key": "k", "model": "m", "base_url": "u"})
|
||
monkeypatch.setattr(nlu, "_call_llm", boom)
|
||
result = await nlu.classify_and_extract("帮我填今天的查课日志，班级计科2301班", knowledge["businesses"])
|
||
assert result["source"] == "fallback"
|
||
assert result["business_id"] == "counselor_worklog"
|
||
assert result["inputs"]["class_name"] == "计科2301班"
|
||
assert "402" in result["error"]
|
||
|
||
async def test_no_api_key_skips_model_entirely(self, knowledge, monkeypatch):
|
||
async def must_not_be_called(prompt, cfg):
|
||
raise AssertionError("没有 Key 时不应该调用模型")
|
||
|
||
monkeypatch.setattr(nlu, "get_llm_config", lambda: {"api_key": "", "model": "m", "base_url": "u"})
|
||
monkeypatch.setattr(nlu, "_call_llm", must_not_be_called)
|
||
result = await nlu.classify_and_extract("帮我填查课日志", knowledge["businesses"])
|
||
assert result["source"] == "fallback"
|
||
|
||
async def test_unknown_business_id_from_model_is_corrected(self, knowledge, monkeypatch):
|
||
async def fake_llm(prompt, cfg):
|
||
return {"business_id": "模型编出来的业务", "confidence": 0.9, "inputs": {}}
|
||
|
||
monkeypatch.setattr(nlu, "get_llm_config", lambda: {"api_key": "k", "model": "m", "base_url": "u"})
|
||
monkeypatch.setattr(nlu, "_call_llm", fake_llm)
|
||
result = await nlu.classify_and_extract("随便说点什么", knowledge["businesses"])
|
||
assert result["business_id"] == "generic_explore"
|
||
|
||
@pytest.mark.parametrize("value", ["null", "None", "未提及", "", " "])
|
||
async def test_placeholder_values_are_dropped(self, knowledge, monkeypatch, value):
|
||
async def fake_llm(prompt, cfg):
|
||
return {"business_id": "counselor_worklog", "confidence": 0.9, "inputs": {"class_name": value}}
|
||
|
||
monkeypatch.setattr(nlu, "get_llm_config", lambda: {"api_key": "k", "model": "m", "base_url": "u"})
|
||
monkeypatch.setattr(nlu, "_call_llm", fake_llm)
|
||
result = await nlu.classify_and_extract("帮我填查课日志", knowledge["businesses"])
|
||
assert "class_name" not in result["inputs"]
|
||
|
||
async def test_confidence_clamped_to_unit_range(self, knowledge, monkeypatch):
|
||
async def fake_llm(prompt, cfg):
|
||
return {"business_id": "counselor_worklog", "confidence": 9.9, "inputs": {}}
|
||
|
||
monkeypatch.setattr(nlu, "get_llm_config", lambda: {"api_key": "k", "model": "m", "base_url": "u"})
|
||
monkeypatch.setattr(nlu, "_call_llm", fake_llm)
|
||
result = await nlu.classify_and_extract("帮我填查课日志", knowledge["businesses"])
|
||
assert result["confidence"] == 1.0
|
||
|
||
async def test_existing_inputs_are_offered_to_the_prompt(self, knowledge, monkeypatch):
|
||
seen = {}
|
||
|
||
async def fake_llm(prompt, cfg):
|
||
seen["prompt"] = prompt
|
||
return {"business_id": "counselor_worklog", "confidence": 0.9, "inputs": {}}
|
||
|
||
monkeypatch.setattr(nlu, "get_llm_config", lambda: {"api_key": "k", "model": "m", "base_url": "u"})
|
||
monkeypatch.setattr(nlu, "_call_llm", fake_llm)
|
||
await nlu.classify_and_extract("补充一下", knowledge["businesses"], existing={"class_name": "计科2301班"})
|
||
assert "计科2301班" in seen["prompt"]
|
||
|
||
|
||
class TestCallLlmKeyFallback:
|
||
"""主 Key 402 时 _call_llm 必须切到兜底 Key 重发，而不是让任务失败。"""
|
||
|
||
@staticmethod
|
||
def _stub_openai(monkeypatch, keys_seen: list[str]):
|
||
import openai
|
||
|
||
class FakeCompletions:
|
||
def __init__(self, key):
|
||
self._key = key
|
||
|
||
async def create(self, **kwargs):
|
||
keys_seen.append(self._key)
|
||
if self._key == "PRIMARY":
|
||
import httpx
|
||
|
||
raise openai.APIStatusError(
|
||
"Error code: 402 - Insufficient Balance",
|
||
response=httpx.Response(402, request=httpx.Request("POST", "http://x")),
|
||
body=None,
|
||
)
|
||
return type("R", (), {"choices": [type("C", (), {"message": type("M", (), {"content": '{"ok": 1}'})()})()]})()
|
||
|
||
class FakeClient:
|
||
def __init__(self, api_key, base_url, timeout):
|
||
self.chat = type("Chat", (), {"completions": FakeCompletions(api_key)})()
|
||
|
||
monkeypatch.setattr(openai, "AsyncOpenAI", FakeClient)
|
||
|
||
async def test_switches_to_fallback_on_402(self, monkeypatch):
|
||
from backend.browser import llm_keys
|
||
|
||
marked = {}
|
||
monkeypatch.setattr(llm_keys, "mark_primary_exhausted", lambda reason="": marked.update(hit=True))
|
||
monkeypatch.setattr(llm_keys, "clear_primary_exhausted", lambda: None)
|
||
keys_seen: list[str] = []
|
||
self._stub_openai(monkeypatch, keys_seen)
|
||
|
||
cfg = {"api_key": "PRIMARY", "primary_api_key": "PRIMARY", "fallback_api_key": "FALLBACK",
|
||
"base_url": "http://x", "model": "m"}
|
||
assert await nlu._call_llm("hi", cfg) == {"ok": 1}
|
||
assert keys_seen == ["PRIMARY", "FALLBACK"]
|
||
assert marked.get("hit") is True
|
||
|
||
async def test_without_fallback_the_402_propagates(self, monkeypatch):
|
||
keys_seen: list[str] = []
|
||
self._stub_openai(monkeypatch, keys_seen)
|
||
cfg = {"api_key": "PRIMARY", "primary_api_key": "PRIMARY", "fallback_api_key": "",
|
||
"base_url": "http://x", "model": "m"}
|
||
with pytest.raises(Exception, match="402"):
|
||
await nlu._call_llm("hi", cfg)
|
||
assert keys_seen == ["PRIMARY"]
|
||
|
||
|
||
def test_prompt_lists_every_business_and_required_field(knowledge):
|
||
prompt = nlu._build_prompt("帮我填查课日志", knowledge["businesses"], {}, None)
|
||
assert "counselor_worklog" in prompt and "generic_explore" in prompt
|
||
assert "class_name" in prompt and "observe_date" in prompt
|
||
assert json.dumps # 提示词要求模型输出 JSON
|
||
assert "JSON" in prompt.upper()
|