Custom Tracker to save conversation history into Firebase Firestore
Hey,I have found a solution to save all the conversation history into the Firebase FireStore Database.
This custom tracker can save all the conversations into the Firestore database.
I followed the same logic of InMemoryTracker to implement this.
How I did?
- First I have Checked this forum Help-with-custom-tracker. But It doesn’t worked because some issues in the tracker_store.py because I’m using Firestore and not using any host parameter but the source code tracker_store in tracker_store.py requires “host” parameter must.And the Exception is not throwing instead ImportError is throwing warning.
- Then After that source code inspection I understand that I need to mention that “host” parameter compulsory, why “host” parameter compulsory
. - Now I’m cleared how to connect custom_tracker and its connected prefectly now.
- Then I started to implement the Firestore tracker, Some one on the forum said to Modify mongodb tracker and use it ,But I don’t understand its logic.
- Now again its time to do reverse engineering
Again I have inspected the tracker_store.py code.
It says implement save(), retrieve() and keys() method.
I think this point should be mentioned in the documentation.
- After that I have followed InMemoryTracker logic.
- Finally its working perfectly
Here’s the custom_tracker.py code (in root directory):
import traceback
import contextlib
import itertools
import json
import logging
import os
import pickle
from datetime import datetime, timezone
from time import sleep
from typing import (
Callable,
Dict,
Iterable,
Iterator,
List,
Optional,
Text,
Union,
TYPE_CHECKING,
)
from boto3.dynamodb.conditions import Key
import rasa.core.utils as core_utils
from rasa.core.actions.action import ACTION_LISTEN_NAME
from rasa.core.brokers.broker import EventBroker
from rasa.core.constants import (
POSTGRESQL_SCHEMA,
POSTGRESQL_MAX_OVERFLOW,
POSTGRESQL_POOL_SIZE,
)
from rasa.core.conversation import Dialogue
from rasa.core.domain import Domain
from rasa.core.events import SessionStarted
from rasa.core.trackers import ActionExecuted, DialogueStateTracker, EventVerbosity
import rasa.cli.utils as rasa_cli_utils
from rasa.utils.common import class_from_module_path, raise_warning, arguments_of
from rasa.utils.endpoints import EndpointConfig
import sqlalchemy as sa
from rasa.core.tracker_store import TrackerStore
# Import custom dbConfig for firebase returns firestore.client() class object
from actionserver.db.dbConfig import db
logger = logging.getLogger(__name__)
# COLLECTION = "restaurant-bot-tracker"
class FirebaseTrackerStore(TrackerStore):
"""Stores conversation history in Firebase"""
def __init__(
self,
domain: Domain,
collection: Optional[Text] = "tracker",
host: Optional[Text] = "localhost",
event_broker: Optional[EventBroker] = None
):
self.store = {}
self.COLLECTION = collection
super().__init__(domain, event_broker)
def save(self, tracker: DialogueStateTracker) -> None:
"""Updates and saves the current conversation state
Args:
tracker: DialogueStateTracker from TrackerStore Class
Returns:
None
Stores data in Firebase and creates tracker
"""
try:
if self.event_broker:
self.stream_events(tracker)
serialised = self.serialiseTracker(tracker)
ref = db.collection(self.COLLECTION).document(tracker.sender_id)
ref.set(serialised)
except Exception as e:
traceback.print_exc()
def checkSenderId(self, sender_id):
"""
Checks if sender Id exists in database (firebase)
Args:
sender_id : takes the sender_id as input
Returns:
Boolean (if the id exists or not)
"""
check = db.collection(self.COLLECTION).document(sender_id).get().exists
return check
def retrieve(self, sender_id: Text) -> Optional[DialogueStateTracker]:
"""
Args:
sender_id: the message owner ID
Returns:
DialogueStateTracker
"""
if self.checkSenderId(sender_id):
logger.debug(f"Recreating tracker for id '{sender_id}'")
deserialised_tracker = self.deserialiseTracker(sender_id)
return deserialised_tracker
else:
logger.debug(f"Creating a new tracker for id '{sender_id}'.")
return None
def keys(self) -> Iterable[Text]:
"""
Returns:
sender_ids of the Tracker Store in Firebase
"""
docs = db.collection(self.COLLECTION).stream()
keys = []
for doc in docs:
keys.append(doc.id)
return keys
def serialiseTracker(self, tracker:DialogueStateTracker):
"""
User defined serialisation
Args:
tracker : takes tracker object as input
Returns:
dialogue
"""
try:
dialogue = tracker.as_dialogue().as_dict()
dialogue = json.dumps(dialogue)
dialogue = json.loads(dialogue)
return dialogue
except Exception as e:
traceback.print_exc()
return None
def deserialiseTracker(self, sender_id):
"""User defined deserialisation
Args:
sender_id : Takes sender_id as input
Returns:
returns tracker
"""
tracker = self.init_tracker(sender_id)
try:
if not tracker:
return None
ref = db.collection(self.COLLECTION).document(sender_id)
dialogue = ref.get().to_dict()
# serialiseTracker(dialogue) no need to pass
dialogue = Dialogue.from_parameters(
json.loads(json.dumps(dialogue)))
except Exception as e:
traceback.print_exc()
tracker.recreate_from_dialogue(dialogue)
return tracker
Here’s the code of endpoints.yml
tracker_store:
type: custom_tracker.FirebaseTrackerStore
host: localhost
collection: restaurant-bot-tracker
Here’s the code of db from this module.
from actionserver.db.dbConfig import db
import firebase_admin
from firebase_admin import credentials
from firebase_admin import firestore
cred = credentials.Certificate('actionserver\db\serviceAccount.json')
firebase_admin.initialize_app(cred)
db = firestore.client()
requirements:
pip install firebase_adminset current working directory in the python path-
Get Firestore serviceAccount.json from the Firebase console.
Thats it firestore custom tracker is ready
Thank you