ComfyUI 的缓存架构和实现

围绕 ComfyUI,大家讨论最多的是节点、工作流、算力这些,真正去看缓存细节的人其实不多。但只要你开始在一台机器上堆多个模型、多个 LoRA、多个 workflow,缓存策略就会直接决定这几件事:

  • 你是算力浪费,还是把显存 / 内存用在刀刃上;
  • 容器会不会莫名其妙 OOM;
  • 工作流切换时,是“秒级热身”还是“从头再来”。

这篇文章只做一件事:把 ComfyUI 当前的缓存架构和实现讲清楚,重点是三类策略(CLASSIC / LRU / RAM_PRESSURE)在「键怎么算、什么时候命中、什么时候过期」上的差异,以及在多模型、多 LoRA、多 workflow 场景下应该怎么选择。

1. 整体架构:两路缓存 + 四种策略

先把整体结构捋清楚。

1.1 两类缓存:outputs 和 objects

在执行一个工作流的时候,ComfyUI 维护了两类缓存:

  • outputs
    存中间结果和 UI 输出。命中时可以直接跳过节点执行,这部分是真正省计算的地方。
  • objects
    存节点对象实例(类实例),避免每次重新构造节点对象。

这两类缓存由一个统一的集合类 CacheSet 来管理。核心结构在 execution.py 中:

class CacheType(Enum):
    CLASSIC = 0
    LRU = 1
    NONE = 2
    RAM_PRESSURE = 3

class CacheSet:
    def __init__(self, cache_type=None, cache_args={}):
        if cache_type == CacheType.NONE:
            self.init_null_cache()
        elif cache_type == CacheType.RAM_PRESSURE:
            cache_ram = cache_args.get("ram"16.0)
            self.init_ram_cache(cache_ram)
        elif cache_type == CacheType.LRU:
            cache_size = cache_args.get("lru"0)
            self.init_lru_cache(cache_size)
        else:
            self.init_classic_cache()
        self.all = [self.outputs, self.objects]

不管是哪种策略,结构都是:

  • outputs按输入签名做 key
  • objects按 (node_id, class_type) 做 key

1.2 四种策略:CLASSIC / LRU / RAM_PRESSURE / NONE

从启动参数到缓存策略的选择,大致是这样的优先级(在 main.py 中):

  • 指定 --cache-lru → 用 LRU;
  • 否则指定 --cache-ram → 用 RAM_PRESSURE;
  • 否则指定 --cache-none → 完全关闭缓存;
  • 都没指定 → 默认 CLASSIC。

CacheType 的定义前面已经贴了。不同策略只决定 outputs 的实现方式,而 objects 基本始终使用层级缓存 HierarchicalCache(CacheKeySetID),后面细讲。


2. 缓存键:输入签名 + 变化指纹

缓存是否命中,首先取决于“key 算得是否合理”。ComfyUI 的设计核心是:

  • 输出缓存(outputs):用“输入签名 + is_changed 指纹(+ 某些情况下的 node_id)”作为 key;
  • 对象缓存(objects):用 (node_id, class_type) 作为 key。

2.1 输出缓存:输入签名是怎么来的

输出键的生成由 CacheKeySetInputSignature 负责,核心逻辑在 comfy_execution/caching.py:100-126

async def get_node_signature(self, dynprompt, node_id):
    signature = []
    ancestors, order_mapping = self.get_ordered_ancestry(dynprompt, node_id)
    signature.append(await self.get_immediate_node_signature(dynprompt, node_id,
    order_mapping))
    for ancestor_id in ancestors:
        signature.append(await self.get_immediate_node_signature(dynprompt,
        ancestor_id, order_mapping))
    return to_hashable(signature)

这里有几个点:

  1. 不只看当前节点
    它会按拓扑顺序把“当前节点 + 所有祖先”的签名拼在一起。这保证了只要整个子图的拓扑和输入一致,输出就能命中缓存。

  2. immediate 签名包含什么
    get_immediate_node_signature 会把这些信息打包进去(位置见同文件 108–126):

    • class_type:节点类型;
    • is_changed 指纹(通过 fingerprint_inputs/IS_CHANGED);
    • 必要时的 node_id
    • 有序的输入值/链接。
  3. 什么时候把 node_id 也算进 key
    当节点被声明为非幂等,或者内部有 UNIQUE_ID 这种隐含输入时,会把 node_id 加进签名(见 comfy_execution/caching.py:18-23, 116),避免“看起来一样”的节点被错误复用。

最后通过 to_hashable 转成可哈希结构(tuple/frozenset 等),作为最终键值。

结果是:

  • 输入完全相同 → key 相同 → 直接命中;
  • 任意一个上游节点输入或参数变化 → 指纹变了 → key 不同 → 不会复用旧结果。

2.2 对象缓存:用 (node_id, class_type)

节点对象的键由 CacheKeySetID 构造(comfy_execution/caching.py:66-80),逻辑简单:

  • key = (node_id, class_type)

读取时:

obj = caches.objects.get(unique_id)
if obj is None:
    obj = class_def()
    caches.objects.set(unique_id, obj)

对象缓存存在的目的只有一个:同一个 workflow 执行过程中,不要重复 new 节点实例。


3. 缓存容器:Basic / Hierarchical / LRU / RAM / Null

有了 key,还需要一个合理的「容器」和「驱逐策略」。

主要类在 comfy_execution/caching.py

  • BasicCache:基础容器,提供 set_prompt / clean_unused / get / set 等;
  • HierarchicalCache:按“父节点 → 子图”构建层级缓存;
  • LRUCache:在 Basic + Hierarchical 的基础上增加代际 LRU;
  • RAMPressureCache:在 LRU 的基础上增加 RAM 压力驱逐;
  • NullCache:空实现(禁用缓存)。

核心接口统一在 BasicCache

def clean_unused(self):
    self._clean_cache()
    self._clean_subcaches()

