Skip to content

API Reference - Core Package#

src #

intentional_core #

Init file for intentional_core.

__about__ #

Package descriptors for intentional-core.

bot_interface #

Functions to load bots from config files.

BotInterface #

Bases: ABC

Tiny base class used to recognize Intentional bots interfaces.

The class name is meant to represent the communication channel you will use to interact with your bot. For example an interface that uses a local command line interface would be called "LocalBotInterface", one that uses Whatsapp would be called "WhatsappBotInterface", one that uses Twilio would be called "TwilioBotInterface", etc.

In order for your bot to be usable, you need to assign a value to the name class variable in the class definition.

Source code in intentional-core/src/intentional_core/bot_interface.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
class BotInterface(ABC):
    """
    Tiny base class used to recognize Intentional bots interfaces.

    The class name is meant to represent the **communication channel** you will use to interact with your bot.
    For example an interface that uses a local command line interface would be called "LocalBotInterface", one that
    uses Whatsapp would be called "WhatsappBotInterface", one that uses Twilio would be called "TwilioBotInterface",
    etc.

    In order for your bot to be usable, you need to assign a value to the `name` class variable in the class definition.
    """

    name: Optional[str] = None
    """
    The name of the bot interface. This should be a unique identifier for the bot interface.
    This string will be used in configuration files to identify the bot interface.

    The bot interface name should directly recall the class name as much as possible.
    For example, the name of "LocalBotInterface" should be "local", the name of "WhatsappBotInterface" should be
    "whatsapp", etc.
    """

    @abstractmethod
    async def run(self):
        """
        Run the bot interface.

        This method should be overridden by the subclass to implement the bot's main loop.
        """
        raise NotImplementedError("BotInterface subclasses must implement the run method.")
name: Optional[str] = None class-attribute instance-attribute #

The name of the bot interface. This should be a unique identifier for the bot interface. This string will be used in configuration files to identify the bot interface.

The bot interface name should directly recall the class name as much as possible. For example, the name of "LocalBotInterface" should be "local", the name of "WhatsappBotInterface" should be "whatsapp", etc.

run() abstractmethod async #

Run the bot interface.

This method should be overridden by the subclass to implement the bot's main loop.

Source code in intentional-core/src/intentional_core/bot_interface.py 
49
50
51
52
53
54
55
56
@abstractmethod
async def run(self):
    """
    Run the bot interface.

    This method should be overridden by the subclass to implement the bot's main loop.
    """
    raise NotImplementedError("BotInterface subclasses must implement the run method.")
load_bot_interface_from_dict(config) #

Load a bot interface, and all its inner classes, from a dictionary configuration.

Parameters:

Name Type Description Default
config Dict[str, Any]

The configuration dictionary.

 required

Returns:

Type Description
BotInterface

The bot interface instance.
Source code in intentional-core/src/intentional_core/bot_interface.py 
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
def load_bot_interface_from_dict(config: Dict[str, Any]) -> BotInterface:
    """
    Load a bot interface, and all its inner classes, from a dictionary configuration.

    Args:
        config: The configuration dictionary.

    Returns:
        The bot interface instance.
    """
    logger.debug("Loading bot interface from configuration:\n%s", json.dumps(config, indent=4))

    # Import all the necessary plugins
    plugins = config.pop("plugins")
    logger.debug("Plugins to import: %s", plugins)
    for plugin in plugins:
        import_plugin(plugin)

    # Initialize the intent router
    intent_router = IntentRouter(config.pop("conversation", {}))

    # Get all the subclasses of Bot
    subclasses: Set[BotInterface] = inheritors(BotInterface)
    logger.debug("Known bot interface classes: %s", subclasses)

    for subclass in subclasses:
        if not subclass.name:
            logger.error("Bot interface class '%s' does not have a name. This bot type will not be usable.", subclass)
            continue

        if subclass.name in _BOT_INTERFACES:
            logger.warning(
                "Duplicate bot interface type '%s' found. The older class (%s) "
                "will be replaced by the newly imported one (%s).",
                subclass.name,
                _BOT_INTERFACES[subclass.name],
                subclass,
            )
        _BOT_INTERFACES[subclass.name] = subclass

    # Identify the type of bot interface and see if it's known
    interface_class_ = config.pop("interface", None)
    if not interface_class_:
        raise ValueError("Bot configuration must contain an 'interface' key to know which interface to use.")

    if interface_class_ not in _BOT_INTERFACES:
        raise ValueError(
            f"Unknown bot interface type '{interface_class_}'. Available types: {list(_BOT_INTERFACES)}. "
            "Did you forget to add the correct plugin name in the configuration file, or to install it?"
        )

    # Handoff to the subclass' init
    logger.debug("Creating bot interface of type '%s'", interface_class_)
    return _BOT_INTERFACES[interface_class_](intent_router=intent_router, config=config)
