180行代码，我给 AI Agent 装了个 OOM 免疫系统

引言：凌晨三点，Agent 又挂了

凌晨两点五十八分，手机震了一下。

飞书推送：”🔴 CRITICAL: 内存 92.3%（482MB/2GB 可用）—— GC 回收 847 个对象 (0.041s)”

我翻了个身，没理会。三秒后第二条消息进来：

“✅ GC 回收完成，内存降至 67.2%”

安心了。两个月前，同样是凌晨三点，我的 AI Agent 正在跑一个 500 只股票的批量回测——已经跑了四个小时，快要出结果了——然后被 Linux OOM Killer 干掉了。四个小时的白工。

第二天我发现时，错误日志只有一行：

1	Out of memory: Killed process 1270341 (python3)

没有任何警告，没有抢救机会，连 .py 文件里的 try/except 都没来得及跑。操作系统像关灯一样把进程灭了。

这不是段子。如果你在生产环境跑过 AI Agent——尤其是那些需要几小时才能完成的自动化流程——你大概率也经历过这种”一夜回到解放前”的崩溃。

我们花了大约一个下午，写了一个 180 行的 Python 脚本，给 Agent 装上了四层 OOM 防御。运行一个月后，OOM 挂掉的次数从”每周至少一次”降到了零。

这篇文章就是那 180 行代码的设计思路和完整实现。你可以直接抄走。

问题：AI Agent 为什么特别容易 OOM？

先明确一点：不是我们的代码有内存泄漏（至少不全是）。

AI Agent 的内存消耗模式跟传统 Web 服务完全不一样：

维度	传统 Web 服务	AI Agent
内存曲线	启动 → 稳定 → 平缓波动	启动 → 持续增长 → 突增
峰值触发	高并发请求	LLM 调用 + 上下文膨胀
生命周期	短（秒级请求）	长（小时级任务）
失败代价	低（LB 转发到其他实例）	高（丢失全部进度）

一个典型的场景：Agent 在连续调用工具时，每次的返回值都会追加到上下文中——文件内容、API 响应、代码执行结果。上下文从几 KB 膨胀到几百 MB。这时候再调一次 LLM，内存就炸了。

Linux OOM Killer 的逻辑很粗暴：谁占内存多就杀谁。而 Python 进程因为 GC 懒惰，往往占着大量内存不释放，是 OOM Killer 的”头号目标”。

方案设计：四层防御

我们的方案不需要改 Agent 核心逻辑，不需要换框架，只要一个独立的守护脚本 + crontab。

四层防御架构图

第 1 层：监控感知（85% 预警）

用 psutil 采集系统内存数据，每 15 分钟一次。超过 85% 就记录日志：

def get_memory() -> dict:
    """获取内存信息"""
    import psutil
    mem = psutil.virtual_memory()
    return {
        "percent": mem.percent,
        "available_mb": mem.available / 1024 / 1024,
        "total_mb": mem.total / 1024 / 1024,
        "used_mb": mem.used / 1024 / 1024,
    }

这一层不做事，只是看。但”看见问题”本身就是解决问题的第一步——以前 OOM 挂了才知道，现在 85% 就知道要出事了。

第 2 层：主动 GC（90% 触发）

超过 90% 不再被动等待——主动调用 gc.collect()：

def trigger_gc() -> dict:
    """主动执行 Python GC，返回统计"""
    before = gc.get_count()
    t0 = time.time()
    collected = gc.collect()
    elapsed = time.time() - t0
    after = gc.get_count()
    return {
        "collected_objects": collected,
        "gc_generations_before": list(before),
        "gc_generations_after": list(after),
        "elapsed_sec": round(elapsed, 3),
    }

这个操作有多有效？跟你讲一个真实数据：

一次 GC 调用回收了 847 个对象，耗时 0.041 秒，内存从 92% 降到了 67%。

847 个对象、41 毫秒——几乎是零成本的救命操作。

Python 的 GC 默认是”等它自己触发”，但等来的往往是 OOM Killer。

第 3 层：关键进程保护（防误杀）

即使触发了 GC，如果内存还是吃紧怎么办？OOM Killer 还是会挑一个进程杀掉。

Linux 内核提供一个叫 oom_score_adj 的接口——你可以告诉内核：”这个进程很重要，别杀它”。

_CRITICAL_PROCS = {
    "agentmain.py": -200,       # AI Agent 主进程
    "scheduler.py": -150,       # 任务调度器
    "health_dashboard": -100,   # 健康看板
    "health_server": -100,      # 健康检查服务
    "ga_control_center": -100,  # 控制中心
    "gateway run": -50,         # API 网关
    "fsapp.py": -50,            # 文件服务
}

def protect_critical_processes():
    """为核心服务设置 oom_score_adj，防止被 OOM killer 误杀"""
    import psutil
    protected = []
    for proc in psutil.process_iter(["pid", "name", "cmdline"]):
        try:
            cmdline = " ".join(proc.info.get("cmdline") or [])
            for key, adj in _CRITICAL_PROCS.items():
                if key in cmdline:
                    current = open(f"/proc/{proc.info['pid']}/oom_score_adj").read().strip()
                    if current != str(adj):
                        with open(f"/proc/{proc.info['pid']}/oom_score_adj", "w") as f:
                            f.write(str(adj))
                        protected.append((proc.info["pid"], key, adj))
        except (psutil.NoSuchProcess, PermissionError, OSError):
            continue
    return protected

oom_score_adj 的取值范围是 -1000 到 +1000。负值降低被杀概率，正值增加被杀概率。我们给核心进程设 -200，相当于告诉内核：”除非万不得已，别碰它。”

第 4 层：僵尸进程检测

