Muscle-Mem: Agent行为记忆和缓存系统

muscle-mem 是一个面向 AI Agent 的行为缓存。

它是一个 Python SDK：在 Agent 解决任务时记录其工具调用模式；当再次遇到相同任务时，确定性地重放已经学习到的轨迹；如果检测到边缘情况，则回退到 Agent 模式。

muscle-mem 的目标，是让 LLM 退出重复性任务的关键执行路径，从而提升速度、减少结果波动，并为大量原本只需一段脚本即可完成的任务消除 token 成本。

这是一个尚未被充分探索的领域，欢迎提供任何反馈！

阅读《Muscle Mem：移除 Agent 中的 LLM 调用》，了解更多背景。

加入 Muscle Mem Discord 反馈意见。

• 2025 年 5 月 7 日——首个可运行的演示
• 2025 年 5 月 8 日——正式开源
• 2025 年 6 月 1 日——Parameterization 和 Tags 功能合并

muscle-mem 不是另一个 Agent 框架

你可以用任何方式实现自己的 Agent，然后将它接入 muscle-mem 的引擎。

接收到任务后，引擎会：

• 使用 Checks 判断当前环境以前是否见过（cache-hit），还是一个新环境（cache-miss）。
• 执行任务：
  • cache-hit 时，使用检索到的轨迹；
  • cache-miss 时，将任务交给你的 Agent。
• 收集工具调用事件，作为一条新轨迹加入缓存。

一切都围绕 Cache Validation

要让 Agent 安全地复用工具调用，最关键的问题是 cache validation。请思考：

对于我们提供给 Agent 的每一个工具，环境中的哪些特征可以用来判断执行该操作是否安全？

如果你能回答这个问题，你的 Agent 就能拥有 Muscle Memory。

pip install muscle-mem

Muscle Mem 的 API 目前处于 v0 阶段，次版本发布也可能包含破坏性变更，因此我们强烈建议在生产环境中固定到具体版本。

引擎会包装你的 Agent，并作为任务的主要执行器。

它会自行管理历史轨迹缓存，并决定何时调用你的 Agent。

from muscle_mem import Engine

engine = Engine()
engine.set_agent(your_agent).finalize()

# your agent is independently callable
your_agent("do some task")

# the engine gives you the same interface, but with muscle memory
engine("do some task")
engine("do some task") # cache hit

你的 Agent 应当是可调用对象，可以是函数，也可以实现 __call__ 方法。传递给 engine() 的所有 args 和 kwargs 都会被直接传给你的 Agent。示例中使用的是字符串，但它也可以是 ChatCompletion 消息列表，或者你的 Agent 可调用对象所需的任何其他内容。

默认情况下，所有轨迹都会存储在同一个缓存中。你可以使用任务描述为轨迹添加 tag，从而创建彼此独立的 bucket。

engine("do some task", tags=["some task"]) # cache miss
engine("do some task", tags=["some task"]) # cache hit

engine("do some task", tags=["different task"]) # cache miss
engine("do some task", tags=["different task"]) # cache hit

记录工具调用

通过 decorator 对执行操作的工具进行插桩，让引擎能够在 Agent 调用工具时记录这些操作。

使用 @engine.function decorator 为简单的函数工具插桩：

from muscle_mem import Engine

engine = Engine()

@engine.function()
def hello(name: str):
	print(f"hello {name}!")
	
hello("world") # invocation of hello is stored, with arg name="world"

使用 @engine.method decorator 为对象上的方法插桩。

这样就可以通过 self 参数，对有状态的 API client 进行依赖注入，例如 self.db.get_user(id) 或 self.model.generate(prompt)。

from muscle_mem import Engine

engine = Engine()

class SomeClient:
   @engine.method()
   def hello(self, name: str):
       print(f"hello {name}!")

client = SomeClient()

client.hello("world") # invocation of SomeClient.hello is stored, with arg name="world"

请注意，由于运行时对象 self 无法序列化，因此轨迹中会省略 self。

为了让引擎能够重放基于方法的工具，必须显式传入该对象的实例，以便重新注入为 self。

使用 engine.set_context() 将运行时对象提供给引擎。

engine = (
    engine
    .set_agent(your_agent)
    .set_context(client) # your client object will be injected as self into SomeClient.hello
    .finalize()
)
engine("say hello world!")