load_configuration_file(path) #

Load an Intentional bot from a YAML configuration file.

Parameters:

Name Type Description Default
path Path

Path to the YAML configuration file.

 required

Returns:

Type Description
BotInterface

The bot instance.
Source code in intentional-core/src/intentional_core/bot_interface.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def load_configuration_file(path: Path) -> BotInterface:
    """
    Load an Intentional bot from a YAML configuration file.

    Args:
        path: Path to the YAML configuration file.

    Returns:
        The bot instance.
    """
    logger.debug("Loading YAML configuration file from '%s'", path)
    with open(path, "r", encoding="utf-8") as file:
        config = yaml.safe_load(file)
    return load_bot_interface_from_dict(config)

bot_structure #

Functions to load bot structure classes from config files.

BotStructure #

Bases: EventListener

Tiny base class used to recognize Intentional bot structure classes.

The bot structure's name is meant to represent the structure of the bot. For example a bot that uses a direct WebSocket connection to a model such as OpenAI's Realtime API could be called "RealtimeAPIBotStructure", one that uses a VAD-STT-LLM-TTS stack could be called "AudioToTextBotStructure", and so on

In order for your bot structure to be usable, you need to assign a value to the name class variable in the bot structure class' definition.

Source code in intentional-core/src/intentional_core/bot_structure.py 
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
class BotStructure(EventListener):
    """
    Tiny base class used to recognize Intentional bot structure classes.

    The bot structure's name is meant to represent the **structure** of the bot. For example a bot that uses a direct
    WebSocket connection to a model such as OpenAI's Realtime API could be called "RealtimeAPIBotStructure", one that
    uses a VAD-STT-LLM-TTS stack could be called "AudioToTextBotStructure", and so on

    In order for your bot structure to be usable, you need to assign a value to the `name` class variable in the bot
    structure class' definition.
    """

    name: Optional[str] = None
    """
    The name of this bot's structure. This should be a unique identifier for the bot structure type.

    The bot structure's name should directly recall the class name as much as possible. For example, the name of
    "WebsocketBotStructure" should be "websocket", the name of "AudioToTextBotStructure" should be "audio_to_text",
    etc.
    """

    def __init__(self) -> None:
        """
        Initialize the bot structure.
        """
        self.event_handlers: Dict[str, Callable] = {}

    async def connect(self) -> None:
        """
        Connect to the bot.
        """

    async def disconnect(self) -> None:
        """
        Disconnect from the bot.
        """

    @abstractmethod
    async def run(self) -> None:
        """
        Main loop for the bot.
        """

    @abstractmethod
    async def send(self, data: Dict[str, Any]) -> Dict[str, Any]:
        """
        Send a message to the bot.
        """

    @abstractmethod
    async def handle_interruption(self, lenght_to_interruption: int) -> None:
        """
        Handle an interruption in the streaming.

        Args:
            lenght_to_interruption: The length of the data that was produced to the user before the interruption.
                This value could be number of characters, number of words, milliseconds, number of audio frames, etc.
                depending on the bot structure that implements it.
        """

    def add_event_handler(self, event_name: str, handler: Callable) -> None:
        """
        Add an event handler for a specific event type.

        Args:
            event_name: The name of the event to handle.
            handler: The handler function to call when the event is received.
        """
        if event_name in self.event_handlers:
            logger.warning(
                "Event handler for '%s' already exists. The older handler will be replaced by the new one.",
                event_name,
            )

        logger.debug("Adding event handler for event '%s'", event_name)
        self.event_handlers[event_name] = handler

    async def handle_event(self, event_name: str, event: Dict[str, Any]) -> None:
        """
        Handle different types of events that the model may generate.
        """
        logger.debug("Received event '%s'", event_name)

        if "*" in self.event_handlers:
            logger.debug("Calling wildcard event handler for event '%s'", event_name)
            await self.event_handlers["*"](event)

        if event_name in self.event_handlers:
            logger.debug("Calling event handler for event '%s'", event_name)
            await self.event_handlers[event_name](event)
name: Optional[str] = None class-attribute instance-attribute #

The name of this bot's structure. This should be a unique identifier for the bot structure type.

The bot structure's name should directly recall the class name as much as possible. For example, the name of "WebsocketBotStructure" should be "websocket", the name of "AudioToTextBotStructure" should be "audio_to_text", etc.

__init__() #

Initialize the bot structure.

Source code in intentional-core/src/intentional_core/bot_structure.py 
45
46
47
48
49
def __init__(self) -> None:
    """
    Initialize the bot structure.
    """
    self.event_handlers: Dict[str, Callable] = {}
