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
69 lines
2.4 KiB
Python
69 lines
2.4 KiB
Python
"""NodeExecutor 接口抽象测试（T1：整体项目部分拆解·先行实现）。"""
|
|
import pytest
|
|
|
|
from router_system.executors import (ModelNodeExecutor, RuleNodeExecutor,
|
|
build_node_executor)
|
|
from router_system.experts import build_expert_pool
|
|
from router_system.knowledge import KnowledgeBase
|
|
from router_system.memory import TaskNode, WorkingMemory
|
|
from router_system.router import build_router
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rule_node_executor_deterministic():
|
|
ex = RuleNodeExecutor(kb=KnowledgeBase())
|
|
node = TaskNode(id="t1", kind="solve", domain="math", query="求解方程 x^2-5x+6=0")
|
|
mem = WorkingMemory()
|
|
r1 = await ex.execute(node, "math", "medium", mem)
|
|
r2 = await ex.execute(node, "math", "medium", mem)
|
|
assert r1.text == r2.text # 确定性
|
|
assert "求解" in r1.text
|
|
assert r1.model_used.startswith("rule:")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_model_node_executor_uses_expert_pool():
|
|
experts = build_expert_pool({}, ["code", "math", "legal", "medical",
|
|
"finance", "life", "education", "general"])
|
|
ex = ModelNodeExecutor(experts)
|
|
node = TaskNode(id="t1", kind="implement", domain="code", query="写一个函数")
|
|
mem = WorkingMemory()
|
|
resp = await ex.execute(node, "code", "medium", mem)
|
|
assert resp.text # mock 专家输出
|
|
assert "mock" in resp.text
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_model_node_executor_fallback_to_general():
|
|
# 未知领域 → general 专家兜底
|
|
experts = {"general": build_expert_pool({}, ["general"])["general"]}
|
|
ex = ModelNodeExecutor(experts)
|
|
node = TaskNode(id="t1", kind="explain", domain="unknown", query="测试")
|
|
resp = await ex.execute(node, "unknown", "easy", WorkingMemory())
|
|
assert resp.text
|
|
|
|
|
|
def test_factory_rule():
|
|
ex = build_node_executor("rule", kb=KnowledgeBase())
|
|
assert isinstance(ex, RuleNodeExecutor)
|
|
|
|
|
|
def test_factory_model():
|
|
experts = build_expert_pool({}, ["code"])
|
|
ex = build_node_executor("api", experts=experts)
|
|
assert isinstance(ex, ModelNodeExecutor)
|
|
|
|
|
|
def test_factory_invalid():
|
|
with pytest.raises(ValueError):
|
|
build_node_executor("bad-backend")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_router_uses_node_executor():
|
|
r = build_router()
|
|
assert r.node_executor.name == "rule"
|
|
res = await r.route("求解方程 x^2 - 5x + 6 = 0")
|
|
assert res.domain == "math"
|
|
assert res.response
|