肌肉记忆

muscle-mem 是一个用于 AI 代理的行为缓存。

它是一个 Python SDK，能够在代理解决任务的过程中记录其工具调用模式，并在再次遇到相同任务时确定性地重放这些已学习的轨迹；如果检测到边缘情况，则会回退到代理模式。

muscle-mem 的目标是让大型语言模型从重复性任务的热路径中脱离出来，从而提高速度、减少波动性，并为那些“本来只需一段脚本就能完成”的场景节省 token 成本。

这仍是一个尚未被充分探索的领域，因此我们非常欢迎任何反馈！

• 欲了解更多信息，请阅读 Muscle Mem - 从代理中移除 LLM 调用。
• 如需交流反馈，请加入 Muscle Mem Discord 服务器。

开发日志

• 2025年5月7日 - 首次可用演示
• 2025年5月8日 - 开源发布
• 2025年6月1日 - 参数化与标签功能合并

工作原理

muscle-mem 并非另一个代理框架。

您可以按照自己习惯的方式实现代理，然后将其接入 muscle-mem 的引擎。

当接收到一项任务时，引擎将执行以下步骤：

1. 使用检查机制判断当前环境是否曾出现过（缓存命中），或是否为新环境（缓存未命中）；
2. 执行任务，具体方式如下：
   • 缓存命中时，使用检索到的轨迹；
   • 缓存未命中时，将任务交由您的代理处理。
3. 收集工具调用事件，作为新的轨迹添加到缓存中。

关键在于缓存验证

要为您的代理引入安全的工具复用功能，最关键的问题就是缓存验证。您需要自问：

对于我们赋予代理的每一种工具，在当前环境中有哪些特征可以用来判断执行该操作是否安全？

如果您能够回答这个问题，那么您的代理就可以具备肌肉记忆功能。

API 说明

安装

pip install muscle-mem

Muscle Mem 的 API 当前处于 v0 版本，且会在小版本更新时发生破坏性变更，因此我们强烈建议在生产环境中固定使用某一特定版本。

引擎

引擎包装您的代理，充当任务的主要执行者。

它管理着自己的历史轨迹缓存，并决定何时调用您的代理。

from muscle_mem import Engine

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

# 您的代理可以直接调用
your_agent("do some task")

# 引擎提供相同的接口，但具备肌肉记忆功能
engine("do some task")
engine("do some task") # 缓存命中

假设您的代理可以通过函数形式或 __call__ 方法进行调用。传递给 engine() 的所有位置参数和关键字参数都会直接传递给您的代理。我们的示例中使用了字符串，但也可以是 ChatCompletion 消息列表，或者您的代理可调用方法所期望的任何内容。

默认情况下，轨迹会被存储在一个统一的缓存中。您可以通过为轨迹添加描述性的标签来创建不同的存储桶。

engine("do some task", tags=["some task"]) # 缓存未命中
engine("do some task", tags=["some task"]) # 缓存命中

engine("do some task", tags=["different task"]) # 缓存未命中
engine("do some task", tags=["different task"]) # 缓存命中

工具插桩

通过装饰器对执行动作的工具进行插桩，以便引擎能够在代理执行操作时记录这些动作。

函数

使用 @engine.function 装饰器为简单的函数工具进行插桩：

from muscle_mem import Engine

engine = Engine()

@engine.function()
def hello(name: str):
    print(f"hello {name}!")
    
hello("world") # 调用 hello 被记录下来，参数 name="world"

方法

使用 @engine.method 装饰器为对象上的方法进行插桩。

这允许通过 self 参数注入有状态的 API 客户端，例如 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") # 调用 SomeClient.hello 被记录下来，参数 name="world"

需要注意的是，由于运行时对象（self）无法序列化，因此 self 不会包含在轨迹中。

为了使引擎能够重放基于方法的工具调用，必须显式地向引擎传递一个实例对象，以重新注入 self。

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

