view_agent_memories.py

#!/usr/bin/env python3
"""
查看智能体的中间产物
包括保存在 /memories/ 路径下的文件和 LangGraph store 中的状态
"""
import json
import sys
from typing import Dict, Any, Optional
from services.ai.agents.deep_agent_service import deep_agent_service
from langgraph.store.memory import InMemoryStore
from utils.logger import analysis_logger


def view_memories_from_store(session_id: Optional[str] = None):
    """从 LangGraph store 中查看中间产物"""
    print("=== 查看智能体中间产物 ===\n")
    
    try:
        service = deep_agent_service.get_service()
        if not service:
            print("❌ Deep Agent 服务未初始化")
            return
        
        if not service.store:
            print("❌ 未配置存储，无法查看中间产物")
            return
        
        print(f"存储类型: {type(service.store).__name__}")
        
        # 如果是 InMemoryStore，尝试查看内容
        if isinstance(service.store, InMemoryStore):
            print("\n⚠️  当前使用 InMemoryStore（内存存储）")
            print("   中间产物只保存在内存中，服务重启后会丢失")
            print("   无法直接查看内存中的内容")
            print("\n建议：")
            print("   1. 查看日志文件获取智能体的操作记录")
            print("   2. 使用流式 API 查看实时操作")
            print("   3. 检查是否有文件系统工具创建的文件")
            return
        
        # 如果有其他类型的 store，尝试列出内容
        print("\n尝试列出存储内容...")
        # 这里需要根据实际的 store 类型来实现
        
    except Exception as e:
        print(f"❌ 查看中间产物失败: {e}")
        import traceback
        traceback.print_exc()


def view_logs():
    """查看日志文件中的智能体操作记录"""
    print("\n=== 查看日志文件 ===\n")
    
    log_files = [
        "logs/deepagent.log",
        "logs/analysis.log"
    ]
    
    for log_file in log_files:
        try:
            import os
            if os.path.exists(log_file):
                print(f"\n📄 {log_file}:")
                print("-" * 60)
                # 读取最后 50 行
                with open(log_file, 'r', encoding='utf-8', errors='ignore') as f:
                    lines = f.readlines()
                    for line in lines[-50:]:
                        if 'TOOL' in line or 'write_file' in line or 'memories' in line.lower():
                            print(line.strip())
            else:
                print(f"⚠️  {log_file} 不存在")
        except Exception as e:
            print(f"❌ 读取 {log_file} 失败: {e}")


def view_recent_sessions():
    """查看最近的会话信息"""
    print("\n=== 查看最近的会话 ===\n")
    
    try:
        # 这里可以扩展为从数据库或文件系统读取会话信息
        print("提示：会话信息保存在 LangGraph store 中")
        print("当前使用 InMemoryStore，无法持久化查看")
        print("\n建议使用流式 API 查看实时操作：")
        print("  POST /api/deep-agent/chat/stream")
        print("  设置 stream_mode=['tasks', 'checkpoints', 'debug']")
    except Exception as e:
        print(f"❌ 查看会话失败: {e}")


def main():
    """主函数"""
    import argparse
    
    parser = argparse.ArgumentParser(description="查看智能体的中间产物")
    parser.add_argument("--session-id", type=str, help="会话ID")
    parser.add_argument("--logs", action="store_true", help="查看日志文件")
    parser.add_argument("--sessions", action="store_true", help="查看会话信息")
    
    args = parser.parse_args()
    
    if args.logs:
        view_logs()
    elif args.sessions:
        view_recent_sessions()
    else:
        # 默认查看所有
        view_memories_from_store(args.session_id)
        view_logs()
        view_recent_sessions()
        
        print("\n" + "=" * 60)
        print("💡 提示：")
        print("   1. 使用 --logs 只查看日志")
        print("   2. 使用 --sessions 查看会话信息")
        print("   3. 使用流式 API 查看实时操作")
        print("   4. 中间产物保存在 /memories/ 路径下，通过文件系统工具访问")


if __name__ == "__main__":
    main()