191 lines
6.0 KiB
Python
191 lines
6.0 KiB
Python
"""
|
||
调试研究流程 - 详细追踪Agent执行情况
|
||
|
||
使用方法：
|
||
export PYTHONIOENCODING=utf-8 && python tests/debug_research.py
|
||
"""
|
||
|
||
import sys
|
||
import os
|
||
import json
|
||
from datetime import datetime
|
||
|
||
# 添加项目根目录到Python路径
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
|
||
from src.agents.coordinator import create_research_coordinator
|
||
from src.config import Config
|
||
|
||
|
||
def print_step(step_num: int, title: str):
|
||
"""打印步骤标题"""
|
||
print("\n" + "="*80)
|
||
print(f"步骤 {step_num}: {title}")
|
||
print("="*80)
|
||
|
||
|
||
def print_substep(title: str):
|
||
"""打印子步骤"""
|
||
print(f"\n>>> {title}")
|
||
print("-"*60)
|
||
|
||
|
||
def print_file_content(file_path: str, content: any, max_length: int = 500):
|
||
"""打印文件内容"""
|
||
print(f"\n📄 文件: {file_path}")
|
||
if isinstance(content, dict) or isinstance(content, list):
|
||
content_str = json.dumps(content, ensure_ascii=False, indent=2)
|
||
else:
|
||
content_str = str(content)
|
||
|
||
if len(content_str) > max_length:
|
||
print(content_str[:max_length] + "...")
|
||
else:
|
||
print(content_str)
|
||
|
||
|
||
def debug_research(question: str, depth: str = "quick"):
|
||
"""
|
||
调试研究流程，显示详细执行日志
|
||
|
||
Args:
|
||
question: 研究问题
|
||
depth: 深度模式（使用quick模式加快调试）
|
||
"""
|
||
print("\n" + "🔬 "* 40)
|
||
print("智能深度研究系统 - 调试模式")
|
||
print("🔬 " * 40)
|
||
|
||
print(f"\n研究问题: {question}")
|
||
print(f"深度模式: {depth}")
|
||
print(f"开始时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
||
|
||
# 验证API配置
|
||
print_step(0, "验证API配置")
|
||
print(f"DashScope API Key: {Config.DASHSCOPE_API_KEY[:20]}..." if Config.DASHSCOPE_API_KEY else "❌ 未配置")
|
||
print(f"Tavily API Key: {Config.TAVILY_API_KEY[:20]}..." if Config.TAVILY_API_KEY else "❌ 未配置")
|
||
print(f"LLM模型: {Config.LLM_MODEL}")
|
||
|
||
# 创建Agent
|
||
print_step(1, "创建ResearchCoordinator Agent")
|
||
try:
|
||
agent = create_research_coordinator(
|
||
question=question,
|
||
depth=depth,
|
||
format="technical",
|
||
min_tier=3
|
||
)
|
||
print("✅ Agent创建成功")
|
||
print(f"Agent类型: {type(agent)}")
|
||
except Exception as e:
|
||
print(f"❌ Agent创建失败: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return
|
||
|
||
# 执行研究
|
||
print_step(2, "执行研究流程")
|
||
print("调用 agent.invoke() ...")
|
||
print("注意：这可能需要几分钟，请耐心等待...\n")
|
||
|
||
try:
|
||
# 记录开始时间
|
||
start_time = datetime.now()
|
||
|
||
# 执行Agent
|
||
result = agent.invoke({
|
||
"messages": [
|
||
{
|
||
"role": "user",
|
||
"content": f"请开始研究这个问题：{question}"
|
||
}
|
||
]
|
||
})
|
||
|
||
# 记录结束时间
|
||
end_time = datetime.now()
|
||
duration = (end_time - start_time).total_seconds()
|
||
|
||
print_step(3, "执行完成")
|
||
print(f"✅ 研究完成！")
|
||
print(f"⏱️ 总耗时: {duration:.2f}秒 ({duration/60:.2f}分钟)")
|
||
|
||
# 显示结果
|
||
print_step(4, "结果分析")
|
||
print(f"结果类型: {type(result)}")
|
||
print(f"结果键: {result.keys() if isinstance(result, dict) else 'N/A'}")
|
||
|
||
# 尝试提取消息
|
||
if isinstance(result, dict) and 'messages' in result:
|
||
messages = result['messages']
|
||
print(f"\n消息数量: {len(messages)}")
|
||
|
||
# 显示最后几条消息
|
||
print("\n最后3条消息:")
|
||
for i, msg in enumerate(messages[-3:], 1):
|
||
print(f"\n--- 消息 {i} ---")
|
||
if hasattr(msg, 'content'):
|
||
content = msg.content
|
||
if len(content) > 300:
|
||
print(content[:300] + "...")
|
||
else:
|
||
print(content)
|
||
else:
|
||
print(msg)
|
||
|
||
# 尝试访问虚拟文件系统
|
||
print_step(5, "虚拟文件系统检查")
|
||
print("注意：需要根据DeepAgents实际API来访问虚拟文件系统")
|
||
print("这部分功能待实现...")
|
||
|
||
# 保存完整结果到文件
|
||
print_step(6, "保存调试结果")
|
||
output_dir = "outputs/debug"
|
||
os.makedirs(output_dir, exist_ok=True)
|
||
|
||
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||
output_file = os.path.join(output_dir, f"debug_{timestamp}.json")
|
||
|
||
debug_data = {
|
||
"question": question,
|
||
"depth": depth,
|
||
"start_time": start_time.isoformat(),
|
||
"end_time": end_time.isoformat(),
|
||
"duration_seconds": duration,
|
||
"result": str(result), # 转换为字符串以便保存
|
||
}
|
||
|
||
with open(output_file, 'w', encoding='utf-8') as f:
|
||
json.dump(debug_data, f, ensure_ascii=False, indent=2)
|
||
|
||
print(f"✅ 调试结果已保存到: {output_file}")
|
||
|
||
except KeyboardInterrupt:
|
||
print("\n\n⚠️ 用户中断执行")
|
||
print(f"已执行时间: {(datetime.now() - start_time).total_seconds():.2f}秒")
|
||
except Exception as e:
|
||
print(f"\n\n❌ 执行失败: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
|
||
# 保存错误信息
|
||
output_dir = "outputs/debug"
|
||
os.makedirs(output_dir, exist_ok=True)
|
||
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||
error_file = os.path.join(output_dir, f"error_{timestamp}.txt")
|
||
|
||
with open(error_file, 'w', encoding='utf-8') as f:
|
||
f.write(f"Question: {question}\n")
|
||
f.write(f"Depth: {depth}\n")
|
||
f.write(f"Error: {str(e)}\n\n")
|
||
f.write(traceback.format_exc())
|
||
|
||
print(f"错误信息已保存到: {error_file}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
# 使用简单的问题和quick模式进行调试
|
||
question = "Python asyncio最佳实践"
|
||
|
||
debug_research(question, depth="quick")
|