tzt
b2fa8c3c81
算法： - RouterCache：语义条目写入时预计算向量范数、语义查找单遍完成（消除命中后二次 O(N) 查找）、 相似度=1.0 提前终止；微基准（3000 条目×200 查询）：3986ms -> 1685ms，2.37x - TaskGraph.topo_order：O(V²logV) 重排序/成员扫描 -> 邻接表+deque 的 O(V+E) Kahn， 输出顺序契约不变（初始就绪层按插入序、循环依赖按插入序兜底、未知依赖忽略） - RuleClassifier：同分决胜按领域名字典序（与规则表排列无关），次高分 O(n) 扫描 工程卫生： - .mimosa/（扫描器工作目录）加入 .gitignore 并移出索引 - test_review 抽样测试改用内联确定性 LCG，消除 2 个低危（不安全随机数） 测试：新增 11 项（topo 契约 6 + 缓存回归 3 + 分类器 2） pytest 230 passed（基线 219 全绿 + 11）
64 lines
2.0 KiB
Python
64 lines
2.0 KiB
Python
"""TaskGraph（黑板/工作记忆）单元测试——拓扑排序契约。
|
|
|
|
契约（与 2026-09 优化前行为一致，复杂度 O(V²logV) -> O(V+E)）：
|
|
- 依赖在前；初始就绪层按插入序稳定输出
|
|
- 未知依赖 id 忽略；重复依赖不重复产出
|
|
- 循环依赖：剩余节点按插入序兜底追加（不崩溃）
|
|
"""
|
|
from router_system.memory import TaskGraph, TaskNode
|
|
|
|
|
|
def _node(nid: str, deps=()) -> TaskNode:
|
|
return TaskNode(id=nid, kind="solve", domain="general", query="q", deps=list(deps))
|
|
|
|
|
|
def test_topo_chain_order():
|
|
g = TaskGraph()
|
|
g.add_node(_node("a"))
|
|
g.add_node(_node("b", ["a"]))
|
|
g.add_node(_node("c", ["b"]))
|
|
assert [n.id for n in g.topo_order()] == ["a", "b", "c"]
|
|
|
|
|
|
def test_topo_diamond_initial_ready_by_insertion():
|
|
"""菱形依赖：初始就绪层按插入序。"""
|
|
g = TaskGraph()
|
|
g.add_node(_node("s"))
|
|
g.add_node(_node("y", ["s"])) # 先插入 y
|
|
g.add_node(_node("x", ["s"]))
|
|
g.add_node(_node("t", ["x", "y"]))
|
|
order = [n.id for n in g.topo_order()]
|
|
assert order == ["s", "y", "x", "t"]
|
|
|
|
|
|
def test_topo_independent_nodes_keep_insertion_order():
|
|
g = TaskGraph()
|
|
g.add_node(_node("n3"))
|
|
g.add_node(_node("n1"))
|
|
g.add_node(_node("n2"))
|
|
assert [n.id for n in g.topo_order()] == ["n3", "n1", "n2"]
|
|
|
|
|
|
def test_topo_unknown_dep_ignored():
|
|
g = TaskGraph()
|
|
g.add_node(_node("a", ["不存在的依赖"]))
|
|
assert [n.id for n in g.topo_order()] == ["a"]
|
|
|
|
|
|
def test_topo_cycle_fallback_by_insertion():
|
|
g = TaskGraph()
|
|
g.add_node(_node("p", ["q"]))
|
|
g.add_node(_node("q", ["p"]))
|
|
g.add_node(_node("r"))
|
|
order = [n.id for n in g.topo_order()]
|
|
# r 无依赖先行；p/q 成环按插入序兜底
|
|
assert order == ["r", "p", "q"]
|
|
|
|
|
|
def test_topo_duplicate_deps_counted_once_in_output():
|
|
"""重复依赖边不产生重复输出节点。"""
|
|
g = TaskGraph()
|
|
g.add_node(_node("a"))
|
|
g.add_node(_node("b", ["a", "a"]))
|
|
assert [n.id for n in g.topo_order()] == ["a", "b"]
|