工具结果缓存与幂等设计：让 Agent 重试不重复执行、不重复计费

Agent 重试策略（463）与并发吞吐（465）落地之后，会自然长出一个新问题：同样的调用发生了两遍怎么办。模型偶尔会拿相同参数再调一次工具；重试机制会重发上一次失败的请求；并发聚合里个别任务超时后重跑。如果工具是查询，这叫浪费钱；如果工具是发消息、下订单、写数据库，这叫事故。本篇给工具层补上两道闸：读操作用缓存挡住重复计算，写操作用幂等键保证「做多次等于做一次」。

💡 本篇是 461（工具注册表）与 463（重试装饰器）的续篇，包装器挂在工具表上即可全局生效，业务代码零改动。

Step 1：认清三类重复，对症下药

1 先分清读和写

重复调用分三种情况：一是「相同参数重复读」——模型翻来覆去查同一个城市天气，重复花钱但无害；二是「重试导致重复读」——超时重试把已成功的调用再跑一遍，浪费一次计费；三是「重试导致重复写」——这条最危险，超时的那个请求可能已经把消息发出去了，重试等于发两遍。

对策完全不同：读操作可以缓存（同参数直接返回上次结果），写操作绝不能简单缓存（必须保证幂等——同一业务动作执行多次，效果与一次相同）。

判断读还是写不要看函数名，看副作用。get_xxx 也可能写日志计数，send_xxx 也可能先查后写。给工具注册表（461 的 TOOLS）加一个 side_effect 布尔字段，登记时人工标注，后续所有机制都靠它分流。

Step 2：设计缓存键——工具名加规范化参数哈希

2 缓存键的核心是「规范化」

同一调用的参数可能有多种写法：city="北京" 与 city='北京' 与 {"city": "北京"} 语义相同。直接拿 repr 当键会漏掉这些命中。规范化的办法：先把参数转成 JSON，键排序，再哈希：

import json, hashlib

def cache_key(tool_name: str, args: dict, kwargs: dict) -> str:
    payload = json.dumps(
        {"t": tool_name, "a": list(args), "k": kwargs},
        ensure_ascii=False, sort_keys=True, separators=(",", ":"),
    )
    return hashlib.sha256(payload.encode("utf-8")).hexdigest()

print(cache_key("weather", ("北京",), {"unit": "c"}))
# 64 位十六进制串，同参数必同键，异参数几乎必异键

注意 sort_keys=True 与 separators 的作用：前者让键序无关，后者去掉空白差异。这样新旧两次调用只要语义相同，键一定相同。

💡 如果模型传参有「半角全角、大小写、多空格」这类噪声，可以在哈希前加一步清洗（去空白、统一小写）。清洗规则要写死在缓存层，别让每个工具自己发明。

Step 3：SQLite 持久缓存 + TTL 过期

3 为什么不用内存字典

内存缓存进程重启即失，Agent 常常一跑几小时或跨天复用，用 SQLite 一张表就能持久化，无需额外部署：

import sqlite3, time

conn = sqlite3.connect("tool_cache.db", check_same_thread=False)
conn.execute('''
CREATE TABLE IF NOT EXISTS cache (
    k TEXT PRIMARY KEY,
    v TEXT NOT NULL,
    exp REAL NOT NULL              -- 过期时间戳
)''')

def cache_get(k: str):
    row = conn.execute("SELECT v, exp FROM cache WHERE k=?", (k,)).fetchone()
    if row and row[1] > time.time():
        return row[0]
    return None                      # 不存在或已过期

def cache_put(k: str, v: str, ttl: float):
    conn.execute(
        "INSERT OR REPLACE INTO cache(k, v, exp) VALUES (?,?,?)",
        (k, v, time.time() + ttl),
    )
    conn.commit()

过期行不主动删也没关系，读的时候跳过即可；量大可以起个定时任务 DELETE WHERE exp < 当前时间。结果值统一 json.dumps 存字符串，读出来 json.loads 还原。

TTL 按工具的「数据新鲜度要求」逐个定：天气 10 分钟，汇率 1 小时，公司内部知识库 1 天，而「当前时间」这类工具必须 ttl=0（即不缓存）——否则模型上午问过一次时间，下午再问拿到的是旧值，后面所有推理全部带偏。

Step 4：负缓存——失败结果也要记下来

4 短 TTL 缓存错误，挡住重试风暴

某个目标站宕机时，每次调用都要等满超时（比如 10 秒）才失败。并发重试叠加起来，Agent 会把大量时间砸在一个注定失败的地址上。负缓存的思路：失败结果也入库，但 TTL 给短一点（比如 60 秒），窗口内的同参数调用直接返回上次错误：