层级相关逻辑在 HierarchicalCache,用来支持“子图单独分区缓存”。

3.1 层级缓存:怎么定位到某个节点的分区

HierarchicalCache 通过 parent 链定位子缓存,代码在 comfy_execution/caching.py:242-269

def _get_cache_for(self, node_id):
    parent_id = self.dynprompt.get_parent_node_id(node_id)
    ...
    for parent_id in reversed(hierarchy):
        cache = cache._get_subcache(parent_id)
        if cache is Nonereturn None
    return cache

def get(self, node_id):
    cache = self._get_cache_for(node_id)
    if cache is Nonereturn None
    return cache._get_immediate(node_id)

def set(self, node_id, value):
    cache = self._get_cache_for(node_id)
    assert cache is not None
    cache._set_immediate(node_id, value)

含义很直接:

  • 根节点存在当前 cache;
  • 某些节点生成子图,会在该节点下挂一个子 cache;
  • 读写时,先通过 parent 链找到对应的子 cache 再读写。

clean_unused() 里除了清除不用的 key,还会删除没用到的子 cache 分区(_clean_subcaches())。

4. 执行循环中的使用路径

缓存不是“挂在那就完事了”,它在执行循环中有比较明确的调用点。

4.1 设置 prompt + 清理

启动一次执行时(execution.py:681-685):

is_changed_cache = IsChangedCache(prompt_id, dynamic_prompt, self.caches.outputs)
for cache in self.caches.all:
    await cache.set_prompt(dynamic_prompt, prompt.keys(), is_changed_cache)
    cache.clean_unused()

这里做了三件事:

  1. 给当前 prompt 生成一个 IsChangedCache,为 key 计算提供 is_changed 结果;
  2. 对 outputs / objects 各自执行一次 set_prompt(不同策略实现不同);
  3. 紧接着执行 clean_unused(),做一次基于“当前 prompt 键集合”的清理。

4.2 节点执行前后:cache 命中与写入

在节点执行路径中:

  • 执行前:优先尝试从 outputs 命中(execution.py:686-701);
  • 执行后:将 (ui, outputs) 作为 CacheEntry 写入 outputsexecution.py:568-571)。

为了简化,这里只看抽象行为:

  • 命中 → 跳过计算,直接拿值;
  • 未命中 → 正常跑一遍,将结果塞回缓存。

4.3 每个节点之后的 RAM 轮询

如果是 RAM_PRESSURE 模式,执行完每个节点都会触发一次内存检查(execution.py:720):

self.caches.outputs.poll(ram_headroom=self.cache_args["ram"])

只有 RAMPressureCache 实现了 poll,其他模式下这个调用等同空操作。


5. CLASSIC:默认层级缓存

先看默认策略:CLASSIC。

5.1 初始化与结构

CacheSet.init_classic_cacheexecution.py:97-126):

class CacheType(Enum):
    CLASSIC = 0
    LRU = 1
    NONE = 2
    RAM_PRESSURE = 3

class CacheSet:
    def init_classic_cache(self):
        self.outputs = HierarchicalCache(CacheKeySetInputSignature)
        self.objects = HierarchicalCache(CacheKeySetID)

可以看到:

  • outputs 用层级缓存 + 输入签名做 key;
  • objects 也用层级缓存 + (node_id, class_type) 做 key;
  • 不涉及 LRU 或 RAM 驱逐。

5.2 CLASSIC 的过期机制:完全由「当前 prompt」驱动

CLASSIC 模式不做容量和时间管理,它只有两种“失效”方式:

  1. 提示切换 / 执行前清理

clean_unused() 的核心逻辑(comfy_execution/caching.py:172-195):

def clean_unused(self):
    self._clean_cache()
    self._clean_subcaches()
  • _clean_cache():把不在“当前 prompt 键集合”的项删掉;
  • _clean_subcaches():把不再需要的子缓存分区删掉。

execution.py:681-685 每次绑定新 prompt 时,都会执行这一步。结果是:

  • 换了一个新的 workflow / prompt,旧 workflow 的 outputs / objects 都会被视为“未使用”,被清理掉;
  • CLASSIC 不会跨不同 prompt 保留旧 workflow 的缓存。
  1. 键不命中(指纹失效)

is_changed 的计算在 execution.py:48-89,当节点输入更新时,指纹会变化;CacheKeySetInputSignature 在构造键时会把这个指纹带进去(comfy_execution/caching.py:115-127)。因此只要:

  • 参数 / 输入 / 上游节点的任一变化 → key 改变 → 旧值自然不命中。

5.3 CLASSIC 明确不会做的事

在 CLASSIC 下:

  • 不做 LRU 容量控制:没有 max_size,也没有代际淘汰逻辑;
  • 不做 RAM 压力驱逐:poll() 是空的,执行循环里即使调用了也什么都不干;
  • 不做 TTL:不看时间,只看 prompt 键集合。

对应的注释已经在参考内容中点得很清楚,这里就不重复堆代码了。

5.4 在频繁切换 workflow / 模型时的表现

结合上面的机制,总结一下 CLASSIC 在多 workflow 场景下的行为:

  • 执行新工作流时:
    set_prompt + clean_unused 会直接把“不在新 prompt 键集合里”的缓存项(包括对象子缓存)全部清掉;
  • 模型 / LoRA 变化:
    即便节点 ID 不变,输入签名和 is_changed 指纹不同,也会生成新键;旧条目先不命中,随后在下一次 prompt 绑定时被清空;
  • 回切旧 workflow:
    因为在上一次切换时已经把旧 workflow 相关缓存清干净了,所以基本等于重新计算。

适用场景

  • workflow 比较固定;
  • 主要想在“一次执行当中”复用中间结果,不在意跨 prompt 的持久化。

6. LRU:代际 LRU 控制 outputs 尺寸

