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
This commit is contained in:
tzt
2026-09-05 08:28:25 +08:00
parent 747d85c3ba
commit ce0f6170d3
82 changed files with 74132 additions and 36 deletions
+164
View File
@@ -0,0 +1,164 @@
"""Skill 技能体系 + RouteAgent 测试(T12Agent-Skill 路由器·先行实现)。"""
import pytest
from router_system.agent import RouteAgent
from router_system.classifier import RuleClassifier
from router_system.fallback import build_fallback
from router_system.judge import build_judge
from router_system.knowledge import KnowledgeBase
from router_system.planner import Planner
from router_system.skills import (FallbackSkill, JudgeSkill, KBAnswerSkill,
KBRetrieveSkill, Skill, SkillContext,
SkillRegistry, TemplateSkill,
build_skill_registry)
def _registry():
kb = KnowledgeBase()
judge = build_judge({"type": "rule"}, 0.70, kb=kb)
fallback = build_fallback({"type": "mock"})
return build_skill_registry(kb=kb, judge=judge, fallback=fallback)
def _agent():
kb = KnowledgeBase()
clf = RuleClassifier()
return RouteAgent(
classifier=clf, planner=Planner(kb), kb=kb,
judge=build_judge({"type": "rule"}, 0.70, kb=kb),
fallback=build_fallback({"type": "mock"}),
)
# ---------------------------------------------------------------
# Skill 注册表
# ---------------------------------------------------------------
def test_registry_registers_builtins():
reg = _registry()
names = reg.list()
assert "es.analyze" in names
assert "es.solve" in names
assert "es.disclaimer" in names
assert "kb.retrieve" in names
assert "kb.answer" in names
assert "judge.evaluate" in names
assert "fallback.call" in names
assert len(names) >= 20 # 17 模板 + kb×2 + judge + fallback
def test_registry_duplicate_rejected():
reg = SkillRegistry()
class S(Skill):
name = "x"
reg.register(S())
with pytest.raises(ValueError):
reg.register(S())
def test_registry_unknown_skill():
reg = _registry()
with pytest.raises(KeyError):
import asyncio
asyncio.run(reg.execute("no.such.skill", SkillContext(
query="q", domain="general", difficulty="easy", memory=None)))
def test_catalog_shape():
reg = _registry()
catalog = reg.catalog()
assert catalog and all({"name", "description", "params"} <= set(c) for c in catalog)
# ---------------------------------------------------------------
# 技能执行
# ---------------------------------------------------------------
@pytest.mark.asyncio
async def test_template_skill_executes():
reg = _registry()
kb = KnowledgeBase()
from router_system.memory import WorkingMemory
text = await reg.execute("es.solve", SkillContext(
query="求解方程 x^2-5x+6=0", domain="math", difficulty="medium",
memory=WorkingMemory(), kb=kb))
assert "求解" in text
@pytest.mark.asyncio
async def test_kb_retrieve_skill():
reg = _registry()
from router_system.memory import WorkingMemory
text = await reg.execute("kb.retrieve", SkillContext(
query="加班费怎么计算", domain="legal", difficulty="easy",
memory=WorkingMemory(), kb=KnowledgeBase()))
assert "150%" in text # 命中加班费知识条目
@pytest.mark.asyncio
async def test_kb_answer_skill():
reg = _registry()
from router_system.memory import WorkingMemory
text = await reg.execute("kb.answer", SkillContext(
query="git 回滚代码怎么操作", domain="code", difficulty="easy",
memory=WorkingMemory(), kb=KnowledgeBase()))
assert "git 核心概念" in text
# ---------------------------------------------------------------
# RouteAgent 端到端(无用户指定:完全自主分析 + 技能调用)
# ---------------------------------------------------------------
@pytest.mark.asyncio
async def test_agent_code_flow_multi_skills():
agent = _agent()
res = await agent.route("用 Python 写一个快速排序函数,并解释时间复杂度")
assert res.domain == "code"
assert res.response
# 多技能调用链:4 个 es.* 技能
skill_steps = [s for s in res.route if s.startswith("skill:es.")]
assert len(skill_steps) == 4
assert any("es.analyze" in s for s in skill_steps)
assert any("es.implement" in s for s in skill_steps)
assert res.request_id
@pytest.mark.asyncio
async def test_agent_legal_uses_retrieve_and_disclaimer():
agent = _agent()
res = await agent.route("加班费怎么计算")
assert res.domain == "legal"
# 技能组合:kb.retrieve(法条)+ es.disclaimer(免责)
assert any("skill:kb.retrieve" in s for s in res.route)
assert any("skill:es.disclaimer" in s for s in res.route)
assert "150%" in res.response
assert "" in res.response
@pytest.mark.asyncio
async def test_agent_low_confidence_fallback():
agent = _agent()
res = await agent.route("今天天气怎么样")
assert res.upgraded is True
assert "direct_fallback" in res.route
assert "fallback" in res.model_used or "mock" in res.model_used
@pytest.mark.asyncio
async def test_agent_no_user_domain_group():
# Agent 模式:不要求 domain_group,完全自主分析
agent = _agent()
res = await agent.route("基金定投的收益率怎么计算")
assert res.domain == "finance"
assert res.subdomain == "investing"
assert res.domain_group is None # 无用户指定概念
@pytest.mark.asyncio
async def test_agent_trace_and_catalog():
agent = _agent()
res = await agent.route("求解方程 x^2 - 5x + 6 = 0")
trace = agent.trace_store.get(res.request_id)
assert trace is not None
assert trace["domain"] == "math"
catalog = agent.skills_catalog()
assert any(s["name"] == "kb.retrieve" for s in catalog)