Agent重复调用工具问题解析：从原理到实践的避坑指南

1次阅读

没有评论

共计 1705 个字符，预计需要花费 5 分钟才能阅读完成。

在基于 Agent 的自动化系统中，工具重复调用是一个高频问题。当 Agent 因网络延迟、超时重试或逻辑缺陷多次触发同一工具时，会产生以下典型问题：

资源浪费 ：数据库写入、API 调用等操作消耗额外计算资源
数据不一致 ：如支付类操作可能因重复执行导致资金损失
状态污染 ：文件系统操作可能因多次执行产生冲突结果
系统雪崩 ：级联调用可能引发链式反应导致服务不可用

原理：通过唯一标识确保相同操作只生效一次
优点：
无状态设计，实现简单
天然支持分布式场景
缺点：
需要业务逻辑支持
可能增加存储开销

原理：在执行前查询操作状态
优点：
直观易理解
可结合业务状态机
缺点：
存在竞态条件风险
需要持久化存储支持

原理：通过共享锁确保串行执行
优点：
强一致性保证
通用性强
缺点：
性能开销大
存在死锁风险

import hashlib
from functools import wraps

class IdempotentExecutor:
    """
    幂等执行器实现
    :param storage: 状态存储后端（需实现 get/set 方法）"""
    def __init__(self, storage):
        self.storage = storage

    def generate_key(self, func, *args, **kwargs):
        """生成唯一操作标识"""
        arg_str = ",".join(map(str, args))
        kwarg_str = ",".join(f"{k}={v}" for k,v in kwargs.items())
        raw_key = f"{func.__name__}|{arg_str}|{kwarg_str}"
        return hashlib.sha256(raw_key.encode()).hexdigest()

    def __call__(self, func):
        @wraps(func)
        def wrapped(*args, **kwargs):
            key = self.generate_key(func, *args, **kwargs)

            # 状态检查
            if self.storage.get(key):
                raise AlreadyExecutedError(f"Operation {key} already executed")

            try:
                result = func(*args, **kwargs)
                # 状态持久化
                self.storage.set(key, "completed", ttl=3600)
                return result
            except Exception as e:
                self.storage.set(key, "failed", ttl=60)
                raise
        return wrapped

# 使用示例
cache = RedisStorage()  # 需实现 get/set 方法
executor = IdempotentExecutor(cache)

@executor
def process_payment(user_id, amount):
    """模拟支付处理"""
    # 实际业务逻辑...
    return {"status": "success"}