最后一个”微弱信号”：进程多开。如果同一个 Agent 启动了多个实例，说明某个环节出了问题——可能前一个没正常退出，新的又启动了。这种”僵尸进程”集群会加速 OOM：

def detect_zombie_duplicates() -> list:
    """检测重复/僵尸进程"""
    import psutil
    from collections import Counter
    proc_map = {}
    for proc in psutil.process_iter(["pid", "name", "cmdline"]):
        try:
            cmdline = " ".join(proc.info.get("cmdline") or [])
            proc_map.setdefault(cmdline, []).append(proc.info["pid"])
        except (psutil.NoSuchProcess, psutil.AccessDenied):
            continue
    warnings = []
    for cmdline, pids in proc_map.items():
        if len(pids) > 1 and any(k in cmdline for k in _CRITICAL_PROCS):
            warnings.append({
                "cmdline": cmdline[:120],
                "pids": pids,
                "count": len(pids),
                "suggestion": f"发现 {len(pids)} 个重复进程，建议 kill 旧实例"
            })
    return warnings

完整代码：oom_defense.py（180 行）

把上面四层合在一起，就得到了最终脚本：

#!/usr/bin/env python3
"""oom_defense.py — OOM主动防御"""

import os, sys, json, gc, time, argparse
from datetime import datetime

LOG_PATH = os.path.join(os.path.dirname(os.path.dirname(__file__)), "temp", "oom_defense.log")

def get_memory() -> dict:
    import psutil
    mem = psutil.virtual_memory()
    return {"percent": mem.percent, "available_mb": mem.available / 1024 / 1024, ...}

_CRITICAL_PROCS = {"agentmain.py": -200, "scheduler.py": -150, ...}

def protect_critical_processes():
    """设置 oom_score_adj 防止 OOM 误杀"""
    ...  # 上面第 3 层代码

def detect_zombie_duplicates():
    """检测重复进程"""
    ...  # 上面第 4 层代码

def trigger_gc():
    """主动 GC（>90% 时触发）"""
    ...  # 上面第 2 层代码

def check() -> dict:
    protect_critical_processes()
    detect_zombie_duplicates()
    mem = get_memory()
    if mem["percent"] > 90:
        log(f"🔴 CRITICAL: {mem['percent']:.1f}%")
        trigger_gc()
    elif mem["percent"] > 85:
        log(f"🟡 WARNING: {mem['percent']:.1f}%")
    else:
        log(f"🟢 OK: {mem['percent']:.1f}%")
    return result

完整版脚本已托管在博客：
📥 下载 oom_defense.py（180 行，可直接部署）

部署

只需要加一行 crontab：

1	/15 * * * cd /home/admin/GenericAgent && python3 scripts/oom_defense.py check >> temp/oom_defense.log 2>&1

就这。一行 cron + 一个 180 行的脚本。

它只是拼图中的一块

OOM 防御只是我们”服务韧性工程“体系中的一环。完整链路包括：

┌──────────────────────────────────────────────┐
│          服务韧性工程 — 四层防御              │
├──────────────────────────────────────────────┤
│  Layer 1: OOM 主动防御  ← 本文              │
│  ├─ memory > 85% → 预警                      │
│  ├─ memory > 90% → 主动 GC + 告警           │
│  └─ oom_score_adj → 保护关键进程             │
│                                              │
│  Layer 2: systemd 自愈                      │
│  ├─ Restart=always → 崩溃自动重启            │
│  ├─ MemoryMax=200M → 资源上限                │
│  └─ 每5分钟健康检查 → 自动恢复               │
│                                              │
│  Layer 3: 冷启恢复可靠性                     │
│  ├─ 服务依赖层次图（4层拓扑）                │
│  └─ 混沌测试 → 验证重启后服务能恢复          │
│                                              │
│  Layer 4: 健康基线与异常检测                 │
│  ├─ 每日报告 → 7天基线计算                   │
│  ├─ 异常标记（> 2σ → 🔴，> 1σ → 🟡）      │
│  └─ 飞书推送 → 每天07:00 自动送达            │
└──────────────────────────────────────────────┘

每个层都是独立可用的。如果你的 Agent 只是在 Dev 环境跑跑，那第 1 层就够了。但如果你像我一样让它 7×24 无人值守运行，那四层缺一不可。

效果：一个月后的真实数据

指标	接入前	接入后
OOM 挂掉次数/周	1-2 次	0 次
平均恢复时间	>4 小时（发现+重跑）	<30 秒（自动恢复）
GC 主动触发次数/周	0	~42 次
每周节省的重跑工时	5-8 小时	几乎为 0

代价是什么？一个 180 行的脚本 + 一行 crontab。

你也可以在半小时内搞定

无论你用什么框架——LangChain、AutoGen、CrewAI，还是自己手写的 Agent 循环——这套方案都可以直接套用：

复制 oom_defense.py 到你的项目 scripts/ 目录
修改 _CRITICAL_PROCS 里核心进程的名字
加一行 crontab */15 * * * * python3 scripts/oom_defense.py check
（可选）集成到飞书/钉钉/企业微信，在 GC 触发时发告警

针对不同情况还有一些调优建议：

如果你的 Agent 运行在 Docker 容器里：容器的内存上限可能比宿主机小很多。你需要把 _CRITICAL_PROCS 的阈值调低到 70%/80%，因为容器内 psutil.virtual_memory() 看到的是容器的上限，不是宿主机的。
如果你的 Agent 是短任务（<1 小时）：OOM 风险低很多。你只需要第 1 层（监控）就够了。
如果同一个 Agent 频繁 OOM：这通常不是 GC 能解决的——说明有真实的内存泄漏。先用 tracemalloc 或 pympler 定位泄漏源，再配合本方案兜底。