第二种策略是 LRU,主要解决的问题是:在允许跨 prompt 复用输出的前提下,限制缓存总量,避免无限膨胀

6.1 初始化:只作用于 outputs

CacheSet.init_lru_cacheexecution.py:127-135):

def init_lru_cache(self, cache_size):
    self.outputs = LRUCache(CacheKeySetInputSignature, max_size=cache_size)
    self.objects = HierarchicalCache(CacheKeySetID)

注意几点:

  • LRU 只作用于 outputs
  • objects 仍然用 HierarchicalCache,不受 LRU 驱逐;
  • 启动方式:--cache-lru N,且 N > 0

6.2 LRU 的代际设计

LRUCache 的骨架逻辑在 comfy_execution/caching.py:299-337

class LRUCache(BasicCache):
    def __init__(self, key_class, max_size=100):
        self.max_size = max_size
        self.min_generation = 0
        self.generation = 0
        self.used_generation = {}
        self.children = {}

    async def set_prompt(...):
        self.generation += 1
        for node_id in node_ids:
            self._mark_used(node_id)

    def get(self, node_id):
        self._mark_used(node_id)
        return self._get_immediate(node_id)

    def _mark_used(self, node_id):
        cache_key = self.cache_key_set.get_data_key(node_id)
        if cache_key is not None:
            self.used_generation[cache_key] = self.generation

含义:

  • 有一个全局代数 generation,每次绑定新 prompt generation += 1
  • 每次:

    • 绑定 prompt 时,会把该 prompt 内所有节点标记为“在当前代被使用”;
    • get / set 时更新条目的 used_generation[key] 为当前代。

6.3 容量驱逐:按「最老代」逐步清理

clean_unused() 的一部分逻辑在 comfy_execution/caching.py:314-323

def clean_unused(self):
    while len(self.cache) > self.max_size and self.min_generation < self.
    generation:
        self.min_generation += 1
        to_remove = [key for key in self.cache if self.used_generation[key] < self.
        min_generation]
        for key in to_remove:
            del self.cache[key]
            del self.used_generation[key]
            if key in self.children:
                del self.children[key]
    self._clean_subcaches()

简单归纳一下:

  • 只要 len(cache) > max_size,就逐步提升 min_generation
  • 每提升一代,就删除“最近使用代 < min_generation”的条目;
  • 同步清除掉和这些 key 绑定的子缓存引用;
  • 清理完再执行 _clean_subcaches() 做层级清扫。

6.4 子图分区与代际配合

子缓存创建时会显式标记父节点和子节点“被使用”,避免刚刚生成的子图被误删。代码在 comfy_execution/caching.py:338-349

async def ensure_subcache_for(self, node_id, children_ids):
    await super()._ensure_subcache(node_id, children_ids)
    await self.cache_key_set.add_keys(children_ids)
    self._mark_used(node_id)
    cache_key = self.cache_key_set.get_data_key(node_id)
    self.children[cache_key] = []
    for child_id in children_ids:
        self._mark_used(child_id)
        self.children[cache_key].append(self.cache_key_set.get_data_key(child_id))
    return self

配合前面提到的层级结构,就形成了“按 workflow 子图分区 + LRU 按代际清理”的整体行为。

6.5 触发时机和行为总结

在 LRU 模式下:

  • 每次绑定 prompt:
    generation += 1,标记当前 prompt 的节点使用代为当前代;
  • 每次 get/set
    更新条目的 used_generation 为当前代;
  • 每次 clean_unused()

    • len(cache) > max_size 时,通过提升 min_generation 清除旧代条目;
    • 额外清理无用子缓存。

特点

  • 可以跨 prompt 保留一部分中间结果;
  • max_size 控制缓存上限;
  • 没有 RAM 压力感知:poll() 依然不做事。

适用场景

  • 希望在多 workflow 之间部分复用缓存;
  • 但机器内存有限,需要给 outputs 一个明确的容量上限;
  • 对 RAM 细粒度控制没有强需求,或使用的是物理机 / 内存足够的环境。

7. RAM_PRESSURE:按可用内存压力驱逐

第三种策略是 RAM_PRESSURE,对应类是 RAMPressureCache。它继承自 LRUCache,但不按 max_size 做驱逐,而是:

  • 通过 poll(ram_headroom),在可用内存不足时按“OOM 评分”驱逐条目。

7.1 初始化:objects 仍然是层级缓存

CacheSet.init_ram_cacheexecution.py:131-133):

def init_ram_cache(self, min_headroom):
    self.outputs = RAMPressureCache(CacheKeySetInputSignature)
    self.objects = HierarchicalCache(CacheKeySetID)

注意两个点:

  • RAM 模式下,只有 outputs 会按 RAM 压力驱逐
  • objects 不参与 RAM 驱逐,逻辑完全和 CLASSIC/LRU 下相同。

7.2 poll:可用 RAM 检测 + OOM 评分驱逐

poll 的主逻辑在 comfy_execution/caching.py:384-454

def poll(self, ram_headroom):
    def _ram_gb():
        # 优先 cgroup v2/v1,失败回退 psutil
        ...
    if _ram_gb() > ram_headroom: return
    gc.collect()
    if _ram_gb() > ram_headroom: return
    clean_list = []
    for key, (outputs, _), in self.cache.items():
        oom_score = RAM_CACHE_OLD_WORKFLOW_OOM_MULTIPLIER ** (self.generation -
        self.used_generation[key])
        ram_usage = RAM_CACHE_DEFAULT_RAM_USAGE
        def scan_list_for_ram_usage(outputs):
            nonlocal ram_usage
            ...
        scan_list_for_ram_usage(outputs)
        oom_score *= ram_usage
        bisect.insort(clean_list, (oom_score, self.timestamps[key], key))
    while _ram_gb() < ram_headroom * RAM_CACHE_HYSTERESIS and clean_list:
        _, _, key = clean_list.pop()
        del self.cache[key]
        gc.collect()

