#!/usr/bin/env python3 """ CognitiveKernel-Pro 日志管理系统 自动生成控制台输出日志到本地文件夹 """ import os import sys import time import json from datetime import datetime from typing import Optional, TextIO from pathlib import Path class ConsoleLogger: """ 控制台输出日志记录器 自动将控制台输出同时写入日志文件 """ def __init__(self, log_dir: str = "logs", task_name: str = None): """ 初始化日志记录器 Args: log_dir: 日志目录 task_name: 任务名称，用于生成日志文件名 """ self.log_dir = Path(log_dir) self.task_name = task_name or "cognitive_kernel" self.log_file = None self.original_stdout = None self.original_stderr = None self.start_time = time.time() # 创建日志目录 self.log_dir.mkdir(exist_ok=True) # 生成日志文件名 timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") self.log_filename = f"{self.task_name}_console_{timestamp}.log" self.log_filepath = self.log_dir / self.log_filename print(f"📁 日志将保存到: {self.log_filepath}") def start_logging(self): """ 开始记录控制台输出 """ try: # 保存原始的stdout和stderr self.original_stdout = sys.stdout self.original_stderr = sys.stderr # 打开日志文件 self.log_file = open(self.log_filepath, 'w', encoding='utf-8') # 写入日志头部信息 self._write_log_header() # 替换stdout和stderr sys.stdout = TeeOutput(self.original_stdout, self.log_file) sys.stderr = TeeOutput(self.original_stderr, self.log_file) print(f"✅ 控制台日志记录已启动") print(f"📝 日志文件: {self.log_filepath}") except Exception as e: print(f"❌ 启动日志记录失败: {e}") self.stop_logging() def stop_logging(self): """ 停止记录控制台输出 """ try: if self.original_stdout: sys.stdout = self.original_stdout if self.original_stderr: sys.stderr = self.original_stderr if self.log_file: # 写入日志尾部信息 self._write_log_footer() self.log_file.close() self.log_file = None print(f"✅ 控制台日志记录已停止") print(f"📁 日志已保存到: {self.log_filepath}") except Exception as e: print(f"❌ 停止日志记录失败: {e}") def _write_log_header(self): """ 写入日志头部信息 """ header = f""" {'='*80} CognitiveKernel-Pro 控制台输出日志 {'='*80} 任务名称: {self.task_name} 开始时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} 日志文件: {self.log_filename} Python版本: {sys.version} 工作目录: {os.getcwd()} {'='*80} """ self.log_file.write(header) self.log_file.flush() def _write_log_footer(self): """ 写入日志尾部信息 """ end_time = time.time() duration = end_time - self.start_time footer = f""" {'='*80} 日志记录结束 {'='*80} 结束时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} 总执行时间: {duration:.2f} 秒 日志文件大小: {self.log_filepath.stat().st_size} 字节 {'='*80} """ self.log_file.write(footer) self.log_file.flush() def __enter__(self): """ 上下文管理器入口 """ self.start_logging() return self def __exit__(self, exc_type, exc_val, exc_tb): """ 上下文管理器出口 """ self.stop_logging() class TeeOutput: """ 同时输出到控制台和文件的包装器 """ def __init__(self, console: TextIO, file: TextIO): self.console = console self.file = file def write(self, text: str): """ 同时写入控制台和文件 """ # 写入控制台 self.console.write(text) self.console.flush() # 写入文件，添加时间戳 if text.strip(): # 只对非空行添加时间戳 timestamp = datetime.now().strftime('[%H:%M:%S] ') # 如果文本不是以换行符开始，添加时间戳 if not text.startswith('\n'): self.file.write(timestamp) self.file.write(text) self.file.flush() def flush(self): """ 刷新缓冲区 """ self.console.flush() self.file.flush() def __getattr__(self, name): """ 代理其他属性到控制台 """ return getattr(self.console, name) class SessionLogger: """ 会话详细日志记录器 记录完整的会话数据 """ def __init__(self, log_dir: str = "logs"): self.log_dir = Path(log_dir) self.log_dir.mkdir(exist_ok=True) def save_session_log(self, session, task_name: str = None) -> str: """ 保存会话详细日志 Args: session: AgentSession对象 task_name: 任务名称 Returns: 日志文件路径 """ try: # 生成文件名 timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") task_name = task_name or "session" filename = f"{task_name}_detailed_{timestamp}.json" filepath = self.log_dir / filename # 转换会话数据 session_data = session.to_dict() # 添加元数据 session_data['_metadata'] = { 'log_created_at': datetime.now().isoformat(), 'log_filename': filename, 'total_steps': len(session_data.get('steps', [])), 'task_summary': session_data.get('task', '')[:100] + '...' if len(session_data.get('task', '')) > 100 else session_data.get('task', '') } # 保存到文件 with open(filepath, 'w', encoding='utf-8') as f: json.dump(session_data, f, ensure_ascii=False, indent=2, default=str) print(f"📁 会话详细日志已保存: {filepath}") return str(filepath) except Exception as e: print(f"❌ 保存会话日志失败: {e}") return None def setup_logging_environment(): """ 设置日志环境 创建日志目录并更新gitignore """ # 创建logs目录 logs_dir = Path("logs") logs_dir.mkdir(exist_ok=True) # 创建子目录 (logs_dir / "console").mkdir(exist_ok=True) (logs_dir / "sessions").mkdir(exist_ok=True) (logs_dir / "api_calls").mkdir(exist_ok=True) # 创建README文件 readme_content = """# CognitiveKernel-Pro 日志目录 这个目录包含了CognitiveKernel-Pro运行时生成的各种日志文件。 ## 目录结构 - `console/`: 控制台输出日志 - `sessions/`: 详细会话日志 - `api_calls/`: API调用日志 ## 日志文件类型 ### 控制台日志 (`*_console_*.log`) - 完整的控制台输出 - 包含时间戳 - 实时执行过程记录 ### 会话日志 (`*_detailed_*.json`) - 完整的会话数据 - 每个步骤的详细信息 - Progress State历史 ### API调用日志 (`*_api_*.log`) - LLM API调用记录 - 请求和响应数据 - 性能统计 ## 注意事项 - 这些日志文件已添加到 .gitignore 中 - 日志文件可能包含敏感信息，请勿提交到版本控制 - 定期清理旧的日志文件以节省磁盘空间 """ with open(logs_dir / "README.md", 'w', encoding='utf-8') as f: f.write(readme_content) print(f"✅ 日志环境设置完成") print(f"📁 日志目录: {logs_dir.absolute()}") # 便捷函数 def start_console_logging(task_name: str = None) -> ConsoleLogger: """ 启动控制台日志记录 Args: task_name: 任务名称 Returns: ConsoleLogger实例 """ logger = ConsoleLogger(task_name=task_name) logger.start_logging() return logger def save_session_log(session, task_name: str = None) -> str: """ 保存会话详细日志 Args: session: AgentSession对象 task_name: 任务名称 Returns: 日志文件路径 """ logger = SessionLogger() return logger.save_session_log(session, task_name)