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
113 lines
4.1 KiB
Python
113 lines
4.1 KiB
Python
"""专家系统内核端到端测试（L0 模式：零参数全链路）。"""
|
|
import pytest
|
|
|
|
from router_system.knowledge import KnowledgeBase
|
|
from router_system.router import Router, build_router
|
|
from router_system.experts import build_expert_pool
|
|
from router_system.judge import build_judge
|
|
from router_system.fallback import build_fallback
|
|
from router_system.cache import RouterCache
|
|
from router_system.planner import Planner
|
|
|
|
|
|
@pytest.fixture()
|
|
def es_router():
|
|
return build_router()
|
|
|
|
|
|
def _hybrid_router(expert_backend: str = "api"):
|
|
"""构造 expert_backend=api 的路由（专家池为 mock，验证切换路径）。"""
|
|
config = {"router": {"judge_fallback_threshold": 0.70},
|
|
"execution": {"expert_backend": expert_backend}}
|
|
kb = KnowledgeBase()
|
|
classifier = build_router().classifier
|
|
experts = build_expert_pool({}, ["code", "math", "legal", "medical", "general"])
|
|
judge = build_judge({"type": "rule"}, 0.70, kb=kb)
|
|
fallback = build_fallback({"type": "mock"})
|
|
return Router(classifier, experts, judge, fallback, RouterCache(),
|
|
None, config, kb=kb, planner=Planner(kb))
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_hybrid_backend_uses_expert_pool():
|
|
r = _hybrid_router("api")
|
|
res = await r.route("用 Python 写一个快速排序函数，并解释时间复杂度")
|
|
# 子任务由专家池执行（mock 专家），而非规则执行器
|
|
assert res.response
|
|
assert any(s.startswith("analyze:") for s in res.route) or "plan:multi" in " ".join(res.route)
|
|
# 专家池 mock 输出包含"mock"标识
|
|
assert "mock" in res.response or "关键点" in res.response
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dag_route_code(es_router):
|
|
r = await es_router.route("用 Python 写一个快速排序函数，并解释时间复杂度")
|
|
assert r.domain == "code"
|
|
assert r.response
|
|
# 拆解轨迹：plan:multi[4] 与 4 个节点执行
|
|
assert any("plan:multi[4]" in step for step in r.route)
|
|
node_steps = [s for s in r.route if s.startswith("analyze:") or s.startswith("design:")
|
|
or s.startswith("implement:") or s.startswith("verify:")]
|
|
assert len(node_steps) == 4
|
|
# 合并响应包含各步骤产出
|
|
assert "【code 分析】" in r.response
|
|
assert "【code 自检】" in r.response
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dag_route_math(es_router):
|
|
r = await es_router.route("求解方程 x^2 - 5x + 6 = 0")
|
|
assert r.domain == "math"
|
|
assert any("plan:multi[3]" in step for step in r.route)
|
|
assert "【math 求解】" in r.response
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_single_node_easy(es_router):
|
|
# code/easy 高置信 → 单节点不拆
|
|
r = await es_router.route("python 排序")
|
|
assert r.upgraded is False
|
|
assert any("plan:single" in step for step in r.route)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_deterministic_output(es_router):
|
|
q = "用 Python 写一个快速排序函数，并解释时间复杂度"
|
|
r1 = await es_router.route(q)
|
|
r2 = await es_router.route(q)
|
|
# 第二次缓存命中（响应一致）
|
|
assert r1.response == r2.response
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_legal_retrieves_facts(es_router):
|
|
r = await es_router.route("劳动合同到期不续签，公司需要支付经济补偿吗")
|
|
assert r.domain == "legal"
|
|
# retrieve 步骤命中知识库事实（竞业/经济补偿类）
|
|
assert "知识检索" in r.response or "经济补偿" in r.response
|
|
# 免责提示存在
|
|
assert "⚠" in r.response
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_medical_disclaimer(es_router):
|
|
r = await es_router.route("高血压患者日常饮食需要注意什么")
|
|
assert r.domain == "medical"
|
|
assert "⚠" in r.response
|
|
assert any("warning:disclaimer" in s or s.startswith("warning:") for s in r.route)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_low_confidence_direct_fallback(es_router):
|
|
r = await es_router.route("今天天气怎么样")
|
|
assert r.upgraded is True
|
|
assert "direct_fallback" in r.route
|
|
|
|
|
|
def test_health_es(es_router):
|
|
h = es_router.health()
|
|
assert h["status"] == "ok"
|
|
assert h["execution_mode"] == "rule"
|
|
assert h["rules"] > 0
|
|
assert "planner" in h
|