解决Agent重复调用工具问题的设计与实现

1次阅读

没有评论

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

在构建基于 Agent 的自动化系统中，重复调用工具是一个常见但容易被忽视的问题。这个问题通常出现在以下几种场景：

多 Agent 协作场景 ：多个 Agent 可能同时或先后调用同一个工具，导致重复执行相同的操作。
异步回调场景 ：Agent 在等待工具返回结果时超时，重新发起调用，但实际上第一次调用已经在处理中。
失败重试场景 ：工具调用失败后，Agent 的重试机制可能导致短时间内多次重复调用。

重复调用带来的负面影响不容小觑：

资源浪费 ：每次工具调用都会消耗计算资源，重复调用直接导致资源利用率下降。
数据不一致 ：特别是对于写操作，重复调用可能导致数据被多次修改，产生不一致状态。
性能瓶颈 ：无节制的重复调用可能使下游工具服务过载，影响整体系统性能。

针对这个问题，业界主要有三种解决方案：

通过在调用前检查是否有相同的请求正在处理，避免重复提交。实现方式包括：

基于请求内容的哈希值建立内存缓存
使用分布式缓存记录正在处理的请求

优点：实现简单，拦截早，资源消耗小。
缺点：无法处理业务层面的重复请求。

维护工具调用的状态机，在状态流转时进行校验。常用方法：

为每个工具调用分配唯一 ID
在数据库或缓存中记录调用状态

优点：能处理更复杂的业务场景。
缺点：实现复杂度较高。

让工具本身具备处理重复调用的能力，确保多次调用产生相同的结果。常见策略：

使用唯一业务 ID
实现 check-and-set 操作

优点：最彻底的解决方案。
缺点：需要改造现有工具，实施成本高。

在实际项目中，我们推荐组合使用请求去重和状态管理，在不对现有工具做大规模改造的情况下获得较好的效果。

下面是一个基于 Python 的实现示例，结合了请求去重和状态管理：

import hashlib
from typing import Dict, Optional
from dataclasses import dataclass

@dataclass
class ToolInvocation:
    tool_name: str
    params: dict
    status: str = 'pending'  # pending/running/completed/failed
    result: Optional[dict] = None

class ToolDispatcher:
    def __init__(self):
        self.active_invocations: Dict[str, ToolInvocation] = {}

    def _generate_request_id(self, tool_name: str, params: dict) -> str:
        """生成请求唯一标识"""
        param_str = str(sorted(params.items()))
        return f"{tool_name}:{hashlib.md5(param_str.encode()).hexdigest()}"

    async def invoke_tool(self, tool_name: str, params: dict) -> dict:
        request_id = self._generate_request_id(tool_name, params)

        # 检查是否已有相同请求
        if request_id in self.active_invocations:
            existing = self.active_invocations[request_id]
            if existing.status == 'completed':
                return existing.result
            elif existing.status == 'running':
                raise Exception('相同请求正在处理中')
            # 失败状态允许重试

        # 记录新请求
        invocation = ToolInvocation(tool_name=tool_name, params=params)
        self.active_invocations[request_id] = invocation

        try:
            invocation.status = 'running'
            # 实际调用工具的逻辑
            result = await self._call_actual_tool(tool_name, params)
            invocation.result = result
            invocation.status = 'completed'
            return result
        except Exception as e:
            invocation.status = 'failed'
            raise e

    async def _call_actual_tool(self, tool_name: str, params: dict) -> dict:
        # 这里是实际调用工具的代码
        # 模拟一个耗时操作
        await asyncio.sleep(1)
        return {"success": True}

这个实现的核心要点：