Langchain-Chatchat如何进行压力测试？Locust模拟高并发-洪萨配资

Langchain-Chatchat 如何进行压力测试？用 Locust 模拟高并发场景

在企业级 AI 应用日益普及的今天，一个“能回答问题”的系统只是起点。真正决定其能否上线运行的关键，在于它能不能扛住几十甚至上百人同时提问——尤其是在内部知识库、客服助手这类高频交互场景中。

Langchain-Chatchat 作为当前最受欢迎的本地化 RAG（检索增强生成）框架之一，凭借对私有文档的支持和完全离线部署的能力，成了许多组织构建智能问答系统的首选。但随之而来的问题是：当你把 PDF、Word 全导入后，面对突发的多人访问，系统会不会卡死？响应时间会不会从 2 秒飙升到 30 秒？

这时候，光靠手动点几下前端界面已经远远不够了。你需要一套自动化、可量化、能模拟真实负载的压力测试方案。而Locust，正是解决这个问题的最佳工具之一。

为什么选择 Locust 来压测 Langchain-Chatchat？

市面上有不少性能测试工具，比如 JMeter、k6 或 Gatling，但为什么我们推荐使用 Locust？

因为它足够“像程序员”。

传统的压测工具大多依赖图形界面或 XML/JSON 配置文件来定义请求流程，一旦逻辑复杂（比如需要动态构造 payload、处理鉴权 token、校验返回内容），就会变得难以维护。而 Locust 直接让你用 Python 写测试脚本——这意味着你可以：

灵活控制请求参数
动态生成问题文本
校验模型返回是否合理
捕获异常并自定义失败条件
轻松集成进 CI/CD 流水线

更重要的是，Locust 基于 gevent 实现协程并发，资源消耗极低。一台普通笔记本就能轻松模拟上千并发用户，非常适合本地部署的 LLM 服务压测。

如何编写一个有效的 Locust 脚本？

下面是一个针对 Langchain-Chatchat/chat接口的典型压测脚本示例：

# locustfile.py from locust import HttpUser, task, between import json import random class ChatbotUser(HttpUser): wait_time = between(1, 3) # 模拟用户思考间隔 questions = [ "什么是Langchain-Chatchat？", "如何配置向量数据库？", "支持哪些大语言模型？", "怎么上传PDF文件？", "RAG的工作原理是什么？" ] def on_start(self): """用户启动时可执行初始化操作""" pass @task def query_knowledge_base(self): """模拟向知识库发起查询""" payload = { "query": random.choice(self.questions), "history": [], "stream": False } headers = {'Content-Type': 'application/json'} with self.client.post( "/chat", data=json.dumps(payload), headers=headers, catch_response=True ) as response: if response.status_code != 200: response.failure(f"HTTP {response.status_code}") return try: result = response.json() if "result" not in result or not result["result"].strip(): response.failure("返回结果为空或缺少 'result' 字段") except Exception as e: response.failure(f"解析JSON失败: {e}")

关键设计说明：

wait_time = between(1, 3)：模拟真实用户的操作节奏，避免瞬间洪峰造成误判。
随机问题池（questions列表）：防止缓存命中率过高导致测试失真。
catch_response=True+ 手动 failure 控制：不仅看 HTTP 状态码，还要验证业务层面的响应质量。例如，即使接口返回 200，但如果答案为空，也应标记为失败。
无 history：初始阶段建议关闭多轮对话，聚焦单次问答性能；后续可扩展测试带历史上下文的场景。

这个脚本能精准反映系统在持续负载下的表现，而不是仅仅“打个接口看看通不通”。

启动压测：两种模式任选

方式一：Web UI 模式（适合调试）

locust -f locustfile.py --host http://localhost:8080

执行后打开http://localhost:8089，你会看到一个简洁的控制台：

设置总用户数（Users to spawn）
设置每秒新增用户数（Spawn rate）
实时查看请求数、响应时间分布、失败率等指标

这种模式特别适合开发阶段反复调整参数、观察趋势。

方式二：命令行无头模式（适合自动化）

locust -f locustfile.py \ --host http://localhost:8080 \ --users 100 \ --spawn-rate 10 \ --run-time 5m \ --headless \ --csv=results

参数解释：
---users 100：最终并发用户数
---spawn-rate 10：每秒启动 10 个新用户，实现渐进加压
---run-time 5m：运行 5 分钟自动停止
---headless：不启用 Web 界面，适用于服务器环境或 CI 流程
---csv=results：输出results_stats.csv等报告文件，便于后续分析