add_event_handler(event_name, handler) #

Add an event handler for a specific event type.

Parameters:

Name Type Description Default
event_name str

The name of the event to handle.

 required
handler Callable

The handler function to call when the event is received.

 required
Source code in intentional-core/src/intentional_core/bot_structure.py 
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
def add_event_handler(self, event_name: str, handler: Callable) -> None:
    """
    Add an event handler for a specific event type.

    Args:
        event_name: The name of the event to handle.
        handler: The handler function to call when the event is received.
    """
    if event_name in self.event_handlers:
        logger.warning(
            "Event handler for '%s' already exists. The older handler will be replaced by the new one.",
            event_name,
        )

    logger.debug("Adding event handler for event '%s'", event_name)
    self.event_handlers[event_name] = handler
connect() async #

Connect to the bot.

Source code in intentional-core/src/intentional_core/bot_structure.py 
51
52
53
54
async def connect(self) -> None:
    """
    Connect to the bot.
    """
disconnect() async #

Disconnect from the bot.

Source code in intentional-core/src/intentional_core/bot_structure.py 
56
57
58
59
async def disconnect(self) -> None:
    """
    Disconnect from the bot.
    """
handle_event(event_name, event) async #

Handle different types of events that the model may generate.

Source code in intentional-core/src/intentional_core/bot_structure.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
async def handle_event(self, event_name: str, event: Dict[str, Any]) -> None:
    """
    Handle different types of events that the model may generate.
    """
    logger.debug("Received event '%s'", event_name)

    if "*" in self.event_handlers:
        logger.debug("Calling wildcard event handler for event '%s'", event_name)
        await self.event_handlers["*"](event)

    if event_name in self.event_handlers:
        logger.debug("Calling event handler for event '%s'", event_name)
        await self.event_handlers[event_name](event)
handle_interruption(lenght_to_interruption) abstractmethod async #

Handle an interruption in the streaming.

Parameters:

Name Type Description Default
lenght_to_interruption int

The length of the data that was produced to the user before the interruption. This value could be number of characters, number of words, milliseconds, number of audio frames, etc. depending on the bot structure that implements it.

 required
Source code in intentional-core/src/intentional_core/bot_structure.py 
73
74
75
76
77
78
79
80
81
82
@abstractmethod
async def handle_interruption(self, lenght_to_interruption: int) -> None:
    """
    Handle an interruption in the streaming.

    Args:
        lenght_to_interruption: The length of the data that was produced to the user before the interruption.
            This value could be number of characters, number of words, milliseconds, number of audio frames, etc.
            depending on the bot structure that implements it.
    """
run() abstractmethod async #

Main loop for the bot.

Source code in intentional-core/src/intentional_core/bot_structure.py 
61
62
63
64
65
@abstractmethod
async def run(self) -> None:
    """
    Main loop for the bot.
    """
send(data) abstractmethod async #

Send a message to the bot.

Source code in intentional-core/src/intentional_core/bot_structure.py 
67
68
69
70
71
@abstractmethod
async def send(self, data: Dict[str, Any]) -> Dict[str, Any]:
    """
    Send a message to the bot.
    """
ContinuousStreamBotStructure #

Bases: BotStructure

Base class for structures that support continuous streaming of data, as opposed to turn-based message exchanges.

Source code in intentional-core/src/intentional_core/bot_structure.py 
116
117
118
119
class ContinuousStreamBotStructure(BotStructure):
    """
    Base class for structures that support continuous streaming of data, as opposed to turn-based message exchanges.
    """
TurnBasedBotStructure #

Bases: BotStructure

Base class for structures that support turn-based message exchanges, as opposed to continuous streaming of data.

Source code in intentional-core/src/intentional_core/bot_structure.py 
122
123
124
125
class TurnBasedBotStructure(BotStructure):
    """
    Base class for structures that support turn-based message exchanges, as opposed to continuous streaming of data.
    """
load_bot_structure_from_dict(intent_router, config) #

Load a bot structure from a dictionary configuration.

Parameters:

Name Type Description Default
config Dict[str, Any]

The configuration dictionary.

 required

Returns:

Type Description
BotStructure

