439 lines
15 KiB
Python
439 lines
15 KiB
Python
"""U6 集成测试: Agent v2 完整生命周期 — ReAct + LLM Gateway + Skill + Quality Gate"""
|
||
|
||
import json
|
||
from datetime import datetime, timezone
|
||
from typing import Any
|
||
|
||
import pytest
|
||
|
||
from agentkit.core.config_driven import AgentConfig, ConfigDrivenAgent
|
||
from agentkit.core.protocol import TaskMessage, TaskResult, TaskStatus
|
||
from agentkit.llm.gateway import LLMGateway
|
||
from agentkit.llm.protocol import LLMProvider, LLMRequest, LLMResponse, TokenUsage
|
||
from agentkit.quality.gate import QualityGate
|
||
from agentkit.quality.output import OutputStandardizer
|
||
from agentkit.skills.base import Skill, SkillConfig, QualityGateConfig, IntentConfig
|
||
from agentkit.tools.function_tool import FunctionTool
|
||
from agentkit.tools.registry import ToolRegistry
|
||
|
||
|
||
# ── Mock LLM Provider ────────────────────────────────────
|
||
|
||
|
||
class MockLLMProvider(LLMProvider):
|
||
"""Mock LLM Provider，返回预设的响应"""
|
||
|
||
def __init__(self, responses: list[str] | None = None):
|
||
self.responses = responses or ['{"result": "mock_llm_response"}']
|
||
self._call_count = 0
|
||
|
||
async def chat(self, request: LLMRequest) -> LLMResponse:
|
||
content = self.responses[self._call_count % len(self.responses)]
|
||
self._call_count += 1
|
||
return LLMResponse(
|
||
content=content,
|
||
model="mock-model",
|
||
usage=TokenUsage(prompt_tokens=10, completion_tokens=20),
|
||
)
|
||
|
||
|
||
class MockReActProvider(LLMProvider):
|
||
"""Mock Provider 模拟 ReAct 循环：先返回 tool_call，再返回 final answer"""
|
||
|
||
def __init__(self):
|
||
self._call_count = 0
|
||
|
||
async def chat(self, request: LLMRequest) -> LLMResponse:
|
||
self._call_count += 1
|
||
if self._call_count == 1:
|
||
# 第一次：返回 tool_call
|
||
return LLMResponse(
|
||
content="",
|
||
model="mock-model",
|
||
usage=TokenUsage(prompt_tokens=50, completion_tokens=30),
|
||
tool_calls=[
|
||
{
|
||
"id": "tc_001",
|
||
"name": "search",
|
||
"arguments": {"query": "test query"},
|
||
}
|
||
],
|
||
)
|
||
else:
|
||
# 第二次：返回最终答案
|
||
return LLMResponse(
|
||
content='{"answer": "found it", "confidence": 0.95}',
|
||
model="mock-model",
|
||
usage=TokenUsage(prompt_tokens=30, completion_tokens=20),
|
||
)
|
||
|
||
|
||
# ── Helpers ──────────────────────────────────────────────
|
||
|
||
|
||
def _make_task(task_type: str = "generate", input_data: dict | None = None) -> TaskMessage:
|
||
return TaskMessage(
|
||
task_id="integration-001",
|
||
agent_name="test_agent",
|
||
task_type=task_type,
|
||
priority=1,
|
||
input_data=input_data or {"query": "test"},
|
||
callback_url=None,
|
||
created_at=datetime.now(timezone.utc),
|
||
)
|
||
|
||
|
||
def _make_gateway_with_provider(provider: LLMProvider) -> LLMGateway:
|
||
"""创建带 mock provider 的 LLMGateway"""
|
||
gateway = LLMGateway()
|
||
gateway.register_provider("mock", provider)
|
||
return gateway
|
||
|
||
|
||
def _make_skill_config(
|
||
name: str = "test_skill",
|
||
execution_mode: str = "react",
|
||
quality_gate: dict | None = None,
|
||
prompt: dict | None = None,
|
||
tools: list[str] | None = None,
|
||
) -> SkillConfig:
|
||
return SkillConfig(
|
||
name=name,
|
||
agent_type="test",
|
||
task_mode="llm_generate",
|
||
prompt=prompt or {"identity": "Test skill", "instructions": "Do test things"},
|
||
execution_mode=execution_mode,
|
||
quality_gate=quality_gate,
|
||
tools=tools,
|
||
)
|
||
|
||
|
||
# ── ConfigDrivenAgent v2 Backward Compat 测试 ────────────
|
||
|
||
|
||
class TestConfigDrivenAgentV2BackwardCompat:
|
||
"""测试 ConfigDrivenAgent 向后兼容"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_llm_client_backward_compat(self):
|
||
"""llm_client 参数仍然可用"""
|
||
|
||
class MockLLMClient:
|
||
async def chat(self, messages, **kwargs):
|
||
return json.dumps({"title": "Test", "content": "Hello"})
|
||
|
||
config = AgentConfig(
|
||
name="test_agent",
|
||
agent_type="test",
|
||
task_mode="llm_generate",
|
||
prompt={"identity": "Test", "instructions": "Do test"},
|
||
)
|
||
agent = ConfigDrivenAgent(config=config, llm_client=MockLLMClient())
|
||
|
||
# llm_client 应该被自动包装为 LLMGateway
|
||
assert agent.llm_gateway is not None
|
||
|
||
task = _make_task()
|
||
result = await agent.handle_task(task)
|
||
assert result["title"] == "Test"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_llm_gateway_param(self):
|
||
"""llm_gateway 参数直接传入"""
|
||
provider = MockLLMProvider()
|
||
gateway = _make_gateway_with_provider(provider)
|
||
|
||
config = AgentConfig(
|
||
name="test_agent",
|
||
agent_type="test",
|
||
task_mode="llm_generate",
|
||
prompt={"identity": "Test", "instructions": "Do test"},
|
||
llm={"model": "mock/mock-model"},
|
||
)
|
||
agent = ConfigDrivenAgent(config=config, llm_gateway=gateway)
|
||
|
||
assert agent.llm_gateway is gateway
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_no_llm_backward_compat(self):
|
||
"""无 LLM 客户端时降级模式仍然正常"""
|
||
config = AgentConfig(
|
||
name="test_agent",
|
||
agent_type="test",
|
||
task_mode="llm_generate",
|
||
prompt={"identity": "Test", "instructions": "Do test"},
|
||
)
|
||
agent = ConfigDrivenAgent(config=config)
|
||
|
||
task = _make_task()
|
||
result = await agent.handle_task(task)
|
||
assert result["mode"] == "llm_generate_no_client"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_llm_gateway_takes_precedence(self):
|
||
"""llm_gateway 和 llm_client 同时传入时，llm_gateway 优先"""
|
||
provider = MockLLMProvider()
|
||
gateway = _make_gateway_with_provider(provider)
|
||
|
||
class MockLLMClient:
|
||
async def chat(self, messages, **kwargs):
|
||
return json.dumps({"source": "llm_client"})
|
||
|
||
config = AgentConfig(
|
||
name="test_agent",
|
||
agent_type="test",
|
||
task_mode="llm_generate",
|
||
prompt={"identity": "Test", "instructions": "Do test"},
|
||
llm={"model": "mock/mock-model"},
|
||
)
|
||
agent = ConfigDrivenAgent(config=config, llm_client=MockLLMClient(), llm_gateway=gateway)
|
||
|
||
# 应该使用 llm_gateway 而非 llm_client
|
||
assert agent.llm_gateway is gateway
|
||
|
||
|
||
# ── ConfigDrivenAgent + SkillConfig 测试 ─────────────────
|
||
|
||
|
||
class TestConfigDrivenAgentWithSkillConfig:
|
||
"""测试 ConfigDrivenAgent 接受 SkillConfig"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_skill_config_creates_skill(self):
|
||
"""传入 SkillConfig 时自动创建 Skill"""
|
||
skill_config = _make_skill_config()
|
||
agent = ConfigDrivenAgent(config=skill_config)
|
||
|
||
assert agent.skill is not None
|
||
assert agent.skill.name == "test_skill"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_agent_config_no_skill(self):
|
||
"""传入 AgentConfig 时不创建 Skill"""
|
||
config = AgentConfig(
|
||
name="test_agent",
|
||
agent_type="test",
|
||
task_mode="llm_generate",
|
||
prompt={"identity": "Test", "instructions": "Do test"},
|
||
)
|
||
agent = ConfigDrivenAgent(config=config)
|
||
assert agent.skill is None
|
||
|
||
|
||
# ── ReAct 模式测试 ──────────────────────────────────────
|
||
|
||
|
||
class TestReActMode:
|
||
"""测试 ConfigDrivenAgent 的 ReAct 执行模式"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_react_mode_uses_react_engine(self):
|
||
"""execution_mode=react 时使用 ReAct 引擎"""
|
||
provider = MockLLMProvider(['{"answer": "react_result"}'])
|
||
gateway = _make_gateway_with_provider(provider)
|
||
|
||
skill_config = _make_skill_config(execution_mode="react")
|
||
agent = ConfigDrivenAgent(config=skill_config, llm_gateway=gateway)
|
||
|
||
task = _make_task()
|
||
result = await agent.handle_task(task)
|
||
|
||
assert result["answer"] == "react_result"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_direct_mode_uses_legacy(self):
|
||
"""execution_mode=direct 时使用传统模式"""
|
||
provider = MockLLMProvider(['{"answer": "direct_result"}'])
|
||
gateway = _make_gateway_with_provider(provider)
|
||
|
||
skill_config = _make_skill_config(execution_mode="direct")
|
||
agent = ConfigDrivenAgent(config=skill_config, llm_gateway=gateway)
|
||
|
||
task = _make_task()
|
||
result = await agent.handle_task(task)
|
||
|
||
# direct 模式走 _handle_llm_generate，但使用 gateway
|
||
assert result is not None
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_agent_config_uses_legacy_mode(self):
|
||
"""AgentConfig（无 execution_mode）使用传统模式"""
|
||
provider = MockLLMProvider()
|
||
gateway = _make_gateway_with_provider(provider)
|
||
|
||
config = AgentConfig(
|
||
name="test_agent",
|
||
agent_type="test",
|
||
task_mode="llm_generate",
|
||
prompt={"identity": "Test", "instructions": "Do test"},
|
||
llm={"model": "mock/mock-model"},
|
||
)
|
||
agent = ConfigDrivenAgent(config=config, llm_gateway=gateway)
|
||
|
||
task = _make_task()
|
||
result = await agent.handle_task(task)
|
||
assert result is not None
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_react_without_gateway_falls_back(self):
|
||
"""ReAct 模式但无 gateway 时回退到传统模式"""
|
||
skill_config = _make_skill_config(execution_mode="react")
|
||
agent = ConfigDrivenAgent(config=skill_config)
|
||
|
||
task = _make_task()
|
||
result = await agent.handle_task(task)
|
||
|
||
# 无 gateway 时降级
|
||
assert result["mode"] == "llm_generate_no_client"
|
||
|
||
|
||
# ── handle_task_with_feedback 测试 ───────────────────────
|
||
|
||
|
||
class TestConfigDrivenFeedback:
|
||
"""测试 ConfigDrivenAgent 的 handle_task_with_feedback"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_feedback_adds_to_input(self):
|
||
"""handle_task_with_feedback 将反馈添加到 input_data"""
|
||
skill_config = _make_skill_config()
|
||
agent = ConfigDrivenAgent(config=skill_config)
|
||
|
||
task = _make_task(input_data={"query": "test"})
|
||
result = await agent.handle_task_with_feedback(task, "quality feedback: missing field")
|
||
|
||
# 应该将 feedback 添加到 enhanced_input 中重新执行
|
||
assert result is not None
|
||
|
||
|
||
# ── 完整生命周期集成测试 ─────────────────────────────────
|
||
|
||
|
||
class TestAgentV2Lifecycle:
|
||
"""完整生命周期：创建 → 注入 Skill → 执行 → 返回结果"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_full_react_lifecycle(self):
|
||
"""完整 ReAct 生命周期"""
|
||
provider = MockLLMProvider(['{"title": "Test Title", "content": "Test content here"}'])
|
||
gateway = _make_gateway_with_provider(provider)
|
||
|
||
skill_config = _make_skill_config(
|
||
execution_mode="react",
|
||
quality_gate={"required_fields": ["title", "content"], "max_retries": 1},
|
||
)
|
||
|
||
agent = ConfigDrivenAgent(config=skill_config, llm_gateway=gateway)
|
||
|
||
task = _make_task()
|
||
result = await agent.execute(task)
|
||
|
||
assert result.status == TaskStatus.COMPLETED
|
||
assert result.output_data is not None
|
||
assert result.output_data.get("title") == "Test Title"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_full_legacy_lifecycle(self):
|
||
"""完整传统模式生命周期（向后兼容）"""
|
||
config = AgentConfig(
|
||
name="legacy_agent",
|
||
agent_type="test",
|
||
task_mode="llm_generate",
|
||
prompt={"identity": "Legacy", "instructions": "Do legacy things"},
|
||
)
|
||
|
||
agent = ConfigDrivenAgent(config=config)
|
||
|
||
task = _make_task()
|
||
result = await agent.execute(task)
|
||
|
||
assert result.status == TaskStatus.COMPLETED
|
||
assert result.output_data is not None
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_tool_call_mode_still_works(self):
|
||
"""tool_call 模式仍然正常"""
|
||
registry = ToolRegistry()
|
||
|
||
async def search(query: str, **kwargs) -> dict:
|
||
return {"results": [f"Result for {query}"]}
|
||
|
||
tool = FunctionTool(name="search", description="Search tool", func=search)
|
||
registry.register(tool)
|
||
|
||
config = AgentConfig(
|
||
name="tool_agent",
|
||
agent_type="test",
|
||
task_mode="tool_call",
|
||
tools=["search"],
|
||
)
|
||
agent = ConfigDrivenAgent(config=config, tool_registry=registry)
|
||
|
||
task = _make_task(input_data={"query": "test"})
|
||
result = await agent.handle_task(task)
|
||
|
||
assert "results" in result
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_custom_mode_still_works(self):
|
||
"""custom 模式仍然正常"""
|
||
config = AgentConfig(
|
||
name="custom_agent",
|
||
agent_type="test",
|
||
task_mode="custom",
|
||
custom_handler="my_handler",
|
||
)
|
||
|
||
async def my_handler(task):
|
||
return {"custom": True, "task_id": task.task_id}
|
||
|
||
agent = ConfigDrivenAgent(config=config, custom_handlers={"my_handler": my_handler})
|
||
|
||
task = _make_task()
|
||
result = await agent.handle_task(task)
|
||
|
||
assert result["custom"] is True
|
||
|
||
|
||
# ── Quality Gate + Output Standardizer 集成 ──────────────
|
||
|
||
|
||
class TestQualityGateOutputIntegration:
|
||
"""Quality Gate 与 Output Standardizer 的集成"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_quality_gate_with_output_standardizer(self):
|
||
"""Quality Gate 检查后使用 OutputStandardizer 标准化输出"""
|
||
skill_config = _make_skill_config(
|
||
quality_gate={"required_fields": ["title"], "max_retries": 0},
|
||
)
|
||
skill = Skill(config=skill_config)
|
||
gate = QualityGate()
|
||
standardizer = OutputStandardizer()
|
||
|
||
output = {"title": "Test", "content": "Some content"}
|
||
quality_result = await gate.validate(output, skill)
|
||
assert quality_result.passed is True
|
||
|
||
standard = await standardizer.standardize(output, skill, quality_result)
|
||
assert standard.skill_name == "test_skill"
|
||
assert standard.data["title"] == "Test"
|
||
assert standard.metadata.quality_score == 1.0
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_quality_gate_fails_then_standardize(self):
|
||
"""Quality Gate 失败后仍可标准化输出"""
|
||
skill_config = _make_skill_config(
|
||
quality_gate={"required_fields": ["missing_field"], "max_retries": 0},
|
||
)
|
||
skill = Skill(config=skill_config)
|
||
gate = QualityGate()
|
||
standardizer = OutputStandardizer()
|
||
|
||
output = {"title": "Test"}
|
||
quality_result = await gate.validate(output, skill)
|
||
assert quality_result.passed is False
|
||
|
||
standard = await standardizer.standardize(output, skill, quality_result)
|
||
assert standard.metadata.quality_score < 1.0
|