Agent调用工具链的工程化实践：从设计模式到性能优化

1次阅读

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

在复杂业务场景中，Agent 调用工具链常面临三个核心痛点：

工具依赖冲突 ：当多个工具依赖同一个库的不同版本时，可能导致运行时错误。例如工具 A 需要 requests==2.25 而工具 B 需要 requests==3.0
上下文丢失 ：在异步调用链中，ThreadLocal 存储的上下文无法跨线程传递，导致鉴权信息等关键数据丢失
同步阻塞 ：工具链中的同步 IO 操作会阻塞整个事件循环，造成性能瓶颈。通过火焰图可以看到 90% 的延迟集中在少数几个阻塞调用上

责任链模式 ：
每个工具处理器独立处理请求
天然支持动态添加 / 移除处理器

典型代码结构：

class ToolHandler:
    def __init__(self, successor=None):
        self._successor = successor

    async def handle(self, request):
        processed = await self._process(request)
        if self._successor:
            return await self._successor.handle(processed)
        return processed

中介者模式 ：
更适合工具间需要复杂交互的场景
会增加中心节点的复杂度

classDiagram
    class ToolRegistry {+register(tool: BaseTool)
        +unregister(tool_name: str)
        +get_tool(tool_name: str) BaseTool
    }

    class BaseTool {
        <<abstract>>
        +name: str
        +version: str
        +execute(input: dict) Awaitable[dict]
    }

    class CircuitBreaker {
        -failure_count: int
        -last_failure_time: float
        +check_state() bool
        +record_failure()}

    ToolRegistry "1" *-- "*" BaseTool
    BaseTool <|-- ConcreteTool
    BaseTool o-- CircuitBreaker

from contextvars import ContextVar
import asyncio
from datetime import datetime, timedelta
import random

request_context = ContextVar('request_context', default={})

class AsyncInvoker:
    def __init__(self, max_concurrency=10):
        self.semaphore = asyncio.Semaphore(max_concurrency)
        self.circuit_breaker = {
            'failure_threshold': 3,
            'recovery_timeout': 30,
            'last_failure': None,
            'failure_count': 0
        }

    async def invoke_with_retry(self, tool, input_data, max_retries=3):
        async with self.semaphore:
            for attempt in range(max_retries + 1):
                try:
                    if self._check_circuit_breaker():
                        return await tool.execute(input_data)
                    else:
                        raise CircuitOpenError("Circuit breaker is open")
                except Exception as e:
                    if attempt == max_retries:
                        self._record_failure()
                        raise
                    await self._exponential_backoff(attempt)

    def _check_circuit_breaker(self):
        cb = self.circuit_breaker
        if cb['failure_count'] < cb['failure_threshold']:
            return True

        if datetime.now() - cb['last_failure'] > timedelta(seconds=cb['recovery_timeout']
        ):
            cb['failure_count'] = 0
            return True
        return False

    def _record_failure(self):
        self.circuit_breaker['failure_count'] += 1
        self.circuit_breaker['last_failure'] = datetime.now()

    async def _exponential_backoff(self, attempt):
        delay = min(2 ** attempt + random.uniform(0, 1), 10)
        await asyncio.sleep(delay)

class ContextAwareTool(BaseTool):
    async def execute(self, input_data):
        ctx = request_context.get()
        # 使用上下文中存储的认证信息等
        headers = ctx.get('headers', {})
        ...

# 调用处设置上下文
async def handle_request(request):
    request_context.set({'headers': request.headers})
    await tool_registry.get_tool('demo').execute({})

为每个工具调用生成唯一 trace_id
在工具接口中强制要求实现 idempotency_key 参数
使用 Redis 记录已处理的请求

重点关注：

工具类中不当的闭包引用
未释放的线程局部存储
缓存未设置 TTL

检测方法：

import tracemalloc

tracemalloc.start()
# 执行压力测试
snapshot = tracemalloc.take_snapshot()
top_stats = snapshot.statistics('lineno')
for stat in top_stats[:10]:
    print(stat)

Prometheus 指标示例：

from prometheus_client import Counter, Histogram

TOOL_INVOKE_COUNT = Counter(
    'tool_invoke_total', 
    'Total tool invocations',
    ['tool_name', 'status']
)

TOOL_LATENCY = Histogram(
    'tool_execute_seconds',
    'Tool execution latency',
    ['tool_name'],
    buckets=(.1, .25, .5, 1, 2.5, 5, 10)
)

class MonitoredTool(BaseTool):
    async def execute(self, input_data):
        start_time = time.time()
        try:
            result = await super().execute(input_data)
            TOOL_INVOKE_COUNT.labels(
                tool_name=self.name,
                status='success'
            ).inc()
            return result
        except Exception:
            TOOL_INVOKE_COUNT.labels(
                tool_name=self.name,
                status='failure'
            ).inc()
            raise
        finally:
            TOOL_LATENCY.labels(tool_name=self.name).observe(time.time() - start_time)

将 CPU 密集型操作放到线程池：

await asyncio.get_event_loop().run_in_executor(
    None, 
    cpu_intensive_task
)

使用 async 版本的库（如 aiohttp 代替 requests）

设置合理的超时时间：

await asyncio.wait_for(tool.execute(input),
    timeout=30.0
)

使用语义化版本控制

工具注册时检查版本约束：

if not tool.check_compatible(registry_version):
    raise VersionConflictError()

通过 API 网关实现灰度路由

示例语法：

pipeline:
  - name: data_fetch
    tool: http_fetcher
    params:
      url: ${input.url}
    timeout: 5s

  - name: data_parse
    tool: json_parser
    depends_on: [data_fetch]

  - name: notify
    tool: slack_notifier
    when: ${data_parse.result.code != 200}

实现思路：