The BotStructure instance.
Source code in intentional-core/src/intentional_core/bot_structure.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
def load_bot_structure_from_dict(intent_router: IntentRouter, config: Dict[str, Any]) -> BotStructure:
    """
    Load a bot structure from a dictionary configuration.

    Args:
        config: The configuration dictionary.

    Returns:
        The BotStructure instance.
    """
    # Get all the subclasses of Bot
    subclasses: Set[BotStructure] = inheritors(BotStructure)
    logger.debug("Known bot structure classes: %s", subclasses)
    for subclass in subclasses:
        if not subclass.name:
            logger.error(
                "BotStructure class '%s' does not have a name. This bot structure type will not be usable.", subclass
            )
            continue

        if subclass.name in _BOT_STRUCTURES:
            logger.warning(
                "Duplicate bot structure type '%s' found. The older class (%s) "
                "will be replaced by the newly imported one (%s).",
                subclass.name,
                _BOT_STRUCTURES[subclass.name],
                subclass,
            )
        _BOT_STRUCTURES[subclass.name] = subclass

    # Identify the type of bot and see if it's known
    class_ = config.pop("type")
    logger.debug("Creating bot of type '%s'", class_)
    if class_ not in _BOT_STRUCTURES:
        raise ValueError(
            f"Unknown bot structure type '{class_}'. Available types: {list(_BOT_STRUCTURES)}. "
            "Did you forget to install your plugin?"
        )

    # Handoff to the subclass' init
    return _BOT_STRUCTURES[class_](config, intent_router)

events #

Base class for very simplified event emitter and listener.

EventEmitter #

Sends any event to the listener. TODO see if there's any scenario where we need more as this pattern is easy to extend but can get messy.

Source code in intentional-core/src/intentional_core/events.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
class EventEmitter:
    """
    Sends any event to the listener.
    TODO see if there's any scenario where we need more as this pattern is easy to extend but can get messy.
    """

    def __init__(self, listener: EventListener):
        """
        Register the listener.
        """
        self._events_listener = listener

    async def emit(self, event_name: str, event: Dict[str, Any]):
        """
        Send the event to the listener.
        """
        logger.debug("Emitting event %s", event_name)
        await self._events_listener.handle_event(event_name, event)
__init__(listener) #

Register the listener.

Source code in intentional-core/src/intentional_core/events.py 
33
34
35
36
37
def __init__(self, listener: EventListener):
    """
    Register the listener.
    """
    self._events_listener = listener
emit(event_name, event) async #

Send the event to the listener.

Source code in intentional-core/src/intentional_core/events.py 
39
40
41
42
43
44
async def emit(self, event_name: str, event: Dict[str, Any]):
    """
    Send the event to the listener.
    """
    logger.debug("Emitting event %s", event_name)
    await self._events_listener.handle_event(event_name, event)
EventListener #

Bases: ABC

Listens to events and handles them.

Source code in intentional-core/src/intentional_core/events.py 
15
16
17
18
19
20
21
22
23
24
class EventListener(ABC):
    """
    Listens to events and handles them.
    """

    @abstractmethod
    async def handle_event(self, event_name: str, event: Dict[str, Any]):
        """
        Handle the event.
        """
handle_event(event_name, event) abstractmethod async #

Handle the event.

Source code in intentional-core/src/intentional_core/events.py 
20
21
22
23
24
@abstractmethod
async def handle_event(self, event_name: str, event: Dict[str, Any]):
    """
    Handle the event.
    """

intent_routing #

Intent routing logic.

IntentRouter #

Bases: Tool

Special tool used to alter the system prompt depending on the user's response.

