跳转至

Agent 系统中每次 LLM 调用都创建一个记录对象,如何优化这类高频小对象的内存占用?

在 Agent 高频调用 LLM 的场景里,每次请求生成一个记录对象,看似无害,但当 QPS 上去后,GC 压力和内存毛刺会让你头疼。我优化这类问题有一个固定思路,不是上来就改代码,而是先分清楚:这对象是“胖”还是“多”,是“短命”还是“长驻”。


🔎 为什么小对象会压垮内存?

每个 Python 对象的固定开销大约 56 字节(PyObject 头部),如果你用普通的 class 且没做 slots,每个属性又有一个 dict 开销(又是 64+ 字节)。一个原本十几字节的数据,包装成对象可能膨胀到 150 字节以上。 1000 个请求/秒 → 150KB/s → 每秒产生大量垃圾,GC 频繁触发,CPU 也会被拖累。


🧱 第一步:把对象“压扁”——slotsdataclass(slots=True)

这是我几乎会第一时间做的事情。

# ❌ 没优化
class LLMRecord:
    def __init__(self, ts, model, tokens, latency):
        self.ts = ts
        self.model = model
        self.tokens = tokens
        self.latency = latency

# ✅ 优化后:无 __dict__,内存直降 50%-70%
from dataclasses import dataclass

@dataclass(slots=True)
class LLMRecord:
    ts: float
    model: str
    tokens: int
    latency: float
属性 普通类 slots
每对象内存 ≈ 200 字节 ≈ 72 字节
属性访问速度 慢(字典查找) 快(偏移量访问)
是否支持动态属性

如果字段更少、值域更小,甚至可以考虑用 namedtuple(不可变,内存更小)或 array 模块分层存储。


♻️ 第二步:对象复用——池化高频短命对象

不要每次调用都 LLMRecord(...) 然后丢弃,用一个预分配的对象池来循环使用,尤其是那些生命周期只在“生成本地日志或缓冲”的记录。

import queue

class RecordPool:
    def __init__(self, size=2000):
        self._pool = queue.Queue(size)
        for _ in range(size):
            self._pool.put(LLMRecord(0, "", 0, 0.0))  # 预置空对象

    def acquire(self, ts, model, tokens, latency):
        r = self._pool.get()          # 取出一个,可能阻塞或直接拿到
        r.ts = ts
        r.model = model
        r.tokens = tokens
        r.latency = latency
        return r

    def release(self, record):
        # 重置敏感字段,避免泄露
        record.model = ""
        self._pool.put(record)

使用 asyncio.Queue 也可以做异步安全的池子。这样把分配/释放变成借/还,内存曲线拉平成一条直线。


⏳ 第三步:缩短对象生命周期——流式处理,不囤积

Agent 产生的记录很多时候是用来“攒到一定量再批量写库”。这会让大量活对象挂在内存里。

改成异步流式写入,生成一条、发往后台队列一条,对象很快就能回收或归还池子。

async def worker(record_queue):
    while True:
        record = await record_queue.get()
        await flush_to_db(record)
        pool.release(record)   # 写入后立即还池

🪞 第四步:避免循环引用和意外延长生命

记录对象如果被放进某个全局 list 或异常堆栈里没清掉,GC 就没法回收。

建议:

  • 记录对象不要持有对 Agent 实例的引用,只保留基础数据。

  • weakref 做回调引用,不阻碍回收。

  • 定期打印 gc.get_stats() 观察代回收情况。


🧹 第五步:如果数据量巨大,考虑“列式存储”思路

对于高频记录,完全面向行的对象可能不是最优解。你可以:

  • arraynumpy 存数值字段(latency、tokens)

  • list 统一存字符串字段

  • 只在需要展示时拼装成临时对象

这虽然增加了编码复杂度,但在每秒数万记录的极限场景下,内存效率有数量级的提升。


📉 我自己的一个真实效果对比

早期不做优化的 Agent 监控模块,1000 QPS 下内存占用 800MB,且 GC 停顿明显。 做了三步:slots → 对象池 → 异步写出,峰值内存降到 90MB 以下,GC 停顿消失。 技术本身不酷,酷的是组合拳。


写代码时我总记着:每一个对象都不止是数据,还是垃圾回收器眼里的一粒沙。让沙粒变细、减少扬沙,内存自然就安静了。