A Rasa tracker implementation for couchDB

README.md

This is a CouchDB implementation for a Rasa (2.0) tracker that makes use of the Cloudant CouchDB API wrapper.

To use, you'll need to install the cloudant package with pip, or add to your requirements.txt (cloudant==2.14.0)

Copy tracker.py into project-root/couch_tracker/ directory. Next, configure your endpoint details based on the example, and insert into your endpoints yml.

That should be it! Let me know if it works, or doesn't- @digitalWestie x

Licence MIT - https://opensource.org/licenses/MIT Copyright 2020 Rory Gianni

endpoints.yml

#Example settings to be inserted into endpoints.yml
tracker_store:
  type: couch_tracker.tracker.CouchTrackerStore
  url: 172.20.128.2
  port: 5984
  username: mycouchuser
  password: mycouchpass
  db: mydbname

#...

tracker.py

import json
import logging
from typing import (
    Any,
    Callable,
    Dict,
    Iterable,
    Iterator,
    List,
    Optional,
    Text,
    Union,
    TYPE_CHECKING,
)
from rasa.core.brokers.broker import EventBroker
from rasa.shared.core.trackers import (
    ActionExecuted,
    DialogueStateTracker,
    EventVerbosity,
)
from rasa.core.tracker_store import TrackerStore

from cloudant.client import CouchDB
from cloudant.database import CouchDatabase
logger = logging.getLogger(__name__)

class CouchTrackerStore(TrackerStore):
    """Stores conversation history in CouchDB"""

    def __init__(
        self,
        domain,
        username,
        host="localhost",
        port=5984,
        db="rasa",
        password: Optional[Text] = None,
        event_broker: Optional[EventBroker] = None,
        timeout: Optional[float] = None,
        use_ssl: bool = False,
    ):
        url = "https://" if use_ssl else "http://"
        url = url + str(host) + ":" + str(port)
        self.client = CouchDB(username, password, url=url, connect=True, auto_renew=True, timeout=timeout)
        self.db = CouchDatabase(self.client, db)
        if self.db.exists():
            logger.debug('Couch database %s already exists, proceeding', db)
        else:
            logger.debug('Couch database %s doesn\'t exist, creating', db)
            self.db.create()
        super().__init__(domain, event_broker)

    def save(self, tracker):
        """Saves the current conversation state"""
        if self.event_broker:
            self.stream_events(tracker)

        serialised_tracker = self.serialise_tracker(tracker)
        document_dict = { "_id": tracker.sender_id, "tracker": serialised_tracker }
        #creates doc if not already exists, otherwise simply instantiates a document object
        doc = self.db.create_document(document_dict)
        if doc.exists():
            doc.fetch() #get latest version of doc
            document_dict['_rev'] = doc['_rev'] #provide latest revision hash
            doc.update(document_dict) #replace all dict values
            doc.save()

    def retrieve(self, sender_id):
        """sender_id: the message owner ID (used as document id in CouchDB), returns: DialogueStateTracker"""
        logger.debug('Attempting to retrieve conversation with sender_id: %s', sender_id)
        try:
            document = self.db[sender_id]
        except Exception as e:
            document = None

        if document is None:
            logger.debug('No document retrieved from Couch DB with id: %s', sender_id)
            return None
        elif 'tracker' in document:
            result = self.deserialise_tracker(sender_id, json.dumps(document.get('tracker')))
            logger.debug('Deserialised document.tracker retrieved from Couch DB with id: %s', sender_id)
            return result
        result = self.deserialise_tracker(sender_id, document.json())
        logger.debug('Deserialised document retrieved from Couch DB with id: %s', sender_id)
        return result

    def keys(self) -> Iterable[Text]:
        """Returns keys of the Couch Tracker Store"""
        return self.db.keys(remote=True)

    def serialise_tracker(self, tracker: DialogueStateTracker) -> Dict:
        """returns representation of the tracker ready for couchdb"""
        return tracker.as_dialogue().as_dict()