Source code in intentional-core/src/intentional_core/intent_routing.py 
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
class IntentRouter(Tool):
    """
    Special tool used to alter the system prompt depending on the user's response.
    """

    name = "classify_response"
    description = "Classify the user's response for later use."
    parameters = [
        ToolParameter(
            "outcome",
            "The outcome the conversation reached, among the ones described in the prompt.",
            "string",
            True,
            None,
        ),
    ]

    def __init__(self, config: Dict[str, Any]) -> None:
        self.background = config.get("background", "You're a helpful assistant.")
        self.initial_message = config.get("initial_message", None)
        self.graph = networkx.MultiDiGraph()

        # Init the stages
        self.stages = {}
        for stage_name, stage_config in config["stages"].items():
            logger.debug("Loading stage %s", stage_name)
            self.stages[stage_name] = Stage(stage_config)
            self.stages[stage_name].tools[self.name] = self  # Add the intent router to the tools of each stage
            self.graph.add_node(stage_name)

        # Connect the stages
        for stage_name, stage in self.stages.items():
            for outcome_name, outcome_config in stage.outcomes.items():
                if outcome_config["move_to"] not in [*self.stages, BACKTRACKING_CONNECTION]:
                    raise ValueError(
                        f"Stage {stage_name} has an outcome leading to an unknown stage {outcome_config['move_to']}"
                    )
                self.graph.add_edge(stage_name, outcome_config["move_to"], key=outcome_name)

        # Initial prompt
        initial_stage = ""
        for stage_name, stage in self.stages.items():
            if START_CONNECTION in stage.accessible_from:
                if initial_stage:
                    raise ValueError("Multiple start stages found!")
                initial_stage = stage_name
        if not initial_stage:
            raise ValueError("No start stage found!")

        self.current_stage_name = initial_stage
        self.backtracking_stack = []

    @property
    def current_stage(self):
        """
        Shorthand to get the current stage instance.
        """
        return self.stages[self.current_stage_name]

    async def run(self, params: Dict[str, Any]) -> str:
        """
        Given the response's classification, returns the new system prompt and the tools accessible in this stage.

        Args:
            params: The parameters for the tool. Contains the `response_type`.

        Returns:
            The new system prompt and the tools accessible in this stage.
        """
        selected_outcome = params["outcome"]
        transitions = self.get_transitions()

        if selected_outcome not in self.current_stage.outcomes and selected_outcome not in transitions:
            raise ValueError(f"Unknown outcome {params['outcome']}")

        if selected_outcome in self.current_stage.outcomes:
            next_stage = self.current_stage.outcomes[params["outcome"]]["move_to"]

            if next_stage != BACKTRACKING_CONNECTION:
                # Direct stage to stage connection
                self.current_stage_name = next_stage
            else:
                # Backtracking connection
                self.current_stage_name = self.backtracking_stack.pop()
        else:
            # Indirect transition, needs to be tracked in the stack
            self.backtracking_stack.append(self.current_stage_name)
            self.current_stage_name = selected_outcome

        return self.get_prompt(), self.current_stage.tools

    def get_prompt(self):
        """
        Get the prompt for the current stage.
        """
        outcomes = "\n".join(f"  - {name}: {data['description']}" for name, data in self.current_stage.outcomes.items())
        transitions = "\n".join(f"  - {stage}: {self.stages[stage].description}" for stage in self.get_transitions())
        return DEFAULT_PROMPT_TEMPLATE.format(
            intent_router_tool=self.name,
            stage_name=self.current_stage_name,
            background=self.background,
            current_goal=self.current_stage.goal,
            outcomes=outcomes,
            transitions=transitions,
        )

    def get_transitions(self):
        """
        Return a list of all the stages that can be reached from the current stage.
        """
        return [
            name
            for name, stage in self.stages.items()
            if (
                (self.current_stage_name in stage.accessible_from or "_all_" in stage.accessible_from)
                and name != self.current_stage_name
            )
        ]
current_stage property #

Shorthand to get the current stage instance.

get_prompt() #

Get the prompt for the current stage.

Source code in intentional-core/src/intentional_core/intent_routing.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
def get_prompt(self):
    """
    Get the prompt for the current stage.
    """
    outcomes = "\n".join(f"  - {name}: {data['description']}" for name, data in self.current_stage.outcomes.items())
    transitions = "\n".join(f"  - {stage}: {self.stages[stage].description}" for stage in self.get_transitions())
    return DEFAULT_PROMPT_TEMPLATE.format(
        intent_router_tool=self.name,
        stage_name=self.current_stage_name,
        background=self.background,
        current_goal=self.current_stage.goal,
        outcomes=outcomes,
        transitions=transitions,
    )
get_transitions() #

Return a list of all the stages that can be reached from the current stage.

Source code in intentional-core/src/intentional_core/intent_routing.py 
149
150
151
152
153
154
155
156
157
158
159
160
def get_transitions(self):
    """
    Return a list of all the stages that can be reached from the current stage.
    """
    return [
        name
        for name, stage in self.stages.items()
        if (
            (self.current_stage_name in stage.accessible_from or "_all_" in stage.accessible_from)
            and name != self.current_stage_name
        )
    ]
run(params) async #

Given the response's classification, returns the new system prompt and the tools accessible in this stage.

Parameters:

Name Type Description Default
params Dict[str, Any]

The parameters for the tool. Contains the response_type.

 required

Returns:

Type Description
str

The new system prompt and the tools accessible in this stage.
Source code in intentional-core/src/intentional_core/intent_routing.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
async def run(self, params: Dict[str, Any]) -> str:
    """
    Given the response's classification, returns the new system prompt and the tools accessible in this stage.

    Args:
        params: The parameters for the tool. Contains the `response_type`.

    Returns:
        The new system prompt and the tools accessible in this stage.
    """
    selected_outcome = params["outcome"]
    transitions = self.get_transitions()

    if selected_outcome not in self.current_stage.outcomes and selected_outcome not in transitions:
        raise ValueError(f"Unknown outcome {params['outcome']}")

    if selected_outcome in self.current_stage.outcomes:
        next_stage = self.current_stage.outcomes[params["outcome"]]["move_to"]

        if next_stage != BACKTRACKING_CONNECTION:
            # Direct stage to stage connection
            self.current_stage_name = next_stage
        else:
            # Backtracking connection
            self.current_stage_name = self.backtracking_stack.pop()
    else:
        # Indirect transition, needs to be tracked in the stack
        self.backtracking_stack.append(self.current_stage_name)
        self.current_stage_name = selected_outcome

    return self.get_prompt(), self.current_stage.tools
