Genie Flow Code Examples

Question and Answer

The first, simple Question and Answer dialogue can be defined as follows:

from statemachine import State

from genie_flow.genie import GenieModel, GenieStateMachine


class QandAModel(GenieModel):

    @classmethod
    def get_state_machine_class(cls) -> type["GenieStateMachine"]:
        return QandAMachine


class QandAMachine(GenieStateMachine):
    # STATES
    intro = State(initial=True, value=000)
    user_enters_query = State(value=100)
    ai_creates_response = State(value=200)

    # EVENTS AND TRANSITIONS
    user_input = intro.to(ai_creates_response) | user_enters_query.to(ai_creates_response)
    ai_extraction = ai_creates_response.to(user_enters_query)

    # TEMPLATES
    templates = dict(
        intro="q_and_a/intro.jinja2",
        user_enters_query="q_and_a/user_input.jinja2",
        ai_creates_response="q_and_a/ai_response.jinja2"
    )

data model

First, the QandAModel data model class is defined. Mark that this is a subclass of GenieModel making sure that all the required properties are available and the newly created data model can be persisted. This new data model class is, in essence, a Pydantic model. All features of the Pydantic framework can be used.

In this example, there is not much happening as we are not extracting any data from the dialogue, except for the dialogue itself, which is a standard feature of Genie Flow.

state machine - states

Secondly, we defined the state machine class QandAMachine, which is a subclass of GenieStateMachine to ensure the base Genie Flow functionality is available. Within this class, we define our states, transitions and templates.

States are class properties, instantiated with a State object. This is the Python State Machine object identifying a state machine state. Here we see the state intro getting the flag initial=True, which identifies this state to be the initial state (of which there can only be one). Since this example shows a never-ending dialogue, there is no state with the predicate final=True.

All states in this QandAMachine class have a unique value. This is a str or int that uniquely identifies the state. These are defined by the developer, and their value is insignificant.

state machine - events and transitions

We also define the transitions that are possible between states. This is done by assigning transitions to the event that triggers them. So in this example, the event user_input will make the machine transition to the state ai_creates_response when it is currently either in the state intro or the state user_enters_query. Here, the two transitions, intro to ai_creates_response and user_enters_query to ai_creates_response are chained together using the | character.

The event ai_extraction is defined to trigger the remaining transition of this state machine, which is the transition between ai_creates_respponse to user_enters_query.

state machine - templates

And finally, we define the templates that are linked to each and every state. This is done by implementing a class property called templates that is instantiated with a dictionary. Every state should be a key in that dictionary, referring to the path to the Jinja2 file with the given template.

At this point in time, the Jinja2 environment needs to be told where to find the templates. This is done at application start-up by calling the following function: register_template_directory("q_and_a", "example_qa/templates"), imported as from ai_state_machine.templates import register_template_directory. The prefix q_and_a serving as a virtual directory when referring to a specific file.

At the initiation of a new Genie Flow state machine, it is checked to see if all states have a template assigned and will raise an exception if not all of them have one.

Templates are used to both render the output that needs to be sent to the user as well as the prompt that needs to be sent to the LLM. They are Jinja2 templates and are rendered with the data that is captured so far. For example, the intro state has the following template:

Welcome to this simple Question and Answer dialogue Genie Flow example!

How can I help? Please go ahead and ask me anything.

This is the template that is sent to the user as a welcome message. When the user then enters a query, a prompt is sent to the LLM. That prompt is constructed using the template that is linked to the state ai_creates_response:

You are a friendly chatbot, aiming to have a dialogue with a human user. You will be given
the dialogue that you have had before, followed by the most recent response from the user.
Your aim is to respond logically, taking the dialogue you had into account.

---
*** DIALOGUE SO FAR ***
{{ chat_history }}
---

---
*** MOST RECENT HUMAN STATEMENT ***
{{ actor_input }}
___

First assert if the most recent human statement indicated that the user wants to stop the dialog.
If the user wants to stop the dialogue, just say **STOP**.

If the user does not want to stop, respond.
Be succinct, to the point, but friendly.
Stick to the language that the user start their conversation in.

Here the power of Jinja2 comes to bear. The template contains "mustache notation" to indicate the placeholders for data attributes. So will {{ actor_input }} be replaced by the most recent statement by the most recent actor. Any attribute of the QandAModel is available inside the template.

That makes the template for state user_enters_query straight forward:

{{ actor_input }}

Meaning that it just prints the output of the LLM. Because when this template gets rendered, the actor was the LLM, and the attribute actor_input will be the output of the LLM.

Developing a subclass of

Question and Answer with Conditions

Putting conditions on transitions is straightforward. In the source code of q_and_a_cond.py, it can be seen that we introduced a state outro, with an attribute final=True, which indicates this is a final state from which no more transitions can be made.