engine.finalize() 是一项可选检查，用于确保你在使用引擎之前，已经提供了它需要的所有依赖。

Check

Check 是 cache validation 的基本构件。它们用于判断执行某个操作是否安全。

每个 Check 都封装了：

• 一个 capture callback，用于从当前环境中提取相关特征；
• 一个 compare callback，用于判断当前环境是否与缓存中的环境匹配。

Check(
    capture: Callable[P, T],
    compare: Callable[[T, T], Union[bool, float]],
):

你可以为每个 @engine.function（或 @engine.method）工具附加 Checks，以强制执行 cache validation。

检查既可以在工具调用前作为 precheck 执行——它也用于查询时验证——也可以在工具调用后作为 postcheck 执行。

下面是一个刻意简化的示例。它会捕获 hello 工具的使用情况，并将时间戳和一秒的过期时间作为 Check 的 cache validation 机制。

# our capture implementation, taking params and returning T
def capture(name: str) -> T:
    now = time.time()
    return T(name=name, time=now)

# our compare implementation, taking current and candidate T
def compare(current: T, candidate: T) -> bool:
    # cache is valid if happened within the last 1 second
    diff = current.time - candidate.time
    passed = diff <= 1
    return passed

# decorate our tool with a precheck
@engine.function(pre_check=Check(capture, compare))
def hello(name: str):
    time.sleep(0.1)
    print(f"hello {name}")

动态参数

默认情况下，Muscle Mem 会将工具调用的所有参数存储为静态值。

这并不一定总是符合需求，因为有些参数可能需要在每次运行时动态变化。例如，一个用于填写表单的 bot 在填写 First Name 文本字段时，会调用 type("John") 工具；但在后续运行中，我们需要一种使用其他姓名的方法。

Muscle Mem 通过顶层 params 系统解决这个问题。你可以用它指定哪些参数应当在每次运行时动态变化。

如果某个值在调用 Agent 前的运行时就已经确定——你很可能已经将它插入 prompt 模板——那么可以通过 engine() 可选的 params 参数，将它标记为顶层参数。

@engine.function()
def type(text: str):
    print(text)

# first run stores invocation of type("John"), with dynamic arg `text` mapped to top level param `name`
engine("fill the form with name: John", params={"name": "John"})

# second run is a cache hit, using Jane instead of John
engine("fill the form with name: Jane", params={"name": "Jane"})

记录轨迹时，如果任何底层工具调用发现某个参数与顶层参数直接匹配，引擎就会将它标记为动态参数，并根据 key 将其映射到对应的顶层参数。

当前实现通过对工具参数进行精确字符串匹配，判断它是否源自某个顶层参数。这里假设所有工具参数都来自 LLM，因此都可以序列化为字符串。

整合所有内容

下面是上述所有代码片段组合而成的完整脚本。

from dataclasses import dataclass
from muscle_mem import Check, Engine
import time

engine = Engine()

# our "environment" features, stored in DB
@dataclass
class T:
    name: str
    time: float

# our capture implementation, taking params and returning T
def capture(name: str) -> T:
    now = time.time()
    return T(name=name, time=now)

# our compare implementation, taking current and candidate T
def compare(current: T, candidate: T) -> bool:
    # cache is valid if happened within the last 1 second
    diff = current.time - candidate.time
    passed = diff <= 1
    return passed

# decorate our tool with a precheck
@engine.function(pre_check=Check(capture, compare))
def hello(name: str):
    time.sleep(0.1)
    print(f"hello {name}")
    
# pretend this is your agent
def agent(name: str):
   for i in range(9):
        hello(name + " + " + str(i))

engine.set_agent(agent).finalize()

# Run once
cache_hit = engine("erik")
assert not cache_hit

# Run again 
cache_hit = engine("erik")
assert cache_hit

# Break cache with a sleep, then run again
time.sleep(3)
cache_hit = engine("erik")
assert not cache_hit

如需查看更贴近真实场景的示例，请参阅一个 computer-use Agent 的实现。

随着这个系统不断发展，欢迎提供任何反馈！

你可以通过以下方式参与：

• 加入 Muscle Mem Discord；
• 测试 muscle-mem repo，并为它点亮 star。