283 lines
10 KiB
Python
283 lines
10 KiB
Python
"""U5: SkillConfig 扩展 + 专业 Agent 执行模式路由测试"""
|
|
|
|
import os
|
|
import pytest
|
|
from datetime import datetime, timezone
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import yaml
|
|
|
|
from agentkit.skills.base import SkillConfig
|
|
from agentkit.core.exceptions import ConfigValidationError
|
|
from agentkit.core.protocol import TaskMessage
|
|
|
|
|
|
def _make_task(**overrides):
|
|
defaults = dict(
|
|
task_id="t1",
|
|
agent_name="test",
|
|
task_type="test",
|
|
priority=1,
|
|
input_data={"query": "test"},
|
|
callback_url=None,
|
|
created_at=datetime.now(timezone.utc),
|
|
)
|
|
defaults.update(overrides)
|
|
return TaskMessage(**defaults)
|
|
|
|
|
|
class TestSkillConfigExecutionModes:
|
|
"""SkillConfig.VALID_EXECUTION_MODES 扩展测试"""
|
|
|
|
def test_rewoo_is_valid_mode(self):
|
|
config = SkillConfig(
|
|
name="test_rewoo",
|
|
agent_type="test",
|
|
execution_mode="rewoo",
|
|
prompt={"identity": "test", "instructions": "test"},
|
|
)
|
|
assert config.execution_mode == "rewoo"
|
|
|
|
def test_plan_exec_is_valid_mode(self):
|
|
config = SkillConfig(
|
|
name="test_plan_exec",
|
|
agent_type="test",
|
|
execution_mode="plan_exec",
|
|
prompt={"identity": "test", "instructions": "test"},
|
|
)
|
|
assert config.execution_mode == "plan_exec"
|
|
|
|
def test_reflexion_is_valid_mode(self):
|
|
config = SkillConfig(
|
|
name="test_reflexion",
|
|
agent_type="test",
|
|
execution_mode="reflexion",
|
|
prompt={"identity": "test", "instructions": "test"},
|
|
)
|
|
assert config.execution_mode == "reflexion"
|
|
|
|
def test_existing_modes_still_valid(self):
|
|
for mode in ("react", "direct", "custom"):
|
|
config = SkillConfig(
|
|
name=f"test_{mode}",
|
|
agent_type="test",
|
|
execution_mode=mode,
|
|
prompt={"identity": "test", "instructions": "test"},
|
|
)
|
|
assert config.execution_mode == mode
|
|
|
|
def test_invalid_mode_raises_error(self):
|
|
with pytest.raises(ConfigValidationError):
|
|
SkillConfig(
|
|
name="test_invalid",
|
|
agent_type="test",
|
|
execution_mode="nonexistent",
|
|
prompt={"identity": "test", "instructions": "test"},
|
|
)
|
|
|
|
def test_all_six_modes_in_valid_set(self):
|
|
expected = {"react", "direct", "custom", "rewoo", "plan_exec", "reflexion"}
|
|
assert SkillConfig.VALID_EXECUTION_MODES == expected
|
|
|
|
|
|
class TestYAMLConfigLoading:
|
|
"""专业 Agent YAML 配置加载测试"""
|
|
|
|
YAML_DIR = "/Users/Chiguyong/Code/Fischer/fischer-agentkit/configs/skills"
|
|
|
|
def _load_yaml(self, filename):
|
|
path = os.path.join(self.YAML_DIR, filename)
|
|
with open(path) as f:
|
|
return yaml.safe_load(f)
|
|
|
|
def test_rewoo_agent_yaml_loads(self):
|
|
data = self._load_yaml("rewoo_agent.yaml")
|
|
config = SkillConfig(**data)
|
|
assert config.execution_mode == "rewoo"
|
|
assert config.agent_type == "parallel_data_fetch"
|
|
assert config.fallback_strategies == ["simplified_rewoo", "react", "direct"]
|
|
|
|
def test_plan_exec_agent_yaml_loads(self):
|
|
data = self._load_yaml("plan_exec_agent.yaml")
|
|
config = SkillConfig(**data)
|
|
assert config.execution_mode == "plan_exec"
|
|
assert config.agent_type == "structured_planning"
|
|
|
|
def test_reflexion_agent_yaml_loads(self):
|
|
data = self._load_yaml("reflexion_agent.yaml")
|
|
config = SkillConfig(**data)
|
|
assert config.execution_mode == "reflexion"
|
|
assert config.agent_type == "high_precision"
|
|
|
|
def test_react_agent_yaml_loads(self):
|
|
data = self._load_yaml("react_agent.yaml")
|
|
config = SkillConfig(**data)
|
|
assert config.execution_mode == "react"
|
|
assert config.agent_type == "dynamic_tool_chain"
|
|
|
|
def test_direct_agent_yaml_loads(self):
|
|
data = self._load_yaml("direct_agent.yaml")
|
|
config = SkillConfig(**data)
|
|
assert config.execution_mode == "direct"
|
|
assert config.agent_type == "simple_generation"
|
|
|
|
def test_all_agents_use_default_model(self):
|
|
"""All agent YAMLs use model: 'default' (LLM gateway resolves the actual provider)."""
|
|
direct_data = self._load_yaml("direct_agent.yaml")
|
|
assert direct_data["llm"]["model"] == "default"
|
|
|
|
plan_data = self._load_yaml("plan_exec_agent.yaml")
|
|
assert plan_data["llm"]["model"] == "default"
|
|
|
|
react_data = self._load_yaml("react_agent.yaml")
|
|
assert react_data["llm"]["model"] == "default"
|
|
|
|
def test_direct_agent_has_no_tools(self):
|
|
data = self._load_yaml("direct_agent.yaml")
|
|
assert data["tools"] == []
|
|
|
|
def test_capabilities_parsed(self):
|
|
data = self._load_yaml("react_agent.yaml")
|
|
config = SkillConfig(**data)
|
|
cap_tags = [c.tag if hasattr(c, "tag") else c for c in config.capabilities]
|
|
assert "dynamic_adaptation" in cap_tags
|
|
|
|
|
|
class TestConfigDrivenAgentRouting:
|
|
"""ConfigDrivenAgent execution_mode 路由测试"""
|
|
|
|
def _make_agent(self, execution_mode):
|
|
from agentkit.core.config_driven import ConfigDrivenAgent
|
|
from agentkit.llm.gateway import LLMGateway
|
|
|
|
config = SkillConfig(
|
|
name=f"test_{execution_mode}",
|
|
agent_type="test",
|
|
execution_mode=execution_mode,
|
|
prompt={"identity": "test", "instructions": "test"},
|
|
)
|
|
|
|
llm_gateway = MagicMock(spec=LLMGateway)
|
|
llm_gateway.chat = AsyncMock()
|
|
|
|
agent = ConfigDrivenAgent(config=config, llm_gateway=llm_gateway)
|
|
return agent
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rewoo_routes_to_handle_rewoo(self):
|
|
agent = self._make_agent("rewoo")
|
|
with patch.object(
|
|
agent, "_handle_rewoo", new_callable=AsyncMock, return_value={"content": "rewoo result"}
|
|
) as mock:
|
|
result = await agent.handle_task(_make_task())
|
|
mock.assert_called_once()
|
|
assert result == {"content": "rewoo result"}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_plan_exec_routes_to_handle_plan_exec(self):
|
|
agent = self._make_agent("plan_exec")
|
|
with patch.object(
|
|
agent,
|
|
"_handle_plan_exec",
|
|
new_callable=AsyncMock,
|
|
return_value={"content": "plan_exec result"},
|
|
) as mock:
|
|
result = await agent.handle_task(_make_task())
|
|
mock.assert_called_once()
|
|
assert result == {"content": "plan_exec result"}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reflexion_routes_to_handle_reflexion(self):
|
|
agent = self._make_agent("reflexion")
|
|
with patch.object(
|
|
agent,
|
|
"_handle_reflexion",
|
|
new_callable=AsyncMock,
|
|
return_value={"content": "reflexion result"},
|
|
) as mock:
|
|
result = await agent.handle_task(_make_task())
|
|
mock.assert_called_once()
|
|
assert result == {"content": "reflexion result"}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_react_still_routes_correctly(self):
|
|
agent = self._make_agent("react")
|
|
with patch.object(
|
|
agent, "_handle_react", new_callable=AsyncMock, return_value={"content": "react result"}
|
|
) as mock:
|
|
result = await agent.handle_task(_make_task())
|
|
mock.assert_called_once()
|
|
assert result == {"content": "react result"}
|
|
|
|
|
|
class TestFallbackStrategiesWiring:
|
|
"""Verify fallback_strategies flows from SkillConfig -> ReWOOEngine (#5)"""
|
|
|
|
@staticmethod
|
|
def _make_fake_engine(captured_kwargs: dict):
|
|
"""Build a FakeReWOOEngine that records kwargs and returns a result with .output."""
|
|
|
|
class FakeReWOOEngine:
|
|
def __init__(self, **kwargs):
|
|
captured_kwargs.update(kwargs)
|
|
|
|
async def execute(self, **kwargs):
|
|
class _Result:
|
|
output = "rewoo result"
|
|
|
|
return _Result()
|
|
|
|
return FakeReWOOEngine
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fallback_strategies_passed_to_rewoo_engine(self):
|
|
"""SkillConfig.fallback_strategies must reach ReWOOEngine constructor."""
|
|
from agentkit.core.config_driven import ConfigDrivenAgent
|
|
from agentkit.llm.gateway import LLMGateway
|
|
|
|
config = SkillConfig(
|
|
name="test_rewoo_wiring",
|
|
agent_type="test",
|
|
execution_mode="rewoo",
|
|
prompt={"identity": "test", "instructions": "test"},
|
|
fallback_strategies=["simplified_rewoo", "direct"],
|
|
)
|
|
|
|
llm_gateway = MagicMock(spec=LLMGateway)
|
|
llm_gateway.chat = AsyncMock()
|
|
agent = ConfigDrivenAgent(config=config, llm_gateway=llm_gateway)
|
|
|
|
captured_kwargs: dict = {}
|
|
FakeReWOOEngine = self._make_fake_engine(captured_kwargs)
|
|
|
|
# ReWOOEngine is imported lazily inside _handle_rewoo, so patch at source.
|
|
with patch("agentkit.core.rewoo.ReWOOEngine", FakeReWOOEngine):
|
|
await agent._handle_rewoo(_make_task())
|
|
|
|
assert captured_kwargs.get("fallback_strategies") == ["simplified_rewoo", "direct"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fallback_strategies_none_when_not_configured(self):
|
|
"""When fallback_strategies is None, ReWOOEngine receives None (uses defaults)."""
|
|
from agentkit.core.config_driven import ConfigDrivenAgent
|
|
from agentkit.llm.gateway import LLMGateway
|
|
|
|
config = SkillConfig(
|
|
name="test_rewoo_no_fallback",
|
|
agent_type="test",
|
|
execution_mode="rewoo",
|
|
prompt={"identity": "test", "instructions": "test"},
|
|
)
|
|
|
|
llm_gateway = MagicMock(spec=LLMGateway)
|
|
llm_gateway.chat = AsyncMock()
|
|
agent = ConfigDrivenAgent(config=config, llm_gateway=llm_gateway)
|
|
|
|
captured_kwargs: dict = {}
|
|
FakeReWOOEngine = self._make_fake_engine(captured_kwargs)
|
|
|
|
with patch("agentkit.core.rewoo.ReWOOEngine", FakeReWOOEngine):
|
|
await agent._handle_rewoo(_make_task())
|
|
|
|
assert captured_kwargs.get("fallback_strategies") is None
|