Dify平台压测：Locust实现流式接口性能测试

Dify平台压测：Locust实现流式接口性能测试

在AI应用从原型走向生产落地的过程中，性能从来不是最后才考虑的问题。尤其当系统需要支撑成百上千的并发用户时，一个看似流畅的对话流程，可能在真实流量冲击下暴露出响应延迟、资源争抢甚至服务崩溃等隐患。

Dify作为一款可视化AI Agent与应用开发平台，让开发者无需深入代码即可编排复杂的RAG系统、智能体和文本生成逻辑。这种“低门槛”特性吸引了大量企业用于构建智能客服、知识问答、自动化内容生成等高并发场景。但正因如此，它的后端稳定性更需经受住严苛考验——毕竟，谁愿意自己的客服机器人在客户咨询高峰期卡成PPT？

为了验证Dify在真实业务负载下的表现，我们开展了一次完整的压力测试。目标很明确：
- 验证8核16G最小资源配置下的最大吞吐能力（TPS）
- 探索16核32G条件下的性能上限
- 构建一套可复用的流式接口压测方案，覆盖/chat-messages等关键SSE接口
- 找出瓶颈所在，并给出切实可行的调优建议

最终，我们不仅跑出了接近百TPS的稳定输出，还总结出一套适用于LLM应用的压力测试方法论。

为什么选择 Locust？

面对市面上琳琅满目的压测工具，我们曾犹豫过JMeter、k6、wrk……但最终选择了Locust。原因很简单：它能原生支持流式响应。

Locust的优势在于：

• 用Python写脚本，灵活度极高：可以动态构造请求头、参数，甚至模拟真实用户的等待行为。
• 原生支持stream=True：轻松处理SSE分块数据，逐帧解析token返回过程。
• Web UI实时监控：RPS、延迟分布、失败率一目了然，适合快速迭代调参。
• 分布式架构友好：未来若需更大规模压测，可通过Master-Worker模式横向扩展。

更重要的是，我们可以借助events.fire()自定义上报指标，比如TTFB（首字节时间）和完整响应时间，在控制台中独立观察它们的变化趋势。

场景设计：不只是“你好，世界”

压测不能只测最简单的路径。我们围绕Dify三大典型业务场景构建测试矩阵：

简单ChatFlow：基础API层性能基线

仅包含“开始”与“直接回复”节点，固定输出一句话。这是评估核心API处理开销的理想起点，排除复杂逻辑干扰，专注测量框架本身的调度延迟。

复杂ChatFlow：真实Agent工作流模拟

包含条件判断、函数调用、插件执行（如天气查询）、向量检索等多个环节。所有外部依赖启用Mock模式，避免网络波动影响结果。这个场景用来检验多步骤任务调度的能力，尤其是上下文管理与异步协调机制。

文件召回场景：知识库检索性能探测

通过/datasets/{id}/retrieve接口发起混合搜索（hybrid_search），测试Weaviate向量数据库与全文检索组件的协同效率。上传一份PDF文档（含文字与图片），查询语义相关性较高的片段。此场景重点关注I/O性能与索引结构对响应速度的影响。

脚本实战：如何捕获流式响应的关键指标

以下是我们使用的完整Locust脚本，已整合三个主要场景：