流程拆一下:

  1. 获取可用 RAM

    _ram_gb() 的实现优先读取 cgroup 的限制:

    • cgroup v2:memory.max / memory.current
    • cgroup v1:memory.limit_in_bytes / memory.usage_in_bytes
    • 都失败才回退 psutil.virtual_memory().available

    这解决了容器环境下“宿主机内存大,容器实际被限制”的常见问题。

  2. 阈值和 GC

    • 如果可用 RAM > ram_headroom,直接返回;
    • 否则先跑一次 gc.collect()
    • 再测一次 RAM,如果还是不足,进入驱逐流程。
  3. 为每个条目计算 OOM 评分

    • 初始 oom_score = RAM_CACHE_OLD_WORKFLOW_OOM_MULTIPLIER ** (generation - used_generation[key])

      • 大概意思是:越久没被用,分数指数级放大(默认倍数为 1.3,见 comfy_execution/caching.py:365);
    • 初始 ram_usage = RAM_CACHE_DEFAULT_RAM_USAGE(0.1,见 360);
    • 递归遍历 outputs 列表:

      • CPU tensor:numel * element_size * 0.5(认为 CPU 上的 tensor 价值更高,折半);
      • 自定义对象:如果实现了 get_ram_usage() 就加上它;
    • 最后 oom_score *= ram_usage,得到综合评分。

    所有条目按 (oom_score, timestamp, key) 排序,放入 clean_list

  4. 按迟滞阈值逐个删除

    • 只要 _ram_gb() < ram_headroom * RAM_CACHE_HYSTERESIS,就从 clean_list 末尾 pop 一个 key 并删除;
    • 每删一个都跑一次 gc.collect()
    • 迟滞倍数 RAM_CACHE_HYSTERESIS 默认 1.1,避免“刚删完又马上触发清理”的抖动。
  5. 访问时间戳

    访问时会更新 timestamps(comfy_execution/caching.py:376-382):

   def set(self, node_id, value):
       self.timestamps[self.cache_key_set.get_data_key(node_id)] = time.time()
       super().set(node_id, value)

   def get(self, node_id):
       self.timestamps[self.cache_key_set.get_data_key(node_id)] = time.time()
       return super().get(node_id)

在 oom_score 一样时,timestamp 起到“最近访问优先保留”的作用。

7.3 提示绑定下的行为:不清理 outputs

RAM 模式下,clean_unused() 的行为与 CLASSIC 不同(见参考说明):

  • RAM 模式:
    clean_unused() 只做子缓存分区清理,不会删掉“当前 prompt 未使用”的 outputs 条目;
  • CLASSIC 模式:
    clean_unused() 会同时删掉当前 prompt 未用到的 outputs 条目。

结果是:

  • RAM 模式可以跨多个 workflow 长时间保留中间结果;
  • 只有在 RAM 不够时才做清退。

7.4 容器环境下需要注意的点

的 AutoDL 中,用 psutil.virtual_memory().available,在容器里看到的是宿主机内存,而不是容器的限额,导致永远“不触发回收”,最后 OOM。

适用场景

  • 多 workflow / 多模型 / 多 LoRA 同时存在,且希望尽可能长时间复用输出结果;
  • 机器内存有限,但更关心“不 OOM”,而不是一个固定的 max_size
  • 特别适合容器环境(K8s / AutoDL 等),配合 --cache-ram <GB>

8. 一些落地建议

最后,用一段比较直接的建议收尾。

8.1 单模型 / workflow 稳定:用 CLASSIC 即可

特点:

  • 工作流基本不换;
  • 主要希望避免一次执行中的重复计算(比如多次 preview)。

用默认 CLASSIC:

  • 结构简单;
  • 不参与复杂的跨 prompt 保留;
  • 不需要担心 LRU 尺寸和 RAM 阈值调参。

8.2 多 workflow + 控制缓存尺寸:用 LRU

场景:

  • 有多套 workflow,在它们之间来回切;
  • 机器内存不是特别大,希望 outputs 不要无限膨胀;
  • 又希望某些常用子图能被复用。

做法:

  • 启动时加 --cache-lru N(N 先给一个相对保守的值,比如几百到几千条,看内存曲线再调);
  • LRUCache 用代际 + max_size 帮你自动做“近期常用保留、早期冷门清理”。

8.3 多模型 / 多 LoRA / 容器环境:优先 RAM_PRESSURE

场景:

  • 容器 / 云平台(K8s、AutoDL 等);
  • 有较多模型、LoRA 和 workflow 混用;
  • 内存被容器限制,容易因爆 RAM 掉进 OOM。

做法:

  • 启动时用 --cache-ram <GB> 配一个“希望保留的 RAM 余量”;

    • 比如容器给了 32GB,就设在 16–24GB 看情况;
  • RAMPressureCache

    • 最大程度保留各个 workflow 的中间结果(包括应用 LoRA 后的模型对象等);
    • 在内存不足时,根据 OOM 评分优先清旧代、大内存条目。

注意一点:即使有容器优化,如果平台本身的 cgroup 挂得不标准,_ram_gb() 的结果还是有可能偏离实际,这一点要结合平台文档确认。

8.4 完全不想折腾:直接关缓存

如果你:

  • 环境受限;
  • 或者调试阶段不想被缓存影响行为理解;

可以直接用:

  • --cache-none
    对应 CacheType.NONECacheSet.init_null_cache()NullCache,所有 get/set 都是 no-op。

代价就是:每次执行都完全重新算一遍。

9. 小结一下

