196 lines
5.1 KiB
Python
196 lines
5.1 KiB
Python
"""
|
||
ResearchCoordinator测试
|
||
|
||
测试主Agent的完整执行流程
|
||
"""
|
||
|
||
import sys
|
||
import os
|
||
|
||
# 添加src目录到Python路径
|
||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
|
||
|
||
from src.agents.coordinator import create_research_coordinator, run_research
|
||
from src.config import Config
|
||
|
||
|
||
def test_coordinator_creation():
|
||
"""测试ResearchCoordinator创建"""
|
||
print("=" * 60)
|
||
print("测试1: ResearchCoordinator创建")
|
||
print("=" * 60)
|
||
|
||
try:
|
||
# 测试默认参数
|
||
agent = create_research_coordinator(
|
||
question="什么是Python asyncio?",
|
||
depth="quick"
|
||
)
|
||
|
||
print("✓ ResearchCoordinator创建成功")
|
||
print(f" Agent类型: {type(agent)}")
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"✗ ResearchCoordinator创建失败: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return False
|
||
|
||
|
||
def test_config_validation():
|
||
"""测试配置验证"""
|
||
print("\n" + "=" * 60)
|
||
print("测试2: 配置验证")
|
||
print("=" * 60)
|
||
|
||
# 测试无效的深度模式
|
||
try:
|
||
agent = create_research_coordinator(
|
||
question="测试问题",
|
||
depth="invalid_depth"
|
||
)
|
||
print("✗ 应该抛出ValueError但没有")
|
||
return False
|
||
except ValueError as e:
|
||
print(f"✓ 正确捕获无效深度模式: {e}")
|
||
|
||
# 测试无效的min_tier
|
||
try:
|
||
agent = create_research_coordinator(
|
||
question="测试问题",
|
||
min_tier=5
|
||
)
|
||
print("✗ 应该抛出ValueError但没有")
|
||
return False
|
||
except ValueError as e:
|
||
print(f"✓ 正确捕获无效min_tier: {e}")
|
||
|
||
# 测试无效的格式
|
||
try:
|
||
agent = create_research_coordinator(
|
||
question="测试问题",
|
||
format="invalid_format"
|
||
)
|
||
print("✗ 应该抛出ValueError但没有")
|
||
return False
|
||
except ValueError as e:
|
||
print(f"✓ 正确捕获无效格式: {e}")
|
||
|
||
return True
|
||
|
||
|
||
def test_simple_research_dry_run():
|
||
"""测试简单研究流程（dry run，不执行真实搜索）"""
|
||
print("\n" + "=" * 60)
|
||
print("测试3: 简单研究流程（模拟）")
|
||
print("=" * 60)
|
||
|
||
print("\n注意: 这个测试需要API密钥才能执行真实的Agent调用")
|
||
print("如果API密钥未配置，将跳过此测试\n")
|
||
|
||
# 检查API密钥
|
||
try:
|
||
Config.validate()
|
||
except ValueError as e:
|
||
print(f"⚠️ 跳过测试：{e}")
|
||
return True # 不算失败
|
||
|
||
try:
|
||
# 创建Agent但不执行
|
||
agent = create_research_coordinator(
|
||
question="Python装饰器的作用",
|
||
depth="quick",
|
||
format="technical"
|
||
)
|
||
|
||
print("✓ Agent创建成功，准备就绪")
|
||
print(" 如需运行完整测试，请确保API密钥已配置")
|
||
print(" 然后运行：python -m tests.test_integration")
|
||
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"✗ 测试失败: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return False
|
||
|
||
|
||
def test_depth_configs():
|
||
"""测试三种深度模式的配置"""
|
||
print("\n" + "=" * 60)
|
||
print("测试4: 深度模式配置")
|
||
print("=" * 60)
|
||
|
||
depth_modes = ["quick", "standard", "deep"]
|
||
|
||
for depth in depth_modes:
|
||
try:
|
||
agent = create_research_coordinator(
|
||
question="测试问题",
|
||
depth=depth
|
||
)
|
||
|
||
depth_config = Config.get_depth_config(depth)
|
||
|
||
print(f"\n✓ {depth}模式配置正确:")
|
||
print(f" - 最大迭代: {depth_config['max_iterations']}")
|
||
print(f" - 置信度阈值: {depth_config['confidence_threshold']}")
|
||
print(f" - 目标来源数: {depth_config['target_sources']}")
|
||
print(f" - 并行搜索数: {depth_config['parallel_searches']}")
|
||
|
||
except Exception as e:
|
||
print(f"✗ {depth}模式配置失败: {e}")
|
||
return False
|
||
|
||
return True
|
||
|
||
|
||
def main():
|
||
"""运行所有测试"""
|
||
print("\n")
|
||
print("=" * 60)
|
||
print("ResearchCoordinator测试套件")
|
||
print("=" * 60)
|
||
print("\n")
|
||
|
||
results = []
|
||
|
||
# 测试1: 创建
|
||
results.append(("创建测试", test_coordinator_creation()))
|
||
|
||
# 测试2: 配置验证
|
||
results.append(("配置验证", test_config_validation()))
|
||
|
||
# 测试3: 简单研究流程
|
||
results.append(("简单研究流程", test_simple_research_dry_run()))
|
||
|
||
# 测试4: 深度模式配置
|
||
results.append(("深度模式配置", test_depth_configs()))
|
||
|
||
# 总结
|
||
print("\n" + "=" * 60)
|
||
print("测试总结")
|
||
print("=" * 60)
|
||
|
||
for test_name, passed in results:
|
||
status = "✓ 通过" if passed else "✗ 失败"
|
||
print(f"{test_name}: {status}")
|
||
|
||
all_passed = all(result[1] for result in results)
|
||
|
||
print("\n" + "=" * 60)
|
||
if all_passed:
|
||
print("✓ 所有测试通过！ResearchCoordinator实现正确。")
|
||
else:
|
||
print("✗ 部分测试失败，请检查实现。")
|
||
print("=" * 60 + "\n")
|
||
|
||
return all_passed
|
||
|
||
|
||
if __name__ == "__main__":
|
||
success = main()
|
||
sys.exit(0 if success else 1)
|