码力全开 / 在Rasa Pro应用中调试执行动作

Created Fri, 08 May 2026 21:13:00 +0800 Modified Fri, 08 May 2026 21:25:51 +0800
2163 Words 4 min

这里将介绍如何在Rasa Pro应用中调试其action,从而更好地知道程序是否正常运行。

假设我们有1个动作是询问天气,我们可以通过如下方式创建应用:

export RASA_LICENSE=xxxxx
rasa init

在创建的项目actions目录下创建1个ask_weather.py的模块,其内容如下:

import random
from typing import Any, Text, Dict, List
from rasa_sdk import Action, Tracker
from rasa_sdk.executor import CollectingDispatcher
import requests

class ActionWeatherForecast(Action):
    """自定义动作:查询指定城市的天气预报"""

    def name(self) -> Text:
        return "action_weather_forecast"

    def run(self, dispatcher: CollectingDispatcher,
            tracker: Tracker,
            domain: Dict[Text, Any]) -> List[Dict[Text, Any]]:
        # 1. 从tracker中获取槽位值(例如城市名称)
        # 假设在domain.yml中定义了名为 'city' 的slot
        city = tracker.get_slot("city")
        if not city:
            dispatcher.utter_message(text="请问您想查询哪个城市的天气?")
            return []
        # 2. 调用外部天气API (此处以模拟数据或真实API为例)
        # 注意:实际生产中请替换为真实的API Key和URL
        try:
            # 模拟返回数据
            weather_info = self._get_mock_weather(city)
            # 3. 构造回复消息
            temperature = weather_info.get('temperature', '未知')
            description = weather_info.get('description', '未知')
            response_text = f"{city}今天的天气是{description},气温约为{temperature}摄氏度。"
            # 4. 发送消息给用户
            dispatcher.utter_message(text=response_text)
        except Exception as e:
            dispatcher.utter_message(text=f"抱歉,查询{city}的天气时出错了,请稍后再试。")

        return []

    def _get_mock_weather(self, city: str) -> dict:
        """模拟获取天气数据,实际项目中应替换为真实的API调用"""
        if city == "北京":
            temperature = 26
            description = "阴天"
        elif city == "广州":
            temperature = 30
            description = "晴朗,紫外线很强"
        else:
            temperature = random.randint(15,40)
            description = random.choices(["晴朗","雨天","大雪","大雨","阴凉","大风","雾霾"])[0]
        # 简单的模拟逻辑
        return {
            "temperature": temperature,
            "description": description
        }

同样在domain目录下新建1个ask_weather.yml的配置文件,其内容如下:

version: "3.1"

intents:
  - ask_weather
  # ... 其他意图

entities:
  - city
  # ... 其他实体

slots:
  city:
    type: text
    influence_conversation: false
    mappings:
      - type: from_entity
        entity: city

responses:
  # ... 其他回复模板

actions:
  - action_weather_forecast  # 这里必须与代码中 name() 返回的值完全一致

之后修改项目目录下endpoints.yml中配置:

action_endpoint:
  url: "http://localhost:5055/webhook"
  actions_module: "actions"

其中url是调试的URL地址,actions_module是动作模块的目录。

接着执行如下的命令开启actions调试服务:

$ rasa run actions --auto-reload
2026-05-08 21:22:10 INFO     rasa.tracing.backend_tracing_config  - [info     ] No backend tracing configuration found in endpoints.yml. Supported backend tracing types are 'jaeger' and 'otlp'. Backend tracing will not be configured. event_key=endpoint.read.no_backend_tracing_config
2026-05-08 21:22:11 INFO     rasa_sdk.executor  - Registered function for 'add_contact'.
2026-05-08 21:22:11 INFO     rasa_sdk.executor  - Registered function for 'action_weather_forecast'.
2026-05-08 21:22:11 INFO     rasa_sdk.executor  - Registered function for 'list_contacts'.
2026-05-08 21:22:11 INFO     rasa_sdk.executor  - Registered function for 'remove_contact'.
2026-05-08 21:22:11 INFO     rasa_sdk.endpoint  - Starting action endpoint server...
2026-05-08 21:22:11 INFO     rasa_sdk.endpoint  - Starting plugins...
2026-05-08 21:22:11 INFO     rasa_sdk.endpoint  - Action endpoint is up and running on http://0.0.0.0:5055
2026-05-08 21:22:11 INFO     rasa_sdk.tracing.config  - No endpoint for tracing type available in endpoints.yml,tracing will not be configured.
2026-05-08 21:22:11 INFO     sanic.server  - Starting worker [19732]

