projectAIpopular/tests/test_skills_agent.py

"""Skill 技能体系 + RouteAgent 测试（T12：Agent-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)