AI Agent 内存泄漏：那些让你的应用越跑越慢的隐藏杀手

做 AI Agent 开发的时候，很多人会遇到一个诡异的问题：刚上线时跑得飞快，过了几天就开始变慢，再过几周应用直接卡死。重启能短暂恢复，但很快又回到老样子。

这种情况，十有八九是内存泄漏在作祟。

今天来聊聊 AI Agent 开发中常见的内存泄漏类型、怎么排查，以及怎么从根上避免。

什么是 AI Agent 的内存泄漏？

传统软件开发里的内存泄漏，指的是程序申请了内存但没有正确释放。AI Agent 的内存泄漏要更复杂一些——它不仅仅是代码层面的内存，还有上下文膨胀、向量数据库污染、session 状态累积这些 AI 场景特有的泄漏形式。

简单说：你的 Agent 持有的"记忆"越来越多，但有些记忆是无用的、甚至是有害的，它们没有被清理掉，导致系统越来越慢。

第一类泄漏：Session 状态的无限膨胀

问题现象：

Agent 跑久了之后，每次对话的响应时间越来越长。从最开始的 200ms，慢慢爬到 2s、5s 甚至超时。

根因分析：

很多 Agent 实现会在每次对话后，把用户输入、模型输出、系统响应全部存入 session history。如果 session 没有做长度限制或者没有截断策略，这些历史数据会一直累积，下一次请求时会原封不动地发送给模型——上下文越来越长，推理自然越来越慢。

看一个典型的问题代码：

# ❌ 有泄漏风险的做法
def chat(user_input, session_id):
    session = get_session(session_id)
    # 没有限制地追加
    session.history.append({"role": "user", "content": user_input})
    
    # 每次都把完整历史发给模型
    response = model.invoke(session.history)
    session.history.append({"role": "assistant", "content": response})
    return response

这段代码看起来完全正常，但问题是：session.history 会无限增长。一个跑了 1000 轮对话的 session，下一次请求要把 1000 轮的内容全部发送给模型——不仅是钱的问题，速度会慢到不可用。

怎么修：

# ✅ 正确做法：固定窗口 + 摘要压缩
def chat(user_input, session_id):
    session = get_session(session_id)
    
    # 追加新输入
    session.history.append({"role": "user", "content": user_input})
    
    # 只保留最近 N 轮
    MAX_HISTORY = 10
    if len(session.history) > MAX_HISTORY * 2:  # user + assistant
        # 对早期内容做摘要，保留核心信息
        summary = summarizeEarlierContent(session.history[:-MAX_HISTORY * 2])
        session.history = [{"role": "system", "content": summary}] + session.history[-MAX_HISTORY * 2:]
    
    response = model.invoke(session.history)
    session.history.append({"role": "assistant", "content": response})
    return response

第二类泄漏：向量数据库里的垃圾

问题现象：

向量搜索的返回结果越来越不相关，检索质量下降，但embedding模型和向量索引都没换。

根因分析：

很多 Agent 用向量数据库做长期记忆，文档一段段地写入向量库。但很少有人定期清理：如果用户上传了一堆临时文档做了一次性分析，或者测试时灌入了一批垃圾数据，这些内容会一直在向量库里躺着，每次检索时这些垃圾会和有效信息一起被召回，污染检索结果。

另一个常见问题：向量化后的数据没有做去重。同一份文档被上传两次，向量库里就有两份几乎一样的向量，检索时它们会同时被召回，浪费计算资源不说，还拉低了结果质量。

怎么修：

# ✅ 定期清理 + 去重策略
def periodicCleanup():
    # 1. 删除超过 N 天的临时文档
    threshold = datetime.now() - timedelta(days=30)
    vectorDB.delete where metadata["is_temporary"] == True and timestamp < threshold
    
    # 2. 向量去重（用余弦相似度检测）
    all_vectors = vectorDB.getAllVectors()
    for i, vec in enumerate(all_vectors):
        for j in range(i+1, len(all_vectors)):
            if cosineSimilarity(vec, all_vectors[j]) > 0.98:
                vectorDB.deleteDuplicates(vec[j])

# ✅ 写入时做去重检测
def addMemory(content, metadata):
    content_hash = hashlib.md5(content.encode()).hexdigest()
    if vectorDB.exists(content_hash):
        return  # 跳过重复内容
    vectorDB.insert(embed(content), metadata, content_hash=content_hash)

第三类泄漏：工具调用产生的副作用

问题现象：

Agent 跑着跑着开始出现奇怪的错误——文件找不到、数据库连接池耗尽、网络端口被占用。最诡异的是这些错误不一定是必现的，有时候正常有时候报错。

根因分析：

AI Agent 的一大特点是工具化——Agent 会调用各种外部工具来完成复杂任务。但很多工具在使用后没有正确清理资源：

