Agent工具多次调用性能优化实战：从并发瓶颈到高效调度

1次阅读

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

在订单处理、风控审核等业务场景中，我们经常需要频繁调用 Agent 工具进行决策或计算。例如，一个电商平台可能需要对每笔订单调用风控 Agent、库存 Agent 和物流 Agent。这种模式在高并发场景下会暴露出明显问题：

网络延迟放大 ：每次调用都需要完整的网络往返时间，100 次串行调用意味着 100 倍延迟
计算资源浪费 ：相同参数的重复调用导致底层计算资源被无意义消耗
系统稳定性风险 ：当 Agent 服务出现波动时，大量重复请求会加剧服务端压力

我们曾遇到一个典型案例：在促销期间，订单处理系统因为同步调用风控 Agent 导致平均响应时间从 200ms 飙升到 2 秒，直接影响了成交转化率。

方案类型	QPS(相同资源)	CPU 占用	实现复杂度	适用场景
纯同步调用	100	65%	★☆☆☆☆	低并发、简单业务
简单异步改造	350	75%	★★☆☆☆	IO 密集型任务
智能调度系统	800+	85%	★★★★☆	高并发、复杂业务流

注：测试环境为 4 核 8G 云主机，Agent 平均响应时间 50ms

智能调度系统的优势在于：
1. 请求合并：将短时间内相同参数的调用合并为单次请求
2. 结果复用：通过缓存层避免重复计算
3. 异步编排：非阻塞地处理多个 Agent 的依赖关系

import asyncio
from typing import Dict, Any
from collections import defaultdict

class BatchProcessor:
    def __init__(self, max_batch_size: int = 50):
        self._queue = asyncio.Queue()
        self._max_batch_size = max_batch_size
        self._pending_requests = defaultdict(list)

    async def process(self, params: Dict[str, Any]) -> Any:
        """将请求参数加入批处理队列"""
        future = asyncio.get_event_loop().create_future()
        await self._queue.put((params, future))
        return await future

    async def start_worker(self):
        """后台批量处理任务"""
        while True:
            batch = []
            # 等待首个请求或达到批量大小
            item = await self._queue.get()
            batch.append(item)

            # 非阻塞获取更多请求
            while len(batch) < self._max_batch_size:
                try:
                    item = self._queue.get_nowait()
                    batch.append(item)
                except asyncio.QueueEmpty:
                    break

            # 按参数分组处理
            param_groups = defaultdict(list)
            for params, future in batch:
                param_key = frozenset(params.items())
                param_groups[param_key].append((params, future))

            # 批量调用 Agent
            tasks = []
            for param_group in param_groups.values():
                sample_params = param_group[0][0]
                task = self._call_agent(sample_params)
                tasks.append((task, param_group))

            # 分发结果
            for task, group in tasks:
                try:
                    result = await task
                    for _, future in group:
                        if not future.done():
                            future.set_result(result)
                except Exception as e:
                    for _, future in group:
                        if not future.done():
                            future.set_exception(e)

    async def _call_agent(self, params: Dict[str, Any]) -> Any:
        """实际调用 Agent 服务"""
        # 实现具体的网络调用逻辑
        ...

import redis
from datetime import timedelta
import pickle

class AgentResultCache:
    def __init__(self, redis_client: redis.Redis):
        self._client = redis_client

    async def get_or_set(self, 
                        key: str, 
                        builder: callable,
                        ttl: timedelta = timedelta(minutes=5)) -> Any:
        """获取缓存或执行 builder 函数"""
        # 尝试获取缓存
        cached = self._client.get(key)
        if cached is not None:
            return pickle.loads(cached)

        # 获取分布式锁防击穿
        lock_key = f"lock:{key}"
        with self._client.lock(lock_key, timeout=5):
            # 双重检查
            cached = self._client.get(key)
            if cached is not None:
                return pickle.loads(cached)

            # 执行实际调用
            result = await builder()

            # 设置缓存及随机过期时间防雪崩
            randomized_ttl = ttl + timedelta(seconds=random.randint(0, 60))
            self._client.setex(
                name=key,
                time=randomized_ttl,
                value=pickle.dumps(result)
            )
            return result

from typing import Optional
from datetime import datetime, timedelta

class CircuitBreaker:
    def __init__(self,
                 max_failures: int = 5,
                 reset_timeout: timedelta = timedelta(seconds=30)):
        self._max_failures = max_failures
        self._reset_timeout = reset_timeout
        self._failure_count = 0
        self._last_failure_time: Optional[datetime] = None
        self._state = "closed"  # closed/half-open/open

    async def execute(self, callable_fn: callable) -> Any:
        current_state = self._get_current_state()

        if current_state == "open":
            raise CircuitOpenError("Service unavailable")

        try:
            result = await callable_fn()
            self._record_success()
            return result
        except Exception as e:
            self._record_failure()
            raise

    def _get_current_state(self) -> str:
        if self._state == "open":
            if datetime.now() - self._last_failure_time > self._reset_timeout:
                self._state = "half-open"
                return "half-open"
            return "open"
        return self._state

    def _record_success(self):
        if self._state == "half-open":
            self._state = "closed"
            self._failure_count = 0

    def _record_failure(self):
        self._failure_count += 1
        self._last_failure_time = datetime.now()

        if self._failure_count >= self._max_failures:
            self._state = "open"

使用 Locust 对优化前后方案进行压测（模拟 1000 并发用户）：