把上面的内容压缩成几句话:

  • ComfyUI 有两路缓存:

    • outputs 存中间结果,真正用来省算力;
    • objects 存节点实例,只减少 Python 对象构造开销。
  • 键体系:

    • 输出:输入签名 + is_changed 指纹(+ 条件下的 node_id)
    • 对象:**(node_id, class_type)**。
  • 四种策略:

    • CLASSIC:默认层级缓存,按当前 prompt 键集合清理,不做 LRU / RAM 驱逐;
    • LRU:只对 outputs 做代际 LRU,配合 max_size 控制容量;
    • RAM_PRESSURE:在 LRU 基础上加 RAM 压力驱逐,在内存不足时按 OOM 评分清理;
    • NONE:彻底关掉缓存。
  • 在多模型 / 多 workflow / 多 LoRA 场景下:

    • 对象缓存 objects 始终是层级缓存,不参与 LRU / RAM 驱逐,在每次绑定 prompt 时按键集合清理;
    • 模型权重 / LoRA 的真实驻留由模型管理层控制;
    • 真正要关心的是:如何选择 outputs 的策略,让中间结果既能有效复用,又不会把内存打爆。

如果我们在一个复杂的工作流环境里跑 ComfyUI,建议先搞清楚自己处于哪种场景,再结合上面的策略选项,把缓存调成我们能控制的状态,而不是让它在后台「自动长草」。

以上。

关于行业 Agent 的思考:「行业 Workflow + Agent」的混合模式

过去一年,AI Agent 从狂热逐渐回归理性。在企业级应用和垂直行业落地中,我们看到了一个趋势:在行业中,纯粹依靠 Agent 自主决策的构想,正在被「Workflow + Agent」的混合模式所取代。

对于我们一线的同学来说,最重要的是要去解决实际问题。

当前我们能看到的行业 Agent 大多数实际落地的逻辑是:行业 Agent 的壁垒在于行业 Know-how,而落地的最佳路径是利用 Agent 做交互与分发,利用 Workflow 做执行与兜底。

1. 行业 Agent 是什么

很多人把 Agent 想象成一个全能的「超级员工」,指望给它一个模糊的目标(比如“帮我提升下季度销售额”),它就能自动拆解任务、调用工具、完成工作。在通用领域或简单场景下(如订机票、写周报),这或许可行。但在垂直行业(金融、制造、医疗、物流等),这种纯 Agent 模式目前是行不通的。

1.1 Agent 是交互方式,不是业务本身

Agent 在行业应用中的本质,是入口交互

它改变了人与系统的互动方式。以前我们需要点击菜单、填写表单、通过 SQL 查询数据库;现在我们可以通过自然语言表达意图。Agent 的核心价值在于它能“听懂”用户的意图,并将其转化为系统能理解的指令。

1.2. 真正的壁垒是行业 Know-how

大模型本身是通用的。GPT-5 或者是 Claude 4.5,它们具备的是通用的逻辑推理能力和语言能力,但它们不懂你们公司的复杂的审批流程,不懂某个特定设备的维修手册,也不懂行业内潜规则式的业务逻辑。

行业 Agent 的「行业」二字,才是重点。

  • 什么是 Know-how? 是我们沉淀了十年的 SOP,是数据库里积累的边缘案例,是针对特定业务场景的异常处理机制。
  • Agent 的角色: 它是这些 Know-how 的「调度员」,而不是「创造者」。

如果脱离了行业 Know-how,Agent 就是一个会说话但办不成事的空壳。

2. 为什么「纯 Agent」模式在企业端走不通?

在 Demo 阶段,我们经常看到这样的演示:用户说一句话,Agent 自动规划了五个步骤,调用了三个 API,完美解决了问题。

但在生产环境中,这种全自动的「纯 Agent」模式面临三个无法回避的死结:

2.1 幻觉与确定性的冲突

企业级应用,尤其是涉及到资金、生产安全、合规的场景,稳定压倒一切。 大模型的本质是概率预测,这意味着它永远存在「幻觉」的可能性。哪怕准确率做到 99%,那剩下的 1% 的不可控对于企业核心流程来说也是灾难。

你无法向审计部门解释,为什么系统批准了一笔违规报销,仅仅因为 Agent 觉得「这看起来没问题」。

2.2 流程的黑盒化

纯 Agent 模式下,决策过程往往隐藏在模型的推理链中。当出现问题时,很难复盘和追责。企业需要的是可审计、可监控、可干预的流程。

2.3 成本与延迟

让大模型去规划每一个微小的步骤(比如“点击确认按钮”、“校验手机号格式”),是对算力的巨大浪费。这些确定性的逻辑,用传统的代码实现既快又准,用 LLM 去推理则是大炮打蚊子,且增加了响应延迟。

3. Workflow + Agent 的混合模式

既然大模型的幻觉无法根除,而传统软件的确定性又是刚需,最务实的方案就是将两者结合:Workflow + Agent

这是一个“动静结合”的架构。

  • Workflow(工作流/RPA): 负责“静”。它是骨架,是肌肉。它包含固定的业务逻辑、SOP、API 调用序列。它保证了核心流程的确定性可靠性
  • Agent(大模型): 负责“动”。它是大脑,是神经。它负责理解非结构化的输入(自然语言),进行意图识别,然后决策应该触发哪一条 Workflow。

3.1 核心逻辑

Agent 不直接去操作底层数据库或核心系统,Agent 的输出对象是 Workflow。

  • 用户 -> 对话 -> Agent (理解意图/参数提取) -> 触发 -> Workflow (执行/校验) -> 返回结果 -> Agent (格式化输出) -> 用户

3.2 这种模式解决了什么问题?

  1. 复用历史沉淀: 企业过去十年建设的 ERP、CRM、以及各种自动化脚本(RPA),不需要推倒重来。它们被封装成一个个 Workflow,成为 Agent 的「工具箱」。
  2. 控制风险: 所有的执行动作(写库、转账、发货)都由 Workflow 控制,Workflow 内部包含严格的校验逻辑(If-Else),这是大模型无法绕过的硬规则。
  3. 降低成本: 只有在需要理解和决策的环节才消耗 Token,大量的执行环节由低成本的代码完成。

