tzt
ce0f6170d3
- 入库历史遗漏源码/测试：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
131 lines
4.5 KiB
Python
131 lines
4.5 KiB
Python
"""知识库扩充验证测试（规则/模板/事实表/output 规则端到端）。"""
|
|
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
|
|
|
|
|
|
def _kb():
|
|
return KnowledgeBase()
|
|
|
|
|
|
def _plan(query: str, domain: str, difficulty: str = "medium"):
|
|
c = Classification(domain=domain, confidence=0.9, difficulty=difficulty)
|
|
return Planner(_kb()).plan(query, c)
|
|
|
|
|
|
# ---------------------------------------------------------------
|
|
# 规则 / 模板 / 事实表规模
|
|
# ---------------------------------------------------------------
|
|
def test_expanded_rules_count():
|
|
kb = _kb()
|
|
assert kb.rules_count() >= 40, f"知识库应扩充至 40+ 条规则，当前 {kb.rules_count()}"
|
|
|
|
|
|
def test_new_task_templates_exist():
|
|
kb = _kb()
|
|
required = ["code-algorithm", "code-refactor", "code-explain", "code-test",
|
|
"math-proof", "math-optimize", "medical-firstaid", "general-writing"]
|
|
for tid in required:
|
|
assert kb.task_template(tid) is not None, f"缺少任务模板 {tid}"
|
|
|
|
|
|
def test_expanded_facts():
|
|
kb = _kb()
|
|
legal = kb.facts("legal")
|
|
medical = kb.facts("medical")
|
|
assert len(legal) >= 10
|
|
assert len(medical) >= 10
|
|
# 新事实条目存在
|
|
legal_text = " ".join(f["statement"] for f in legal)
|
|
assert "150%" in legal_text # 加班费
|
|
assert "押金" in legal_text # 租房押金
|
|
assert "无理由" in legal_text # 消费者退货
|
|
assert "继承" in legal_text # 继承顺序
|
|
medical_text = " ".join(f["statement"] for f in medical)
|
|
assert "烫伤" in medical_text # 烫伤急救
|
|
assert "抗生素" in medical_text # 抗生素
|
|
assert "失眠" in medical_text # 失眠
|
|
|
|
|
|
# ---------------------------------------------------------------
|
|
# 新拆解行为
|
|
# ---------------------------------------------------------------
|
|
def test_algorithm_split_five():
|
|
g = _plan("用动态规划实现背包问题", "code", "hard")
|
|
ids = [n.id for n in g.nodes()]
|
|
assert ids == ["analyze", "design", "implement", "complexity", "verify"]
|
|
|
|
|
|
def test_refactor_split_three():
|
|
g = _plan("这段代码重复太多，帮我重构", "code", "hard")
|
|
ids = [n.id for n in g.nodes()]
|
|
assert ids == ["analyze", "refactor", "verify"]
|
|
|
|
|
|
def test_code_explain_split():
|
|
g = _plan("帮我解释这段代码什么意思", "code", "medium")
|
|
kinds = [n.kind for n in g.nodes()]
|
|
assert kinds == ["analyze", "explain", "verify"]
|
|
|
|
|
|
def test_testcase_split():
|
|
g = _plan("给这个函数写单元测试", "code", "medium")
|
|
kinds = [n.kind for n in g.nodes()]
|
|
assert "testcase" in kinds
|
|
|
|
|
|
def test_math_proof_and_optimize():
|
|
g = _plan("证明勾股定理", "math", "hard")
|
|
assert len(g) == 3
|
|
g2 = _plan("求函数 f(x)=x^2 的最小值", "math", "medium")
|
|
kinds = [n.kind for n in g2.nodes()]
|
|
assert "optimize" in kinds
|
|
|
|
|
|
def test_firstaid_split():
|
|
g = _plan("烫伤后怎么处理", "medical", "medium")
|
|
kinds = [n.kind for n in g.nodes()]
|
|
assert kinds == ["analyze", "advise", "disclaimer"]
|
|
|
|
|
|
def test_writing_split():
|
|
g = _plan("写一封请假邮件", "general", "medium")
|
|
kinds = [n.kind for n in g.nodes()]
|
|
assert kinds == ["analyze", "draft", "polish"]
|
|
|
|
|
|
# ---------------------------------------------------------------
|
|
# output 知识规则端到端
|
|
# ---------------------------------------------------------------
|
|
@pytest.mark.asyncio
|
|
async def test_git_knowledge_in_response():
|
|
from router_system.router import build_router
|
|
r = build_router()
|
|
res = await r.route("git 回滚代码怎么操作")
|
|
assert "git 核心概念" in res.response
|
|
assert "git revert" in res.response
|
|
# 推理轨迹记录规则触发
|
|
assert any("code-git-knowledge" in s for s in res.route)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_docker_knowledge_in_response():
|
|
from router_system.router import build_router
|
|
r = build_router()
|
|
res = await r.route("docker 部署一个服务")
|
|
assert "镜像 vs 容器" in res.response
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_medical_hypertension_full_flow():
|
|
from router_system.router import build_router
|
|
r = build_router()
|
|
res = await r.route("高血压患者可以吃哪些降压药，副作用是什么")
|
|
assert res.domain == "medical"
|
|
assert any("plan:multi[3]" in s for s in res.route)
|
|
assert "⚠" in res.response
|
|
assert res.upgraded is False
|