Stage #

Describes a stage in the bot's conversation.

Source code in intentional-core/src/intentional_core/intent_routing.py 
163
164
165
166
167
168
169
170
171
172
173
174
175
class Stage:
    """
    Describes a stage in the bot's conversation.
    """

    def __init__(self, config: Dict[str, Any]) -> None:
        self.goal = config["goal"]
        self.description = config.get("description", "--no description provided--")
        self.accessible_from = config.get("accessible_from", [])
        if isinstance(self.accessible_from, str):
            self.accessible_from = [self.accessible_from]
        self.tools = load_tools_from_dict(config.get("tools", {}))
        self.outcomes = config.get("outcomes", {})

model_client #

Functions to load model client classes from config files.

ContinuousStreamModelClient #

Bases: ModelClient

Base class for model clients that support continuous streaming of data, as opposed to turn-based message exchanges.

Source code in intentional-core/src/intentional_core/model_client.py 
113
114
115
116
class ContinuousStreamModelClient(ModelClient):
    """
    Base class for model clients that support continuous streaming of data, as opposed to turn-based message exchanges.
    """
ModelClient #

Bases: ABC, EventEmitter

Tiny base class used to recognize Intentional model clients.

In order for your client to be usable, you need to assign a value to the _name class variable in the client class' definition.

Source code in intentional-core/src/intentional_core/model_client.py 
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
class ModelClient(ABC, EventEmitter):
    """
    Tiny base class used to recognize Intentional model clients.

    In order for your client to be usable, you need to assign a value to the `_name` class variable
    in the client class' definition.
    """

    name: Optional[str] = None
    """
    The name of the client. This should be a unique identifier for the client type.
    This string will be used in configuration files to identify the type of client to serve a model from.
    """

    def __init__(self, parent: "BotStructure", intent_router: IntentRouter) -> None:
        """
        Initialize the model client.

        Args:
            parent: The parent bot structure.
        """
        super().__init__(parent)
        self.intent_router = intent_router

    async def connect(self) -> None:
        """
        Connect to the model.
        """
        await self.emit("on_model_connection", {})

    async def disconnect(self) -> None:
        """
        Disconnect from the model.
        """
        await self.emit("on_model_disconnection", {})

    @abstractmethod
    async def run(self) -> None:
        """
        Handle events from the model by either processing them internally or by translating them into higher-level
        events that the BotStructure class can understand, then re-emitting them.
        """

    @abstractmethod
    async def send(self, data: Dict[str, Any]) -> None:
        """
        Send a unit of data to the model. The response is streamed out as an async generator.
        """

    @abstractmethod
    async def update_system_prompt(self) -> None:
        """
        Update the system prompt in the model.
        """

    @abstractmethod
    async def handle_interruption(self, lenght_to_interruption: int) -> None:
        """
        Handle an interruption while rendering the output to the user.

        Args:
            lenght_to_interruption: The length of the data that was produced to the user before the interruption.
                This value could be number of characters, number of words, milliseconds, number of audio frames, etc.
                depending on the bot structure that implements it.
        """
name: Optional[str] = None class-attribute instance-attribute #

The name of the client. This should be a unique identifier for the client type. This string will be used in configuration files to identify the type of client to serve a model from.

__init__(parent, intent_router) #

Initialize the model client.

Parameters:

Name Type Description Default
parent BotStructure

The parent bot structure.

 required
Source code in intentional-core/src/intentional_core/model_client.py 
54
55
56
57
58
59
60
61
62
def __init__(self, parent: "BotStructure", intent_router: IntentRouter) -> None:
    """
    Initialize the model client.

    Args:
        parent: The parent bot structure.
    """
    super().__init__(parent)
    self.intent_router = intent_router
connect() async #

Connect to the model.

Source code in intentional-core/src/intentional_core/model_client.py 
64
65
66
67
68
async def connect(self) -> None:
    """
    Connect to the model.
    """
    await self.emit("on_model_connection", {})
disconnect() async #

Disconnect from the model.

Source code in intentional-core/src/intentional_core/model_client.py 
70
71
72
73
74
async def disconnect(self) -> None:
    """
    Disconnect from the model.
    """
    await self.emit("on_model_disconnection", {})
handle_interruption(lenght_to_interruption) abstractmethod async #

Handle an interruption while rendering the output to the user.

Parameters:

Name Type Description Default
lenght_to_interruption int