from locust import HttpUser, TaskSet, task, between, events import time import json class SimpleChatFlow(TaskSet): @task def chat_stream(self): url = "/v1/chat-messages" headers = { "Authorization": "Bearer app-KqXrYsNtLmJpZaWcVbQeTdUo", "Content-Type": "application/json" } payload = { "inputs": {}, "query": "请简单介绍一下你自己", "response_mode": "streaming", "user": "load_test_user_001" } start_time = time.time() try: with self.client.post( url, json=payload, headers=headers, stream=True, catch_response=True, timeout=60 ) as resp: if resp.status_code != 200: resp.failure(f"Status {resp.status_code}") return ttfb = time.time() - start_time print(f"[Simple] TTFB: {ttfb:.2f}s") chunk_count = 0 for line in resp.iter_lines(decode_unicode=True): if line.startswith("data:"): chunk_count += 1 if chunk_count == 1: events.fire( "request", name="TTFB", response_time=int(ttfb * 1000), response_length=0, exception=None ) total_time = time.time() - start_time print(f"[Simple] Chunk #{chunk_count}: {line[:50]}... ({total_time:.2f}s)") if '"message_end"' in line: total_time_ms = int((time.time() - start_time) * 1000) events.fire( "request", name="Full Response", response_time=total_time_ms, response_length=len(line), exception=None ) break except Exception as e: print(f"[Simple] Request failed: {e}") class ComplexChatFlow(TaskSet): @task def complex_chat(self): url = "/v1/chat-messages" headers = { "Authorization": "Bearer app-ZxWnMpLkQjHgTrFeSaNdVcBi", "Content-Type": "application/json" } payload = { "inputs": {"dept": "IT"}, "query": "我需要申请一台新电脑，请指导我完成流程", "response_mode": "streaming", "user": "load_test_user_002" } start_time = time.time() try: with self.client.post( url, json=payload, headers=headers, stream=True, catch_response=True, timeout=120 ) as resp: if resp.status_code != 200: resp.failure(f"Status {resp.status_code}") return ttfb = time.time() - start_time print(f"[Complex] TTFB: {ttfb:.2f}s") for line in resp.iter_lines(decode_unicode=True): if line.startswith("data:"): data_str = line[5:].strip() if data_str == "[DONE]": continue try: data_json = json.loads(data_str) event = data_json.get("event") if event == "message_token": print(f"[Complex] Streaming token received.") elif event == "agent_thought": print(f"[Complex] Agent is thinking...") except: pass if '"message_end"' in line: total_time = time.time() - start_time print(f"[Complex] Full response time: {total_time:.2f}s") break except Exception as e: print(f"[Complex] Request failed: {e}") class FileRetrieval(TaskSet): @task def retrieve_file(self): url = "/v1/datasets/da0bcf35-abc5-4c77-8e2b-4e890b93b61c/retrieve" headers = { "Authorization": "Bearer dataset-pqrDBWoy9UILq7zbHnCkN3dY", "Content-Type": "application/json" } payload = { "query": "流程审批是什么？如果有图片，请一起返回", "retrieval_model": { "search_method": "hybrid_search", "reranking_enable": False, "score_threshold_enabled": False } } start_time = time.time() try: resp = self.client.post(url, json=payload, headers=headers, timeout=30) if resp.status_code == 200: ttfb = time.time() - start_time print(f"[Retrieve] Success, TTFB: {ttfb:.2f}s") else: print(f"[Retrieve] Failed: {resp.status_code}, {resp.text}") except Exception as e: print(f"[Retrieve] Request failed: {e}") class ChatUser(HttpUser): tasks = [SimpleChatFlow, ComplexChatFlow] wait_time = between(1, 3) class RetrieveUser(HttpUser): tasks = [FileRetrieval] wait_time = between(2, 5)

📝 关键点说明：
- 使用events.fire()将TTFB和完整响应时间单独上报，便于在Web UI中对比分析。
-iter_lines(decode_unicode=True)实现逐行解析SSE流，精准捕捉第一个token到达时刻。
- 支持多个用户类并行运行不同场景任务，贴近真实混合负载。

压测执行：从50到150并发的演进

启动命令如下：

locust -f locustfile.py --host=http://your-dify-api-host.com

访问http://localhost:8089进入控制台，设置参数：

• User Count: 依次测试 50 / 100 / 150 并发用户
• Spawn Rate: 每秒启动5个用户，避免瞬间冲击
• User Class: 切换为ChatUser或RetrieveUser
• 持续时间: 每轮压测运行5分钟，确保进入稳态

同时开启Prometheus + Grafana监控集群资源使用情况，重点关注：

这些数据帮助我们在性能下降时快速定位瓶颈。

8核16G调优之路：一步步逼近极限

测试环境部署于Kubernetes集群，节点规格为8C16G，存储采用NFS共享盘。

第一次压测：初步暴露瓶颈

初始资源配置较为保守：

压测100并发，结果如下：

问题很明显：dify-apiCPU利用率一度飙至98%，成为首要瓶颈。

第二次压测：提升API层资源

将dify-api提升至1.2核，其他不变。

结果：RPS上升至30.8，提升约43%。证实计算资源确实不足。

第三次压测：双核+双Worker进程

进一步将dify-api设为2核2GB，并设置SERVER_WORKER_AMOUNT=2，启用多进程处理。

结果：RPS跃升至49.5，几乎翻倍。但再往上增长乏力。

第四次压测：优化PostgreSQL配置

尝试调整数据库参数：

