Zero-dependency library for chat-bot creators with deadlines. It allows you to describe a conversation, talk with user according to your schema and restore it, if something went wrong.
Project description
PyConversation
Zero-dependency library for chat-bot creators with deadlines.
It allows you to describe a conversation, talk with user according to your schema and restore it, if something went wrong.
Table of contents
Quickstart
First we need to create a message schema, which consists of messages.
Messages describe chat-bot's actions. For instance, send a text message, which doesn't need any feedback or ask a question. Each message has a unique id. Most common types of messages are Group, Text and Ask.
For full details about different message types see Messages
Group is a kind of container, that holds list of other messages.
Text sends a text message, which doesn't require any feedback.
Ask sends a text message and waits for an answer
Enough theory, let's see an example!
from pyconversation import Group, Text, Ask
fruit_bot_conversation = Group(
id="root",
children=[
Text(id="root.hello", text="Hello!"),
Ask(id="root.fruits", text="What fruits do you like?"),
Text(id="root.bye", text="Bye"),
],
)
In this example, we create a schema for simple bot, who asks which fruits does user like. Root message is a Group. It holds a block of messages. First of them is a Text which sends user a greeting message. Second one (Ask) asks about user's favorite fruit and waits for answer. And finally, third Text message sends Bye to user.
Second step - we need a logger
Logger is an object, which stores user's answers and message history. This library exposes 2 loggers:
DictLogger- stores data in a dictionaryJsonFileLogger- takes file path as a parameter and stores json data in this file
If you need something different, see Creating Own Loggers
But now let's use DictLogger
from pyconversation import DictLogger
logger = DictLogger()
That's all!
For full loggers documentation see Loggers
But how to send those messages?
The answer's simple - using a MessageSender!
Example code:
from pyconversation import Group, Text, Ask, DictLogger, MessageSender
# Conversation from step 1
fruit_bot_conversation = Group(
id="root",
children=[
Text(id="root.hello", text="Hello!"),
Ask(id="root.fruits", text="What fruits do you like?"),
Text(id="root.bye", text="Bye"),
],
)
# Logger from step 2
logger = DictLogger()
# Initialize a message sender
sender = MessageSender(
root=fruit_bot_conversation, # Our conversation
logger=logger, # Our logger
send=print # A send function, which takes a string and sends the message. In this case, we use print to log messages to console
)
# Answer to the question before the first one is always empty
answer = None
# Send messages!
while True:
# Send messages one by one, until we run into a message, which requires an answer
# This function takes answer to previous question as a parameter
sender.send_all_skippable(answer)
# If all messages sent
if sender.finished:
# Dispose of sender's resources (like open files) and get the result!
print("\nResult:", sender.finalize())
break
# If not all messages have been sent and we still need an answer, ask!
answer = input()
Done! If you run this, you'll get the following in the console:
Hello!
What fruits do you like?
<your answer'll be here>
Bye
Result: {'root.fruits': '<your answer>'}
And one more example with decorated function (like in real chat-bots):
bot = ... # Initialize chat bot
sender = None
@bot.connection
def on_connection(user_id):
sender = MessageSender(
root=conversation, # Our conversation
logger=logger, # Our logger
send=lambda text: bot.send(user_id, text)
)
sender.send_all_skippable(None)
@bot.message
def on_message(user_id, message):
sender.send_all_skippable(message)
if sender.finished:
print("\nResult:", sender.finalize())
sender = None
For full message sender documentation see Message Sender
You've created your first chat-bot with clever conversation! Here quick tutorial ends.
Messages
Text
Text message sends some text, which doesn't require user's answer
Constructor parameters:
id(str) - unique message idtext(str) - text to send
Usage example:
Text(id="hello", text="Hello, user!")
Group
Group is a message, which doesn't send anything and doesn't require an answer. It's just a container for a list of messages
Constructor parameters:
id(str) - unique message idchildren(list[message]) - list of messages to send
Usage example:
Group(
id="group",
children=[
Text(id="hello", text="Hello!"),
Text(id="bye", text="Good bye!"),
],
)
Ask
Ask message send some text to user and waits for an answer
Constructor parameters:
id(str) - unique message idtext(str) - question text
Usage example:
Ask(id="name", text="What's your name?")
Switch
Switch message asks user a question and sends a message depending on user's answer.
Constructor parameters:
id(str) - unique message idtext(str) - question textanswer_map(dict[str, message]) - dict, where key is user's answer and value is a messagefallback(message?) - message, which'll be sent if answer doesn't match anything inanswer_mapdictrepeat_on_fallback(bool?) - if true, after fallback was sent question is asked over and over again until answer matches something inanswer_mapdict
Usage example:
Switch(
id="fruit"
text="What fruit do you like?"
answer_map={
"apple": Text(id="apple", text="Yeah, apples are delicious!"),
"peach": Text(id="peach", text="Me too!"),
"feijoa": Text(id="feijoa", text="I don't know that fruit!"),
},
fallback=Text(id="dont_understand", text="Sorry, I didn't understand you"),
repeat_on_fallback=True
)
ListAsk
ListAsk asks user a question and waits for several answers.
In result dictionary it's represented by an array.
Constructor parameters:
id(str) - unique message idtext(str) - question textstop_command(str) - if user sends this string as an answer, ListAsk finishes waiting for answersmax_count(int?) - maximal count of answers
Usage example:
ListAsk(
id="fruits",
text="What fruits do you like? Enter 'that's all' if you can't remember any more",
stop_command="that's all",
max_count=10,
)
TerminateGroup
TerminateGroup sends another message and then terminates sending group, inside which it is located
Constructor parameters:
id(str) - unique message idchild(message?) - message to send before terminating the group
Usage example:
Group( # This group's gonna be terminated
id="group",
children=[
Text(id="hello", text="Hello!"),
Switch(
id="bye_condition",
text="Can I say bye?",
answer_map={
"yes": Text(id="bye", text="Good bye!")
},
fallback=TerminateGroup(
id="terminate",
child=Text(id="eh", text="Eh..."),
),
),
Text(id="what", text="What?!"), # This will not be sent,
],
)
Creating Own Messages
Every message is a class, so to create your own message, you just need to inherit BaseMessage class (It can be imported like this: from pyconversation import BaseMessage)
Usage example:
from pyconversation import Text, BaseMessage, BaseLogger, MessageTransfer, MessageTransferGenerator
class HelloMessage(BaseMessage):
username: str
def __init__(self, *, id: str, username: str) -> None:
super().__init__(id=id) # BaseMessage takes one parameter - id
self.username = username
def _base_iterator(self, logger: BaseLogger) -> MessageTransferGenerator: # This is an abstract method
text_message = Text(id=f"{self.id}.text", text=f"Hello, {self.username}!")
yield from text_message.iterator(logger)
answer = yield MessageTransfer(
id=self.id,
text="Is it your real name?",
)
logger.log(self.id, answer)
As you can see, each message has an iterator method, which takes logger as a parameter and returns a generator. Also, this message gets an answer and logs it to logger. Details on how to interact with logger and log answers will be explained in Loggers
But what is that MessageTransfer object? It's used to pass string message to sender and get an answer. Details in next article.
MessageTransfer
Message transfer is used to pass string message to sender and get an answer. It can be yielded from message's generator.
Constructor parameters:
id(str) - message's unique idtext(str?) - text, which'll be sent to user or None, if you don't want to ask any questions, you just need an answerskip(bool?) - if true, this question doesn't need an answer and won't wait for it.terminate_group(bool?) - when this is true, group which intercepted such transfer processes it and terminates.
Usage example in upper Creating Own Messages section
Loggers
Loggers are used to store users' answers and message history.
Message history is a list, where question ids are stored. It's used to restore conversation. For example, if user has already answered several questions and suddenly the server stops, last sent message id will be taken from history, and conversation will begin from the last message.
DictLogger
DictLogger stores answers and history in-memory (in a dictionary). So it's just an example to play with the library. Don't use it in production code.
No constructor parameters.
Usage example:
logger = DictLogger()
JsonFileLogger
JsonFileLogger stores everything in a JSON file. JSON file stays on the computer anyway, so when server suddenly stops and the reboots, your bot'll be able to continue conversation from the right place.
Constructor parameters:
file_path(str) - JSON file's absolute path. It must be unique between all conversations on this server.
Usage example:
logger = JsonFileLogger(pathlib.Path(__file__).parent / "conversation.json")
Creating Own Loggers
If you need to create your own logger (and you'll need it more often, than creating own messages) you need to inherit the BaseLogger class.
It has the following abstract methods:
-
log(-> None) - stores answer by message's unique idParameters:
id(str) - message unique idvalue(str) - answer
-
set_array(-> None) - initializes empty list in answer dictionary using message unique id as a keyParameters:
id(str) - message unique id
-
add_array_item(-> None) - add item to existing list using message id as answer dictionary keyParameters:
id(str) - message unique idvalue(str) - value to add to list
-
get(-> union[str, list[str], None]) - get message answer or list of answers by message id if existsParameters:
id(str) - message unique id
-
get_result_dict(-> dict[str, union[str, list[str]]]) - get full answer dictionaryNo parameters
And also the following virtual methods (not necessary to implement):
-
reset_history(-> None) - remove all elements from message history listNo parameters
-
log_last_id(-> None) - add message id to message history listParameters:
id(str) - message unique id
-
get_last_id(-> str?) - get last sent message id (last element in message history list)No parameters
-
finalize(-> None) - dispose of logger's resources (open files, socket connections, etc.)Note: This method is called when the conversation is finished. So, for instance,
JsonFileLoggerdeletes it's data file in this method.No parameters
Usage example:
from typing import Union, List, Dict
from pyconversation import BaseLogger
class MySocketLogger(BaseLogger):
socket: Socket
def __init__(self) -> None:
super().__init__()
self._connect_socket()
def log(self, id: str, value: str) -> None:
self.socket.emit("SET_OR_REPLACE", {"id": id, "value": value})
def set_array(self, id: str) -> None:
self.socket.emit("SET_OR_REPLACE", {"id": id, "value": []})
def add_array_item(self, id: str, value: str) -> None:
self.socket.emit("ADD_ARRAY_ITEM", {"id": id, "value": value})
def get(self, id: str) -> Union[str, List[str]]:
return self.socket.emit("GET", {"id": id})
def get_result_dict(self) -> Dict[str, Union[str, List[str]]]:
return self.socket.emit("GET_ALL")
def reset_history(self) -> None:
self.socket.emit("SET_HISTORY", [])
def log_last_id(self, id: str) -> None:
self.socket.emit("ADD_HISTORY", id)
def get_last_id(self, id: str) -> Union[str, None]:
if not self.socket.emit("HISTORY_EMPTY"):
return self.socket.emit("GET_LAST_IN_HISTORY")
def finalize(self) -> None:
self.socket.emit("CLEAR_EVERYTHING")
self._disconnect_socket()
def _connect_socket(self) -> None:
self.socket = ... # We'll log our data using a socket
def _disconnect_socket(self) -> None:
self.socket.disconnect()
self.socket = None
Message Sender
Message sender is used to simplify conversation restoring and message sending.
Constructor parameters:
root(message) - root message (aka message schema)logger(logger) - loggersend(function (str) -> None) - send function (takes string and sends it to user)headline_text(str?) - text, which'll be sent to user whent message sender is constructed. Whether conversation is constructed or restored, it's sent anyway.stop_command(str?) - if user sends this as an answer, conversation terminates.
Exposed properties:
finished(bool) - is conversation finished (true if all messages have been sent or conversation has been stopped by stop command)terminated(bool) - is conversation terminated (true if conversation was stopped by stop command)
Exposed methods:
-
send_all_skippableSend all messages until sender runs into a message, which requires an answer.Parameters:
prev_answer(str?) - answer to previous message
See usage example in Quickstart
Compatibility
This library is compatible with Python>=3.6
© 2021 Roman Melamud
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Hashes for pyconversation-1.0.7-py3-none-any.whl
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 | 53ac4fb70708db84967595bb0e400b8ba5ebc8dbe30966a7a87fa1b14e49da41 |
|
| MD5 | 6fded226558575ca4a20699261f80457 |
|
| BLAKE2b-256 | d38f9293edcc56ca53b8be2083a4bc0b371f28274d1d29ab1f3c46ecacbd96f4 |