The length of the data that was produced to the user before the interruption. This value could be number of characters, number of words, milliseconds, number of audio frames, etc. depending on the bot structure that implements it.

 required
Source code in intentional-core/src/intentional_core/model_client.py 
 95
 96
 97
 98
 99
100
101
102
103
104
@abstractmethod
async def handle_interruption(self, lenght_to_interruption: int) -> None:
    """
    Handle an interruption while rendering the output to the user.

    Args:
        lenght_to_interruption: The length of the data that was produced to the user before the interruption.
            This value could be number of characters, number of words, milliseconds, number of audio frames, etc.
            depending on the bot structure that implements it.
    """
run() abstractmethod async #

Handle events from the model by either processing them internally or by translating them into higher-level events that the BotStructure class can understand, then re-emitting them.

Source code in intentional-core/src/intentional_core/model_client.py 
76
77
78
79
80
81
@abstractmethod
async def run(self) -> None:
    """
    Handle events from the model by either processing them internally or by translating them into higher-level
    events that the BotStructure class can understand, then re-emitting them.
    """
send(data) abstractmethod async #

Send a unit of data to the model. The response is streamed out as an async generator.

Source code in intentional-core/src/intentional_core/model_client.py 
83
84
85
86
87
@abstractmethod
async def send(self, data: Dict[str, Any]) -> None:
    """
    Send a unit of data to the model. The response is streamed out as an async generator.
    """
update_system_prompt() abstractmethod async #

Update the system prompt in the model.

Source code in intentional-core/src/intentional_core/model_client.py 
89
90
91
92
93
@abstractmethod
async def update_system_prompt(self) -> None:
    """
    Update the system prompt in the model.
    """
TurnBasedModelClient #

Bases: ModelClient

Base class for model clients that support turn-based message exchanges, as opposed to continuous streaming of data.

Source code in intentional-core/src/intentional_core/model_client.py 
107
108
109
110
class TurnBasedModelClient(ModelClient):
    """
    Base class for model clients that support turn-based message exchanges, as opposed to continuous streaming of data.
    """
load_model_client_from_dict(parent, intent_router, config) #

Load a model client from a dictionary configuration.

Parameters:

Name Type Description Default
config Dict[str, Any]

The configuration dictionary.

 required

Returns:

Type Description
ModelClient

The ModelClient instance.
Source code in intentional-core/src/intentional_core/model_client.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
def load_model_client_from_dict(
    parent: "BotStructure", intent_router: IntentRouter, config: Dict[str, Any]
) -> ModelClient:
    """
    Load a model client from a dictionary configuration.

    Args:
        config: The configuration dictionary.

    Returns:
        The ModelClient instance.
    """
    # Get all the subclasses of ModelClient
    subclasses: Set[ModelClient] = inheritors(ModelClient)
    logger.debug("Known model client classes: %s", subclasses)
    for subclass in subclasses:
        if not subclass.name:
            logger.error(
                "Model client class '%s' does not have a name. This model client type will not be usable.", subclass
            )
            continue

        if subclass.name in _MODELCLIENT_CLASSES:
            logger.warning(
                "Duplicate model client type '%s' found. The older class (%s) "
                "will be replaced by the newly imported one (%s).",
                subclass.name,
                _MODELCLIENT_CLASSES[subclass.name],
                subclass,
            )
        _MODELCLIENT_CLASSES[subclass.name] = subclass

    # Identify the type of bot and see if it's known
    class_ = config.pop("client")
    logger.debug("Creating model client of type '%s'", class_)
    if class_ not in _MODELCLIENT_CLASSES:
        raise ValueError(
            f"Unknown model client type '{class_}'. Available types: {list(_MODELCLIENT_CLASSES)}. "
            "Did you forget to install your plugin?"
        )

    # Handoff to the subclass' init
    return _MODELCLIENT_CLASSES[class_](parent=parent, intent_router=intent_router, config=config)

tools #

Tools baseclass for Intentional.

Tool #

Bases: ABC

Tools baseclass for Intentional.

Source code in intentional-core/src/intentional_core/tools.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
class Tool(ABC):
    """
    Tools baseclass for Intentional.
    """

    name: str = None
    description: str = None
    parameters: List[ToolParameter] = None

    @abstractmethod
    async def run(self, params: dict) -> Any:
        """
        Run the tool.
        """
run(params) abstractmethod async #

Run the tool.

Source code in intentional-core/src/intentional_core/tools.py 
42
43
44
45
46
@abstractmethod
async def run(self, params: dict) -> Any:
    """
    Run the tool.
    """
ToolParameter dataclass #

A parameter for an Intentional tool.

