deepagents----/tests/debug_research.py

"""
调试研究流程 - 详细追踪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")