119 lines
3.8 KiB
Python
Executable File
119 lines
3.8 KiB
Python
Executable File
"""
|
|
Agent协作流程集成测试
|
|
"""
|
|
|
|
import pytest
|
|
from gfa_core.mas.agents import (
|
|
RouterAgent, PlanningAgent, ExecutionAgent,
|
|
DiagnosisAgent, KnowledgeAgent
|
|
)
|
|
from gfa_core.mas.agents.base import AgentState, AgentMessage
|
|
|
|
|
|
@pytest.mark.integration
|
|
class TestAgentFlow:
|
|
"""测试Agent协作流程"""
|
|
|
|
@pytest.fixture
|
|
def llm_manager(self, mock_llm_manager):
|
|
"""LLM管理器"""
|
|
return mock_llm_manager
|
|
|
|
@pytest.fixture
|
|
def agents(self, llm_manager):
|
|
"""创建所有Agent"""
|
|
return {
|
|
'router': RouterAgent(llm_manager),
|
|
'planning': PlanningAgent(llm_manager),
|
|
'execution': ExecutionAgent(llm_manager, None),
|
|
'diagnosis': DiagnosisAgent(llm_manager),
|
|
'knowledge': KnowledgeAgent(llm_manager),
|
|
}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_full_analysis_flow(self, agents):
|
|
"""测试完整分析流程"""
|
|
# 1. Router识别意图
|
|
router_state = AgentState(session_id="test_flow_001")
|
|
router_msg = AgentMessage(
|
|
sender="user",
|
|
receiver="router",
|
|
message_type="request",
|
|
payload={"query": "帮我分析拟南芥SPL基因家族"}
|
|
)
|
|
|
|
router_result = await agents['router'].execute(router_state, router_msg)
|
|
assert router_result.mode.value == "analysis"
|
|
|
|
# 2. Planning生成计划
|
|
planning_state = AgentState(
|
|
session_id="test_flow_001",
|
|
results={"intent": {"family": "SPL", "species": "Arabidopsis"}}
|
|
)
|
|
planning_msg = AgentMessage(
|
|
sender="router",
|
|
receiver="planning",
|
|
message_type="request",
|
|
payload={"task": "分析SPL基因家族"}
|
|
)
|
|
|
|
planning_result = await agents['planning'].execute(planning_state, planning_msg)
|
|
assert "plan" in planning_result.results or len(planning_result.messages) > 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_error_diagnosis_flow(self, agents):
|
|
"""测试错误诊断流程"""
|
|
# 模拟执行失败后的诊断
|
|
diagnosis_state = AgentState(
|
|
session_id="test_error_001",
|
|
results={
|
|
"error": {
|
|
"stage": "gene-identify",
|
|
"message": "HMMER not found"
|
|
}
|
|
}
|
|
)
|
|
diagnosis_msg = AgentMessage(
|
|
sender="execution",
|
|
receiver="diagnosis",
|
|
message_type="error",
|
|
payload={"error": "HMMER not found"}
|
|
)
|
|
|
|
diagnosis_result = await agents['diagnosis'].execute(diagnosis_state, diagnosis_msg)
|
|
assert len(diagnosis_result.messages) > 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_knowledge_query_flow(self, agents):
|
|
"""测试知识查询流程"""
|
|
knowledge_state = AgentState(session_id="test_knowledge_001")
|
|
knowledge_msg = AgentMessage(
|
|
sender="user",
|
|
receiver="knowledge",
|
|
message_type="request",
|
|
payload={"query": "SPL基因家族有什么功能？"}
|
|
)
|
|
|
|
knowledge_result = await agents['knowledge'].execute(knowledge_state, knowledge_msg)
|
|
assert len(knowledge_result.messages) > 0
|
|
|
|
|
|
@pytest.mark.integration
|
|
class TestMemoryIntegration:
|
|
"""测试记忆系统集成"""
|
|
|
|
def test_memory_manager_import(self):
|
|
"""测试MemoryManager可导入"""
|
|
from gfa_core.mas.memory import MemoryManager
|
|
assert MemoryManager is not None
|
|
|
|
def test_memory_manager_creation(self):
|
|
"""测试MemoryManager可创建"""
|
|
from gfa_core.mas.memory import MemoryManager
|
|
|
|
mm = MemoryManager()
|
|
assert mm is not None
|
|
assert hasattr(mm, 'session')
|
|
assert hasattr(mm, 'case')
|
|
assert hasattr(mm, 'knowledge')
|