Custom Actions
Note: This is material for Rasa Open Source 3.x. If you're interested in the content for Rasa Open Source 2.x, please see the archived version of this lesson here.
Video
Let's say that we've predicted what next action to take on behalf of the user. Then we may need to send an appropriate response at times, but at other times we may need to perform other tasks. Things like:
- Send an email
- Make a calendar appointment
- Fetch relevant information from a database
- Check information from an API
- Calculate something specific
For all of these examples we need to do more than just send back a response. Instead, we'd like to run a bit of code on behalf of our users. To handle these kinds of tasks Rasa offers you to write "Custom Actions". Instead of triggering a response text we can trigger a custom python function to run on our behalf. This opens up a lot of options for our assistant!
Custom Actions as a System
Before moving on to syntax, it helps to understand how these custom actions interact with your Rasa assistant. If we think of our "Rasa service" then we're thinking about a system that can infer what a user wants and what action to take next. If we think about our "Custom Actions" then we're typically concerned about other things. We'd likely need to connect to a database, which brings in security concerns as well as unit tests.
That's why Rasa introduce a separation of concerns when it comes to custom actions. The main Rasa process is distinct from the process that handles the custom Python code.
Code
Github
We will share code snippets below. But we also have a Github repository with the entire Rasa project. If you'd prefer to explore the code that way, simply go to the repository here.
Basic Custom Action
The code below shows how you might define a custom action that tells you the time.
from typing import Any, Text, Dict, List
import arrow import dateparserfrom rasa_sdk import Action, Trackerfrom rasa_sdk.events import SlotSetfrom rasa_sdk.executor import CollectingDispatcher
city_db = { 'brussels': 'Europe/Brussels', 'zagreb': 'Europe/Zagreb', 'london': 'Europe/Dublin', 'lisbon': 'Europe/Lisbon', 'amsterdam': 'Europe/Amsterdam', 'seattle': 'US/Pacific'}
class ActionTellTime(Action):
def name(self) -> Text: return "action_tell_time"
def run(self, dispatcher: CollectingDispatcher, tracker: Tracker, domain: Dict[Text, Any]) -> List[Dict[Text, Any]]: current_place = next(tracker.get_latest_entity_values("place"), None) utc = arrow.utcnow() if not current_place: msg = f"It's {utc.format('HH:mm')} utc now. You can also give me a place." dispatcher.utter_message(text=msg) return [] tz_string = city_db.get(current_place, None) if not tz_string: msg = f"I didn't recognize {current_place}. Is it spelled correctly?" dispatcher.utter_message(text=msg) return [] msg = f"It's {utc.to(city_db[current_place]).format('HH:mm')} in {current_place} now." dispatcher.utter_message(text=msg) return []
The code uses the arrow library for convenience, so ensure that it's installed beforehand.
There are a few parts worth diving into.
- We're defining a class
ActionTellTime
that inherits from theAction
class found in therasa_sdk
module. Whenever you write a custom action, you need to inherit from this class because it handles a lot of boilerplate on your behalf. - The
name(self)
method defines the name of the action. This name also needs to be copied to yourdomain.yml
file and you need to make sure that you don't misspell it when you're using it in yourstories.yml
/rules.yml
/domain.yml
files. - The
run
method contains the logic for your custom action. This method received information from the conversation so far from thetracker
input. You can learn more about the information it can provide by checking the docs. - To send messages to the end user you'll want to use the
dispatcher.utter_message
method. You can also send images or buttons with this method, but in our example we only use it to send text. - We're mocking a database here with our
city_db
dictionary. We're using this "database" to convert a city name to a timezone. - The custom action tries to fetch a detected entity for a place. If it cannot find one in the tracker, it tries to fail gracefully. This is a pattern that you'll see a lot in custom actions. You should always consider how to best catch unexpected behavior.
Running the Action
If you want to interact with an assistant that has this custom action, remember that
before you run rasa shell
you'll want to start the action server first.
# Run this in a separate terminalrasa run actions --port 5055
You should also check your endpoints.yml
file before running the Rasa shell. You want
to make sure that your Rasa shell can find the custom actions. Just make sure that you
have an actions endpoint properly configured. If you're running the custom actions on
port 5055, this should suffice:
action_endpoint: url: "http://localhost:5055/webhook"
More Advanced Custom Actions
Let's add two more custom actions to show some more features.
Setting Slots
First, we'll add a custom action that can register information in a slot on behalf of the user.
class ActionRememberWhere(Action):
def name(self) -> Text: return "action_remember_where"
def run(self, dispatcher: CollectingDispatcher, tracker: Tracker, domain: Dict[Text, Any]) -> List[Dict[Text, Any]]: current_place = next(tracker.get_latest_entity_values("place"), None) utc = arrow.utcnow() if not current_place: msg = "I didn't get where you lived. Are you sure it's spelled correctly?" dispatcher.utter_message(text=msg) return [] tz_string = city_db.get(current_place, None) if not tz_string: msg = f"I didn't recognize {current_place}. Is it spelled correctly?" dispatcher.utter_message(text=msg) return [] msg = f"Sure thing! I'll remember that you live in {current_place}." dispatcher.utter_message(text=msg) return [SlotSet("location", current_place)]
You may have noticed before that a custom action returns a list. This is meant as
a way for you to tell Rasa to run specific events. In our example we're showing how
to emit a SlotSet
-event, but Rasa supports other events as well. Check
the docs
for more information.
Retrieving Slots
The action that can fetch the slot value is shown below.
class ActionTimeDifference(Action):
def name(self) -> Text: return "action_time_difference"
def run(self, dispatcher: CollectingDispatcher, tracker: Tracker, domain: Dict[Text, Any]) -> List[Dict[Text, Any]]: timezone_to = next(tracker.get_latest_entity_values("place"), None) timezone_in = tracker.get_slot("location") if not timezone_in: msg = "To calculuate the time difference I need to know where you live." dispatcher.utter_message(text=msg) return [] if not timezone_to: msg = "I didn't the timezone you'd like to compare against. Are you sure it's spelled correctly?" dispatcher.utter_message(text=msg) return [] tz_string = city_db.get(timezone_to, None) if not tz_string: msg = f"I didn't recognize {timezone_to}. Is it spelled correctly?" dispatcher.utter_message(text=msg) return [] t1 = arrow.utcnow().to(city_db[timezone_to]) t2 = arrow.utcnow().to(city_db[timezone_in]) max_t, min_t = max(t1, t2), min(t1, t2) diff_seconds = dateparser.parse(str(max_t)[:19]) - dateparser.parse(str(min_t)[:19]) diff_hours = int(diff_seconds.seconds/3600) msg = f"There is a {min(diff_hours, 24-diff_hours)}H time difference." dispatcher.utter_message(text=msg) return []
Again you can see that this custom action really tries to be robust against unexpected events. Even if the slot value is empty, or if the slot value doesn't match a database entry, we're able to "fail gracefully".
Links
Exercises
Try to answer the following questions to test your knowledge.
- What does the
tracker
contain in therun
method of a Custom Action? - What does the
dispatcher
contain in therun
method of a Custom Action? - How can you set a slot in a Custom Action?