error_catcher.py•5.57 kB
#!/usr/bin/env python3
"""
全局错误捕获器
自动捕获Python异常并记录到MemOS的error_log_mem
"""
import sys
import os
import traceback
from pathlib import Path
from datetime import datetime
import threading
# 添加项目根目录到Python路径
project_root = Path(__file__).parent.parent
sys.path.insert(0, str(project_root))
# 全局锁，防止并发记录错误时冲突
_error_lock = threading.Lock()
_error_catcher_enabled = True
try:
# 使用SimpleMemoryOps，现在支持持久化存储
from simple_memory_ops_sdk import SimpleMemoryOps as MemoryOps
except ImportError:
try:
from memory_ops_sdk import MemoryOps
except ImportError:
MemoryOps = None
def record_error(exc_type, exc_value, exc_traceback, context_info=None):
"""
记录错误到error_log_mem
Args:
exc_type: 异常类型
exc_value: 异常值
exc_traceback: 异常追踪信息
context_info: 额外的上下文信息
"""
if not _error_catcher_enabled or MemoryOps is None:
return
try:
with _error_lock:
# 初始化Memory Ops SDK
mem = MemoryOps(verbose=False)
# 获取异常信息
error_type = exc_type.__name__ if exc_type else "Unknown"
error_message = str(exc_value) if exc_value else "No message"
# 获取堆栈跟踪
if exc_traceback:
tb_lines = traceback.format_exception(exc_type, exc_value, exc_traceback)
stack_trace = ''.join(tb_lines)
# 获取错误发生的文件和行号
tb = exc_traceback
while tb.tb_next:
tb = tb.tb_next
error_file = tb.tb_frame.f_code.co_filename
error_line = tb.tb_lineno
error_function = tb.tb_frame.f_code.co_name
else:
stack_trace = "No traceback available"
error_file = "Unknown"
error_line = 0
error_function = "Unknown"
# 构造记忆内容
memory_content = f"""Python Error: {error_type}
Message: {error_message}
File: {error_file}
Line: {error_line}
Function: {error_function}
Time: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
Stack Trace:
{stack_trace}"""
if context_info:
memory_content += f"\nContext: {context_info}"
# 构造元数据
metadata = {
"type": "event",
"source": "error_catcher",
"error_type": error_type,
"error_message": error_message,
"error_file": error_file,
"error_line": error_line,
"error_function": error_function,
"severity": "high" if error_type in ["SystemExit", "KeyboardInterrupt", "MemoryError"] else "medium",
"timestamp": datetime.now().isoformat()
}
# 构造标签
tags = ["error", "python", "exception", error_type.lower()]
# 根据文件路径添加更多标签
if "test" in error_file.lower():
tags.append("testing")
if "script" in error_file.lower():
tags.append("script")
# 检查是否有error_log_mem类型
available_types = mem.list_memory_types() if hasattr(mem, 'list_memory_types') else {}
memory_type = 'error_log_mem' if 'error_log_mem' in available_types else 'general_mem'
# 添加记忆
success = mem.add(
text=memory_content,
memory_type=memory_type,
tags=tags,
metadata=metadata
)
if success:
print(f"🔍 Error recorded to {memory_type}: {error_type}")
# 关闭连接（如果支持）
if hasattr(mem, 'close'):
mem.close()
except Exception as e:
# 避免错误处理器本身出错导致无限递归
print(f"Error in error_catcher: {e}", file=sys.stderr)
def global_exception_handler(exc_type, exc_value, exc_traceback):
"""
全局异常处理器
"""
# 记录错误到MemOS
record_error(exc_type, exc_value, exc_traceback)
# 调用原始的异常处理器
sys.__excepthook__(exc_type, exc_value, exc_traceback)
def install_error_catcher():
"""
安装全局错误捕获器
"""
global _error_catcher_enabled
_error_catcher_enabled = True
sys.excepthook = global_exception_handler
print("✅ Global error catcher installed")
def uninstall_error_catcher():
"""
卸载全局错误捕获器
"""
global _error_catcher_enabled
_error_catcher_enabled = False
sys.excepthook = sys.__excepthook__
print("❌ Global error catcher uninstalled")
def test_error_catcher():
"""
测试错误捕获器
"""
print("Testing error catcher...")
try:
# 故意触发一个错误
result = 1 / 0
except ZeroDivisionError:
print("ZeroDivisionError caught and should be recorded")
# 自动安装错误捕获器（当模块被导入时）
if __name__ != "__main__":
install_error_catcher()
def main():
"""主函数 - 用于测试"""
install_error_catcher()
test_error_catcher()
if __name__ == "__main__":
main()