aaron/stock-ai-agent

Fork 0

aaron 5a56076858 增加异常处理

2026-02-22 11:52:14 +08:00

5.4 KiB

Raw Permalink Blame History

系统异常通知功能使用说明

功能概述

系统异常通知功能会自动捕获所有未处理的异常，并发送通知到飞书，让你及时了解系统错误。

主要特性

✅ 自动捕获异常 - 捕获系统中所有未处理的异常 ✅ 飞书通知 - 自动发送异常详情到飞书 ✅ 完整堆栈信息 - 包含异常类型、错误信息和完整堆栈跟踪 ✅ 冷却机制 - 避免重复通知（默认5分钟冷却时间） ✅ 可配置 - 可以启用/禁用通知，调整冷却时间

工作原理

1. 异常捕获流程

系统异常 → 全局异常处理器 → 记录日志 → 检查冷却时间 → 发送飞书通知

2. 通知内容

飞书通知包含以下信息：

🚨 异常类型（如 ValueError, RuntimeError）
📝 异常信息（错误描述）
⏰ 发生时间
📚 完整堆栈跟踪（代码格式化显示）

3. 冷却机制

为了避免同一错误重复通知，系统实现了冷却机制：

默认冷却时间：300秒（5分钟）
在冷却期内的异常只记录日志，不发送飞书通知
冷却时间过后再次出现异常才会发送通知

配置说明

代码配置

异常通知已在 main.py 中自动初始化：

# 初始化飞书错误通知
from app.services.feishu_service import get_feishu_service
feishu_service = get_feishu_service()
init_error_notifier(
    feishu_service=feishu_service,
    enabled=True,   # 启用异常通知
    cooldown=300    # 5分钟冷却时间
)

动态调整配置

你可以在运行时动态调整异常通知配置：

from app.utils.error_handler import get_exception_handler

handler = get_exception_handler()

# 启用/禁用异常通知
handler.set_enabled(True)   # 启用
handler.set_enabled(False)  # 禁用

# 调整冷却时间（秒）
handler.set_cooldown(300)   # 5分钟
handler.set_cooldown(60)    # 1分钟
handler.set_cooldown(0)     # 禁用冷却（每次都通知）

使用示例

示例1: 手动触发异常通知

from app.utils.error_handler import get_exception_handler

handler = get_exception_handler()

# 手动报告异常
try:
    # 你的代码
    result = 1 / 0
except Exception as e:
    import sys
    exc_type, exc_value, exc_traceback = sys.exc_info()
    handler.handle_exception(exc_type, exc_value, exc_traceback)

示例2: 在函数中装饰器使用

from functools import wraps
from app.utils.error_handler import get_exception_handler

def notify_on_error(func):
    """异常时发送通知的装饰器"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except Exception as e:
            handler = get_exception_handler()
            import sys
            exc_type, exc_value, exc_traceback = sys.exc_info()
            handler.handle_exception(exc_type, exc_value, exc_traceback)
            raise
    return wrapper

@notify_on_error
def risky_function():
    # 可能出错的代码
    pass

测试

运行测试脚本验证功能：

source backend/venv/bin/activate
python scripts/test_error_notification.py

测试脚本会：

触发各种类型的异常
发送飞书通知
测试冷却机制
验证通知格式

飞书通知示例

🚨 **系统异常报警**

**异常类型**: ZeroDivisionError
**异常信息**: division by zero
**发生时间**: 2026-02-22 11:50:59

**堆栈跟踪**:

Traceback (most recent call last): File "/app/main.py", line 100, in process_data result = calculate_ratio(x, y) File "/app/utils.py", line 50, in calculate_ratio return x / y ZeroDivisionError: division by zero


⚠️ 请及时处理系统异常

注意事项

敏感信息: 异常堆栈可能包含敏感信息（如API密钥、密码等），请注意飞书安全性
网络依赖: 发送飞书通知需要网络连接，如果网络异常会只记录日志
性能影响: 异常处理本身对性能影响很小，但频繁异常可能表示系统有问题
日志级别: 异常信息会记录到 ERROR 级别日志，可通过日志系统查询

故障排查

问题1: 没有收到飞书通知

检查项：

✅ 飞书 webhook URL 是否正确配置（.env 文件中的 FEISHU_WEBHOOK_URL）
✅ FEISHU_ENABLED=true 是否设置
✅ 网络连接是否正常
✅ 是否在冷却期内
✅ 查看日志中是否有 "飞书消息发送成功" 或相关错误信息

问题2: 通知太频繁

解决方案：

增加冷却时间：handler.set_cooldown(600) # 10分钟
或者临时禁用通知：handler.set_enabled(False)

问题3: 想临时禁用通知

from app.utils.error_handler import get_exception_handler
handler = get_exception_handler()
handler.set_enabled(False)

技术实现

异常处理器使用 Python 的 sys.excepthook 机制，可以捕获所有未处理的异常：

import sys

def exception_hook(exc_type, exc_value, exc_traceback):
    # 处理异常
    handler.handle_exception(exc_type, exc_value, exc_traceback)

sys.excepthook = exception_hook

这种方式可以捕获：

主线程中的所有未处理异常
异步任务中的异常（如果未正确捕获）
脚本运行时的异常

但不能捕获：

已被 try-except 捕获的异常（这是设计行为）
子线程中的异常（需要在子线程中单独处理）
系统级别的严重错误（如 Segmentation Fault）

5.4 KiB Raw Permalink Blame History Unescape Escape