engine = (
    engine
    .set_agent(your_agent)
    .set_context(client) # 您的 client 对象会被注入为 self，用于调用 SomeClient.hello
    .finalize()
)
engine("say hello world!")

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

检查机制

检查机制是缓存验证的基本构建模块。它们决定了执行某项操作是否安全。

每个检查机制封装了：

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

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

您可以将检查机制附加到每个工具上（使用 @engine.function 或 @engine.method），以强制执行缓存验证。这些检查可以在工具调用之前作为预检（也可用于查询时验证），或在工具调用之后作为后检。

下面是一个示例，它捕捉了 hello 工具的使用情况，并利用时间戳和一秒钟的过期机制作为缓存验证的检查逻辑。

# 我们的捕获实现，接收参数并返回 T
def capture(name: str) -> T:
    now = time.time()
    return T(name=name, time=now)

# 我们的比较实现，接收当前和候选的 T
def compare(current: T, candidate: T) -> bool:
    # 如果发生在最近一秒钟内，则缓存有效
    diff = current.time - candidate.time
    passed = diff <= 1
    return passed

# 为我们的工具添加预检
@engine.function(pre_check=Check(capture, compare))
def hello(name: str):
    time.sleep(0.1)
    print(f"hello {name}")

顶层参数

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

但这并不总是理想的，因为某些参数可能需要根据每次运行动态变化。例如，一个填写表单的机器人可能会调用 type("John") 工具来填充“名字”字段，但在后续运行中，我们希望使用不同的名字。

为此，Muscle Mem 提供了一个顶层 params 系统，允许您指定那些需要根据每次运行动态变化的参数。

如果某个值在调用代理之前就已经在运行时确定（例如您已经将其模板化到提示词中），则可以通过 engine() 的可选 params 参数将其标记为顶层参数。

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

# 第一次运行会存储对 type("John") 的调用，其中动态参数 `text` 被映射到顶级参数 `name`
engine("用名字：John 填写表单", params={"name": "John"})

# 第二次运行命中缓存，使用 Jane 代替 John
engine("用名字：Jane 填写表单", params={"name": "Jane"})

在记录轨迹时，如果任何底层工具调用中的参数与顶级参数直接匹配，引擎会将其标记为动态，并按键名将其映射到顶级参数。

当前的实现通过工具参数的精确字符串匹配来判断其是否源自顶级参数。假设所有工具参数都来自 LLM，因此都可以被序列化为字符串。

整体整合

以下是上述所有代码片段的合并脚本。

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

engine = Engine()

# 我们的“环境”特征，存储在数据库中
@dataclass
class T:
    name: str
    time: float

# 我们的捕获实现，接收参数并返回 T
def capture(name: str) -> T:
    now = time.time()
    return T(name=name, time=now)

# 我们的比较实现，接收当前和候选的 T
def compare(current: T, candidate: T) -> bool:
    # 如果发生时间在最近1秒内，则缓存有效
    diff = current.time - candidate.time
    passed = diff <= 1
    return passed

# 使用预检查装饰我们的工具
@engine.function(pre_check=Check(capture, compare))
def hello(name: str):
    time.sleep(0.1)
    print(f"hello {name}")
    
# 假设这是你的智能体
def agent(name: str):
   for i in range(9):
        hello(name + " + " + str(i))

engine.set_agent(agent).finalize()

# 第一次运行
cache_hit = engine("erik")
assert not cache_hit

# 再次运行
cache_hit = engine("erik")
assert cache_hit

# 通过睡眠打破缓存，然后再次运行
time.sleep(3)
cache_hit = engine("erik")
assert not cache_hit

如需更真实的示例，请参阅计算机使用代理的实现：

https://github.com/pig-dot-dev/muscle-mem/blob/main/examples/cua.py

---

行动号召

随着该系统的不断发展，诚邀大家提出反馈！

请考虑：

1. 加入 Muscle Mem Discord 社区
2. 测试 muscle-mem 仓库，并给它点个赞

Muscle Memory 快速上手指南

