本文档说明如何对 RAG 系统进行性能测试和基准测试。
| 端点类型 | 目标响应时间 | 说明 |
|---|---|---|
| 健康检查 | < 50ms | /health, /metrics |
| 简单查询 | < 200ms | 文档 CRUD 操作 |
| 搜索查询 | < 500ms | 向量搜索、混合搜索 |
| 批量操作 | < 2s | 批量导入、批量更新 |
项目提供了两个基准测试脚本:
# 运行模拟测试(不需要启动服务器)
python scripts/benchmark_simple.py
这个脚本会:
# 1. 启动应用
python main.py
# 2. 在另一个终端运行基准测试
python scripts/benchmark.py
这个脚本会:
benchmark_results.json# 测试健康检查端点
ab -n 1000 -c 10 http://localhost:8000/health
# 测试搜索端点(POST 请求)
ab -n 100 -c 5 -p search_payload.json -T application/json \
http://localhost:8000/api/v1/documents/search
# 安装 wrk
# Ubuntu: sudo apt-get install wrk
# macOS: brew install wrk
# 运行负载测试
wrk -t4 -c100 -d30s http://localhost:8000/health
# 使用 Lua 脚本测试 POST 请求
wrk -t4 -c100 -d30s -s search.lua http://localhost:8000/api/v1/documents/search
创建 locustfile.py:
from locust import HttpUser, task, between
class RAGSystemUser(HttpUser):
wait_time = between(1, 3)
@task(3)
def health_check(self):
self.client.get("/health")
@task(2)
def search_documents(self):
self.client.post("/api/v1/documents/search", json={
"query_text": "test query",
"top_k": 10
})
@task(1)
def get_metrics(self):
self.client.get("/metrics")
运行 Locust:
# 安装 Locust
pip install locust
# 启动 Locust Web UI
locust -f locustfile.py
# 或命令行模式
locust -f locustfile.py --headless -u 100 -r 10 -t 60s
目的: 验证系统基本可用性
python scripts/benchmark.py
预期结果:
目的: 测试核心搜索功能
步骤:
预期结果:
目的: 测试系统在高并发下的表现
# 使用 wrk 进行 30 秒压力测试
wrk -t8 -c200 -d30s http://localhost:8000/api/v1/documents/search
预期结果:
目的: 验证系统长时间运行的稳定性
# 使用 Locust 运行 1 小时测试
locust -f locustfile.py --headless -u 50 -r 5 -t 1h
监控指标:
查看基准测试结果:
cat benchmark_results.json | jq '.results[] | {endpoint, mean: .response_time.mean}'
启用 SQLAlchemy 查询日志:
# 在 config/settings.py 中
DB_ECHO = True # 打印所有 SQL 查询
使用 EXPLAIN ANALYZE 分析慢查询:
EXPLAIN ANALYZE
SELECT * FROM documents WHERE ...;
使用 Python profiler:
import cProfile
import pstats
# 分析特定函数
cProfile.run('your_function()', 'profile_stats')
# 查看结果
stats = pstats.Stats('profile_stats')
stats.sort_stats('cumulative')
stats.print_stats(20)
使用 py-spy 进行实时分析:
# 安装 py-spy
pip install py-spy
# 分析运行中的进程
py-spy top --pid <PID>
# 生成火焰图
py-spy record -o profile.svg --pid <PID>
查询优化: 避免 N+1 查询,使用 JOIN
# 配置连接池
SQLALCHEMY_POOL_SIZE = 20
SQLALCHEMY_MAX_OVERFLOW = 40
functools.lru_cacheHTTP 缓存: 设置合适的 Cache-Control 头
from functools import lru_cache
@lru_cache(maxsize=1000)
def get_document(doc_id: str):
# 缓存文档查询结果
pass
async/await 处理 I/O 操作orjson 替代标准 json压缩响应: 启用 gzip 压缩
# 使用 orjson
from fastapi.responses import ORJSONResponse
app = FastAPI(default_response_class=ORJSONResponse)
使用 Prometheus + Grafana:
from prometheus_client import Counter, Histogram
# 定义指标
request_count = Counter('http_requests_total', 'Total HTTP requests')
request_duration = Histogram('http_request_duration_seconds', 'HTTP request duration')
监控系统资源:
# CPU 和内存
htop
# 网络
iftop
# 磁盘 I/O
iotop
分析访问日志:
# 统计响应时间
cat access.log | awk '{print $10}' | sort -n | tail -100
# 统计最慢的端点
cat access.log | sort -k10 -n | tail -20
在部署前,确保完成以下性能测试:
# 性能测试报告
## 测试环境
- 日期: YYYY-MM-DD
- 版本: v1.0.0
- 硬件: CPU, RAM, Disk
- 数据量: 文档数量
## 测试结果
### 响应时间
| 端点 | 平均值 | P95 | P99 | 目标 | 状态 |
|-----|--------|-----|-----|------|------|
| /health | 10ms | 15ms | 20ms | <50ms | ✓ |
| /search | 150ms | 300ms | 450ms | <500ms | ✓ |
### 吞吐量
- 健康检查: 1200 RPS
- 搜索查询: 150 RPS
### 资源使用
- CPU: 45%
- 内存: 1.2GB
- 数据库连接: 25
## 问题和建议
1. 搜索查询在高并发下响应时间增加
2. 建议添加缓存层
## 结论
系统性能符合预期目标。
注意: 性能测试应该在与生产环境相似的环境中进行,以获得准确的结果。