# RAG 系统并发测试指南 本项目提供了完整的并发测试套件,用于验证 RAG 系统在高并发环境下的性能和稳定性。 ## 📁 测试文件结构 ``` tests/ ├── __init__.py # 测试包初始化 ├── config.py # 测试配置参数 ├── utils.py # 测试工具和辅助函数 ├── test_api.py # 基础 API 功能测试(同步版本) ├── test_concurrent.py # 完整的异步并发测试套件 ├── quick_test.py # 快速功能验证脚本 └── performance_monitor.py # 性能监控工具 ``` ## 🚀 快速开始 ### 1. 安装依赖 在 VS Code 中按 `Ctrl+Shift+P`,然后选择 `Tasks: Run Task` → `Install Test Dependencies` 或者手动安装: ```bash pip install aiohttp requests psutil ``` ### 2. 启动服务器 选择 `Tasks: Run Task` → `Start RAG Service` 启动后台服务 ### 3. 运行测试 #### 基础功能测试 ```bash python tests/test_api.py ``` #### 快速验证测试 ```bash python tests/quick_test.py ``` #### 完整并发测试 ```bash python tests/test_concurrent.py ``` #### 性能监控测试 ```bash python tests/performance_monitor.py ``` ## 🎯 VS Code 任务 通过 VS Code 的任务系统,您可以轻松运行各种测试: ### 基础任务 - **`Setup RAG Environment`** - 设置 Python 环境 - **`Start RAG Service`** - 启动 RAG 服务(后台) - **`Start RAG Service (Foreground)`** - 启动 RAG 服务(前台) - **`Check Server Health`** - 检查服务器状态 ### 测试任务 - **`Run API Tests`** - 运行基础 API 测试 - **`Run Concurrent Tests`** - 运行完整并发测试 - **`Run Light Concurrent Tests`** - 运行轻量级并发测试 - **`Run Stress Tests`** - 运行压力测试 - **`Run Quick Test`** - 运行快速验证测试 - **`Run Performance Monitor`** - 运行性能监控 ### 维护任务 - **`Clean Test Data`** - 清理测试数据 - **`Install Test Dependencies`** - 安装测试依赖 ## 📊 测试类型详解 ### 1. 健康检查测试 - 验证服务器基本可用性 - 测试并发健康检查请求 - 评估响应时间和成功率 ### 2. 文档上传测试 - 测试并发文档上传 - 验证文件处理能力 - 检查向量化和存储性能 ### 3. 聊天查询测试 - 测试并发问答功能 - 验证检索和生成性能 - 评估响应质量和速度 ### 4. 混合操作测试 - 同时进行多种操作 - 测试系统综合处理能力 - 验证资源竞争处理 ### 5. 性能监控 - 实时监控 CPU 和内存使用 - 生成性能报告 - 识别性能瓶颈 ## 📈 测试参数说明 ### 并发级别配置 - **轻量级测试**: 2-5 个并发请求 - **标准测试**: 10-15 个并发请求 - **压力测试**: 20-50 个并发请求 ### 测试参数 - `num_uploads`: 并发上传文档数量 - `num_queries`: 并发查询请求数量 - `top_k`: 检索文档数量 (默认 3) - `temperature`: LLM 温度参数 (默认 0.7) ## 🔍 结果分析 ### 成功率指标 - **>95%**: 优秀 - **90-95%**: 良好 - **80-90%**: 可接受 - **<80%**: 需要优化 ### 响应时间指标 - **文档上传**: < 5 秒 - **聊天查询**: < 3 秒 - **健康检查**: < 100ms - **文档列表**: < 500ms ### 性能指标 - **CPU 使用率**: 平均 < 80% - **内存使用率**: < 85% - **QPS**: 根据硬件配置而定 ## 📝 生成的报告文件 测试完成后会生成以下文件: - `concurrent_test_report.md` - 并发测试报告 - `performance_metrics_YYYYMMDD_HHMMSS.json` - 性能数据 - `performance_chart.png` - 性能图表(如果安装了 matplotlib) ## 🛠️ 故障排除 ### 常见问题 1. **连接错误** ``` ConnectionError: 无法连接到服务器 ``` 解决方案:确保 RAG 服务器正在运行 2. **依赖缺失** ``` ModuleNotFoundError: No module named 'aiohttp' ``` 解决方案:运行 `Install Test Dependencies` 任务 3. **内存不足** ``` MemoryError ``` 解决方案:减少并发数量或增加系统内存 4. **超时错误** ``` TimeoutError ``` 解决方案:检查网络连接和服务器性能 ### 调试模式 在测试文件中添加调试信息: ```python import logging logging.basicConfig(level=logging.DEBUG) ``` ## 🎨 自定义测试 ### 创建自定义测试脚本 ```python import asyncio import sys import os sys.path.append(os.path.dirname(__file__)) from tests.test_concurrent import ConcurrentRAGTester async def my_custom_test(): async with ConcurrentRAGTester() as tester: # 自定义测试逻辑 result = await tester.chat_query("我的问题") print(f"回答: {result['answer']}") asyncio.run(my_custom_test()) ``` ### 使用测试工具 ```python from tests.utils import TestReporter, TestDataGenerator, PerformanceAnalyzer from tests.config import CONCURRENT_CONFIG, PERFORMANCE_THRESHOLDS # 生成测试数据 docs = TestDataGenerator.generate_test_documents(5) questions = TestDataGenerator.generate_test_questions(10) # 分析性能 analyzer = PerformanceAnalyzer() time_stats = analyzer.analyze_response_times(results) # 生成报告 reporter = TestReporter() report_files = reporter.generate_report(test_results, "my_test") ``` ### 修改测试参数 编辑 `tests/config.py` 中的配置: ```python # 增加并发数量 CONCURRENT_CONFIG["custom"] = { "health_checks": 15, "uploads": 8, "queries": 20, "doc_lists": 4 } ``` ## 📞 支持 如果遇到问题或需要帮助,请: 1. 检查服务器日志 2. 查看测试输出的错误信息 3. 确认所有依赖已正确安装 4. 验证系统资源充足 --- **注意**: 请在充足的系统资源环境下运行压力测试,避免影响其他应用程序。