For those chatbots with complex Finite State Machine (FSM) behaviours, you will probably want to run them on a separate process, that is what for the SDK is made for. Its primary function is to execute the FSM's computations (transition's conditions and states) by running Remote Procedure Call (RPC) server that listen to the back-end requests.
Make sure the following list of packages are installed on your system:
- Python 3.10
- python3.10-dev
- python3.10-distutils
- poetry
poetry add chatfaq-sdk
Install project dependencies:
poetry install
First of all, create a .env file with the needed variables set. You can see an example of those on .env_example file. Next you can see the explanation of each variable:
CHATFAQ_RETRIEVAL_HTTP: The address for the HTTP of the back-end server.
CHATFAQ_BACKEND_WS: The address for the WS of the back-end server.
CHATFAQ_TOKEN: The token to authenticate with the back-end server. You can retrieve the auth token from the backend server:
curl -X POST -u username:password http://localhost:8000/back/api/login/
Run the example:
make run_example
This will run the example FSM that is located in ./examples/model_example/init.py file. You can modify this file to test your own FSMs.
Alternatively you can simply run the server using docker.
docker build -t chatfaq-sdk .
docker run chatfaq-sdk
This is just a dummy example that displays the basic usage of the library.
We are going to build the next FSM:
Import basic modules to build your first FMS:
import os
import random
from chatfaq_sdk import ChatFAQSDK
from chatfaq_sdk.fsm import FSMDefinition, State, Transition
from chatfaq_sdk.conditions import Condition
from chatfaq_sdk.layers import MessageDeclare the 3 possible states of our FSM:
def send_greeting(ctx: dict):
yield Message("Hello!")
yield Message("How are you?")
greeting_state = State(name="Greeting", events=[send_greeting], initial=True)
def send_answer(ctx: dict):
last_payload = ctx["conv_mml"][-1]["stack"][0]["payload"]["content"]
yield Message(
f'My answer to your message: "{last_payload}" is: {random.randint(0, 999)}'
)
yield Message("Tell me more")
answering_state = State(
name="Answering",
events=[send_answer],
)
def send_goodbye(ctx: dict):
yield Message("Byeeeeeeee!")
goodbye_state = State(
name="Goodbye",
events=[send_goodbye],
)Declare the only computable condition for the transitions of our FSM:
def is_saying_goodbye(ctx: dict):
if ctx["conv_mml"][-1]["stack"][0]["payload"]["content"] == "goodbye":
return Condition(1)
return Condition(0)Now lets glue everything together:
Declare our transitions
any_to_goodbye = Transition(dest=goodbye_state, conditions=[is_saying_goodbye])
greeting_to_answer = Transition(
source=greeting_state,
dest=answering_state,
unless=[is_saying_goodbye],
)
answer_to_answer = Transition(
source=answering_state, dest=answering_state, unless=[is_saying_goodbye]
)Build the final instance of our FSM:
fsm_definition = FSMDefinition(
states=[greeting_state, answering_state, goodbye_state],
transitions=[greeting_to_answer, any_to_goodbye, answer_to_answer],
)Finally, run the RPC server loop with the previously built FSM:
import os
sdk = ChatFAQSDK(
chatfaq_retrieval_http="http://localhost:8000",
chatfaq_ws="ws://localhost:8000",
token=os.getenv("CHATFAQ_TOKEN"),
fsm_name="my_first_fsm",
fsm_definition=fsm_definition,
)
sdk.connect()