构建高可靠agent基准测试框架：从技术选型到生产环境实践

1次阅读

没有评论

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

在 AI agent 开发过程中，我最初尝试用 time.time() 手动测量函数执行时间，很快发现三个致命缺陷：

并发测试缺失：真实场景中 agent 往往需要处理高并发请求，简单串行测试无法暴露竞争条件
指标单一：仅测量耗时无法反映 CPU/ 内存波动、网络 IO 等关键维度
环境干扰：未隔离的测试环境会导致结果波动（如其他进程抢占资源）

更痛苦的是，当需要对比不同算法版本时，每次都要重写测试逻辑。这促使我设计一个标准化测试框架。

采用三层分治结构，各层通过接口解耦：

测试定义层
用 YAML 定义测试场景（如并发用户数、请求参数模板）
支持参数化测试（同一测试不同输入组合）
执行引擎层
异步任务调度（asyncio + concurrent.futures）
资源隔离（每个测试用例独立进程）
超时熔断机制
数据收集层
通过装饰器自动捕获耗时、异常等指标
实时写入时序数据库（InfluxDB）
聚合计算 P99 延迟、QPS 等

关键创新点在于用装饰器实现非侵入式监控。例如记录函数执行时间的装饰器：

def metric_collector(func):
    @wraps(func)
    async def wrapper(*args, **kwargs):
        start_time = time.perf_counter()
        try:
            result = await func(*args, **kwargs)
            latency = time.perf_counter() - start_time
            MetricsClient.record_latency(func.__name__, latency)
            return result
        except Exception as e:
            MetricsClient.record_error(func.__name__, str(e))
            raise
    return wrapper

class AsyncRunner:
    def __init__(self, max_workers=10):
        self.semaphore = asyncio.Semaphore(max_workers)

    async def run_with_timeout(self, coro, timeout_sec):
        try:
            async with async_timeout.timeout(timeout_sec):
                async with self.semaphore:
                    return await coro
        except asyncio.TimeoutError:
            logger.warning(f"Task timed out after {timeout_sec} seconds")
            raise

class MetricAnalyzer:
    @staticmethod
    def calculate_stats(raw_data: pd.DataFrame) -> dict:
        return {"throughput": len(raw_data) / raw_data["duration"].sum(),
            "p99_latency": raw_data["latency"].quantile(0.99),
            "error_rate": raw_data["is_error"].mean()}

@app.get("/report/{test_id}")
async def get_report(test_id: str):
    data = load_test_data(test_id)
    return {"summary": MetricAnalyzer.calculate_stats(data),
        "raw_data": data.to_dict(orient="records")
    }

框架自身性能至关重要，我们通过以下手段控制开销：