Core: 3. Responses#
This tutorial shows different options for setting responses.
Here, responses that allow giving custom answers to users are shown.
Let’s do all the necessary imports from DFF.
[1]:
# installing dependencies
%pip install -q dff
Note: you may need to restart the kernel to use updated packages.
[2]:
import re
import random
from dff.script import TRANSITIONS, RESPONSE, Context, Message
import dff.script.responses as rsp
import dff.script.conditions as cnd
from dff.pipeline import Pipeline
from dff.utils.testing.common import (
check_happy_path,
is_interactive_mode,
run_interactive_mode,
)
The response can be set by Callable or *Message:
Callable objects. If the object is callable it must have a special signature:
func(ctx: Context, pipeline: Pipeline) -> Message
Message objects. If the object is Message it will be returned by the agent as a response.
The functions to be used in the toy_script
are declared here.
[3]:
def cannot_talk_about_topic_response(ctx: Context, _: Pipeline) -> Message:
request = ctx.last_request
if request is None or request.text is None:
topic = None
else:
topic_pattern = re.compile(r"(.*talk about )(.*)\.")
topic = topic_pattern.findall(request.text)
topic = topic and topic[0] and topic[0][-1]
if topic:
return Message(f"Sorry, I can not talk about {topic} now.")
else:
return Message("Sorry, I can not talk about that now.")
def upper_case_response(response: Message):
# wrapper for internal response function
def func(_: Context, __: Pipeline) -> Message:
if response.text is not None:
response.text = response.text.upper()
return response
return func
def fallback_trace_response(ctx: Context, _: Pipeline) -> Message:
return Message(
misc={
"previous_node": list(ctx.labels.values())[-2],
"last_request": ctx.last_request,
}
)
[4]:
toy_script = {
"greeting_flow": {
"start_node": { # This is an initial node,
# it doesn't need a `RESPONSE`.
RESPONSE: Message(),
TRANSITIONS: {"node1": cnd.exact_match(Message("Hi"))},
# If "Hi" == request of user then we make the transition
},
"node1": {
RESPONSE: rsp.choice(
[
Message("Hi, what is up?"),
Message("Hello, how are you?"),
]
),
# Random choice from candidate list.
TRANSITIONS: {
"node2": cnd.exact_match(Message("I'm fine, how are you?"))
},
},
"node2": {
RESPONSE: Message("Good. What do you want to talk about?"),
TRANSITIONS: {
"node3": cnd.exact_match(Message("Let's talk about music."))
},
},
"node3": {
RESPONSE: cannot_talk_about_topic_response,
TRANSITIONS: {"node4": cnd.exact_match(Message("Ok, goodbye."))},
},
"node4": {
RESPONSE: upper_case_response(Message("bye")),
TRANSITIONS: {"node1": cnd.exact_match(Message("Hi"))},
},
"fallback_node": { # We get to this node
# if an error occurred while the agent was running.
RESPONSE: fallback_trace_response,
TRANSITIONS: {"node1": cnd.exact_match(Message("Hi"))},
},
}
}
# testing
happy_path = (
(
Message("Hi"),
Message("Hello, how are you?"),
), # start_node -> node1
(
Message("I'm fine, how are you?"),
Message("Good. What do you want to talk about?"),
), # node1 -> node2
(
Message("Let's talk about music."),
Message("Sorry, I can not talk about music now."),
), # node2 -> node3
(Message("Ok, goodbye."), Message("BYE")), # node3 -> node4
(Message("Hi"), Message("Hello, how are you?")), # node4 -> node1
(
Message("stop"),
Message(
misc={
"previous_node": ("greeting_flow", "node1"),
"last_request": Message("stop"),
}
),
),
# node1 -> fallback_node
(
Message("one"),
Message(
misc={
"previous_node": ("greeting_flow", "fallback_node"),
"last_request": Message("one"),
}
),
), # f_n->f_n
(
Message("help"),
Message(
misc={
"previous_node": ("greeting_flow", "fallback_node"),
"last_request": Message("help"),
}
),
), # f_n->f_n
(
Message("nope"),
Message(
misc={
"previous_node": ("greeting_flow", "fallback_node"),
"last_request": Message("nope"),
}
),
), # f_n->f_n
(
Message("Hi"),
Message("Hi, what is up?"),
), # fallback_node -> node1
(
Message("I'm fine, how are you?"),
Message("Good. What do you want to talk about?"),
), # node1 -> node2
(
Message("Let's talk about music."),
Message("Sorry, I can not talk about music now."),
), # node2 -> node3
(Message("Ok, goodbye."), Message("BYE")), # node3 -> node4
)
[5]:
random.seed(31415) # predestination of choice
pipeline = Pipeline.from_script(
toy_script,
start_label=("greeting_flow", "start_node"),
fallback_label=("greeting_flow", "fallback_node"),
)
if __name__ == "__main__":
check_happy_path(pipeline, happy_path)
if is_interactive_mode():
run_interactive_mode(pipeline)
(user) >>> text='Hi'
(bot) <<< text='Hello, how are you?'
(user) >>> text='I'm fine, how are you?'
(bot) <<< text='Good. What do you want to talk about?'
(user) >>> text='Let's talk about music.'
(bot) <<< text='Sorry, I can not talk about music now.'
(user) >>> text='Ok, goodbye.'
(bot) <<< text='BYE'
(user) >>> text='Hi'
(bot) <<< text='Hello, how are you?'
(user) >>> text='stop'
(bot) <<< misc='{'previous_node': ('greeting_flow', 'node1'), 'last_request': {'text': 'stop'}}'
(user) >>> text='one'
(bot) <<< misc='{'previous_node': ('greeting_flow', 'fallback_node'), 'last_request': {'text': 'one'}}'
(user) >>> text='help'
(bot) <<< misc='{'previous_node': ('greeting_flow', 'fallback_node'), 'last_request': {'text': 'help'}}'
(user) >>> text='nope'
(bot) <<< misc='{'previous_node': ('greeting_flow', 'fallback_node'), 'last_request': {'text': 'nope'}}'
(user) >>> text='Hi'
(bot) <<< text='Hi, what is up?'
(user) >>> text='I'm fine, how are you?'
(bot) <<< text='Good. What do you want to talk about?'
(user) >>> text='Let's talk about music.'
(bot) <<< text='Sorry, I can not talk about music now.'
(user) >>> text='Ok, goodbye.'
(bot) <<< text='BYE'