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 版本	验证吞吐量是否提升
调整文本分块策略	观察召回率与速度平衡

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

---

结语：让 AI 系统从“能用”走向“可靠”

Langchain-Chatchat 让我们能够快速搭建一个功能完整的本地知识库问答系统，但这只是第一步。真正的挑战在于让它在真实业务场景中稳定运行。

通过 Locust 进行科学的压力测试，不仅能提前暴露潜在风险，还能为资源采购、架构优化和上线决策提供强有力的数据支撑。它不是“锦上添花”，而是保障 AI 工程落地的必要环节。

未来，随着更多企业将 AI 深度融入工作流，谁能更好地掌握性能调优与稳定性保障的能力，谁就能真正实现从“技术可用”到“体验可信”的跨越。

而这套基于代码的压测方法，正是通往这一目标的重要一步。