Rasa 2.8配置Viber自定义连接器报错的技术求助

问题原因 #

报错Failed to find input channel class for 'Viber.ViberInput'的核心是Rasa无法正确识别自定义的ViberInput类，涉及模块导入路径、配置引用格式，以及自定义代码本身的逻辑问题。

分步解决方案 #

1. 修正credentials.yml配置格式 #

配置中Viber.ViberInput的首字母大写不符合模块引用规则（自定义文件名为viber.py，模块名应为小写），修改为：

viber.ViberInput:
  viber_name: "Rasa_demo_one"
  avatar: ""
  auth_token: "4f831f01ef34ad38-9d057b2fd4ba804-8dd0cf1cdf5e39dc"

2. 确保自定义模块被Python识别 #

将viber.py放在Rasa项目根目录（与credentials.yml、config.yml同目录）后，需要让Python能读取到该模块。在项目根目录执行以下命令添加路径：

export PYTHONPATH="${PYTHONPATH}:$(pwd)"

之后再启动Rasa服务。

3. 修复自定义通道代码问题 #

原代码存在全局实例初始化错误、请求判断逻辑无效、未对接Rasa消息处理流程等问题，修正后的代码如下：

from http.client import HTTPResponse
from viberbot import Api
from viberbot.api.bot_configuration import BotConfiguration
from viberbot.api.messages import TextMessage
from viberbot.api.viber_requests import ViberMessageRequest
from typing import Text, List, Dict, Any, Optional, Callable, Awaitable
from sanic.request import Request
from rasa.core.channels import InputChannel
from rasa.core.channels.channel import UserMessage, CollectingOutputChannel
from sanic import Blueprint, response


class ViberInput(InputChannel):
    """Viber input channel implementation. Based on the HTTPInputChannel."""

    @classmethod
    def name(cls) -> Text:
        return "viber"

    @classmethod
    def from_credentials(cls, credentials: Optional[Dict[Text, Any]]) -> InputChannel:
        if not credentials:
            cls.raise_missing_credentials_exception()

        return cls(
            credentials.get("viber_name"),
            credentials.get("avatar"),
            credentials.get("auth_token")
        )

    def __init__(self, viber_name: Text, avatar: Text, auth_token: Text) -> None:
        """Create a Viber input channel."""
        self.viber_name = viber_name
        self.avatar = avatar
        self.auth_token = auth_token
        self.viber = Api(
            BotConfiguration(
                name=self.viber_name,
                avatar=self.avatar,
                auth_token=self.auth_token
            )
        )

    def blueprint(self, on_new_message: Callable[[UserMessage], Awaitable[Any]]) -> Blueprint:
        viber_webhook = Blueprint("viber_webhook", __name__)

        @viber_webhook.route("/", methods=["POST"])
        async def incoming(request: Request) -> HTTPResponse:
            viber_request = self.viber.parse_request(request.get_data())
            
            # 仅处理用户消息请求
            if isinstance(viber_request, ViberMessageRequest):
                sender_id = viber_request.sender.id
                message = viber_request.message
                
                # 处理文本消息并传给Rasa处理
                if isinstance(message, TextMessage):
                    await on_new_message(UserMessage(
                        message.text,
                        CollectingOutputChannel(),
                        sender_id,
                        input_channel=self.name()
                    ))
            
            return response.text("success")

        # 新增手动设置Webhook的路由
        @viber_webhook.route("/set_webhook", methods=["GET"])
        async def set_webhook(request: Request) -> HTTPResponse:
            # 替换为你的实际公网域名
            webhook_url = "https://your-public-domain.com/webhooks/viber/webhook"
            self.viber.set_webhook(webhook_url)
            return response.text("Webhook set successfully")

        return viber_webhook

4. 启动Rasa服务并设置Webhook #

在项目根目录执行启动命令：

rasa run --enable-api --cors "*"

服务启动后，访问http://localhost:5005/webhooks/viber/set_webhook完成Webhook配置。

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