view_agent_memories_simple.py

#!/usr/bin/env python3
"""
查看智能体的中间产物（简化版）
直接查看日志文件中的智能体操作记录
"""
import os
import re
from typing import List, Dict, Any
from datetime import datetime


def parse_log_line(line: str) -> Dict[str, Any]:
    """解析日志行"""
    # 匹配日志格式: 2025-12-03 10:50:27 - logger_name - LEVEL - message
    pattern = r'(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}) - (\S+) - (\S+) - (.+)'
    match = re.match(pattern, line)
    if match:
        return {
            "timestamp": match.group(1),
            "logger": match.group(2),
            "level": match.group(3),
            "message": match.group(4)
        }
    return None


def view_logs(log_file: str, max_lines: int = 100, filter_keywords: List[str] = None):
    """查看日志文件"""
    if not os.path.exists(log_file):
        print(f"⚠️  {log_file} 不存在")
        return []
    
    filter_keywords = filter_keywords or [
        "TOOL", "write_file", "read_file", "memories", 
        "task(", "write_todos", "search", "traversal"
    ]
    
    relevant_lines = []
    try:
        with open(log_file, 'r', encoding='utf-8', errors='ignore') as f:
            lines = f.readlines()
            # 读取最后 max_lines 行
            recent_lines = lines[-max_lines:] if len(lines) > max_lines else lines
            
            for line in recent_lines:
                line = line.strip()
                if not line:
                    continue
                
                # 检查是否包含关键词
                if any(keyword.lower() in line.lower() for keyword in filter_keywords):
                    parsed = parse_log_line(line)
                    if parsed:
                        relevant_lines.append(parsed)
                    else:
                        relevant_lines.append({"raw": line})
    
    except Exception as e:
        print(f"❌ 读取 {log_file} 失败: {e}")
        return []
    
    return relevant_lines


def format_log_entry(entry: Dict[str, Any]) -> str:
    """格式化日志条目"""
    if "raw" in entry:
        return entry["raw"]
    
    timestamp = entry.get("timestamp", "")
    level = entry.get("level", "")
    message = entry.get("message", "")
    
    level_prefix = {
        "INFO": "[INFO]",
        "WARNING": "[WARN]",
        "ERROR": "[ERROR]",
        "DEBUG": "[DEBUG]"
    }
    
    prefix = level_prefix.get(level, "[LOG]")
    return f"{prefix} [{timestamp}] {message}"


def main():
    """主函数"""
    print("=" * 60)
    print("智能体中间产物查看工具")
    print("=" * 60)
    print()
    
    log_files = [
        ("logs/deepagent.log", "Deep Agent 操作日志"),
        ("logs/analysis.log", "分析服务日志"),
        ("logs/api.log", "API 请求日志")
    ]
    
    all_entries = []
    
    for log_file, description in log_files:
        print(f"\n[{description}] ({log_file}):")
        print("-" * 60)
        
        entries = view_logs(log_file, max_lines=200)
        all_entries.extend(entries)
        
        if entries:
            for entry in entries[-20:]:  # 只显示最后20条
                print(format_log_entry(entry))
        else:
            print("   (无相关记录)")
    
    # 统计信息
    print("\n" + "=" * 60)
    print("[统计信息]:")
    print(f"   找到 {len(all_entries)} 条相关记录")
    
    # 按类型分类
    tool_calls = [e for e in all_entries if "TOOL" in str(e.get("message", ""))]
    file_ops = [e for e in all_entries if "write_file" in str(e.get("message", "")) or "read_file" in str(e.get("message", ""))]
    task_ops = [e for e in all_entries if "task(" in str(e.get("message", ""))]
    
    print(f"   工具调用: {len(tool_calls)} 条")
    print(f"   文件操作: {len(file_ops)} 条")
    print(f"   任务操作: {len(task_ops)} 条")
    
    print("\n[提示]:")
    print("   1. 使用流式 API 查看实时操作:")
    print("      POST /api/deep-agent/chat/stream")
    print("      设置 stream_mode=['tasks', 'checkpoints', 'debug']")
    print("   2. 中间产物保存在 /memories/ 路径下:")
    print("      - /memories/current_project.txt")
    print("      - /memories/search_results.json")
    print("      - /memories/call_chains.json")
    print("      - /memories/exploration_results.json")
    print("   3. 当前使用 InMemoryStore，中间产物只保存在内存中")
    print("   4. 查看 API 文档: http://127.0.0.1:8000/docs")


if __name__ == "__main__":
    main()