这种方式可以直接嵌入 Jenkins、GitHub Actions 等自动化流程中，实现每日性能基线检测。

Langchain-Chatchat 的性能瓶颈在哪里？

很多人以为压测只是“看看能撑多少人”，其实更关键的是定位瓶颈环节。通过结合 Locust 输出与服务端日志，我们可以清晰识别出以下常见瓶颈点：

1. LLM 推理延迟（最常见）

当并发增加时，平均响应时间显著上升，但 CPU 使用率不高，GPU 显存占满 → 说明瓶颈在模型推理。

✅优化方向：
- 升级 GPU 或启用 TensorRT 加速
- 使用更轻量模型（如 Qwen2-7B 替代 70B）
- 开启批处理推理（batching），合并多个请求一起送入模型

小贴士：ChatGLM3 在 batch_size=4 时吞吐量可提升近 3 倍，但需注意显存是否够用。

2. 向量检索变慢

随着知识库增大，Top-K 检索耗时从几十毫秒升至数秒 → 影响整体响应速度。

✅优化方向：
- 使用高效索引结构（如 FAISS 的 IVF-PQ 或 HNSW）
- 降低 embedding 维度（从 1024 维压缩到 768）
- 对文档预分块并建立关键词索引，先过滤再检索

3. 内存溢出或 OOM Killer 触发

系统突然中断，dmesg 显示内存不足 → 多发生在大文档加载或高并发场景。

✅优化方向：
- 控制文本切片大小（建议 256~512 token）
- 启用磁盘缓存而非全内存加载
- 设置合理的超时与连接上限（FastAPI + Uvicorn 可配置 workers 和 timeout）

4. FastAPI 线程阻塞

尽管 FastAPI 是异步框架，但 LangChain 中某些组件（如旧版 embedding 调用）仍是同步阻塞的，导致无法充分利用并发能力。

✅优化方向：
- 将耗时操作包装成async函数
- 使用run_in_threadpool避免事件循环卡顿
- 检查是否有全局锁（如 shared model instance）

架构视角下的压测实践

典型的压测环境架构如下：

+------------------+ +----------------------------+ | | | | | Locust Client |<----->| Langchain-Chatchat Server | | (Load Generator) | HTTP | - FastAPI Backend | | | | - Vector DB (e.g., FAISS) | | | | - LLM (e.g., ChatGLM3) | +------------------+ +----------------------------+ ↑ Local Network / LAN

几点关键建议：

Locust 客户端尽量不在同一台机器运行，避免资源竞争影响测试结果。
确保网络稳定，最好在同一局域网内测试，排除公网抖动干扰。
服务端开启详细日志（如DEBUG级别），记录每个阶段耗时（文档加载 → 向量化 → 检索 → LLM 输入构造 → 推理 → 返回）。
监控硬件状态：使用nvidia-smi查看 GPU 利用率，htop观察 CPU 和内存，iotop检查磁盘 IO。

更贴近真实的测试策略

要让压测数据真正有参考价值，就不能只跑“理想情况”。以下是几个提升测试代表性的技巧：

✅ 渐进式加压（Ramp-up）

不要一开始就拉满 100 并发。应该像这样逐步增加：

--users 100 --spawn-rate 5

每秒加 5 个用户，观察系统在不同负载层级的表现。你会发现很多系统在 30 并发时尚可，到了 50 就开始积压请求。

✅ 混合任务权重（Multi-task Simulation）

现实中用户行为多样。可以定义多个任务，并设置不同权重：

@task(3) def short_query(self): # 短问题，占比高 payload = {"query": "登录密码怎么改？", ...} @task(1) def long_query_with_history(self): # 长问题+多轮对话，较耗资源 payload = { "query": "根据前面提到的技术路线，下一步我们应该怎么做风险评估？", "history": [...], ... }

这样更能模拟真实流量分布。

✅ 引入错误容忍机制

有些请求失败是正常的（如网络抖动）。可以在脚本中设置重试逻辑或允许一定比例的失败率，避免因个别波动误判系统崩溃。

数据之外的价值：建立性能基线

一次成功的压测，不只是出一份报告那么简单。它的真正价值在于帮助团队建立起性能基线（Performance Baseline）。

你可以定期执行以下动作：

场景	压测目标
新增大量文档后	检查检索延迟是否恶化
更换 embedding 模型	对比响应时间和准确率变化
升级 LLM 版本	验证吞吐量是否提升
调整文本分块策略	观察召回率与速度平衡

把这些数据记录下来，形成一张“性能演化图”，就能清楚地回答一个问题：这次改动，到底是变好了还是变差了？