DEV Community

Mukesh
Mukesh

Posted on

Give Your AI Agent a Memory So It Stops Repeating the Same Failed Tool Call

Your agent calls a flaky API, gets a 429, retries with the same arguments, and gets rate-limited again. Ten minutes later, in a fresh session, it does the exact same thing. Nothing about the failure got remembered — the agent has no way to know it already learned this lesson, because the lesson lived in a transcript that got thrown away when the process exited.

This week a post about manually gatekeeping AI agent tool calls hit the front page of dev.to with 48 comments — mostly developers arguing over how much you can trust an agent's tool-calling loop. The honest answer is: you can trust it exactly as much as it remembers what already went wrong. Below is a complete, runnable gatekeeper that wraps any tool call, checks Mem0 for similar past failures before executing, and records the outcome afterward — so the second time your agent is about to make the same mistake, it knows.

What we're building

A MemoryGatekeeper class that sits between your agent's decision to call a tool and the actual execution:

  1. Before calling a tool, search memory for semantically similar past attempts.
  2. If a similar attempt failed recently, block the call and return the remembered reason instead of burning a real API call.
  3. After every call — success or failure — write the outcome back to memory.

You'll run the same script twice: the first run fails and gets recorded, the second run gets blocked before it wastes a request.

Setup

You need Python 3.10+, an OpenAI API key (Mem0's default extraction pipeline uses it to turn raw text into structured memories — no separate Mem0 account required for this local setup), and the mem0ai package.

pip install mem0ai
export OPENAI_API_KEY=sk-...
Enter fullscreen mode Exit fullscreen mode

Mem0 defaults to a local, on-disk vector store, so nothing here talks to a hosted Mem0 service — it's a fully self-contained memory layer you own.

Step 1: The 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
Enter fullscreen mode Exit fullscreen mode

The check step uses semantic search, not exact string matching — fetch_weather(city="NYC") and fetch_weather(city="New York") will match each other, which is exactly the kind of near-duplicate an exact-match cache would miss.

Step 2: A tool that fails predictably

# 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}
Enter fullscreen mode Exit fullscreen mode

Step 3: Run it twice

# 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}")
Enter fullscreen mode Exit fullscreen mode

First run:

First attempt failed as expected: 429: rate limit exceeded, retry after 3600s
Enter fullscreen mode Exit fullscreen mode

Run python run.py again — same process, same tool, same arguments, but now the agent has memory of the earlier failure:

[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
Enter fullscreen mode Exit fullscreen mode

No second API call, no second rate-limit hit, and the agent gets a reason it can reason about (or relay to a human) instead of a raw stack trace.

Tuning it for real use

Three knobs matter once you move past the toy example:

  • block_threshold — 0.75 is conservative on purpose. Push it lower and you'll block near-misses that would have actually succeeded; push it higher and you'll only catch near-identical repeats. Log every block decision for a week before trusting the default.
  • Scope memory per agent, not globally. user_id=self.agent_id keeps one agent's bad luck from blocking a different agent's legitimate call to the same tool. If multiple agents genuinely share risk (same downstream API, same rate limit bucket), give them a shared agent_id.
  • Expire failures, don't keep them forever. A 429 from an hour ago should block a retry; a 429 from three weeks ago probably shouldn't. Mem0 memories carry timestamps in their metadata — add a check in check() that discards hits older than your retry window (e.g., time.time() - hit["created_at"] < 3600) so stale failures don't permanently disable a tool that's since recovered. This one-line filter is the difference between a gatekeeper and a agent that's afraid of everything it's ever failed at once.

Why this beats an in-process cache

A plain dict or functools.lru_cache would catch the exact-repeat case in a single run, but it dies with the process and can't do semantic matching — it won't know that retrying /reports/daily and /reports/daily/ are the same mistake. The value of routing this through Mem0 specifically is that the memory persists across restarts and deployments, and the same store can also hold successful patterns — tool calls that worked, arguments that were well-formed, sequences that completed cleanly — so the gatekeeper isn't just a blocklist, it's the beginning of an agent that actually gets better at using its tools over time instead of relearning the same lesson every cold start.

The full example above is under 80 lines and runs with nothing but a Python environment and an OpenAI key — clone it, run it twice, and you'll see the block happen live before you've read the rest of this sentence.

Top comments (0)