4. 如何设计混合模式

在具体落地时,我们需要构建一个分层的架构体系。

4.1 意图理解与分发

这是系统的入口。用户输入的往往是模糊的、非结构化的自然语言。 这一层的核心任务不是「解决问题」,而是「定义问题」。

  • 意图识别: 判断用户是想「查询库存」、「发起退款」还是「投诉建议」。
  • 参数提取: 从对话中提取执行 Workflow 所需的关键参数(如订单号、日期、金额)。如果参数缺失,Agent 需要反问用户进行补全。
  • 路由分发: 基于意图,将任务指派给具体的 Workflow 或下一级更专业的 Agent。

关键点: 这一层需要极强的语义理解能力,通常需要配合 RAG 来理解特定领域的术语。

4.2 动态决策与 RAG

在某些复杂场景下,直接映射到 Workflow 是不够的。 比如用户问:“我的设备报警代码是 E03,我该怎么办?”

这里不能直接触发一个“维修流程”,因为 Agent 首先需要知道 E03 代表什么。

  • RAG 的介入: Agent 调用知识库,检索 E03 对应的故障原因和处理手册。
  • 初步决策: 基于检索到的 Know-how,Agent 判断是建议用户重启(触发“重启指引 Workflow”),还是必须派人维修(触发“工单提交 Workflow”)。

关键点: RAG 在这里不仅仅是用来回答问题的,更是用来辅助 Agent 做路由决策的。

4.3 确定性执行(Workflow / RPA)

这是系统的执行层,也是“行业 Know-how”固化最深的地方。 这一层严禁幻觉

  • 形式: 它可以是一个 API 接口,一个 Python 脚本,或者是一个复杂的 BPM(业务流程管理)实例,甚至是一个 RPA 机器人。
  • 逻辑: 这里面充满了 If-ElseTry-Catch 和数据库事务。
  • 反馈: Workflow 执行完毕后,必须返回明确的状态码和结果数据(JSON 格式),而不是一段模糊的文本。

4.4 结果综合与反馈

Workflow 返回的是结构化数据(例如:{"status": "success", "order_id": "12345", "delivery_date": "2023-12-01"})。 Agent 的最后一步工作,是将这些冷冰冰的数据,转化为符合人类阅读习惯的自然语言,反馈给用户。

5. 多级 Agent 与 RAG 的协同

在简单的场景下,一个 Agent 配合几个 Workflow 就够了。但在复杂的行业场景(如供应链管理、大型设备运维)中,我们需要更复杂的拓扑结构。

5.1 多级 Agent 架构

不要试图训练一个全知全能的上帝 Agent。应该采用“主帅-将军-士兵”的层级结构。

  • L1 调度 Agent(主帅): 只负责宏观分类。例如,判断是“售前咨询”还是“售后维修”。
  • L2 领域 Agent(将军): 专注于特定领域。例如,“售后 Agent” 拥有查询保修、解读故障码、预约工程师的能力。
  • L3 执行单元(士兵): 具体的 Workflow 或特定的单一功能 Agent。

这种结构的好处是解耦。当售后流程发生变化时,只需要调整 L2 Agent 和对应的 Workflow,不会影响到售前部分。

5.2 RAG 的逻辑化应用

传统的 RAG 主要是为了解决“回答知识性问题”。在混合模式中,RAG 的作用被放大了。

  • 动态 Prompt 注入: 在执行 Workflow 之前,系统可以根据当前的上下文,利用 RAG 从知识库中检索出相关的规则或注意事项,动态注入到 Agent 的 Prompt 中。

    • 例子: 在处理一笔“退款”请求时,RAG 检索到“该用户是 VIP 且信用极好”,将此信息注入 Prompt,Agent 可能会选择触发“极速退款 Workflow”而不是“常规审核 Workflow”。

6. 落地实战中的思考

在实施“行业 Workflow + Agent”模式时,有几个非技术性的坑需要注意。

6.1 人机协同

在很长一段时间内,Agent 不会完全取代人,而是成为人的 Copilot 在设计 Workflow 时,必须预留人工介入的节点。 当 Agent 的置信度低于某个阈值,或者 Workflow 执行遇到异常时,系统应自动升级为人工服务,并将之前的上下文完整传递给人工客服。

6.2 存量资产的价值

很多技术团队在做 AI 转型时,倾向于重构一切。这是错误的。 你们公司遗留的那些看起来陈旧的 API、跑了五年的定时脚本、甚至 Excel 里的宏,都是宝贵的资产。 Agent 的落地应当是「局部改造」而非「推倒重来」。 我们要做的,是给这些老旧的系统加上一个 AI 适配层,让 Agent 能够调用它们,而不是替换它们。

6.3 结构化数据的回流

Agent 与用户的交互过程,产生了大量高质量的数据。 不要让这些数据只停留在对话日志里。需要设计机制,将 Agent 收集到的信息(如用户的新需求、报错的高频词、Workflow 的执行结果)结构化地回流到业务系统中,用于优化 SOP 和微调模型。

7. 小结

行业 Agent 的未来,不是科幻电影里的全自动机器人,而是严谨的工程化实践

我们不需要一个会写诗的 AI,我们需要的是一个能准确理解工单意图,并由后台的 Workflow 准确执行的系统。

  • Agent 是面子:提供极简的交互,理解复杂的意图。
  • Workflow 是里子:承载行业壁垒,保证执行的绝对可靠。
  • RAG 是底子:提供动态的上下文和知识支撑。

降本增效不是靠引入一个昂贵的大模型来实现的,而是靠大模型把过去那些难以被自动化的“非结构化需求”,转化为了可以被低成本代码执行的“结构化指令”。

这才是行业 Agent 的落地。

聊下 AI Agent 的上下文记忆和遗忘

最近 DeepSeek 的 OCR 论文里有个有趣的想法:用光学压缩来模拟人类的记忆遗忘机制。

