AI Agent 50 用户并发,选什么异步框架最优?我用 locust + aiohttp 压测 200 并发,数据说话。

1. 测试环境

• AWS c5.xlarge(4 核 8GB)
• Python 3.11
• 网络:Tokyo → serpbase(美西)
• 测试目标:1 万次请求,200 并发

2. 压测脚本

用 locust:

from locust import HttpUser, task, between
import random

class SERPUser(HttpUser):
    wait_time = between(1, 3)

    @task
    def search(self):
        queries = ['上海装修', 'BTC价格', '今日新闻', 'Python教程', 'AI应用']
        q = random.choice(queries)

        self.client.post(
            '/google/search',
            json={'q': q, 'hl': 'zh-CN', 'gl': 'cn', 'num': 10},
            headers={'X-API-Key': 'your-key'},
            name='/google/search'
        )

locust -f load_test.py --host=https://api.serpbase.dev --users 200 --spawn-rate 20 --run-time 5m --headless

5 分钟,200 用户逐步 ramp up。

3. 测试维度

5 种实现对比:

4. 测试结果

5 分钟 1 万次请求:

httpx + HTTP/2 吞吐量 800 req/s,延迟 30ms,完胜。

5. 不同并发数

httpx HTTP/2 100 池,50-500 并发:

200 并发是甜点。再高 QPS 触顶 + 错误率升。

6. 池大小优化

httpx 池 50-200:

100 池足够,再加提升小。

7. 限流影响

serpbase 100 QPS 限制,200 并发会被限流:

要突破 100 QPS,需要:

1. 升级套餐 (serpbase Growth 200 QPS)
2. 多 key 轮换 (违规,不推荐)
3. 主动限流到 80 QPS,留余量

8. 推荐配置

AI Agent 50 用户,200 并发峰值:

import httpx
import asyncio

class OptimizedSERPClient:
    def __init__(self):
        self.client = httpx.AsyncClient(
            http2=True,
            limits=httpx.Limits(
                max_connections=100,
                max_keepalive_connections=20,
                keepalive_expiry=30
            ),
            timeout=httpx.Timeout(connect=2.0, read=5.0),
            headers={'X-API-Key': 'your-key'}
        )
        self.semaphore = asyncio.Semaphore(80)  # 主动限流到 80 QPS

    async def search(self, query):
        async with self.semaphore:
            r = await self.client.post(
                'https://api.serpbase.dev/google/search',
                json={'q': query, 'hl': 'zh-CN', 'gl': 'cn', 'num': 10}
            )
            r.raise_for_status()
            return r.json()

    async def search_all(self, queries):
        tasks = [self.search(q) for q in queries]
        return await asyncio.gather(*tasks, return_exceptions=True)

    async def close(self):
        await self.client.aclose()

主动限流 + 异步 + HTTP/2 + 池化,综合最优。

9. 监控

压测时盯 4 个指标:

from prometheus_client import Histogram, Counter, Gauge

latency = Histogram('serp_latency', 'Latency', buckets=[0.01, 0.05, 0.1, 0.5, 1, 5])
errors = Counter('serp_errors', 'Errors', ['type'])
qps = Gauge('serp_qps', 'QPS')
active = Gauge('serp_active', 'Active requests')

@latency.time()
async def monitored_search(self, query):
    active.inc()
    try:
        r = await self.client.post(...)
        r.raise_for_status()
        return r.json()
    except Exception as e:
        errors.labels(type=type(e).__name__).inc()
        raise
    finally:
        active.dec()

Grafana 实时看 P50 / P99 / QPS / 错误率。

10. 实战数据

我项目跑 6 个月:

• 50 用户,峰值 200 并发
• 平均 QPS 80
• 月调用 30 万次
• 月成本 serpbase $50
• 0 故障

11. 进阶:WebSocket 模式

对实时性要求高的场景,WebSocket 替代 HTTP:

# 客户端
import websockets
import asyncio

async def stream_search(queries):
    async with websockets.connect('wss://api.serpbase.dev/stream') as ws:
        for q in queries:
            await ws.send(json.dumps({'q': q}))
            result = await ws.recv()
            yield json.loads(result)

WebSocket 减少 HTTP 头开销,延迟再降 30%。

但 SERP API 厂商支持少,目前不是主流。

12. 总结

200 并发压测结果:

AI Agent 推荐 httpx + HTTP/2 + 池化 + 主动限流。

代码 + 压测脚本 GitHub 公开,clone 跑起来。

代码参考

MCP 集成代码参考 serpbase-mcp 开源仓库,Issue 和 PR 欢迎在 GitHub 提交。