Source code in intentional-core/src/intentional_core/tools.py 
20
21
22
23
24
25
26
27
28
29
30
@dataclass
class ToolParameter:
    """
    A parameter for an Intentional tool.
    """

    name: str
    description: str
    type: Any
    required: bool
    default: Any
load_tools_from_dict(config) #

Load a list of tools from a dictionary configuration.

Parameters:

Name Type Description Default
config List[Dict[str, Any]]

The configuration dictionary.

 required

Returns:

Type Description
Dict[str, Tool]

A list of Tool instances.
Source code in intentional-core/src/intentional_core/tools.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
def load_tools_from_dict(config: List[Dict[str, Any]]) -> Dict[str, Tool]:
    """
    Load a list of tools from a dictionary configuration.

    Args:
        config: The configuration dictionary.

    Returns:
        A list of Tool instances.
    """
    # Get all the subclasses of Tool
    if not _TOOL_CLASSES:
        subclasses: Set[Tool] = inheritors(Tool)
        logger.debug("Known tool classes: %s", subclasses)
        for subclass in subclasses:
            if not subclass.name:
                logger.error("Tool class '%s' does not have a name. This tool will not be usable.", subclass)
                continue

            if subclass.name in _TOOL_CLASSES:
                logger.warning(
                    "Duplicate tool '%s' found. The older class (%s) will be replaced by the newly imported one (%s).",
                    subclass.name,
                    _TOOL_CLASSES[subclass.name],
                    subclass,
                )
            _TOOL_CLASSES[subclass.name] = subclass

    # Initialize the tools
    tools = {}
    for tool_config in config:
        class_ = tool_config.pop("name")
        logger.debug("Creating tool of type '%s'", class_)
        if class_ not in _TOOL_CLASSES:
            raise ValueError(
                f"Unknown tool '{class_}'. Available tools: {list(_TOOL_CLASSES)}. "
                "Did you forget to install a plugin?"
            )
        tool_instance = _TOOL_CLASSES[class_](**tool_config)
        tools[tool_instance.name] = tool_instance

    return tools

utils #

Utilities for Intentional.

importing #

Module import functions to handle dynamic plugins import.

import_all_plugins() #

Imports all the intentional-* packages found in the current environment.

Source code in intentional-core/src/intentional_core/utils/importing.py 
36
37
38
39
40
41
42
43
44
45
46
47
def import_all_plugins():
    """
    Imports all the `intentional-*` packages found in the current environment.
    """
    for dist in importlib.metadata.distributions():
        if not hasattr(dist, "_path"):
            logger.debug("'_path' not found in '%s', ignoring", dist)
        path = dist._path  # pylint: disable=protected-access
        if path.name.startswith("intentional_"):
            with open(path / "top_level.txt", encoding="utf-8") as file:
                for name in file.read().splitlines():
                    import_plugin(name)
import_plugin(name) #

Imports the specified package. It does NOT check if this is an Intentional package or not.

Source code in intentional-core/src/intentional_core/utils/importing.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
def import_plugin(name: str):
    """
    Imports the specified package. It does NOT check if this is an Intentional package or not.
    """
    try:
        logger.debug("Importing module %s", name)
        module = importlib.import_module(name)
        # Print all classes in the module
        class_found = False
        for _, obj in inspect.getmembers(module):
            if inspect.isclass(obj):
                logger.debug("Class found: %s", obj)
                class_found = True
        if not class_found:
            logger.debug(
                "No classes found in module %s: are they imported in the top-level __init__ file of the plugin?", name
            )
    except ModuleNotFoundError:
        logger.exception("Module '%s' not found for import, is it installed?", name)
inheritance #

Utils for inheritance checks, to discover subclasses of a base Intentional class.

inheritors(class_, include_abstract=False) #

Find all subclasses of a class, regardless of depth.

Parameters:

Name Type Description Default
class_ Any

The class to find subclasses of.

 required
include_abstract bool

Whether to include abstract classes in the results.

 False
Source code in intentional-core/src/intentional_core/utils/inheritance.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
def inheritors(class_: Any, include_abstract: bool = False) -> Set[Any]:
    """
    Find all subclasses of a class, regardless of depth.

    Args:
        class_: The class to find subclasses of.
        include_abstract: Whether to include abstract classes in the results.
    """
    subclasses = set()
    to_process = [class_]
    while to_process:
        parent = to_process.pop()
        for child in parent.__subclasses__():
            if child not in subclasses:
                to_process.append(child)
                if not include_abstract and inspect.isabstract(child):
                    logger.debug(
                        "Skipping abstract class from inheritor's list: %s. Abstract methods: %s",
                        child,
                        list(child.__abstractmethods__),
                    )
                else:
                    subclasses.add(child)

    return subclasses