There are also added conditions to some of the transitions, as can be seen from this snippet:

    user_input = (
            intro.to(ai_creates_response) |
            user_enters_query.to(ai_creates_response, unless="user_says_stop") |
            user_enters_query.to(outro, cond="user_says_stop")
    )

Here, the event user_input will trigger a transition from state intro to ai_creates_response without any condition. But the transition from user_enters_query will only be towards state ai_creates_response unless user_says_stop and towards the state outro under the condition user_says_stop. In normal words: if the user says stop, we transition directly to the outro state; otherwise, we go to state ai_creates_response

These conditions are plain Python methods that we defined on our state machine class:

from statemachine.event_data import EventData

from genie_flow.genie import GenieStateMachine


class QandACondMachine(GenieStateMachine):
    ...

    def user_says_stop(self, event_data: EventData):
        return (
                event_data.args is not None and
                len(event_data.args) != 0 and
                event_data.args[0] == "*STOP*"
        )

This method is called as the very first method when our state machine receives an event that results in a transition with a condition. In the order of play, our Genie Flow code by then has not yet had a chance to do anything, so we need to deal with the raw EventData object that is passed by the Python State Machine framework.

This method checks the data that is received with the event. The EventData class carries a list of arguments that were passed when the event was sent. So it is checked to see if it is not None, does not have zero length and if the first parameter has the value *STOP*. This is the text that a user can enter to indicate they are done with the dialogue and want to stop.

The condition is linked to the transition by stating the name of the method, as can be seen from the above snippet. They can be stated "positively" (as in: this must be True to make the transition), by making it a cond on the transition. They can also be stated negatively (as in: this must be False to make the transition), by making it an unless condition.

Value Injection

The Python State Machine engine takes care of injecting relevant values into condition methods. As described in Dependency Injection, the following values are available:

event: The name of the event that triggered the transition
event_data: An EventData object containing all event-related values
source: The State that the engine transitions out of
target: The State that the engine transitions into
state: The current State the machine is in
model: A reference to the model attached to the state machine
transition: A Transition object carrying the transition information

Question and Answer with data capture

The final Question and Answer example captures the username. This means we now want to capture that name and be able to use it in our templates. Example code for this can be found in q_and_a_capture.py.

data model

We now need to add the user_name as a property to our data model. It is the data attribute that we want to capture and carry across the dialogue.

from typing import Optional

from pydantic import Field

from genie_flow.genie import GenieModel


class QandACaptureModel(GenieModel):
    user_name: Optional[str] = Field(None, description="The name of the user")

    ...

In true Pydantic style, we define the class property user_name to be an optional str, with a default value of None, and we give it a short description. This now means that our data model has a property user_name that is persisted and available when templates get rendered.

data capture

We now also need to program the capturing of the data. In this example, the user enters their name, which is then extracted using an LLM prompt. If the LLM cannot make out the name, the user is asked again to state their name. This validation is done by commanding the LLM to respond with a given token if it cannot make out a username, using the following LLM template:

You will be given a human response to the question "what is your name".
Your task is to extract the name of the user from that response.
If you can not determine that name, just respond with UNDEFINED.
If you can determine that name, just response with the name, nothing else.

This means we can define a condition as follows:

    def name_is_defined(self, event_data: EventData) -> bool:
        return (
            event_data.args is not None and
            len(event_data.args) != 0 and
            event_data.args[0] != "UNDEFINED"
        )

But the main meal here is the definition of a method that gets called when the LLM has extracted the username. This is done by the following method on our QandACaptureMachine class:

from genie_flow.genie import GenieStateMachine


class QandACaptureMachine(GenieStateMachine):
    ...

    def on_exit_ai_extracts_name(self):
        self.model.user_name = self.model.actor_input

At the time the state ai_extracts_name is exited, the actor_input property of our data model contains the value that is returned by the LLM. In the sunny day scenario, the name that is stated by the user. We therefore assign it to the data model property user_name.

Bear in mind that at this stage, we only know we are transitioning out of the ai_extracts_name state, not if the username has been extracted. That is determined by the condition. So we could be assigning the value UNDEFINED to model.user_name. An alternative would be to create the following method:

from genie_flow.genie import GenieStateMachine


class QandACaptureMachine(GenieStateMachine):
    ...

    def on_enter_welcome_message(self):
        self.model.user_name = self.model.actor_input

This message would be called when we enter the state welcome_message, a point in the flow where we know we have a username that is different from UNDEFINED.

Either of these methods is called after the Genie Flow framework has had a chance to process information. So we can refer to model.actor_input.

conclusion

We have now seen how the concepts laid out in the previous chapter can be expressed in code.