Event Brokers

Rasa Core allows you to stream events to a message broker. The event broker emits events into the event queue. It becomes part of the TrackerStore which you use when starting an Agent or launch rasa.core.run.

All events are streamed to the broker as serialised dictionaries every time the tracker updates its state. An example event emitted from the default tracker looks like this:

{
    "sender_id": "default",
    "timestamp": 1528402837.617099,
    "event": "bot",
    "text": "what your bot said",
    "data": "some data"
}

The event field takes the event’s type_name (for more on event types, check out the Events docs).

Rasa enables three possible broker types:

• Pika Event Broker
• Kafka Event Broker
• SQL Event Broker

Pika Event Broker

The example implementation we’re going to show you here uses Pika, the Python client library for RabbitMQ.

Adding a Pika Event Broker Using the Endpoint Configuration

You can use an endpoint configuration file to instruct Rasa Core to stream all events to your event broker. To do so, add the following section to your endpoint configuration, e.g. endpoints.yml:

event_broker:
  url: localhost
  username: username
  password: password
  queue: queue
  type: pika

Then instruct Rasa Core to use the endpoint configuration and Pika producer by adding --endpoints <path to your endpoint configuration as the following example:

rasa run -m models --endpoints endpoints.yml

Adding a Pika Event Broker in Python

Here is how you add it using Python code:

from rasa.core.brokers.pika import PikaEventBroker
from rasa_platform.core.tracker_store import InMemoryTrackerStore

pika_broker = PikaEventBroker('localhost',
                              'username',
                              'password',
                              queue='rasa_core_events')

tracker_store = InMemoryTrackerStore(db=db, event_broker=pika_broker)

Implementing a Pika Event Consumer

You need to have a RabbitMQ server running, as well as another application that consumes the events. This consumer needs to implement Pika’s start_consuming() method with a callback action. Here’s a simple example:

import json
import pika

def _callback(self, ch, method, properties, body):
        print('Received event {}'.format(json.loads(body)))

if __name__ == '__main__':
    credentials = pika.PlainCredentials('username', 'password')
    connection = pika.BlockingConnection(
        pika.ConnectionParameters('rabbit', credentials=credentials))
    channel = connection.channel()
    channel.basic_consume(_callback,
                          queue='rasa_core_events',
                          no_ack=True)
    channel.start_consuming()

Kafka Event Broker

It is possible to use Kafka as the main broker for your events. In this example, we are going to use the python-kafka library, a Kafka client written in Python.

Note: In order to use the Kafka event broker, rasa has to be installed with the kafka option:

$ pip install rasa[kafka]

Adding a Kafka Event Broker Using the Endpoint Configuration

As for the other brokers, you can use an endpoint configuration file to instruct Rasa Core to stream all events to this event broker. To do it, add the following section to your endpoint configuration.

event_broker:
  url: localhost
  sasl_username: username
  sasl_password: password
  topic: topic
  security_protocol: SASL_PLAINTEXT
  type: kafka

In the case of using the SSL protocol, the endpoints file must look like:

event_broker:
  url: localhost
  topic: topic
  security_protocol: SSL
  ssl_cafile: CARoot.pem
  ssl_certfile: certificate.pem
  ssl_keyfile: key.pem
  ssl_check_hostname: True
  type: kafka

Adding a Kafka Broker in Python

The code below shows an example of how to instantiate a Kafka producer in your script:

from rasa.core.brokers.kafka import KafkaEventBroker
from rasa.core.tracker_store import InMemoryTrackerStore

kafka_broker = KafkaEventBroker(host='localhost:9092',
                                topic='rasa_core_events')

tracker_store = InMemoryTrackerStore(event_broker=kafka_broker)

Authentication and authorization

Rasa Core’s Kafka producer accepts two types of security protocols - SASL_PLAINTEXT and SSL.

For development environments, you can use simple authentication with SASL_PLAINTEXT.

kafka_broker = KafkaEventBroker(host='kafka_broker:9092',
                                sasl_plain_username='kafka_username',
                                sasl_plain_password='kafka_password',
                                security_protocol='SASL_PLAINTEXT',
                                topic='rasa_core_events')

For production, you must use SSL protocol to assure encryption of data and client authentication.

kafka_broker = KafkaEventBroker(host='kafka_broker:9092',
                                ssl_cafile='CARoot.pem',
                                ssl_certfile='certificate.pem',
                                ssl_keyfile='key.pem',
                                ssl_check_hostname=True,
                                security_protocol='SSL',
                                topic='rasa_core_events')

Implementing a Kafka Event Consumer

The following implementation shows an example:

from kafka import KafkaConsumer
from json import loads

consumer = KafkaConsumer('rasa_core_events',
                          bootstrap_servers=['localhost:29093'],
                          value_deserializer=lambda m: json.loads(m.decode('utf-8')),
                          security_protocol='SSL',
                          ssl_check_hostname=False,
                          ssl_cafile='CARoot.pem',
                          ssl_certfile='certificate.pem',
                          ssl_keyfile='key.pem')

for message in consumer:
    print(message.value)

SQL Event Broker

It is possible to use an SQL database as an event broker using SQLAlchemy.

Adding a SQL Event Broker Using the Endpoint Configuration

You can use the endpoint configuration file to instruct Rasa to save all events to your SQL event broker. For example, a valid SQLite configuration could look like:

event_broker:
  type: SQL
  dialect: sqlite
  db: events.db

PostgreSQL databases can also be used:

event_broker:
  type: SQL
  host: 127.0.0.1
  port: 5432
  dialect: postgresql
  username: myuser
  password: mypassword
  db: mydatabase

With this configuration applied, Rasa will create a table called events in the database, where all events will be added.

👋 I can help you get started with Rasa and answer your technical questions.