Dynamic KnowledgeBase : update inmemoryknowledgebase using custom action

Rasa Open Source
1

Hi Rasa Members,

I am working on building a chatbot where my requirement is to update InMemoryKnowledgeBase data based on requirement.

Currently, whenever I start action server InMemoryKnowledgeBase gets updated using below class.

class MyKnowledgeBaseAction(ActionQueryKnowledgeBase):
    def __init__(self):
        knowledge_base = InMemoryKnowledgeBase("userdata.json")
        super().__init__(knowledge_base)

Above class is reading data from productlist.json file. I have a custom actions which updates “userdata.json” based on user’s ask.

Now, I am not able to reload data in InMemoryKnowledgeBase whenever that file is updated. Can someone share on how to update InMemoryKnowledgeBase using any other custom action or code snippet.

2

@yashjn72 Could you please share your actions.py file to understand how you are updating the userdata.json file currently, in addition to your training data files?

Looking through the codebase, I can see InMemoryKnowledgeBase has an instance method self.load() that loads the json file given to its constructor and stores the data in a dictionary self.data. You might need to override the implementation of run method in your custom action extending ActionQueryKnowledgeBase by first calling self.knowledge_base.load() and then copying the rest of the code currently in the run method to maintain existing functionality.

3

Hi @anca ,

Please follow below code snippet for actions.py

class MyKnowledgeBaseAction(ActionQueryKnowledgeBase):
    def __init__(self):
        knowledge_base = InMemoryKnowledgeBase("userdata.json")
        super().__init__(knowledge_base)


class somecustomjob(Action):
    def name(self) -> Text:
        return "action_somecustomaction"

    def run(self, dispatcher: CollectingDispatcher,
                tracker: Tracker,
                domain: Dict[Text, Any]) -> List[Dict[Text, Any]]:
        ## some code to generate userdata.json file 
        with open('userdata.json', 'w') as convert_file: 
            convert_file.write(json.dumps({"user_list":userdata}))
        
        knowledge_base = InMemoryKnowledgeBase("productlist.json")
        ActionQueryKnowledgeBase(knowledge_base)
       
        return [SlotSet(SLOT_OBJECT_TYPE, "user_list")]

I am using above method to reload inMemoryKnowledgeBase. But is is not working.

4

Yup, this probably doesn’t work because the main knowledge base action used by the bot is MyKnowledgeBaseAction whose userdata.json file never gets updated. Please try the suggestion I recommended.

5

Could you please share a sample snippet.

6

Well I am not able to implement what you have suggested. It will be very helpful if you can share a snippet.

My custom action is already extended by Action call, how can I extend it with another class.

userdata.json file is same. As it is created by previous rasa session before restart.

7

@yashjn72 Please try this but beware that this is an untested recommendation:

class MyKnowledgeBaseAction(ActionQueryKnowledgeBase):
    def __init__(self):
        knowledge_base = InMemoryKnowledgeBase("userdata.json")
        super().__init__(knowledge_base)

     async def run(
        self,
        dispatcher: CollectingDispatcher,
        tracker: Tracker,
        domain: "DomainDict",
      ) -> List[Dict[Text, Any]]:
        # this is the only line you need to introduce
        self.knowledge_base.load()
        
        # this is copied from current Rasa SDK Python implementation
        object_type = tracker.get_slot(SLOT_OBJECT_TYPE)
        last_object_type = tracker.get_slot(SLOT_LAST_OBJECT_TYPE)
        attribute = tracker.get_slot(SLOT_ATTRIBUTE)

        new_request = object_type != last_object_type

        if not object_type:
            dispatcher.utter_message(response="utter_ask_rephrase")
            return []

        if not attribute or new_request:
            return await self._query_objects(dispatcher, object_type, tracker)
        elif attribute:
            return await self._query_attribute(
                dispatcher, object_type, attribute, tracker
            )

        dispatcher.utter_message(response="utter_ask_rephrase")
        return []