Dify SSE 工作流压测实战：基于 Locust 的流式性能基准测试套件

Dify SSE 工作流压测实战基于 Locust 的流式性能基准测试套件【免费下载链接】difyBuild Agentic workflows, RAG pipelines, with rich AI model and tool support on one collaborative workspace. Deploy on cloud, VPC, or self-hosted, so teams move from prototype to production without rebuilding the stack.项目地址: https://gitcode.com/GitHub_Trending/di/dify本文以 Dify 仓库自带的压测套件scripts/stress-test/为主体完整讲解如何用Locust对 Dify 工作流的 Server-Sent EventsSSE流式执行接口/v1/workflows/run做性能基准测试。读完你能掌握一键完成环境预置管理员、插件、工作流、API Key的全流程、四个核心 SSE 指标活跃连接数、建连速率、TTFE、事件吞吐的含义与判定标准以及如何用 Gunicorn、PostgreSQL 与系统参数对服务端做针对性调优。一、为什么专门测 SSE 流式性能Dify 的工作流执行接口在response_modestreaming下返回的不是一个一次性 JSON而是一条持续的 Server-Sent Events 事件流。压测这类接口和普通 REST 接口有本质区别连接在收到最后一个事件前不能关闭性能瓶颈更多体现在首事件延迟和单位时间事件投递速率上而不是单纯的请求-响应往返时间。因此sse_benchmark.py没有简单复用 Locust 的默认统计而是自己维护了一套面向流的指标采集器MetricsTracker并实现了符合 W3C 规范的SSEParser。整个套件的四个核心观测指标为Active SSE Connections活跃 SSE 连接数——任意时刻仍处于打开状态的 SSE 连接数量。New Connection Rate建连速率conn/sec——每秒新建的 SSE 连接数。Time to First Event (TTFE)首事件时间——从发出请求到收到第一条 SSE 事件的延迟。Event Throughput事件吞吐events/sec——所有连接上每秒投递的 SSE 事件数。说明普通 Locust 统计里的req/s、Avg/Min/Max/Med仍然会保留但它们对 SSE 场景的参考价值有限真正的判定要看下面这套流式指标。二、被测端点与请求形态压测集中打的是单一端点/v1/workflows/run这一定义在 sse_benchmark.py 顶部通过环境变量给出并附带若干可调项WORKFLOW_PATH os.getenv(WORKFLOW_PATH, /v1/workflows/run) CONNECT_TIMEOUT float(os.getenv(CONNECT_TIMEOUT, 10)) READ_TIMEOUT float(os.getenv(READ_TIMEOUT, 60)) TERMINAL_EVENTS [e.strip() for e in os.getenv(TERMINAL_EVENTS, workflow_finished,error).split(,) if e.strip()] QUESTIONS_FILE os.getenv(QUESTIONS_FILE, )WORKFLOW_PATH默认为/v1/workflows/run。CONNECT_TIMEOUT/READ_TIMEOUT分别是连接与读取超时秒默认 10s / 60s。TERMINAL_EVENTS是判定一条流正常结束的终止事件集合默认workflow_finished,error。QUESTIONS_FILE允许从外部文件加载自定义问题池缺省时使用代码内置的默认问题。每个虚拟用户DifyWorkflowUser发起的请求体形如下见 sse_benchmark.py 的test_workflow_stream任务headers { Authorization: fBearer {self.api_token}, Content-Type: application/json, Accept: text/event-stream, Cache-Control: no-cache, } data WorkflowRequestData( inputsWorkflowInputs(questionquestion), response_modestreaming, userfuser_{self.user_counter}, )要点认证走Bearer api_token这个 token 由前置 setup 流程创建并写进状态文件下文详述。response_modestreaming触发 SSE 流式响应。user字段用一个递增计数器保证每用户不同便于在服务端日志里区分。请求通过self.client.request(..., streamTrue, catch_responseTrue)发出streamTrue是关键——它让 Locust 不一次性读满响应而是按行迭代从而真实模拟流式消费。被测的工作流本身非常简单是一个Start → LLM → End的三节点 DSL见 workflow_llm.ymlLLM 节点使用gpt-4oprovider 为langgenius/openai/openaiprompt 直接引用开始节点的question变量。正因为工作流足够简单压出来的指标主要反映平台与基础设施的流式承载能力而非复杂编排逻辑。三、环境预置setup_all.py 一键装好依赖压测前置依赖一个能真正跑通流式响应的 Dify 实例。setup_all.py 负责把这件事自动化其执行顺序见 setup_all.py为login_admin.py - 登录拿到 access token install_openai_plugin.py - 安装 OpenAI 插件 configure_openai_plugin.py- 用 Mock 服务器配置 OpenAI 插件 import_workflow_app.py - 从 DSL 导入工作流应用 create_api_key.py - 为应用创建 API Key publish_workflow.py - 发布工作流如果检测到/console/api/setup返回的step不是finished即全新实例会在最前面插入setup_admin.py创建首个管理员账号。3.1 管理员账号的两种情形全新实例会用默认值创建第一个管理员可以通过环境变量覆盖见 setup_all.py 的build_admin_config默认testdify.ai/dify/password123STRESS_TEST_ADMIN_EMAILyour-adminexample.com \ STRESS_TEST_ADMIN_USERNAMEdify \ STRESS_TEST_ADMIN_PASSWORDyour-password \ python scripts/stress-test/setup_all.py对于已初始化、已有管理员账号的实例只需提供现有账号登录信息STRESS_TEST_ADMIN_EMAILyour-adminexample.com \ STRESS_TEST_ADMIN_PASSWORDyour-password \ python scripts/stress-test/setup_all.pySTRESS_TEST_ADMIN_USERNAME仅在全新实例走/console/api/setup创建首个管理员时才会用到。3.2 状态文件 stress_test_state.jsonsetup 各步骤的产物统一写入 common/config_helper.py 管理的状态文件setup/config/stress_test_state.json。它包含四个分区admin、auth、app、api_key。其中压测真正要用的是api_key.token——run_locust_stress_test.sh会读取它来做可用性校验见 run_locust_stress_test.sh而sse_benchmark.py里每个用户通过ConfigHelper().get_api_key()取出见 sse_benchmark.pyconfig_helper ConfigHelper() self.api_token config_helper.get_api_key() if not self.api_token: raise ValueError(API key not found. Please run setup_all.py first.)问题池的加载也在这里若指定了QUESTIONS_FILE且文件存在则逐行读取非空行否则回退到内置的 5 个默认问题见 sse_benchmark.py。3.3 Mock OpenAI 服务器为了让压测不依赖真实 OpenAI 配额与网络套件自带一个 Mock 服务器 mock_openai_server.py。它监听http://localhost:5004提供与 OpenAI 兼容的端点GET /v1/models POST /v1/chat/completions POST /v1/completions POST /v1/embeddings GET /v1/models/model_id GET /health其中/v1/chat/completions在streamTrue时会按 OpenAI 的 chunk 格式逐词吐出data: {json}\n\n每个词之间time.sleep(0.05)模拟真实流式延迟最后以data: [DONE]\n\n收尾见 mock_openai_server.py。这套可控、可复现、无外部依赖的 mock 是压测结果可比性的关键。四、服务端启动必须用 Gunicorn 生产模式压测结果的准确性高度依赖服务端的启动方式。README 明确强调不要用 Flask debug 模式而要用 Gunicorn gevent worker 生产模式README 中亦被 run_locust_stress_test.sh 检测到 werkzeug/flask 监听 5001 端口时主动告警拦截。# Run from the api directory cd api uv run gunicorn \ --bind 0.0.0.0:5001 \ --workers 4 \ --worker-class gevent \ --timeout 120 \ --keep-alive 5 \ --log-level info \ --access-logfile - \ --error-logfile - \ app:app各参数含义README 给出的解释--workers 4worker 进程数按 CPU 核心数调整。--worker-class gevent异步 worker用于处理并发连接。--timeout 120长耗时请求的 worker 超时。--keep-alive 5保持连接存活以支撑 SSE 流式。这里有一个值得注意的实现细节Dify 的 Gunicorn 配置 api/gunicorn.conf.py 会在 gevent worker 里做 monkey-patching把psycopg2psycogreen和 gRPC 一并 patch 成 gevent 协程从而让数据库调用与 gRPC 调用都不阻塞事件循环。这正是高并发下用 gevent worker 能扛住 SSE 长连接的底层原因——从源码结构看若换成 sync worker每条 SSE 流都会独占一个线程并发承载会显著下降。不推荐用于压测的方式# Debug mode - DO NOT use for stress testing (slow performance) ./dev/start-api # 运行 Flask debug 单线程模式Mock 服务器同样需要启动python scripts/stress-test/setup/mock_openai_server.py五、运行压测5.1 推荐方式封装脚本# 默认配置headless ./scripts/stress-test/run_locust_stress_test.sh # 直接 uvx 运行 uvx --from locust locust -f scripts/stress-test/sse_benchmark.py --host http://localhost:5001 # 带 Web UI访问 http://localhost:8089 uvx --from locust locust -f scripts/stress-test/sse_benchmark.py --host http://localhost:5001 --web-port 8089run_locust_stress_test.sh 会自动完成四件事校验 Dify APIhttp://localhost:5001/health与 Mock 服务器http://localhost:5004/v1/models在运行并在检测到 debug 模式时告警从stress_test_state.json读取并校验 API token交互式选择 headless 或 Web UI 模式然后用uvx --from locust locust执行 sse_benchmark.py在reports/YYYYMMDD_HHMMSS/目录生成报告并回显关键指标。注意Locust 通过uvx --from locust运行不装在 API 项目环境里README 的 Troubleshooting 也据此解释ModuleNotFoundError: No module named locust并非问题而sseclient-py属于 API 项目依赖。5.2 配置文件 locust.conf压测参数集中在 locust.confhost http://localhost:5001 # 目标地址 users 10 # 并发用户数 spawn-rate 2 # 每秒生成用户数 run-time 1m # 测试时长30s / 5m / 1h locustfile scripts/stress-test/sse_benchmark.py headless true # 无 Web UI print-stats true loglevel INFO # csv reports/locust_results # 取消注释启用 CSV # html reports/locust_report.html # 取消注释启用 HTML 报告脚本会用grep解析其中的users、spawn-rate、run-time再拼进--users --spawn-rate --run-time命令行参数。5.3 自定义问题池直接改sse_benchmark.py里的self.questionsself.questions [ Your custom question 1, Your custom question 2, # Add more questions... ]或者更推荐的方式准备一个每行一个问题的文本文件然后用环境变量QUESTIONS_FILE/path/to/questions.txt运行sse_benchmark.py会优先读取该文件见 sse_benchmark.py无需改动源码。5.4 高级用法# 指定用户数与生成速率 uvx --from locust locust -f scripts/stress-test/sse_benchmark.py \ --host http://localhost:5001 --users 50 --spawn-rate 5 # 生成 CSV 报告 uvx --from locust locust -f scripts/stress-test/sse_benchmark.py \ --host http://localhost:5001 --csv reports/results # 固定运行时长 headless uvx --from locust locust -f scripts/stress-test/sse_benchmark.py \ --host http://localhost:5001 --run-time 5m --headless多次迭代对比for i in {1..3}; do echo Run $i of 3 ./scripts/stress-test/run_locust_stress_test.sh sleep 60 done六、报告结构与指标解读6.1 报告目录每次运行在reports/下生成一个按时间戳命名的子目录YYYYMMDD_HHMMSS/locust_summary.txt—— 完整控制台输出含指标YYYYMMDD_HHMMSS/locust_report.html—— 带图表的交互式 HTML 报告YYYYMMDD_HHMMSS/locust_stats.csv—— 详细统计 CSVYYYYMMDD_HHMMSS/locust_stats_history.csv—— 时序数据YYYYMMDD_HHMMSS/sse_metrics_YYYYMMDD_HHMMSS.json—— 自定义 SSE 指标机器可读其中 JSON 报告由on_test_stop钩子调用export_json_report写出见 sse_benchmark.py结构为{ timestamp, duration_seconds, metrics, locust_stats }metrics即MetricsSnapshot的全部字段方便 CI 做回归分析。6.2 核心指标与健康阈值指标含义健康参考值Active Connections任意时刻打开的 SSE 连接数负载下应保持稳定、不塌落Connection Rate (conn/sec)每秒新建连接数轻载 5–10中载 20–50重载 100TTFE (ms)首事件延迟优秀 50ms良好 50–100可接受 100–500差 500Event Throughput (events/sec)全连接每秒事件数单连接 10–2010 连接 50–100100 连接 200–500RPS每秒请求数优秀 50良好 20–50可接受 10–20待改进 10分位数P50/P95/P99含义P50 表示 50% 请求在该时间内完成以此类推。成功率方面生产就绪应 99%偏低通常意味着错误或超时。6.3 示例输出 DIFY SSE STRESS TEST [2025-09-12 15:45:44,468] Starting test run with 10 users at 2 users/sec SSE Metrics | Active: 8 | Total Conn: 142 | Events: 2841 Rates: 2.4 conn/s | 47.3 events/s | TTFE: 43ms Type Name # reqs # fails | Avg Min Max Med | req/s failures/s POST /v1/workflows/run 142 0(0.00%) | 41 18 192 38 | 2.37 0.00 Aggregated 142 0(0.00%) | 41 18 192 38 | 2.37 0.00 FINAL RESULTS Total Connections: 142 Total Events: 2841 Average TTFE: 43 ms 实时指标框每 5 秒刷新一次on_test_start里report_stats线程time.sleep(5)见 sse_benchmark.py。各字段含义Active当前打开的 SSE 连接数Total Conn累计建立连接数Events累计收到事件数。conn/s建连速率events/s事件投递速率TTFE平均首事件时间。注意 README 正文有两处口径一处说实时报告每 5 秒一次一处把实时框写作Updates every 10 seconds。从源码看刷新频率是 5 秒而10 秒指的是速率计算用的滑动窗口长度MetricsTracker.get_stats中time_window 10.0见 sse_benchmark.py。也就是说指标每 5 秒重算一次但 conn/s、events/s 是基于最近 10 秒窗口内的事件数除窗口时长得到的速率。6.4 结果判读良好表现零失败0.00%TTFE 100ms活跃连接稳定事件吞吐一致预警信号失败率 1%TTFE 500ms活跃连接下降事件速率随时间走低七、测试场景与性能调优7.1 四档负载场景# 轻载 concurrency: 10 iterations: 100 # 正常 concurrency: 100 iterations: 1000 # 重载 concurrency: 500 iterations: 5000 # 极限 concurrency: 1000 iterations: 100007.2 Gunicorn 按负载分档调参# 轻载10-50 并发 uv run gunicorn --bind 0.0.0.0:5001 --workers 2 --worker-class gevent app:app # 中载50-200 并发 uv run gunicorn --bind 0.0.0.0:5001 --workers 4 --worker-class gevent --worker-connections 1000 app:app # 重载200-1000 并发 uv run gunicorn --bind 0.0.0.0:5001 --workers 8 --worker-class gevent --worker-connections 2000 --max-requests 1000 app:appworker 数经验公式Workers (2 × CPU 核心数) 1SSE/WebSocket 场景用 gevent workerCPU 密集型任务用 sync worker。7.3 PostgreSQL 连接池高并发压测时可上调docker/middleware.env里的POSTGRES_MAX_CONNECTIONS默认 100# Edit docker/middleware.env POSTGRES_MAX_CONNECTIONS200 # 默认 100 # 分档参考 # 轻载10-50 用户: 100 # 中载50-200 用户: 200 # 重载200-1000 用户: 500改完重启数据库容器docker compose -f docker/docker-compose.middleware.yaml down db docker compose -f docker/docker-compose.middleware.yaml up -d db从仓库配置可以印证这个链路docker-compose.middleware.yaml 中 PostgreSQL 启动命令为postgres -c max_connections${POSTGRES_MAX_CONNECTIONS:-100}即最终传给postgres的max_connections就是该环境变量缺省 100docker/.env.example 则给了POSTGRES_MAX_CONNECTIONS200的示例值。内存占用经验每个连接约 10MB RAM。100 连接约 1GB、200 连接约 2GB、500 连接约 5GB需确保数据库服务器内存充足。7.4 系统层优化提高文件描述符上限ulimit -n 65536Linux TCP 调优sudo sysctl -w net.core.rmem_max134217728 sudo sysctl -w net.core.wmem_max134217728 sudo sysctl -w net.ipv4.tcp_fastopen3macOS 提高最大连接数sudo sysctl -w kern.ipc.somaxconn2048八、排障速查ModuleNotFoundError: No module named locustLocust 本就通过uvx --from locust在 API 项目环境之外运行属正常。可用uvx --from locust locust --version验证。API key configuration not found先跑python scripts/stress-test/setup_all.py生成stress_test_state.json。服务未运行按上文用 Gunicorn 启动 Dify API5001并启动 Mock 服务器5004。错误率偏高降低并发、检查 CPU/内存、查看 API 服务端日志、必要时增大超时。脚本无执行权限chmod x run_benchmark.sh实际入口是run_locust_stress_test.sh。性能问题定位响应时间高查数据库查询性能、外部 API 延迟、服务器资源、网络拥塞。吞吐低RPS 10查 CPU 瓶颈、内存约束、数据库连接池、API 限流。错误率高查服务端错误日志、资源耗尽、超时配置、连接数上限。九、为什么选 LocustREADME 给出选型理由相较 Drill正确的 SSE 支持能处理流式响应而不过早关闭连接自定义指标可跟踪 TTFE、流时长等 SSE 专属指标Web UI实时可视化监控与控制Python 集成与现有 Python setup 代码无缝衔接可扩展便于按具体测试场景定制。从 sse_benchmark.py 的SSEParser可以看出这种可扩展落地得很具体它按 W3C 规范逐行解析data/event/id字段空行判定一条事件结束:开头的行按注释忽略多行data会用换行拼接后再触发一次回调。正是这种边解析边统计的能力让 TTFE、事件间隔、流时长这些指标能在压测过程中被实时、逐事件地采集下来。十、如何改进这套压测套件配置类改动调整 locust.conf 的users/spawn-rate/run-time。流程改进修改 run_locust_stress_test.sh 的校验与报告逻辑。问题覆盖用QUESTIONS_FILE环境变量指向更大规模的问题池或直接改 sse_benchmark.py 的默认self.questions。指标扩展在 sse_benchmark.py 的MetricsTracker与on_test_start/on_test_stop钩子中新增采集项与导出字段。工作流调整更换 workflow_llm.yml 以压测不同复杂度的编排。适用前提与限制本套件面向本地自托管 Dify默认目标为http://localhost:5001依赖 Mock OpenAI 服务器提供可控的流式响应因此测得的是平台 基础设施在模拟 LLM 下的 SSE 承载能力而非接入真实模型服务时的端到端性能。指标健康阈值如 TTFE、RPS 分档是 README 给出的经验参考值实际目标仍应结合自身 SLA 与硬件规格校准。【免费下载链接】difyBuild Agentic workflows, RAG pipelines, with rich AI model and tool support on one collaborative workspace. Deploy on cloud, VPC, or self-hosted, so teams move from prototype to production without rebuilding the stack.项目地址: https://gitcode.com/GitHub_Trending/di/dify创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考