shared_buffers = 512MB work_mem = 32MB max_connections = 200 checkpoint_timeout = 10min log_min_duration_statement = 500ms

效果微弱，RPS仅小幅提升至54.1。怀疑I/O受限。

第五次压测：升级数据库硬件

将dify-postgres改用本地SSD挂载，资源提升至1核2GB。

意外发生了：RPS反而跌至48.4。查看pg_stat_activity发现大量事务锁等待。

原来是高频短事务导致连接竞争加剧，原有连接池策略已不适用。

第六次压测：引入PgBouncer连接池

部署PgBouncer，配置：

pool_mode = transaction max_client_conn = 150 default_pool_size = 50

再次压测，性能趋于稳定，确认已达当前资源配置下的极限。

8核16G最优配置与实测TPS

经过六轮迭代，得出该配置下的最佳实践：

实测性能汇总

简单ChatFlow场景

• TPS：48.4
• 平均响应时间：1,437 ms
• 95%响应时间：3,900 ms

复杂ChatFlow场景

• TPS：20.7
• 平均响应时间：5,689 ms
• 主要耗时集中在插件调用与向量检索

文件召回场景

• TPS：17.6
• 平均响应时间：5,287 ms
• Weaviate段合并频繁，影响查询效率

💡 当前瓶颈主要在dify-api计算能力和NFS存储I/O性能。

16核32G探索：突破百TPS大关

为进一步挖掘潜力，我们将节点升级至16核32G，开启新一轮调优。

第一次压测：盲目扩容反致性能下滑

将dify-api提升至4核4GB，SERVER_WORKER_AMOUNT=4，却发现RPS降至22.9。

排查发现dify-plugin-daemon出现大量慢SQL，拖累整体表现。

第二次压测：强化数据库性能

将PostgreSQL升级至2核4GB，优化work_mem=64MB、effective_cache_size=8GB。

RPS回升至91.3，说明数据库已成为新瓶颈。

第三~六次压测：实例数与资源配置博弈

我们尝试多种组合：
- 多实例 vs 高配单实例
- Worker数量调整
- Weaviate独立部署

最终确定最优方案：

• dify-api: 3实例 × (2核 / 2GB)，每实例SERVER_WORKER_AMOUNT=2
• dify-postgres: 2核 / 3GB
• dify-weaviate: 1核 / 2GB（独立部署）
• xinference: 1核 / 2GB

结果令人振奋：

🎉 成功突破百TPS！

16核32G最终推荐配置

剩余资源可用于部署Redis缓存或边缘网关。

更高TPS的进阶优化建议

若希望进一步提升吞吐能力，可考虑以下方向：

1. 横向扩展API层：结合负载均衡实现水平扩容，彻底突破单机限制。
2. 数据库独立部署：将PostgreSQL与Weaviate迁移到专用高性能实例，避免资源争抢。
3. 引入Redis缓存：对高频知识库查询结果进行缓存，降低数据库压力。
4. 优化Weaviate HNSW索引参数：调整ef,max_connections提升检索效率。
5. 异步化长任务：将复杂Agent流程拆分为后台任务，前端仅返回任务ID，提高接口响应速度。
6. CDN或边缘计算前置：对于静态资源或轻量查询，可在边缘节点缓存响应，减少核心服务负担。

⚠️ 注意：以上测试未包含实际LLM推理耗时。若接入Qwen、Llama3等模型，需根据显存需求合理分配xinference或ollama资源，否则将成为新的性能黑洞。

Dify的价值在于“可视化编排”，但这并不意味着我们可以忽视底层性能工程。相反，正是因为它封装了太多细节，才更需要我们主动去揭开盖子，看清每一毫秒的消耗来自何处。

本次压测不仅是对Dify的一次体检，更是为所有LLM应用开发者提供了一份通用的方法论：
- 用Locust做流式接口测试是完全可行且高效的；
- 性能优化是一个渐进过程，必须结合监控数据持续迭代；
- 架构设计比参数调优更重要——合理的服务拆分与资源隔离，往往比单纯加CPU更有意义。

如果你正在基于Dify构建企业级AI应用，不妨从一次系统的压测开始。别等到上线那天才发现，“低代码”背后藏着“高风险”。

👉 后续我们将继续分享《Dify高可用部署架构》《多租户性能隔离实践》《成本-性能平衡模型》等内容，欢迎关注。