AICoding代码异常实时监控告警技术实现文档

风骏时光少年

发布于 2026-07-16 20:35:54

230

AICoding代码异常实时监控告警技术实现文档

一、概述

1.1 文档目的

本文档介绍一套轻量化AICoding代码异常实时监控告警方案，覆盖代码运行报错、接口超时、语法异常、资源占用超标四类核心监控场景，实现异常毫秒级捕获、多渠道推送告警，适用于线上AI代码推理、本地AICoding编辑器、云端代码运行容器等环境。方案基于Python监控采集服务+日志解析引擎+告警分发模块搭建，低侵入、易部署，可快速集成现有AICoding平台。

1.2 监控范围

1. 代码语法、运行时、空指针、数组越界等程序异常；
2. AI模型推理代码超时、显存/内存占用超限；
3. AICoding服务接口请求失败、响应延迟过高；
4. 批量代码执行任务崩溃、进程意外退出。

1.3 告警渠道

支持控制台打印、WebHook（企业微信/钉钉）、本地日志文件三类推送渠道，可按需扩展短信、邮件告警能力。

二、整体架构

1. 采集层：实时监听AICoding运行日志、进程资源指标；
2. 解析层：正则匹配异常关键词、阈值判定资源超限；
3. 告警层：封装统一告警分发接口，多渠道同步推送异常信息；
4. 持久层：异常记录写入本地日志，用于事后排查复盘。

三、完整代码实现

3.1 环境依赖

pip install psutil requests

3.2 监控告警核心代码

import re
import time
import psutil
import requests
from datetime import datetime

# 配置项
CONFIG = {
    "monitor_log_path": "./aicoding_runtime.log",  # AICoding运行日志路径
    "mem_threshold": 80,  # 内存占用告警阈值(%)
    "gpu_threshold": 85,  # 显存占用告警阈值(%)
    "webhook_url": "https://oapi.dingtalk.com/robot/send?access_token=xxx",
    "alert_switch": {"console": True, "webhook": True, "log_file": True}
}

# 异常正则匹配规则
ERROR_RULES = [
    re.compile(r"SyntaxError.*"),
    re.compile(r"IndexError.*"),
    re.compile(r"RuntimeError.*"),
    re.compile(r"TimeoutError.*"),
    re.compile(r"进程意外退出"),
]

class AICodingAlert:
    def __init__(self):
        self.log_file = open("./aicoding_alert.log", "a", encoding="utf-8")

    def send_alert(self, alert_content: str):
        """统一告警分发入口"""
        alert_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        full_msg = f"【AICoding代码异常实时告警】{alert_time}\n异常详情：{alert_content}"

        # 控制台告警
        if CONFIG["alert_switch"]["console"]:
            print(f"\033[31m{full_msg}\033[0m")

        # 日志持久化
        if CONFIG["alert_switch"]["log_file"]:
            self.log_file.write(full_msg + "\n")
            self.log_file.flush()

        # WebHook推送
        if CONFIG["alert_switch"]["webhook"]:
            data = {"msgtype": "text", "text": {"content": full_msg}}
            try:
                requests.post(CONFIG["webhook_url"], json=data, timeout=3)
            except Exception:
                print("WebHook推送失败，请检查链接")

    def check_resource(self):
        """检测内存资源占用"""
        mem = psutil.virtual_memory()
        mem_usage = mem.percent
        if mem_usage >= CONFIG["mem_threshold"]:
            self.send_alert(f"内存占用超标，当前使用率：{mem_usage}%")

    def parse_log_line(self, line: str):
        """解析单行日志，匹配异常"""
        for rule in ERROR_RULES:
            if rule.search(line):
                self.send_alert(f"捕获代码异常日志：{line.strip()}")
                break

    def monitor_log(self):
        """实时监听日志文件"""
        with open(CONFIG["monitor_log_path"], "r", encoding="utf-8") as f:
            f.seek(0, 2)
            while True:
                line = f.readline()
                if not line:
                    time.sleep(0.2)
                    self.check_resource()
                    continue
                self.parse_log_line(line)

    def close(self):
        self.log_file.close()

if __name__ == "__main__":
    alert = AICodingAlert()
    try:
        print("AICoding代码异常实时监控告警服务已启动")
        alert.monitor_log()
    except KeyboardInterrupt:
        alert.send_alert("监控服务手动停止")
        alert.close()