给 AI Agent 加记忆模块避免重复踩坑

你的 AI Agent 调用了一个不稳定的 API，得到 429 错误，用同样的参数重试，再次触发限流。十分钟后，在一个新的会话中，它做了完全相同的事。失败的经历什么都没被记住——Agent 无法知道它已经学到了这个教训，因为这段"教训"存在于一个随进程退出就被丢弃的 transcript 中。

本周，一篇关于手动 gatekeeping AI Agent 工具调用的帖子登上了 dev.to 首页，收到了 48 条评论——大多是开发者们在争论你对 Agent 的 tool-calling 循环能有多大信任度。老实说：你能信任它的程度，取决于它记住了多少已经出错过的经历。下面是一个完整的、可运行的 gatekeeper实现，它封装了任意工具调用，在执行前检查 Mem0 中相似的过往失败记录，并在之后将结果写回内存——这样当你的 Agent 第二次要犯同样的错误时，它就知道了。

一个 MemoryGatekeeper 类，位于 Agent 决定调用工具和实际执行之间：

在调用工具前，搜索语义上相似的过往尝试记录。

如果相似的尝试最近失败过，阻止这次调用，并返回记忆中的原因，而不是浪费一次真实的 API 调用。

每次调用后——无论成功还是失败——将结果写回内存。

你会运行同一个脚本两次：第一次运行失败并被记录，第二次运行在浪费请求前就被拦截了。

你需要 Python 3.10+、一个 OpenAI API key（Mem0 的默认提取管道用它把原始文本转化为结构化记忆——这个本地设置不需要单独的 Mem0 账户）和 mem0ai 包。

pip install mem0ai
export OPENAI_API_KEY=sk-...

Mem0 默认使用本地磁盘向量存储，所以这里的任何操作都不会与托管版 Mem0 服务通信——它是一个完全自包含的、属于你自己的记忆层。

第一步：gatekeeper

# gatekeeper.py
from mem0 import Memory
from datetime import datetime, timezone

class MemoryGatekeeper:
    def __init__(self, agent_id="default-agent", block_threshold=0.75):
        self.memory = Memory()
        self.agent_id = agent_id
        self.block_threshold = block_threshold

    def _describe_call(self, tool_name, args):
        return f"tool call: {tool_name} with args {args}"

    def check(self, tool_name, args):
        """Returns (allowed: bool, reason: str | None)."""
        query = self._describe_call(tool_name, args)
        hits = self.memory.search(query, user_id=self.agent_id, limit=3)

        for hit in hits.get("results", []):
            score = hit.get("score", 0)
            memory_text = hit.get("memory", "")
            if score >= self.block_threshold and "failed" in memory_text.lower():
                return False, memory_text
        return True, None

    def record(self, tool_name, args, success, detail):
        outcome = "succeeded" if success else "failed"
        text = (
            f"{self._describe_call(tool_name, args)} {outcome} "
            f"at {datetime.now(timezone.utc).isoformat()}: {detail}"
        )
        self.memory.add(text, user_id=self.agent_id)

    def call(self, tool_name, args, fn):
        allowed, reason = self.check(tool_name, args)
        if not allowed:
            print(f"[BLOCKED] {tool_name}({args}) — remembered: {reason}")
            return {"blocked": True, "reason": reason}

        try:
            result = fn(*args.values()) if isinstance(args, dict) else fn(args)
            self.record(tool_name, args, True, "ok")
            return {"blocked": False, "result": result}
        except Exception as exc:
            self.record(tool_name, args, False, str(exc))
            raise

check 步骤使用的是语义搜索而非精确字符串匹配——fetch_weather(city="NYC") 和 fetch_weather(city="New York") 会相互匹配，这正是精确匹配缓存会漏掉的那类近似重复。

第二步：一个可预测地失败的工具

# flaky_tool.py
class RateLimitError(Exception):
    pass

def call_flaky_api(endpoint):
    if endpoint == "/reports/daily":
        raise RateLimitError("429: rate limit exceeded, retry after 3600s")
    return {"status": "ok", "endpoint": endpoint}

# run.py
from gatekeeper import MemoryGatekeeper
from flaky_tool import call_flaky_api

gate = MemoryGatekeeper(agent_id="report-agent")

try:
    gate.call("call_flaky_api", {"endpoint": "/reports/daily"}, call_flaky_api)
except Exception as exc:
    print(f"First attempt failed as expected: {exc}")

First attempt failed as expected: 429: rate limit exceeded, retry after 3600s

再次运行 python run.py——同一个进程、同样的工具、同样的参数，但现在 Agent 有了之前失败的记忆：

[BLOCKED] call_flaky_api({'endpoint': '/reports/daily'}) — remembered: tool call: call_flaky_api with args {'endpoint': '/reports/daily'} failed at 2026-08-18T09:12:04+00:00: 429: rate limit exceeded, retry after 3600s

没有第二次 API 调用，没有第二次限流触发，而且 Agent 得到了一个它可以推理（或转发给人类）的理由，而不是一个原始的堆栈跟踪。

调优以适配真实使用

一旦你超越玩具示例，有三个旋钮值得关注：

block_threshold — 0.75 是刻意保守的设置。调低它，你会拦截那些本来可能成功的近似请求；调高它，你只会捕获几乎完全相同的重复。信任默认值之前，先记录一周的所有拦截决策。

按 Agent 作用域管理记忆，而不是全局。 user_id=self.agent_id 可以防止一个 Agent 的坏运气阻塞另一个 Agent 对同一工具的合法调用。如果多个 Agent 确实共享风险（同一个下游 API、同一限流 bucket），给它们一个共享的 agent_id。

让失败过期，不要永久保留。 一小时前的 429 应该阻止重试；三周前的 429 可能就不应该了。Mem0 的记忆在其 metadata 中携带时间戳——在 check() 中添加一个检查，丢弃超过重试窗口的命中（例如 time.time() - hit["created_at"] < 3600），这样陈旧的失败就不会永久禁用一个已经恢复的工具。这一行过滤逻辑，是 gatekeeper 和"害怕所有它曾经失败过的一切"的 Agent 之间的区别。

为什么这比进程内缓存更好

一个普通的 dict 或 functools.lru_cache 可以在单次运行中捕获完全重复的情况，但它随进程消亡，而且无法做语义匹配——它不会知道重试 /reports/daily 和 /reports/daily/ 是同一个错误。将这个路由到 Mem0 的特别价值在于：记忆在重启和部署之间持久化，而且同一个存储也可以保存成功的模式——成功过的工具调用、格式良好的参数、成功完成的序列——所以这个 gatekeeper 不只是一个黑名单，它是一个 Agent 的开端——这个 Agent 会随着时间真正学会更好地使用它的工具，而不是每次冷启动都重新学习同样的教训。

上面的完整示例不到 80 行，只需要一个 Python 环境和 OpenAI key 就能运行——克隆它，运行两次，在你还没读完这句话之前就会看到拦截实时发生。