这个思路很巧妙。他们注意到人类记忆和视觉感知都有个共同特点:距离越远,信息越模糊。一小时前的事记得清楚,一年前的事几乎忘光;10 厘米的东西看得清楚,20 米外的东西就模糊了。

DeepSeek 把这个生物学现象转化成了工程实现:近期对话用高分辨率保存,一周前的对话降到中等分辨率,久远的记忆压缩到最小。信息随时间自然衰减,就像人类遗忘一样。

这让我想到 Agent 记忆系统设计的本质问题。

1. 为什么 Agent 需要记忆

上下文窗口和记忆是两码事。

上下文窗口只是让模型一次看到更多对话,像是扩大了工作台。但记忆不同,它让 Agent 能保存、更新和选择性回忆信息。没有记忆,Agent 就像得了失忆症,每次对话都从零开始。

现在大家都在追求更大的上下文窗口,从 8K 到 32K,再到 128K、1M。但这种暴力扩张有个问题:计算成本呈二次方增长。处理 100K tokens 的成本是 10K tokens 的百倍。而且,把所有信息一股脑塞给模型,反而可能让它迷失在细节中。

记忆系统的价值在于选择性保留。

不是所有信息都值得记住,也不是所有信息都需要同样的精度。

2. Agent 记忆的层次结构

AI Agent 的记忆系统可以分成几个层次:

短期记忆(工作记忆)
这是 Agent 的记事本,保存当前对话和正在处理的任务。典型容量在几千到几万 tokens。没有它,Agent 会在对话中途失去思路,问你刚才说了什么。

长期记忆
这是跨会话的持久化存储。用户下次回来,Agent 还记得之前的交互。长期记忆又可以细分:

  • 事实记忆:保存确定的信息,比如用户姓名、偏好、角色定义。这些信息需要随情况更新,保持”当前真实状况”。
  • 情景记忆:记录具体经历,什么时候发生了什么,结果如何。这给 Agent 一种时间连续感,能回顾过去的决策。
  • 语义记忆:组织概念和关系的知识网络。让 Agent 理解”数据库慢”和”查询延迟高”是相关问题。

3. 遗忘机制是有用的

人类会遗忘,不是大脑容量不够,而是遗忘让我们更高效。

想象一下,如果你记得生活中的每个细节:每顿饭的味道、每次呼吸的感觉、路过的每个行人的脸。这些信息会淹没真正重要的记忆。遗忘是一种过滤机制,帮我们保留有价值的信息。

Agent 也需要这种机制。随着交互增加,历史数据会无限增长。如果不加选择地保存所有内容,会面临几个问题:

  1. 存储成本:每个用户的历史数据都完整保存,存储需求会爆炸式增长。
  2. 检索效率:在海量历史中找到相关信息越来越慢。
  3. 注意力分散:太多无关信息会干扰 Agent 的决策。
  4. 隐私风险:永久保存所有对话增加了数据泄露的风险。

4. 用分辨率模拟时间衰减

DeepSeek 的做法是:把历史对话渲染成图像,然后用不同的分辨率来编码。

近期对话用高分辨率(Gundam 模式,800+ tokens),保留完整细节。

一周前的对话用中等分辨率(Base 模式,256 tokens),保留主要内容。

久远的历史用低分辨率(Tiny 模式,64 tokens),只留个大概印象。

这样做的好处是:

  1. 不需要丢弃历史信息,所有对话都保留着
  2. 但远期信息自然”淡化”,占用的 token 越来越少
  3. 模拟了人类记忆的衰减曲线

具体实现上,他们会:

  1. 把超过一定长度的历史对话渲染成图像
  2. 第一次压缩,让 token 减少 10 倍
  3. 上下文再次超长时,进一步降低分辨率,再压缩 10 倍
  4. 随着时间推移,图像越来越小,内容越来越模糊,模型也就逐渐”读不清”了

这种方式不是简单的截断或删除,而是让信息随时间衰减——就像人类记忆一样。

5. 理论上无限的上下文

如果这个思路走通了,就能实现”理论上无限的 context window”。

这里的无限不是真的无限,而是通过分层压缩,让历史信息的存储成本趋近于常数。

我们不需要保持所有信息的高保真度,只需要让信息按重要性和时效性衰减。

最近的对话,全保留。
一周前的,压缩一次。
一个月前的,再压缩一次。
半年前的,只留个印象。

这样,计算资源始终聚焦在最重要的”近期”信息上,而久远的历史虽然还在,但只占用很少的 token。

从成本角度看,这比无脑扩大上下文窗口要合理得多。

6. 记忆管理的工程实践

在实际的 Agent 系统里,记忆管理通常分几个层次:

会话级记忆——当前对话的上下文,存在内存里,对话结束就清空。这部分可以直接用模型的上下文窗口。

用户级记忆——持久化存储,用向量数据库或 KV 存储。每次对话时,根据当前问题检索相关的历史记忆,注入到 prompt 里。

全局知识库——所有用户共享的知识,比如产品文档、技术规范、FAQ。这部分通常用 RAG(检索增强生成)来处理。

关键是如何在这些层次之间做好信息流转和优先级管理。

比如,用户刚说过的话,优先级最高,直接放在 prompt 前面。一周前的对话,需要检索后才注入。一个月前的,可能只保留摘要。

这种分层策略,和 DeepSeek 的分辨率衰减思路是一致的——让资源消耗和信息重要性成正比。

7. 遗忘不是丢失

这里需要明确的是,遗忘不等于删除。

人类的遗忘,是提取变难了,不是信息消失了。在特定的提示下,很多”遗忘”的记忆还能被唤起。

Agent 的记忆也应该这样设计。

低分辨率的历史对话,在大部分场景下不会被激活,但如果用户明确提到某个时间点的事情,系统可以重新加载那段历史的高分辨率版本。

