如何用Locust.io与Python测量Botpress消息从发送到展示的响应时间（避免重复调用List API）

核心思路 #

Botpress的listenConversation本质是通过WebSocket连接实现实时消息推送，我们可以用Python的WebSocket客户端库模拟该逻辑，既避免轮询List Message API带来的额外负载，又能精准测量从发送请求到Bot消息显示的完整耗时。

实现步骤 #

1. 安装依赖 #

pip install websocket-client locust requests

2. 完整Locust测试代码 #

from locust import HttpUser, task, between
import websocket
import json
import time
from threading import Thread

class BotpressPerformanceUser(HttpUser):
    wait_time = between(1, 3)
    
    def on_start(self):
        # 初始化对话，获取conversationId和userId（根据你的实际业务流程调整）
        init_response = self.client.post("/api/v1/conversations", json={
            "userId": "test-user-123",
            "botId": "your-bot-id"
        })
        self.conversation_id = init_response.json()["id"]
        self.user_id = "test-user-123"  # 与初始化时的userId保持一致
        self.bot_response_received = False
        self.bot_response_time = None

    def _listen_for_bot_response(self):
        # 构建WebSocket连接URL（替换为你的Botpress部署域名）
        ws_url = f"wss://your-botpress-domain.com/api/v1/chat/conversations/{self.conversation_id}/listen?userId={self.user_id}"
        
        def on_message(ws, message):
            event = json.loads(message)
            # 过滤出Bot发送的message_created事件
            if event["type"] == "message_created" and event["payload"]["userId"] != self.user_id:
                self.bot_response_time = time.time()
                self.bot_response_received = True
                ws.close()
        
        def on_error(ws, error):
            print(f"WebSocket error: {error}")
        
        def on_close(ws):
            pass
        
        ws = websocket.WebSocketApp(
            ws_url,
            on_message=on_message,
            on_error=on_error,
            on_close=on_close
        )
        ws.run_forever()

    @task
    def test_message_roundtrip(self):
        # 重置状态
        self.bot_response_received = False
        self.bot_response_time = None
        
        # 启动WebSocket监听线程，避免阻塞主线程的HTTP请求
        listen_thread = Thread(target=self._listen_for_bot_response)
        listen_thread.start()
        
        # 记录发送Create Message请求的时间
        start_time = time.time()
        
        # 发送消息请求
        self.client.post(f"/api/v1/conversations/{self.conversation_id}/messages", json={
            "userId": self.user_id,
            "payload": {
                "type": "text",
                "text": "Hello, bot!"
            }
        })
        
        # 等待Bot响应（设置超时，防止无响应时测试卡住）
        timeout = 10
        while not self.bot_response_received and timeout > 0:
            time.sleep(0.1)
            timeout -= 0.1
        
        if self.bot_response_time:
            roundtrip_time = self.bot_response_time - start_time
            print(f"从发消息到Bot消息显示耗时: {roundtrip_time:.3f}s")
            # 将自定义指标上报到Locust统计面板
            self.environment.events.request.fire(
                request_type="WEBSOCKET",
                name="message_roundtrip",
                response_time=roundtrip_time*1000,  # 转换为毫秒
                response_length=0,
                exception=None
            )
        else:
            print("等待Bot响应超时")

关键细节说明 #

• WebSocket URL构造：遵循Botpress API规则，URL需包含会话ID和用户ID，确保只接收目标对话的事件
• 事件过滤：通过userId判断排除自身发送的消息，仅监听Bot的响应事件
• Locust适配：用线程启动WebSocket监听，避免阻塞HTTP请求流程；通过自定义事件上报耗时，方便在Locust面板查看统计数据
• 超时处理：设置等待超时逻辑，防止因Bot无响应导致测试任务挂起

内容的提问来源于stack exchange，提问作者Hong Phi Nguyen