如何在RASA自定义动作服务器初始化阶段提前加载依赖库

首先明确：完全可以在RASA自定义动作服务器启动、接收请求前完成所有重依赖库、模型的预加载，从根源避免每次动作调用重复加载的耗时问题。

实现逻辑 #

RASA 动作SDK启动时，会先加载actions目录下所有模块的顶层代码，完成所有全局实例初始化后，才会启动HTTP服务监听请求。只要将耗时的库导入、模型加载、资源初始化逻辑放在动作类的外部（模块顶层作用域），相关逻辑就只会在服务启动时执行一次，后续所有动作请求都会复用已经加载好的实例。

代码示例 #

错误写法（每次调用重复加载） #

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

class ActionProcessUserInput(Action):
    def name(self) -> str:
        return "action_process_user_input"

    def run(self, dispatcher: CollectingDispatcher, tracker: Tracker, domain: Dict[str, Any]) -> List[Dict[str, Any]]:
        # 错误逻辑：每次触发动作都会重新加载库和模型，耗时极高
        import your_heavy_nlp_lib
        nlp_model = your_heavy_nlp_lib.load_local_model("./model_path")

        user_text = tracker.latest_message.get("text")
        result = nlp_model.process(user_text)
        dispatcher.utter_message(text=result)
        return []

正确写法（启动阶段预加载） #

from rasa_sdk import Action, Tracker
from rasa_sdk.executor import CollectingDispatcher
from typing import Any, Dict, List
# 1. 模块顶层导入重依赖库
import your_heavy_nlp_lib
# 2. 模块顶层完成模型/资源初始化，全局复用
GLOBAL_NLP_MODEL = your_heavy_nlp_lib.load_local_model("./model_path")

class ActionProcessUserInput(Action):
    def name(self) -> str:
        return "action_process_user_input"

    def run(self, dispatcher: CollectingDispatcher, tracker: Tracker, domain: Dict[str, Any]) -> List[Dict[str, Any]]:
        user_text = tracker.latest_message.get("text")
        # 直接复用预加载好的实例，无额外加载耗时
        result = GLOBAL_NLP_MODEL.process(user_text)
        dispatcher.utter_message(text=result)
        return []

特殊场景适配 #

如果你的预加载逻辑需要读取动态配置、或依赖运行时参数，可以重写动作服务器的启动入口，在服务启动前执行自定义加载逻辑：

from rasa_sdk import ActionServer
import os

def custom_preload_logic():
    # 自定义预加载逻辑，比如读取环境变量拉取模型、初始化全局连接等
    global GLOBAL_NLP_MODEL
    model_path = os.getenv("NLP_MODEL_PATH", "./default_model")
    GLOBAL_NLP_MODEL = your_heavy_nlp_lib.load_local_model(model_path)

if __name__ == "__main__":
    # 先执行所有预加载逻辑，完成后再启动服务
    custom_preload_logic()
    action_server = ActionServer(
        action_package="actions",
        port=5055
    )
    action_server.run()

注意事项 #

• 多worker部署动作服务器时，每个worker进程会独立执行一次预加载逻辑，属于正常现象
• 若使用的第三方库不是线程安全的，可在run方法调用实例的位置加线程锁，避免并发请求异常
• 预加载阶段的报错会直接导致动作服务器启动失败，方便提前发现资源加载问题，不会等到线上请求时才触发报错

内容的提问来源于stack exchange，提问作者caaax