DOCSEARCHbyAlgolia
Submit Docs

Rasa SDK

repository·main·Indexed 43 days ago

https://github.com/rasahq/rasa-sdk

A specialized Python toolkit for developing custom actions to extend Rasa conversational AI. It provides the infrastructure to run an action server via HTTP or gRPC, manage conversation state using SlotSet and AllSlotsReset, and implement custom validation logic with ValidationAction and FormValidationAction. Version 3.20.0.dev2.

Tokens
13.4K
Snippets
46
Records
59
Agent score
55%

What's inside Rasa SDK

  1. Run the Rasa SDK action server using Docker

    main

    You can use the official rasa/rasa-sdk Docker image to start an action server.

    Prerequisite: The folder containing your custom actions must be a Python module and must contain an __init__.py file.

    Run the following command to start the server, mapping the local actions directory to the container:

    http://localhost:5055/webhook is the endpoint where the action server will be available.

    docker run -p 5055:5055 --mount type=bind,source=<ABSOLUTE_PATH_TO_YOUR_ACTIONS>,target=/app/actions \
	rasa/rasa-sdk:<version>

  2. Add custom dependencies to the Rasa SDK Docker image

    main

    To include system packages or PyPI dependencies in your action server, create a custom Dockerfile that extends rasa/rasa-sdk.

    Note that you must switch to USER root to perform installations and then switch back to a non-root user (e.g., USER 1001) to run the code securely.

    # Extend the official Rasa SDK image
FROM rasa/rasa-sdk:<version>

# Change back to root user to install dependencies
USER root