muscle-mem 是一个专为 AI Agent 设计的“行为缓存”Python SDK。它能记录 Agent 解决任务时的工具调用模式，并在遇到相同任务时确定性重放这些轨迹，从而将 LLM 从重复性工作中解放出来，显著提升速度、降低波动并节省 Token 成本。

环境准备

• 系统要求：支持 Python 3.8+ 的操作系统（Linux, macOS, Windows）
• 前置依赖：
  • Python 3.8 或更高版本
  • pip 包管理工具
  • 一个已实现的 AI Agent（需为可调用对象，即函数或实现了 __call__ 方法的类）

注意：Muscle Mem 目前处于 v0 阶段，API 可能在次版本更新中发生变化。生产环境使用时，强烈建议锁定具体版本号。

安装步骤

使用 pip 直接安装：

pip install muscle-mem

注：若国内下载速度较慢，可尝试配置清华或阿里镜像源：

pip install muscle-mem -i https://pypi.tuna.tsinghua.edu.cn/simple

基本使用

1. 初始化引擎并接入 Agent

muscle-mem 不是一个新的 Agent 框架，而是包裹在你现有 Agent 外层的执行引擎。

from muscle_mem import Engine

# 假设这是你现有的 Agent
def your_agent(task: str):
    print(f"Agent is working on: {task}")
    # ... 你的 Agent 逻辑 ...

# 初始化引擎
engine = Engine()

# 将你的 Agent 注入引擎
engine.set_agent(your_agent).finalize()

# 现在可以通过 engine 调用，首次运行会记录轨迹
engine("do some task") 

# 再次运行相同任务，将直接命中缓存并重放，不再调用 LLM
engine("do some task") # cache hit

2. 工具插桩 (Tool Instrumentation)

为了让引擎记录具体的工具调用行为，需要使用装饰器对工具函数进行标记。

标记普通函数：

@engine.function()
def hello(name: str):
    print(f"hello {name}!")

# 调用会被自动记录
hello("world") 

标记类方法（支持依赖注入）：

如果工具是类的方法（例如需要访问 self.db），引擎会在重放时重新注入实例对象。

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

client = SomeClient()

# 设置上下文，以便重放时注入 self
engine.set_context(client).finalize()

client.hello("world") # 调用被记录

3. 缓存验证 (Cache Validation)

为了安全地重用工具，必须定义何时缓存失效。通过 Check 机制，你可以指定如何捕获环境特征以及如何比较这些特征。

以下示例展示了一个基于时间的预检查（Pre-check）：如果距离上次执行超过 1 秒，则视为缓存未命中，重新执行 Agent。

from dataclasses import dataclass
from muscle_mem import Check
import time

@dataclass
class EnvState:
    name: str
    timestamp: float

# 捕获当前环境状态
def capture(name: str) -> EnvState:
    return EnvState(name=name, timestamp=time.time())

# 比较当前状态与缓存状态
def compare(current: EnvState, cached: EnvState) -> bool:
    # 如果在 1 秒内，认为缓存有效
    return (current.timestamp - cached.timestamp) <= 1

# 将 Check 绑定到工具上
@engine.function(pre_check=Check(capture, compare))
def hello(name: str):
    time.sleep(0.1)
    print(f"hello {name}")

# 模拟 Agent 逻辑
def agent(name: str):
    hello(name)

engine.set_agent(agent).finalize()

# 第一次运行：记录轨迹
engine("erik") # Cache Miss

# 第二次运行：立即重放
engine("erik") # Cache Hit

# 等待超过 1 秒后运行：缓存失效，重新执行
time.sleep(2)
engine("erik") # Cache Miss

4. 动态参数支持

如果某些参数在每次运行时都需要变化（例如表单填写中的不同姓名），可以使用 params 参数将其标记为动态变量。

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

# 第一次运行：记录轨迹，将 "John" 映射为动态参数 `name`
engine("fill form with John", params={"name": "John"})

# 第二次运行：命中缓存，但使用新的参数值 "Jane" 重放
engine("fill form with Jane", params={"name": "Jane"})