483 lines
19 KiB
Python
483 lines
19 KiB
Python
"""ExpertTeamRouter 单元测试"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from agentkit.experts.config import ExpertConfig, ExpertTemplate
|
||
from agentkit.experts.registry import ExpertTemplateRegistry
|
||
from agentkit.experts.router import (
|
||
ExpertTeamRouter,
|
||
ExpertTeamRoutingResult,
|
||
TEAM_PREFIX_PATTERN,
|
||
)
|
||
|
||
|
||
# ── 辅助函数 ──────────────────────────────────────────────
|
||
|
||
|
||
def _make_template(
|
||
name: str = "test_template",
|
||
persona: str = "测试专家",
|
||
bound_skills: list[str] | None = None,
|
||
) -> ExpertTemplate:
|
||
"""创建测试用 ExpertTemplate 实例"""
|
||
config = ExpertConfig(
|
||
name=name,
|
||
agent_type="expert",
|
||
persona=persona,
|
||
thinking_style="analytical",
|
||
bound_skills=bound_skills or [],
|
||
task_mode="llm_generate",
|
||
prompt={"identity": persona},
|
||
)
|
||
return ExpertTemplate(
|
||
name=name,
|
||
config=config,
|
||
is_builtin=True,
|
||
description=f"{name} 模板",
|
||
)
|
||
|
||
|
||
def _make_registry_with_templates() -> ExpertTemplateRegistry:
|
||
"""创建包含预注册模板的注册中心"""
|
||
registry = ExpertTemplateRegistry()
|
||
registry.register(_make_template("analyst", persona="数据分析师", bound_skills=["data_query"]))
|
||
registry.register(_make_template("strategist", persona="策略专家", bound_skills=["planning"]))
|
||
registry.register(_make_template("reviewer", persona="代码审查员", bound_skills=["code_review"]))
|
||
return registry
|
||
|
||
|
||
# ── TEAM_PREFIX_PATTERN 正则测试 ──────────────────────────
|
||
|
||
|
||
class TestTeamPrefixPattern:
|
||
"""TEAM_PREFIX_PATTERN 正则匹配测试"""
|
||
|
||
def test_match_team_only(self):
|
||
"""@team 前缀匹配"""
|
||
match = TEAM_PREFIX_PATTERN.match("@team 分析这个数据")
|
||
assert match is not None
|
||
assert match.group(1) is None
|
||
assert match.group(2) == "分析这个数据"
|
||
|
||
def test_match_team_with_experts(self):
|
||
"""@team:expert1,expert2 前缀匹配"""
|
||
match = TEAM_PREFIX_PATTERN.match("@team:analyst,strategist 分析这个数据")
|
||
assert match is not None
|
||
assert match.group(1) == "analyst,strategist"
|
||
assert match.group(2) == "分析这个数据"
|
||
|
||
def test_match_team_single_expert(self):
|
||
"""@team:expert 前缀匹配"""
|
||
match = TEAM_PREFIX_PATTERN.match("@team:analyst 分析数据")
|
||
assert match is not None
|
||
assert match.group(1) == "analyst"
|
||
assert match.group(2) == "分析数据"
|
||
|
||
def test_match_team_no_task(self):
|
||
"""@team 无后续任务内容"""
|
||
match = TEAM_PREFIX_PATTERN.match("@team")
|
||
assert match is not None
|
||
assert match.group(1) is None
|
||
assert match.group(2) == ""
|
||
|
||
def test_no_match_without_prefix(self):
|
||
"""无 @team 前缀不匹配"""
|
||
match = TEAM_PREFIX_PATTERN.match("分析这个数据")
|
||
assert match is None
|
||
|
||
def test_no_match_team_in_middle(self):
|
||
"""@team 在中间不匹配（必须开头）"""
|
||
match = TEAM_PREFIX_PATTERN.match("请 @team 分析数据")
|
||
assert match is None
|
||
|
||
def test_match_team_with_leading_whitespace(self):
|
||
"""@team 前有空白字符时正则不匹配（由 resolve() 中 strip() 处理）"""
|
||
match = TEAM_PREFIX_PATTERN.match(" @team:analyst 任务内容")
|
||
# 正则使用 ^ 锚定，前导空白不匹配
|
||
assert match is None
|
||
|
||
|
||
# ── ExpertTeamRoutingResult 默认值测试 ────────────────────
|
||
|
||
|
||
class TestExpertTeamRoutingResult:
|
||
"""ExpertTeamRoutingResult 数据类测试"""
|
||
|
||
def test_default_values(self):
|
||
"""默认值验证"""
|
||
result = ExpertTeamRoutingResult()
|
||
assert result.matched is False
|
||
assert result.team_mode is False
|
||
assert result.specified_experts == []
|
||
assert result.task_content == ""
|
||
assert result.auto_compose is False
|
||
assert result.match_method == ""
|
||
|
||
|
||
# ── ExpertTeamRouter.resolve 测试 ──────────────────────────
|
||
|
||
|
||
class TestExpertTeamRouterResolve:
|
||
"""ExpertTeamRouter.resolve 方法测试"""
|
||
|
||
def test_team_prefix_triggers_team_mode(self):
|
||
"""@team 前缀触发团队模式"""
|
||
router = ExpertTeamRouter()
|
||
result = router.resolve("@team 分析这个数据")
|
||
assert result.matched is True
|
||
assert result.team_mode is True
|
||
assert result.match_method == "explicit_team"
|
||
assert result.task_content == "分析这个数据"
|
||
|
||
def test_team_with_experts_specifies_members(self):
|
||
"""@team:analyst,strategist 指定专家成员"""
|
||
router = ExpertTeamRouter()
|
||
result = router.resolve("@team:analyst,strategist 分析数据")
|
||
assert result.matched is True
|
||
assert result.team_mode is True
|
||
assert result.specified_experts == ["analyst", "strategist"]
|
||
assert result.auto_compose is False
|
||
assert result.match_method == "explicit_team"
|
||
|
||
def test_team_no_experts_uses_default_template(self):
|
||
"""@team 无指定专家时使用默认 dev_team 模板"""
|
||
router = ExpertTeamRouter()
|
||
result = router.resolve("@team 分析数据")
|
||
assert result.matched is True
|
||
assert result.team_mode is True
|
||
# 默认 dev_team 模板包含 5 个成员
|
||
assert result.specified_experts == [
|
||
"tech_lead",
|
||
"frontend_engineer",
|
||
"backend_engineer",
|
||
"qa_engineer",
|
||
"code_reviewer",
|
||
]
|
||
assert result.auto_compose is False
|
||
|
||
def test_team_with_expert_extracts_task(self):
|
||
"""@team:analyst 正确提取任务内容"""
|
||
router = ExpertTeamRouter()
|
||
result = router.resolve("@team:analyst 请分析这份报告")
|
||
assert result.matched is True
|
||
assert result.specified_experts == ["analyst"]
|
||
assert result.task_content == "请分析这份报告"
|
||
|
||
def test_no_team_prefix_no_team_mode(self):
|
||
"""无 @team 前缀不触发团队模式"""
|
||
router = ExpertTeamRouter()
|
||
result = router.resolve("普通问题")
|
||
assert result.matched is False
|
||
assert result.team_mode is False
|
||
assert result.task_content == "普通问题"
|
||
assert result.match_method == ""
|
||
|
||
def test_nonexistent_expert_still_included(self):
|
||
"""指定不存在的专家名仍包含在列表中"""
|
||
router = ExpertTeamRouter()
|
||
result = router.resolve("@team:analyst,nonexistent 任务")
|
||
assert result.specified_experts == ["analyst", "nonexistent"]
|
||
|
||
def test_team_with_empty_task_uses_full_content(self):
|
||
"""@team 无任务内容时 task_content 使用原始内容"""
|
||
router = ExpertTeamRouter()
|
||
result = router.resolve("@team")
|
||
assert result.task_content == "@team"
|
||
# 无指定专家时使用默认模板
|
||
assert result.auto_compose is False
|
||
assert len(result.specified_experts) == 5
|
||
|
||
def test_resolve_strips_leading_whitespace(self):
|
||
"""resolve() 对前导空白做 strip()"""
|
||
router = ExpertTeamRouter()
|
||
result = router.resolve(" @team:analyst 任务内容")
|
||
assert result.matched is True
|
||
assert result.team_mode is True
|
||
assert result.specified_experts == ["analyst"]
|
||
assert result.task_content == "任务内容"
|
||
|
||
def test_invalid_expert_names_rejected(self):
|
||
"""无效专家名被过滤掉"""
|
||
router = ExpertTeamRouter()
|
||
# 包含特殊字符的名称应被过滤
|
||
result = router.resolve("@team:analyst,inva!id,bad/name 任务")
|
||
# 仅保留合法名称
|
||
assert "analyst" in result.specified_experts
|
||
assert "inva!id" not in result.specified_experts
|
||
assert "bad/name" not in result.specified_experts
|
||
|
||
def test_max_experts_limit(self):
|
||
"""指定超过 MAX_EXPERTS 数量的专家时截断"""
|
||
router = ExpertTeamRouter()
|
||
# 构造 15 个专家名
|
||
names = ",".join(f"expert{i}" for i in range(15))
|
||
result = router.resolve(f"@team:{names} 任务")
|
||
from agentkit.experts.router import MAX_EXPERTS
|
||
|
||
assert len(result.specified_experts) == MAX_EXPERTS
|
||
|
||
|
||
# ── ExpertTeamRouter.resolve_expert_configs 测试 ───────────
|
||
|
||
|
||
class TestExpertTeamRouterResolveExpertConfigs:
|
||
"""ExpertTeamRouter.resolve_expert_configs 方法测试"""
|
||
|
||
def test_resolve_existing_templates(self):
|
||
"""解析已注册模板返回对应 ExpertConfig"""
|
||
registry = _make_registry_with_templates()
|
||
router = ExpertTeamRouter(template_registry=registry)
|
||
|
||
configs = router.resolve_expert_configs(["analyst", "strategist"])
|
||
assert len(configs) == 2
|
||
assert configs[0].name == "analyst"
|
||
assert configs[0].persona == "数据分析师"
|
||
assert configs[1].name == "strategist"
|
||
assert configs[1].persona == "策略专家"
|
||
|
||
def test_resolve_nonexistent_creates_dynamic_config(self):
|
||
"""解析不存在的名称创建动态 ExpertConfig"""
|
||
registry = _make_registry_with_templates()
|
||
router = ExpertTeamRouter(template_registry=registry)
|
||
|
||
configs = router.resolve_expert_configs(["analyst", "unknown_expert"])
|
||
assert len(configs) == 2
|
||
assert configs[0].name == "analyst"
|
||
assert configs[0].persona == "数据分析师"
|
||
# 动态生成的配置
|
||
assert configs[1].name == "unknown_expert"
|
||
assert configs[1].persona == "Expert in unknown_expert"
|
||
assert configs[1].agent_type == "expert"
|
||
assert configs[1].thinking_style == "analytical"
|
||
assert configs[1].bound_skills == []
|
||
assert configs[1].is_lead is False
|
||
assert configs[1].task_mode == "llm_generate"
|
||
|
||
def test_resolve_all_nonexistent(self):
|
||
"""所有名称都不存在时全部动态生成"""
|
||
router = ExpertTeamRouter()
|
||
configs = router.resolve_expert_configs(["role_a", "role_b"])
|
||
assert len(configs) == 2
|
||
assert configs[0].name == "role_a"
|
||
assert configs[0].persona == "Expert in role_a"
|
||
assert configs[1].name == "role_b"
|
||
assert configs[1].persona == "Expert in role_b"
|
||
|
||
def test_resolve_empty_list(self):
|
||
"""空列表返回空结果"""
|
||
router = ExpertTeamRouter()
|
||
configs = router.resolve_expert_configs([])
|
||
assert configs == []
|
||
|
||
def test_resolve_preserves_template_skills(self):
|
||
"""解析已注册模板保留 bound_skills"""
|
||
registry = _make_registry_with_templates()
|
||
router = ExpertTeamRouter(template_registry=registry)
|
||
|
||
configs = router.resolve_expert_configs(["analyst"])
|
||
assert configs[0].bound_skills == ["data_query"]
|
||
|
||
def test_resolve_first_expert_is_lead(self):
|
||
"""第一个专家被指定为 lead"""
|
||
registry = _make_registry_with_templates()
|
||
router = ExpertTeamRouter(template_registry=registry)
|
||
|
||
configs = router.resolve_expert_configs(["analyst", "strategist", "reviewer"])
|
||
assert configs[0].is_lead is True
|
||
assert configs[1].is_lead is False
|
||
assert configs[2].is_lead is False
|
||
|
||
def test_resolve_skips_invalid_names(self):
|
||
"""跳过无效的专家名"""
|
||
router = ExpertTeamRouter()
|
||
# 包含无效字符的名称应被跳过
|
||
configs = router.resolve_expert_configs(["valid_name", "inva!id", "also_valid"])
|
||
assert len(configs) == 2
|
||
assert configs[0].name == "valid_name"
|
||
assert configs[1].name == "also_valid"
|
||
|
||
|
||
# ── ExpertTeamRouter 构造测试 ─────────────────────────────
|
||
|
||
|
||
class TestExpertTeamRouterInit:
|
||
"""ExpertTeamRouter 构造函数测试"""
|
||
|
||
def test_default_registry(self):
|
||
"""无参构造创建默认注册中心"""
|
||
router = ExpertTeamRouter()
|
||
assert router._registry is not None
|
||
|
||
def test_custom_registry(self):
|
||
"""传入自定义注册中心"""
|
||
registry = ExpertTemplateRegistry()
|
||
router = ExpertTeamRouter(template_registry=registry)
|
||
assert router._registry is registry
|
||
|
||
|
||
# ── ExpertTeamRouter.can_handle 测试 ──────────────────────
|
||
|
||
|
||
class TestExpertTeamRouterCanHandle:
|
||
"""ExpertTeamRouter.can_handle 方法测试"""
|
||
|
||
def test_can_handle_with_matching_template_name(self):
|
||
"""模板名出现在内容中时返回 True"""
|
||
registry = _make_registry_with_templates()
|
||
router = ExpertTeamRouter(template_registry=registry)
|
||
|
||
assert router.can_handle("请 analyst 帮我分析") is True
|
||
|
||
def test_can_handle_with_matching_description(self):
|
||
"""模板描述词出现在内容中时返回 True"""
|
||
registry = _make_registry_with_templates()
|
||
router = ExpertTeamRouter(template_registry=registry)
|
||
|
||
# 描述为 "analyst 模板" 等，"模板" 长度 > 2 应匹配
|
||
assert router.can_handle("我需要一个模板来参考") is True
|
||
|
||
def test_can_handle_no_templates(self):
|
||
"""无注册模板时返回 False"""
|
||
router = ExpertTeamRouter()
|
||
# 默认注册中心可能为空
|
||
# 强制清空以测试
|
||
router._registry._templates.clear()
|
||
assert router.can_handle("任何内容") is False
|
||
|
||
def test_can_handle_with_templates_no_match(self):
|
||
"""有模板但内容不匹配时仍返回 True（auto-compose 可组建团队）"""
|
||
registry = _make_registry_with_templates()
|
||
router = ExpertTeamRouter(template_registry=registry)
|
||
|
||
# 内容与任何模板名/描述都不匹配，但有模板存在 → auto-compose 可用
|
||
assert router.can_handle("完全无关的内容 xyz123") is True
|
||
|
||
|
||
# ── ExpertTeamRouter 团队模板展开测试 (U7) ─────────────────
|
||
|
||
|
||
class TestExpertTeamRouterTemplateExpansion:
|
||
"""ExpertTeamRouter 团队模板展开测试 — @team:dev_team 自动展开为成员列表"""
|
||
|
||
def test_team_template_expands_to_members(self):
|
||
"""@team:dev_team 展开为 dev_team 模板的 bound_skills 成员列表"""
|
||
registry = ExpertTemplateRegistry()
|
||
# 注册 dev_team 模板（bound_skills 存储成员列表）
|
||
registry.register(
|
||
_make_template(
|
||
"dev_team",
|
||
persona="编程团队",
|
||
bound_skills=["tech_lead", "frontend_engineer", "backend_engineer"],
|
||
)
|
||
)
|
||
router = ExpertTeamRouter(template_registry=registry)
|
||
|
||
result = router.resolve("@team:dev_team 实现用户登录功能")
|
||
assert result.matched is True
|
||
assert result.team_mode is True
|
||
assert result.specified_experts == ["tech_lead", "frontend_engineer", "backend_engineer"]
|
||
assert result.auto_compose is False
|
||
assert result.match_method == "explicit_team"
|
||
|
||
def test_team_template_respects_max_experts_limit(self):
|
||
"""团队模板展开时遵守 MAX_EXPERTS 上限"""
|
||
registry = ExpertTemplateRegistry()
|
||
# 注册一个 bound_skills 超过 MAX_EXPERTS 的模板
|
||
registry.register(
|
||
_make_template(
|
||
"dev_team",
|
||
persona="编程团队",
|
||
bound_skills=[f"expert{i}" for i in range(15)],
|
||
)
|
||
)
|
||
router = ExpertTeamRouter(template_registry=registry)
|
||
|
||
result = router.resolve("@team:dev_team 任务")
|
||
from agentkit.experts.router import MAX_EXPERTS
|
||
|
||
assert len(result.specified_experts) == MAX_EXPERTS
|
||
|
||
def test_team_template_not_found_falls_back_to_default_members(self):
|
||
"""@team:unknown_template 时未知模板名作为普通专家处理"""
|
||
registry = ExpertTemplateRegistry()
|
||
router = ExpertTeamRouter(template_registry=registry)
|
||
|
||
result = router.resolve("@team:unknown_template 任务")
|
||
# 未知模板名（无 bound_skills）→ 作为普通专家名处理
|
||
assert result.specified_experts == ["unknown_template"]
|
||
|
||
def test_no_experts_uses_default_dev_team_template(self):
|
||
"""@team 无指定专家时从 dev_team 模板加载成员"""
|
||
registry = ExpertTemplateRegistry()
|
||
registry.register(
|
||
_make_template(
|
||
"dev_team",
|
||
persona="编程团队",
|
||
bound_skills=["tech_lead", "frontend_engineer", "backend_engineer"],
|
||
)
|
||
)
|
||
router = ExpertTeamRouter(template_registry=registry)
|
||
|
||
result = router.resolve("@team 实现功能")
|
||
assert result.specified_experts == ["tech_lead", "frontend_engineer", "backend_engineer"]
|
||
assert result.auto_compose is False
|
||
|
||
def test_no_experts_no_dev_team_template_uses_hardcoded_fallback(self):
|
||
"""无 dev_team 模板时使用硬编码 fallback 列表"""
|
||
registry = ExpertTemplateRegistry()
|
||
router = ExpertTeamRouter(template_registry=registry)
|
||
|
||
result = router.resolve("@team 任务")
|
||
assert result.specified_experts == [
|
||
"tech_lead",
|
||
"frontend_engineer",
|
||
"backend_engineer",
|
||
"qa_engineer",
|
||
"code_reviewer",
|
||
]
|
||
|
||
def test_team_template_with_empty_bound_skills_treated_as_expert_name(self):
|
||
"""bound_skills 为空的模板名作为普通专家处理"""
|
||
registry = ExpertTemplateRegistry()
|
||
registry.register(_make_template("empty_team", persona="空团队", bound_skills=[]))
|
||
router = ExpertTeamRouter(template_registry=registry)
|
||
|
||
result = router.resolve("@team:empty_team 任务")
|
||
# bound_skills 为空 → 不展开，作为普通专家名
|
||
assert result.specified_experts == ["empty_team"]
|
||
|
||
def test_default_template_class_attribute(self):
|
||
"""DEFAULT_TEMPLATE 类属性为 'dev_team'"""
|
||
from agentkit.experts.router import ExpertTeamRouter as Router
|
||
|
||
assert Router.DEFAULT_TEMPLATE == "dev_team"
|
||
|
||
def test_load_default_template_members_with_template(self):
|
||
"""_load_default_template_members 从 dev_team 模板加载"""
|
||
registry = ExpertTemplateRegistry()
|
||
registry.register(
|
||
_make_template(
|
||
"dev_team",
|
||
persona="编程团队",
|
||
bound_skills=["a", "b", "c"],
|
||
)
|
||
)
|
||
router = ExpertTeamRouter(template_registry=registry)
|
||
|
||
members = router._load_default_template_members()
|
||
assert members == ["a", "b", "c"]
|
||
|
||
def test_load_default_template_members_without_template(self):
|
||
"""_load_default_template_members 无模板时返回硬编码列表"""
|
||
registry = ExpertTemplateRegistry()
|
||
router = ExpertTeamRouter(template_registry=registry)
|
||
|
||
members = router._load_default_template_members()
|
||
assert members == [
|
||
"tech_lead",
|
||
"frontend_engineer",
|
||
"backend_engineer",
|
||
"qa_engineer",
|
||
"code_reviewer",
|
||
]
|