def call_with_cache(tool, ttl=600, neg_ttl=60, **call):
    key = cache_key(tool.__name__, call.get("args", ()), call.get("kwargs", {}))
    hit = cache_get(key)
    if hit is not None:
        return json.loads(hit), True          # 第二个元素：是否命中

    try:
        value = tool(*call.get("args", ()), **call.get("kwargs", {}))
        cache_put(key, json.dumps(value, ensure_ascii=False), ttl)
        return value, False
    except Exception as e:
        # 负缓存：错误短 TTL，窗口内不再硬撞
        cache_put(key, json.dumps({"__error__": str(e)}, ensure_ascii=False), neg_ttl)
        raise

读到带 __error__ 标记的缓存命中时，直接把异常还原抛出（或返回错误字符串），让上层按 463 的降级逻辑换路径，而不是傻等超时。

Step 5：写操作的幂等键——先登记，再执行

5 幂等的本质是「动作先领号」

写操作的正确姿势：每次业务动作用「工具名 + 业务参数哈希」生成幂等键，执行前先在幂等表里登记，登记成功才真正执行，执行完写入结果。重复请求拿着同一个键来查，发现已执行过，直接返回上次结果：

conn.execute('''
CREATE TABLE IF NOT EXISTS idem (
    k TEXT PRIMARY KEY,
    status TEXT NOT NULL,          -- running / done
    result TEXT,
    created REAL NOT NULL
)''')

def call_idempotent(tool, ttl_result=86400, **call):
    key = cache_key(tool.__name__, call.get("args", ()), call.get("kwargs", {}))
    row = conn.execute("SELECT status, result FROM idem WHERE k=?", (key,)).fetchone()
    if row:
        if row[0] == "done":
            return json.loads(row[1])          # 已完成：返回上次结果
        raise RuntimeError("同键调用正在执行，拒绝并发重入")

    try:
        conn.execute("INSERT INTO idem(k, status, created) VALUES (?,?,?)",
                     (key, "running", time.time()))
        conn.commit()
    except sqlite3.IntegrityError:
        raise RuntimeError("同键调用正在执行，拒绝并发重入")

    value = tool(*call.get("args", ()), **call.get("kwargs", {}))
    conn.execute("UPDATE idem SET status='done', result=? WHERE k=?",
                 (json.dumps(value, ensure_ascii=False), key))
    conn.commit()
    return value

running 状态会因进程崩溃变成孤儿锁。补一条兜底：created 超过某个时限（如 10 分钟）且仍为 running 的记录，视为失败，允许删除重试。时限必须大于工具的最大正常耗时，否则会把慢任务误杀成重入。

Step 6：把缓存与幂等挂进工具注册表

6 按副作用字段自动分流

回到 461 的工具表，登记时补两个字段，包装器按字段自动选择策略：

TOOLS = {
    "weather":  {"fn": fetch_weather,  "side_effect": False, "ttl": 600},
    "kb_query": {"fn": query_kb,       "side_effect": False, "ttl": 86400},
    "now":      {"fn": now_iso,        "side_effect": False, "ttl": 0},
    "send_sms": {"fn": send_sms,       "side_effect": True,  "ttl": None},
}

def dispatch(name, *args, **kwargs):
    spec = TOOLS[name]
    if not spec["side_effect"]:
        if spec["ttl"] == 0:
            return spec["fn"](*args, **kwargs)          # 不缓存
        v, _ = call_with_cache(spec["fn"], ttl=spec["ttl"], args=args, kwargs=kwargs)
        return v
    return call_idempotent(spec["fn"], args=args, kwargs=kwargs)

Agent 主循环里所有工具调用统一走 dispatch，缓存与幂等就全局生效了。这也是把 461 注册表设计成「中心登记」的回报：横切能力只改一处。

💡 上线前拿日志做一次回放验证：把过去一天的调用记录（467 的事件流正好可用）喂给新缓存层，统计命中率。命中率低于 15% 说明模型参数噪声大，先做 Step 2 的参数清洗再上缓存。

Step 7：边界情况清点

7 四个最容易翻车的点

其一，参数里混入时间戳或随机数——每次键都不同，缓存形同虚设。识别这类参数并在哈希前剔除，但要小心：剔除后语义变化时宁可不缓存。其二，大结果对象——缓存里存了几 MB 的 JSON，读写都慢，超过阈值（比如 64KB）直接跳过缓存。其三，鉴权类参数——不同用户的相同查询语义不同，键里必须包含用户标识，否则会出现跨用户读到他人结果的越权事故。其四，写操作参数里有「重试方生成的随机请求号」时，要改用业务语义键（收件人 + 内容哈希），别拿随机号当幂等键，否则重试拿新号照样重发。

缓存层是事故高发区，任何修改都要过一遍「跨用户隔离、时间参数、写读分流」三问。保守做法：先只对无副作用且 TTL 明确的工具开缓存，跑稳两周再扩大范围。

到这里，工具层的三件套齐了：463 管失败恢复，465 管并发吞吐，本篇管重复浪费。生产环境的 Agent 工具层，靠的就是这一层层薄而清晰的横切设计，而不是把逻辑糊进每个业务函数。

🐍 Python 💾 缓存 🔁 幂等 🛠️ 工具调用 🧱 工程实战 📚 更多教程

← 返回教程中心