Accessing SLOTS from Custom Connector

Rasa Open Source
1

Hello!

I am writing this forum ticket to seek assistance with an issue I am currently facing while attempting to access a Rasa slot in a custom output channel. I hope that someone from the community can help me find a solution or provide guidance on how to resolve this problem.

The specific problem I am encountering is that I am unable to access a Rasa slot within a custom output channel implementation. I have successfully created a custom output channel by extending the InputChannel class, and it is working as expected for sending messages to the user. However, when I try to access the slots using the tracker.get_slot(slot_name) method within the custom output channel, i get an error that tracker isn’t defined although i am importing it as following: from rasa_sdk import Action, Tracker

I would greatly appreciate any insights or suggestions on how to overcome this issue. Is there anything specific I need to consider when accessing slots within a custom output channel? Are there any additional steps or modifications that I should make to ensure the slots are accessible?

2

Hi @vicesan, can you share your implementation?

1 Like
3

Hey, thanks for replying!

I understood that the error occurs because I am trying to call the get_slot method on the Tracker class directly, without creating an instance of the class. The question now is how to do that.

This is how the structure of my code looks like:

import asyncio
import inspect
import logging
from sanic import Sanic, Blueprint, response
from rasa_sdk import Tracker

from rasa.core.channels.channel import (
    InputChannel,
    CollectingOutputChannel,
    UserMessage,
)


class CustomOutputConnector(InputChannel):
    def __init__(self, agent: Agent) -> None:
        self.agent = agent

    @classmethod
    def name(cls) -> Text:
        """Name of your custom channel."""
        return "first"

    def blueprint(
        self, on_new_message: Callable[[UserMessage], Awaitable[None]]
    ) -> Blueprint:

        custom_webhook = Blueprint(
            "custom_webhook_{}".format(type(self).__name__),
            inspect.getmodule(self).__name__,
        )

        @custom_webhook.route("/", methods=["GET"])
        async def health(request: Request) -> HTTPResponse:
            return response.json({"status": "ok"})

        @custom_webhook.route("/webhook", methods=["POST"])
        async def receive(request: Request) -> HTTPResponse:

            text = request.json.get("text")  # method to fetch text
            input_channel = self.name()  # method to fetch input channel
            metadata = self.get_metadata(request)  # method to get metadata

            collector = CollectingOutputChannel()

            # include exception handling
            try:
                await on_new_message(
                    UserMessage(
                        text,
                        collector,
                        sender_id,
                        input_channel=input_channel,
                        metadata=metadata,
                    )
                )
            except CancelledError:
                logger.error(
                    f"Message handling timed out for " f"user message '{text}'."
                )
            except Exception:
                logger.exception(
                    f"An exception occurred while handling "
                    f"user message '{text}'."
                )

            # Get the current tracker instance for the sender_id

            translation_language = tracker.get_slot('translation_language')
            ....

        return custom_webhook