• 打开文件没关
• 数据库连接用完没释放
• 临时文件没删
• 后台线程/进程没终止

这些资源泄漏在单独一次调用时不会造成明显问题，但 Agent 在长时间运行中会积累大量这类"半开"资源，最终导致系统可用资源耗尽。

一个典型例子：

# ❌ 文件泄漏
def analyzeDocument(path):
    # 打开文件
    with open(path, 'r') as f:  # 如果这行前面 early return，文件就没关
        content = f.read()
    
    if len(content) == 0:
        return None  # 这里直接返回，with 块不会执行，文件实际没关
    
    # 但实际上 with 语句在函数结束时会自动关闭文件
    # 这个例子不太准确，看看下面的数据库例子

# ❌ 数据库连接泄漏
def queryDB(sql):
    conn = getConnection()  # 从连接池获取
    cursor = conn.cursor()
    cursor.execute(sql)
    results = cursor.fetchall()
    # 没有 cursor.close() 也没有 conn.close()
    # 连接池耗尽后，新请求只能等待
    return results

怎么修：

用 context manager 和 proper cleanup：

# ✅ 正确做法
def queryDB(sql):
    with getConnection() as conn:  # context manager 确保退出时释放
        with conn.cursor() as cursor:
            cursor.execute(sql)
            return cursor.fetchall()
    # with 块结束，自动还回连接池

# ✅ 对于 Agent 框架，统一管理工具生命周期
class ToolManager:
    def __init__(self):
        self.openResources = []
    
    def register(self, resource):
        self.openResources.append(resource)
    
    def cleanup(self):
        for resource in self.openResources:
            try:
                resource.close()
            except:
                pass
        self.openResources = []
    
    def __del__(self):
        self.cleanup()  # 确保进程退出时清理

第四类泄漏：Callback 和 Event Handler 的累积

问题现象：

Agent 在长时间运行后，响应开始出现"串话"——这次请求的回复里偶尔会出现上一次请求的内容。

根因分析：

很多 Agent 框架支持 callback 和 event handler，允许开发者注册各种钩子来处理特定事件。如果在每次 session 初始化时都注册新的 handler 但没有注销旧的，多个 session 跑下来，handler 列表会越来越长——一个事件触发时，多个过期的 handler 同时响应，造成"串话"。

这个问题在 Flask/Django 这类 Web 框架配合 Agent 使用时特别常见：每次请求都初始化一个新的 Agent 实例，如果 Agent 内部注册了全局 handler，这些 handler 会在多次请求间累积。

怎么修：

# ✅ 统一管理 handler，用 WeakSet 自动清理
import weakref

class AgentEventBus:
    def __init__(self):
        self._handlers = weakref.WeakSet()  # 引用消失时自动清理
    
    def subscribe(self, handler):
        self._handlers.add(handler)
    
    def unsubscribe(self, handler):
        self._handlers.discard(handler)
    
    def emit(self, event):
        for handler in self._handlers:  # 只会调用活着的 handler
            handler(event)

# ✅ 或者在 Agent 生命周期结束时明确清理
def createAgent(session_id):
    agent = Agent()
    agent.on("message", handleMessage)  # 注册 handler
    
    # 注册清理钩子
    atexit.register(lambda: agent.cleanup())
    
    return agent

怎么系统地排查内存泄漏？

遇到疑似泄漏问题时，按这个顺序排查：

第一步：量化指标先行

不要凭感觉，先把数据跑出来：

• 进程内存占用（ps aux | grep your_process）
• Session 平均长度分布
• 向量数据库记录总数
• 活跃连接数/文件描述符数量

如果可能，上报给监控平台，画出时间曲线——内存泄漏的特征是曲线单调递增（或锯齿递增），正常应该是平稳的。

第二步：用 Profilers 定位具体泄漏点

Python 的话：

• tracemalloc - 追踪 Python 对象内存分配
• objgraph - 可视化对象引用关系
• memory_profiler - 逐行内存分析

import tracemalloc

tracemalloc.start()

# 你的代码...

snapshot = tracemalloc.take_snapshot()
top_stats = snapshot.statistics('lineno')

print("[ Top 10 memory consumers ]")
for stat in top_stats[:10]:
    print(stat)

第三步：针对性修复

找到泄漏点后，根据类型选择对应的修复策略——上面已经分类型讲过了。

写在最后

AI Agent 的内存泄漏比传统软件的更隐蔽，因为它混合了计算内存（代码层面的）和"认知内存"（上下文、向量库里的内容）两种不同的泄漏形式。

核心的防御思路是：给所有累积型资源都加上明确的边界和清理策略。Session 有长度限制、向量库有生命周期管理、工具调用有资源管理、handler 有注册和注销机制——在设计阶段就把这些边界画好，上线后才能安心。

你的 Agent 有没有遇到类似的性能劣化问题？用什么方法排查和解决的？欢迎留言说说。