这需要一个索引机制,能根据时间、主题、关键词快速定位到历史片段。

像 Manus 所使用的文件系统就是这样一种索引机制。

遗忘是常态,召回是特例。这样才能在效率和完整性之间找到平衡。

8. 什么值得记住

并不是所有对话都需要长期保存。

大量的对话是重复的、临时的、没有上下文依赖的。比如简单的问候、重复的确认、无关紧要的闲聊。

这些内容可以在会话结束后直接丢弃,不需要进入长期记忆。

真正值得记住的,是那些包含决策、偏好、关键信息的交互。

比如:

  • 用户明确表达的需求和偏好
  • 重要的决策节点和原因
  • 反复出现的问题和解决方案
  • 用户的角色、职责、技术栈等基础信息

这需要在记忆写入时做判断和过滤。可以用一个小模型或规则引擎,评估每轮对话的重要性,决定是否持久化。 Gemini Code 就是这么干的。

不是所有东西都值得记住,筛选本身就是一种优化。

9. 记忆的更新和冲突

长期记忆不是只写不改的日志,它需要能更新。

用户的偏好会变,角色会变,之前的信息可能过时或错误。

比如用户之前说喜欢中古风的家具,后来又说喜欢北欧风,这个信息需要更新,而不是简单地追加一条新记录。

如果只追加不更新,记忆会越来越冗余,甚至出现矛盾。

一个好的记忆系统,需要能识别冲突,做合并和覆盖。

这可以通过实体识别和关系抽取来实现。把对话内容结构化成三元组(主体-关系-客体),然后在写入时检查是否和已有记忆冲突。

如果冲突,可以根据时间戳、置信度等因素决定是更新还是保留多个版本。

记忆管理的复杂度,不亚于写一个小型数据库。

10. 压缩不只是技术问题

回到 DeepSeek 的光学压缩思路,它的价值不只是技术实现,更重要的是提供了一个思维框架。

我们习惯于把上下文长度当作硬指标——越长越好。但这个论文提醒我们,长度和质量不是一回事。

有时候,适度的遗忘反而能提升系统的整体表现。

有如下的好处:

  1. 减少了无关信息的干扰
  2. 降低了计算成本
  3. 让模型专注于最相关的内容

这和人类的认知机制是一致的。我们不会带着所有历史记忆去做每一个决策,而是根据当前情境激活最相关的那部分。

Agent 应该学会做同样的事。

11. 成本和效果的平衡

从成本角度看,无限扩展上下文是不可持续的。

假设一个 Agent 系统,每天处理 1000 次对话,每次对话平均 10 轮,每轮 500 tokens。

如果所有历史对话都全量保留,一个月下来,单个用户的上下文就会达到 1500k tokens。

如果有 1000 个活跃用户,每次推理都带上完整上下文,token 消耗会是天文数字。

但如果用分层记忆 + 遗忘机制:

  • 最近 3 天的对话,全保留
  • 一周到一个月的,压缩 50%
  • 一个月以上的,压缩 90%

token 消耗可能降低到原来的 10%-20%,而对实际效果的影响很小。

因为大部分场景下,用户关心的就是最近的对话。

12. 记忆应该是系统能力

很多团队把记忆功能当作一个独立的我来开发——加个数据库,存一下历史,检索一下注入。

但这样做的效果往往不好。

因为记忆不是一个功能模块,而是整个 Agent 系统的底层能力。

它需要和 prompt 设计、工具调用、推理链路、错误处理深度集成。

比如:

  • Prompt 需要为记忆预留位置和格式
  • 工具调用的结果需要写回记忆
  • 推理过程中需要动态检索记忆
  • 错误发生时需要回溯历史上下文

记忆管理做得好,Agent 的整体表现会有质的提升。做得不好,就只是一个会话机器人。

13. 一些建议

以下是一些可以尝试落地的建议:

1. 不要把所有历史都塞进 prompt

很多团队的做法是,每次调用都把所有历史对话拼接到 prompt 里。这在对话轮数少的时候没问题,但规模上去后会成为瓶颈。

改成检索式注入——根据当前问题,从历史记忆里检索最相关的几条,而不是全量加载。(这里就会考量检索的能力了)

2. 区分会话记忆和持久记忆

当前对话的上下文,存在内存里就行,不需要持久化。

只有那些需要跨会话保留的信息,才写入数据库。

这样可以大幅减少存储和检索的开销。

3. 给记忆加上过期时间

就像缓存一样,记忆也可以有 TTL(Time To Live)。

一般性的对话,可以设置 7 天或 30 天过期。重要的信息,可以标记为永久保留。

定期清理过期记忆,防止数据无限膨胀。

4. 用好向量数据库

向量检索是目前最适合做语义记忆的技术。

把历史对话做 embedding,存到向量数据库里,每次根据当前问题做相似度检索。

但要注意,向量检索的召回率不是 100%,需要结合关键词索引和时间过滤。

5. 记忆要能解释

用户问”你为什么这么回答”,Agent 应该能说清楚是基于哪段历史记忆做的判断。

这需要在记忆检索时保留溯源信息——这条记忆来自什么时间、什么对话、可信度如何。

可解释性不仅是用户体验问题,也是调试和优化的基础。

14. 最后

上下文记忆和遗忘,本质上是资源分配问题。

我们只有有限的 token 预算,有限的计算资源,有限的响应时间,需要在这些约束下做出最优决策。

无限扩展上下文听起来很美好,但实际上不可行。

真正的解决方案,是学习人类的记忆机制——让重要的信息留下来,让次要的信息自然衰减。

DeepSeek 的光学压缩思路提供了一个很好的启发:遗忘不是缺陷,而是一种优化策略。

如果能把这个思路落地到 Agent 系统里,不仅能降低成本,还能提升整体的智能水平。

因为真正的智能,不是记住所有东西,而是知道什么值得记住,什么可以忘记。