Files
projectAIpopular/tests/test_domains_ext.py
T
tzt ce0f6170d3 chore: T-P-1 工作区收敛——并行会话成果与历史未入库文件整理入库
- 入库历史遗漏源码/测试:router_system 9 模块(agent/executors/inference/knowledge/
  memory/planner/skills/trace)、tests 11 个测试文件、config/knowledge 领域知识
- 入库根目录方案文档(v2/v3/可行性×2)、references 文献(arxiv 14-18/cnki_open/
  参考文献清单)、research 论文素材(routerarena/paper/中文文献 PDF)
- 前端构建产物刷新(新 hash);webapp 误写文档删除
- gitignore 增补:deepseek-harness、research/_refs、.mimosa/.zcode、网关日志/pid、
  临时调试脚本、tests/e2e/node_modules、AI代理功能开发/prefix
- 基线确认:318 passed
2026-09-05 08:28:25 +08:00

168 lines
5.6 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""新领域(finance/life/education)与子领域(subdomain)测试。"""
import pytest
from router_system.classifier import RuleClassifier
from router_system.knowledge import KnowledgeBase
from router_system.models import Classification
from router_system.planner import Planner
from router_system.router import build_router
# ---------------------------------------------------------------
# 分类
# ---------------------------------------------------------------
def test_finance_classification():
clf = RuleClassifier()
for q in ["基金定投的收益率怎么计算", "房贷利率是多少", "信用卡逾期了怎么办"]:
r = clf.classify(q)
assert r.domain == "finance", f"{q} -> {r.domain}"
def test_life_classification():
clf = RuleClassifier()
for q in ["日本旅行攻略", "健身增肌计划", "家常菜谱推荐", "宠物驱虫怎么做"]:
r = clf.classify(q)
assert r.domain == "life", f"{q} -> {r.domain}"
def test_education_classification():
clf = RuleClassifier()
for q in ["考研英语怎么备考", "高效学习方法", "面试技巧有哪些"]:
r = clf.classify(q)
assert r.domain == "education", f"{q} -> {r.domain}"
# ---------------------------------------------------------------
# 知识库加载
# ---------------------------------------------------------------
def test_new_domain_rules_loaded():
kb = KnowledgeBase()
for d in ("finance", "life", "education"):
hits = kb.match("测试", domain=d) # 触发加载检查(domain 存在即可)
assert kb.rules_count() >= 60, f"规则总数应 60+,当前 {kb.rules_count()}"
# 新领域任务模板
assert kb.task_template("finance-advice") is not None
assert kb.task_template("life-guide") is not None
assert kb.task_template("edu-guide") is not None
# 新领域事实表
assert len(kb.facts("finance")) >= 5
assert len(kb.facts("life")) >= 5
assert len(kb.facts("education")) >= 5
# ---------------------------------------------------------------
# 拆解
# ---------------------------------------------------------------
def _plan(query: str, domain: str, difficulty: str = "medium"):
c = Classification(domain=domain, confidence=0.9, difficulty=difficulty)
return Planner(KnowledgeBase()).plan(query, c)
def test_finance_split_four():
g = _plan("基金定投的收益率怎么计算", "finance", "medium")
ids = [n.id for n in g.nodes()]
assert ids == ["facts", "retrieve", "conclude", "disclaimer"]
def test_life_split_three():
g = _plan("日本旅行攻略", "life", "medium")
kinds = [n.kind for n in g.nodes()]
assert kinds == ["analyze", "advise", "conclude"]
def test_edu_split_three():
g = _plan("考研英语怎么备考", "education", "medium")
kinds = [n.kind for n in g.nodes()]
assert kinds == ["analyze", "design", "conclude"]
# ---------------------------------------------------------------
# subdomain / subdomain2(三级子领域)端到端
# ---------------------------------------------------------------
@pytest.mark.asyncio
async def test_subdomain_devops():
r = build_router()
res = await r.route("git 回滚代码怎么操作")
assert res.subdomain == "devops"
assert res.subdomain2 == "git"
assert any("subdomain:devops" in s for s in res.route)
assert any("subdomain2:git" in s for s in res.route)
@pytest.mark.asyncio
async def test_subdomain_labor():
r = build_router()
res = await r.route("加班费怎么计算")
assert res.domain == "legal"
assert res.subdomain == "labor"
assert res.subdomain2 == "labor"
@pytest.mark.asyncio
async def test_subdomain_investing():
r = build_router()
res = await r.route("基金定投的收益率怎么计算")
assert res.domain == "finance"
assert res.subdomain == "investing"
# finance 强制拆解 + 风险免责
assert any("plan:multi[4]" in s for s in res.route)
assert "⚠" in res.response
@pytest.mark.asyncio
async def test_finance_facts_in_response():
r = build_router()
res = await r.route("信用卡逾期了怎么办")
assert res.domain == "finance"
assert "征信" in res.response or "逾期" in res.response
# ---------------------------------------------------------------
# 三级子领域(subdomain2
# ---------------------------------------------------------------
@pytest.mark.asyncio
async def test_subdomain2_sorting():
r = build_router()
res = await r.route("用 Python 写一个快速排序函数")
assert res.domain == "code"
assert res.subdomain == "algorithm"
assert res.subdomain2 == "sorting"
@pytest.mark.asyncio
async def test_subdomain2_first_aid():
r = build_router()
res = await r.route("烫伤后怎么处理")
assert res.domain == "medical"
assert res.subdomain == "firstaid"
assert res.subdomain2 == "first-aid"
@pytest.mark.asyncio
async def test_subdomain2_loan():
r = build_router()
res = await r.route("房贷利率是 LPR 加多少")
assert res.domain == "finance"
assert res.subdomain == "loan"
assert res.subdomain2 == "loan"
@pytest.mark.asyncio
async def test_subdomain2_cache_roundtrip():
r = build_router()
q = "信用卡逾期了怎么办"
r1 = await r.route(q)
r2 = await r.route(q) # 缓存命中
assert r2.cache_hit is True
assert r2.subdomain == r1.subdomain
assert r2.subdomain2 == r1.subdomain2
assert r2.subdomain2 == "credit"
def test_rule_subdomain2_from_map():
kb = KnowledgeBase()
r = kb.rule("code-sort")
assert r.subdomain2 == "sorting"
r2 = kb.rule("medical-firstaid")
assert r2.subdomain2 == "first-aid"