# To install system dependencies
RUN apt-get update -qq && \
    apt-get install -y <NAME_OF_REQUIRED_PACKAGE> && \
    apt-get clean && \
    rm -rf /var/lib/apt/lists/* /tmp/* /var/tmp/*

# To install packages from PyPI
RUN pip install --no-cache-dir <A_REQUIRED_PACKAGE_ON_PYPI>

# Switch back to non-root to run code
USER 1001

  4. Build the Rasa SDK from source

    main

    The Rasa SDK uses Poetry for packaging and dependency management.

    1. Install Poetry (if not already installed):
      curl -sSL https://install.python-poetry.org | python3 -
    2. Install dependencies and the SDK in editable mode:
      make install
    # Install dependencies and rasa-sdk in editable mode
make install

  5. Lint and format code with ruff

    main

    The project uses ruff for code formatting. To maintain code quality:

    • Check and reformat files: Run make lint.
    • Automatic formatting on commit: Install pre-commit via pip install pre-commit and then run pre-commit install.
    # Check and reformat files
make lint

  6. How slot extraction and validation work in ValidationAction

    main

    The ValidationAction follows a specific execution flow:

    1. Extraction: It calls get_extraction_events(), which iterates through slots identified by required_slots(). For each slot, it attempts to call extract_<slot_name>. Extracted values are updated in both the returned events and the tracker to ensure sequential consistency.
    2. Validation: It calls get_validation_events(), which iterates through slots currently in the tracker. For each slot, it looks for a method named validate_<slot_name>. If found, it executes the method and applies the returned dictionary of updates to the tracker.

    This dual process allows you to both pull information from the user's input (extraction) and verify the correctness of existing slot values (validation) in a single action.

  7. Use CollectingDispatcher to send messages and stream responses

    main

    The CollectingDispatcher is the primary object used within custom actions to communicate with the user. It supports two modes of operation:

    1. Unary API: Use utter_message() to send complete messages (text, images, buttons, etc.). These are accumulated and returned as a single payload at the end of the action.
    2. Streaming API: Use stream_start(), stream_chunk(), and stream_end() to emit responses incrementally (e.g., for LLM token streaming).

    Crucially, action authors do not need to worry about the transport layer. If the transport supports streaming, chunks are sent immediately. If not, stream_end() automatically replays all accumulated chunks as individual utter_message() calls, ensuring your action works correctly on all platforms.

    async def run(
    self,
    dispatcher: CollectingDispatcher,
    tracker: Tracker,
    domain: Domain,
    )
    # Unary: Send a complete message
    dispatcher.utter_message(text="Hello!", buttons=[{"button": {"text": "Yes", "payload": "confirm"}}])

    # Streaming: Send incremental chunks
    await dispatcher.stream_start()
    await dispatcher.stream_chunk(text="This ")
    await dispatcher.stream_chunk(text="is ")
    await dispatcher.stream_chunk(text="streaming.")
    await dispatcher.stream_end()
    
    return []

  8. Handle streaming barge-ins in `WebhookStream`

    main

    The WebhookStream method supports streaming responses that can be interrupted (barge-in).

    The Streaming Lifecycle:

    1. chunk_start: The action begins streaming. A unique response_id is generated.
    2. chunk: One or more response chunks (text, images, etc.) are yielded.
    3. chunk_end: The current streaming sequence is complete.
    4. final_result: The full WebhookResponse is sent after the action finishes.

    Barge-in Mechanism: When a user interrupts the assistant, the client calls AckStreamChunks with the active response_id. This triggers dispatcher.cancel_stream(). The server then enters a 'graceful drain' mode: it stops yielding new chunks but waits for the action to finish naturally to yield a final_result (or an error event), ensuring the client can complete its state tracking. The maximum time to wait for this drain is controlled by the ACTION_SERVER_STREAM_BARGE_IN_TIMEOUT_SECONDS environment variable (default 30s).

  9. Configure barge-in timeout via environment variables

    main

    You can control the maximum time the gRPC server waits for a streaming action to finish after a barge-in event using the following environment variable:

    • ACTION_SERVER_STREAM_BARGE_IN_TIMEOUT_SECONDS: The timeout in seconds. If the action does not finish within this window during the 'drain' phase, the task is cancelled and an empty final_result is yielded to allow the client to complete the RPC.

  10. Check Rasa SDK and Rasa version compatibility

    main

    Ensure your rasa-sdk version is compatible with your installed Rasa version.

    rasa-sdk package compatibility:

    SDK versioncompatible Rasa version
    1.0.x>=1.0.x

    Old rasa_core_sdk package compatibility:

    SDK versioncompatible Rasa Core version
    0.12.x>=0.12.x
    0.11.x0.11.x
    not compatible<=0.10.x

  11. Implement form validation with FormValidationAction

    main

    Use the FormValidationAction class to create specialized validation logic for Rasa forms. This class is a subclass of ValidationAction and is designed to work within the form lifecycle.

    Key requirements:

    1. Implement name(): Returns the unique identifier for the action. The form name is automatically derived by removing the validate_ prefix from this name.
    2. The domain_slots() method determines which slots the form should fill. By default, it looks at the required_slots defined for the form in your domain file.
    3. The class automatically handles the requested_slot logic to determine which slot to ask for next, unless you override next_requested_slot() or required_slots().
    from rasa_sdk import FormValidationAction

class MyFormValidationAction(FormValidationAction):
    def name(self) -> str:
        return "validate_my_form"

    # The form name is inferred as 'my_form'

    async def validate_email(self, slot_value, dispatcher, tracker, domain):
        if "@" in slot_value:
            return {}
        else:
            dispatcher.utter_message(text="That doesn't look like a valid email.")
            return {"email": None}

  12. Override required_slots in ValidationAction

    main

    By default, required_slots() returns all slots mapped in the domain. You can override this method to restrict validation or extraction to a specific subset of slots.

    async def required_slots(
    self, 
    domain_slots: List[Text], 
    dispatcher: "CollectingDispatcher", 
    tracker: "Tracker", 
    domain: "DomainDict"
) -> List[Text]:
    # Only validate these specific slots
    return ["email", "phone_number"]