可以看到成功注册了action_weather_forecast动作。

最后编写如下的脚本:

import requests
import json

def test_weather_action():
    """
    测试 Rasa Action Server 中的 action_weather_forecast 接口。
    """
    # 1. Action Server 的端点 URL (根据您的附件内容)
    action_server_url = "http://localhost:5055/webhook"

    # 2. 构建符合 OpenAPI 规范的请求体
    # 模拟 Rasa Core 发送的请求,其中包含对话状态 (tracker) 和要执行的动作名。
    payload = {
        "next_action": "action_weather_forecast",  # 要执行的自定义动作名称
        "sender_id": "test_user_001",              # 模拟用户ID
        "tracker": {
            "sender_id": "test_user_001",
            "slots": {
                "city": "上海"  # 假设用户已提供了城市信息,并已存入名为'city'的槽位
            },
            # "latest_message": {
            #     "intent": {"name": "ask_weather", "confidence": 0.99},
            #     "entities": [{"entity": "city", "value": "北京"}],
            #     "text": "北京天气怎么样?"
            # },
            # "latest_action_name": "action_listen",
            # "events": [],
            # "paused": False,
            # "followup_action": None,
            # "active_loop": {},
            # "latest_input_channel": "rest"
        },
        "domain": {
            # "config": {"store_entities_as_slots": True},
            # "intents": [{"ask_weather": {"use_entities": True}}],
            # "entities": ["city"],
            # "slots": {
            #     "city": {
            #         "type": "text",
            #         "auto_fill": True,
            #         "initial_value": None
            #     }
            # },
            # "responses": {},
            # "actions": ["action_weather_forecast", "utter_greet", "action_listen"]
        }
    }

    # 3. 设置请求头
    headers = {
        'Content-Type': 'application/json',
        'Accept': 'application/json'
    }

    try:
        # 4. 发送 POST 请求
        print(f"正在向 Action Server 发送测试请求: {action_server_url}")
        response = requests.post(action_server_url, data=json.dumps(payload), headers=headers, timeout=10)

        # 5. 处理响应
        print(f"响应状态码: {response.status_code}")
        if response.status_code == 200:
            result = response.json()
            print("✅ 动作执行成功!")
            print("响应内容:")
            print(json.dumps(result, indent=2, ensure_ascii=False))

            # 解析并打印机器人回复
            if 'responses' in result and result['responses']:
                for resp in result['responses']:
                    if 'text' in resp:
                        print(f"\n🤖 机器人回复: {resp['text']}")
            else:
                print("⚠️  响应中未包含直接回复文本,请检查动作返回的 `responses` 字段。")

        elif response.status_code == 400:
            error = response.json()
            print(f"❌ 动作执行被拒绝: {error.get('error', '未知错误')}")
        elif response.status_code == 500:
            print("❌ Action Server 内部执行出错,请检查动作代码日志。")
        else:
            print(f"❌ 收到意外状态码: {response.status_code}, 响应文本: {response.text}")

    except requests.exceptions.ConnectionError:
        print(f"❌ 无法连接到 Action Server,请确保已在运行 `rasa run actions` 并监听在 {action_server_url}")
    except requests.exceptions.Timeout:
        print("❌ 请求超时,请检查 Action Server 状态。")
    except Exception as e:
        print(f"❌ 测试过程中发生未知错误: {e}")

if __name__ == "__main__":
    # 运行测试前,请确保:
    # 1. 您的自定义动作代码已正确放置在 actions/ 目录下。
    # 2. 已通过 `rasa run actions` 启动了 Action Server。
    test_weather_action()

执行上述脚本可以看到类似如下的输出:

正在向 Action Server 发送测试请求: http://localhost:5055/webhook
响应状态码: 200
✅ 动作执行成功!
响应内容:
{
  "events": [],
  "responses": [
    {
      "text": "上海今天的天气是大雨,气温约为26摄氏度。",
      "buttons": [],
      "elements": [],
      "custom": {},
      "template": null,
      "response": null,
      "image": null,
      "attachment": null
    }
  ]
}

� 机器人回复: 上海今天的天气是大雨,气温约为26摄氏度。

这样我们就成功在Rasa应用中通过RESTful API的方式对actions进行了执行。

如果喜欢这篇文章或对您有帮助,可以:[☕] 请我喝杯咖啡 | [💓] 小额赞助