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 "TerminalBotInterface", 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 "TerminalBotInterface", 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
129
130
131
132
133
134
135
136
137
138
139
140
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.
    """
    log.debug(
        "Loading bot interface from configuration:",
        bot_interface_config=json.dumps(config, indent=4),
    )

    # Import all the necessary plugins
    plugins = config.pop("plugins", None)
    if plugins:
        log.debug("Found plugins to import", plugins=plugins)
        for plugin in plugins:
            log.debug("Importing plugin", plugin=plugin)
            import_plugin(plugin)
    else:
        import_all_plugins()

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

    # Get all the subclasses of Bot
    subclasses: Set[BotInterface] = inheritors(BotInterface)
    log.debug("Collected bot interface classes", bot_interfaces=subclasses)

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

        if subclass.name in _BOT_INTERFACES:
            log.warning(
                "Duplicate bot interface type '%s' found. The older class will be replaced by the newly imported one.",
                subclass.name,
                old_bot_interface_name=subclass.name,
                old_bot_interface_class=_BOT_INTERFACES[subclass.name],
                new_bot_interface_class=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
    log.debug("Creating bot interface", bot_interface_class=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.
    """
    log.debug("Loading YAML configuration file", config_file_path=path)
    with open(path, "r", encoding="utf-8") as file:
        config = yaml.safe_load(file)
    return load_bot_interface_from_dict(config)

bot_structures #

Bot structures supported by Intentional.

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 LLM 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_structures/bot_structure.py 
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
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 LLM 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.
    """

    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.
        """
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.

connect() async #

Connect to the bot.

Source code in intentional-core/src/intentional_core/bot_structures/bot_structure.py 
46
47
48
49
async def connect(self) -> None:
    """
    Connect to the bot.
    """
disconnect() async #

Disconnect from the bot.

Source code in intentional-core/src/intentional_core/bot_structures/bot_structure.py 
51
52
53
54
async def disconnect(self) -> None:
    """
    Disconnect from the bot.
    """
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_structures/bot_structure.py 
68
69
70
71
72
73
74
75
76
77
@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_structures/bot_structure.py 
56
57
58
59
60
@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_structures/bot_structure.py 
62
63
64
65
66
@abstractmethod
async def send(self, data: Dict[str, Any]) -> Dict[str, Any]:
    """
    Send a message to the bot.
    """
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_structures/bot_structure.py 
 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
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)
    log.debug("Collected bot structure classes", bot_structure_classes=subclasses)
    for subclass in subclasses:
        if not subclass.name:
            log.error(
                "BotStructure class '%s' does not have a name. This bot structure type will not be usable.",
                subclass,
                bot_structure_class=subclass,
            )
            continue

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

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

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

Bot structure to support text chat for Intentional.

DirectToLLMBotStructure #

Bases: BotStructure

Bot structure implementation for text chat.

Source code in intentional-core/src/intentional_core/bot_structures/direct_to_llm.py 
17
18
19
20
21
22
23
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
class DirectToLLMBotStructure(BotStructure):
    """
    Bot structure implementation for text chat.
    """

    name = "direct_to_llm"

    def __init__(self, config: Dict[str, Any], intent_router: IntentRouter):
        """
        Args:
            config:
                The configuration dictionary for the bot structure.
        """
        super().__init__()
        log.debug("Loading bot structure from config", bot_structure_config=config)

        # Init the model client
        llm_config = config.pop("llm", None)
        if not llm_config:
            raise ValueError(f"{self.__class__.__name__} requires a 'llm' configuration key.")
        self.llm: LLMClient = load_llm_client_from_dict(parent=self, intent_router=intent_router, config=llm_config)

    async def connect(self) -> None:
        """
        Initializes the model and connects to it as/if necessary.
        """
        await self.llm.connect()

    async def disconnect(self) -> None:
        """
        Disconnects from the model and unloads/closes it as/if necessary.
        """
        await self.llm.disconnect()

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

    async def send(self, data: Dict[str, Any]) -> AsyncGenerator[Dict[str, Any], None]:
        """
        Sends a message to the model and forward the response.

        Args:
            data: The message to send to the model in OpenAI format, like {"role": "user", "content": "Hello!"}
        """
        await self.llm.send(data)

    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.
        """
        await self.llm.handle_interruption(lenght_to_interruption)
__init__(config, intent_router) #

Parameters:

Name Type Description Default
config Dict[str, Any]

The configuration dictionary for the bot structure.

 required
Source code in intentional-core/src/intentional_core/bot_structures/direct_to_llm.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
def __init__(self, config: Dict[str, Any], intent_router: IntentRouter):
    """
    Args:
        config:
            The configuration dictionary for the bot structure.
    """
    super().__init__()
    log.debug("Loading bot structure from config", bot_structure_config=config)

    # Init the model client
    llm_config = config.pop("llm", None)
    if not llm_config:
        raise ValueError(f"{self.__class__.__name__} requires a 'llm' configuration key.")
    self.llm: LLMClient = load_llm_client_from_dict(parent=self, intent_router=intent_router, config=llm_config)
connect() async #

Initializes the model and connects to it as/if necessary.

Source code in intentional-core/src/intentional_core/bot_structures/direct_to_llm.py 
39
40
41
42
43
async def connect(self) -> None:
    """
    Initializes the model and connects to it as/if necessary.
    """
    await self.llm.connect()
disconnect() async #

Disconnects from the model and unloads/closes it as/if necessary.

Source code in intentional-core/src/intentional_core/bot_structures/direct_to_llm.py 
45
46
47
48
49
async def disconnect(self) -> None:
    """
    Disconnects from the model and unloads/closes it as/if necessary.
    """
    await self.llm.disconnect()
handle_interruption(lenght_to_interruption) 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_structures/direct_to_llm.py 
66
67
68
69
70
71
72
73
74
75
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.
    """
    await self.llm.handle_interruption(lenght_to_interruption)
run() async #

Main loop for the bot.

Source code in intentional-core/src/intentional_core/bot_structures/direct_to_llm.py 
51
52
53
54
55
async def run(self) -> None:
    """
    Main loop for the bot.
    """
    await self.llm.run()
send(data) async #

Sends a message to the model and forward the response.

Parameters:

Name Type Description Default
data Dict[str, Any]

The message to send to the model in OpenAI format, like {"role": "user", "content": "Hello!"}

 required
Source code in intentional-core/src/intentional_core/bot_structures/direct_to_llm.py 
57
58
59
60
61
62
63
64
async def send(self, data: Dict[str, Any]) -> AsyncGenerator[Dict[str, Any], None]:
    """
    Sends a message to the model and forward the response.

    Args:
        data: The message to send to the model in OpenAI format, like {"role": "user", "content": "Hello!"}
    """
    await self.llm.send(data)

end_conversation #

Tool that ends and resets the conversation. Once a conversation reaches this stage, the bot should restart from the start stage. See IntentRouter.

EndConversationTool #

Bases: Tool

Tool to end the conversation. Resets the intent router to its initial stage.

Source code in intentional-core/src/intentional_core/end_conversation.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
class EndConversationTool(Tool):
    """
    Tool to end the conversation. Resets the intent router to its initial stage.
    """

    id = "end_conversation"
    name = "end_conversation"
    description = "End the conversation."
    parameters = []

    def __init__(self, intent_router: "IntentRouter"):
        self.router = intent_router

    async def run(self, params=None) -> str:
        """
        Ends the conversation and resets the intent router to its initial stage.
        """
        log.debug("The conversation has ended.")
        self.router.current_stage_name = self.router.initial_stage
        log.debug(
            "Intent router reset to initial stage.",
            initial_stage=self.router.initial_stage,
        )
run(params=None) async #

Ends the conversation and resets the intent router to its initial stage.

Source code in intentional-core/src/intentional_core/end_conversation.py 
32
33
34
35
36
37
38
39
40
41
async def run(self, params=None) -> str:
    """
    Ends the conversation and resets the intent router to its initial stage.
    """
    log.debug("The conversation has ended.")
    self.router.current_stage_name = self.router.initial_stage
    log.debug(
        "Intent router reset to initial stage.",
        initial_stage=self.router.initial_stage,
    )

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 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
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.
        """
        log.debug("Emitting event", event_name=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 
66
67
68
69
70
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 
72
73
74
75
76
77
async def emit(self, event_name: str, event: Dict[str, Any]):
    """
    Send the event to the listener.
    """
    log.debug("Emitting event", event_name=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 
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
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
class EventListener(ABC):
    """
    Listens to events and handles them.
    """

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

    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:
            log.debug(
                "Event handler for '%s' was already assigned. The older handler will be replaced by the new one.",
                event_name,
                event_name=event_name,
                event_handler=self.event_handlers[event_name],
            )
        log.debug("Adding event handler", event_name=event_name, event_handler=handler)
        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 LLM may generate.
        """
        if "*" in self.event_handlers:
            log.debug("Calling wildcard event handler", event_name=event_name)
            await self.event_handlers["*"](event)

        if event_name in self.event_handlers:
            log.debug("Calling event handler", event_name=event_name)
            await self.event_handlers[event_name](event)
        else:
            log.debug("No event handler for event", event_name=event_name)
__init__() #

Initialize the bot structure.

Source code in intentional-core/src/intentional_core/events.py 
21
22
23
24
25
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/events.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
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:
        log.debug(
            "Event handler for '%s' was already assigned. The older handler will be replaced by the new one.",
            event_name,
            event_name=event_name,
            event_handler=self.event_handlers[event_name],
        )
    log.debug("Adding event handler", event_name=event_name, event_handler=handler)
    self.event_handlers[event_name] = handler
handle_event(event_name, event) async #

Handle different types of events that the LLM may generate.

Source code in intentional-core/src/intentional_core/events.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
async def handle_event(self, event_name: str, event: Dict[str, Any]) -> None:
    """
    Handle different types of events that the LLM may generate.
    """
    if "*" in self.event_handlers:
        log.debug("Calling wildcard event handler", event_name=event_name)
        await self.event_handlers["*"](event)

    if event_name in self.event_handlers:
        log.debug("Calling event handler", event_name=event_name)
        await self.event_handlers[event_name](event)
    else:
        log.debug("No event handler for event", event_name=event_name)

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 
 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
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
class IntentRouter(Tool):
    """
    Special tool used to alter the system prompt depending on the user's response.
    """

    id = "classify_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 = {}
        if "stages" not in config or not config["stages"]:
            raise ValueError("The conversation must have at least one stage.")
        for name, stage_config in config["stages"].items():
            log.debug("Adding stage", stage_name=name)
            self.stages[name] = Stage(name, stage_config)
            self.stages[name].tools[self.name] = self  # Add the intent router to the tools list of each stage
            self.graph.add_node(name)

        # Add end stage
        name = "_end_"
        log.debug("Adding stage", stage_name=name)
        end_tool = EndConversationTool(intent_router=self)
        self.stages[name] = Stage(
            name,
            {"custom_template": f"The conversation is over. Call the '{end_tool.name}' tool."},
        )
        self.stages[name].tools[end_tool.name] = end_tool
        self.graph.add_node("_end_")

        # Connect the stages
        for 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 '{name}' has an outcome leading to an unknown stage '{outcome_config['move_to']}'"
                    )
                log.debug(
                    "Adding connection",
                    origin=name,
                    target=outcome_config["move_to"],
                    outcome=outcome_name,
                )
                self.graph.add_edge(name, outcome_config["move_to"], key=outcome_name)

        # Find initial stage
        self.initial_stage = ""
        for name, stage in self.stages.items():
            if START_CONNECTION in stage.accessible_from:
                if self.initial_stage:
                    raise ValueError("Multiple start stages found!")
                log.debug("Found start stage", stage_name=name)
                self.initial_stage = name
        if not self.initial_stage:
            raise ValueError("No start stage found!")

        self.current_stage_name = self.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: Optional[Dict[str, Any]] = None) -> 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_external_transitions()

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

        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 = "You need to reach one of these situations:\n" + "\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_external_transitions()
        )
        template = self.current_stage.custom_template or DEFAULT_PROMPT_TEMPLATE
        return 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_external_transitions(self):
        """
        Return a list of all the stages that can be reached from the current stage that are not direct connections.
        """
        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_external_transitions() #

Return a list of all the stages that can be reached from the current stage that are not direct connections.

Source code in intentional-core/src/intentional_core/intent_routing.py 
179
180
181
182
183
184
185
186
187
188
189
190
def get_external_transitions(self):
    """
    Return a list of all the stages that can be reached from the current stage that are not direct connections.
    """
    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
        )
    ]
get_prompt() #

Get the prompt for the current stage.

Source code in intentional-core/src/intentional_core/intent_routing.py 
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
def get_prompt(self):
    """
    Get the prompt for the current stage.
    """
    outcomes = "You need to reach one of these situations:\n" + "\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_external_transitions()
    )
    template = self.current_stage.custom_template or DEFAULT_PROMPT_TEMPLATE
    return 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,
    )
run(params=None) async #

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

Parameters:

Name Type Description Default
params Optional[Dict[str, Any]]

The parameters for the tool. Contains the response_type.

 None

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 
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
async def run(self, params: Optional[Dict[str, Any]] = None) -> 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_external_transitions()

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

    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 
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
class Stage:
    """
    Describes a stage in the bot's conversation.
    """

    def __init__(self, stage_name, config: Dict[str, Any]) -> None:
        self.custom_template = config.get("custom_template", None)
        self.goal = config.get("goal", None)
        self.description = config.get("description", None)
        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", {})

        # If a custom template is given, nothing else is strictly needed
        if not self.custom_template:
            # Make sure the stage has a goal
            if not self.goal:
                raise ValueError(f"Stage '{stage_name}' is missing a goal.")
            # If the stage is accessible from somewhere else than a direct transition, it needs a description
            if self.accessible_from and self.accessible_from != ["_start_"] and not self.description:
                raise ValueError(
                    "Stages that set the 'accessible_from' field also need a description. "
                    f"'{stage_name}' has 'accessible_from' set to {self.accessible_from}, but no 'description' field."
                )
        # Make sure all outcomes have a description
        for name, outcome in self.outcomes.items():
            if "description" not in outcome:
                raise ValueError(f"Outcome '{name}' in stage '{stage_name}' is missing a description.")
            if "move_to" not in outcome:
                raise ValueError(f"Outcome '{name}' in stage '{stage_name}' is missing a 'move_to' field.")

        log.debug(
            "Stage loaded",
            custom_template=self.custom_template,
            stage_goal=self.goal,
            stage_description=self.description,
            stage_accessible_from=self.accessible_from,
            stage_tools=self.tools,
            outcomes=self.outcomes,
        )

llm_client #

Functions to load LLM client classes from config files.

LLMClient #

Bases: ABC, EventEmitter

Tiny base class used to recognize Intentional LLM 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/llm_client.py 
 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
class LLMClient(ABC, EventEmitter):
    """
    Tiny base class used to recognize Intentional LLM 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 LLM from.
    """

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

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

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

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

    @abstractmethod
    async def run(self) -> None:
        """
        Handle events from the LLM 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 LLM. The response is streamed out as an async generator.
        """

    @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 LLM from.

__init__(parent, intent_router) #

Initialize the LLM client.

Parameters:

Name Type Description Default
parent BotStructure

The parent bot structure.

 required
Source code in intentional-core/src/intentional_core/llm_client.py 
59
60
61
62
63
64
65
66
67
def __init__(self, parent: "BotStructure", intent_router: IntentRouter) -> None:
    """
    Initialize the LLM client.

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

Connect to the LLM.

Source code in intentional-core/src/intentional_core/llm_client.py 
69
70
71
72
73
async def connect(self) -> None:
    """
    Connect to the LLM.
    """
    await self.emit("on_llm_connection", {})
disconnect() async #

Disconnect from the LLM.

Source code in intentional-core/src/intentional_core/llm_client.py 
75
76
77
78
79
async def disconnect(self) -> None:
    """
    Disconnect from the LLM.
    """
    await self.emit("on_llm_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/llm_client.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
@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 LLM 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/llm_client.py 
81
82
83
84
85
86
@abstractmethod
async def run(self) -> None:
    """
    Handle events from the LLM 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 LLM. The response is streamed out as an async generator.

Source code in intentional-core/src/intentional_core/llm_client.py 
88
89
90
91
92
@abstractmethod
async def send(self, data: Dict[str, Any]) -> None:
    """
    Send a unit of data to the LLM. The response is streamed out as an async generator.
    """
load_llm_client_from_dict(parent, intent_router, config) #

Load a LLM client from a dictionary configuration.

Parameters:

Name Type Description Default
config Dict[str, Any]

The configuration dictionary.

 required

Returns:

Type Description
LLMClient

The LLMClient instance.
Source code in intentional-core/src/intentional_core/llm_client.py 
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
def load_llm_client_from_dict(parent: "BotStructure", intent_router: IntentRouter, config: Dict[str, Any]) -> LLMClient:
    """
    Load a LLM client from a dictionary configuration.

    Args:
        config: The configuration dictionary.

    Returns:
        The LLMClient instance.
    """
    # Get all the subclasses of LLMClient
    subclasses: Set[LLMClient] = inheritors(LLMClient)
    log.debug("Collected LLM client classes", llm_client_classes=subclasses)
    for subclass in subclasses:
        if not subclass.name:
            log.error(
                "LLM client class '%s' does not have a name. This LLM client type will not be usable.",
                subclass,
                llm_client_class=subclass,
            )
            continue

        if subclass.name in _LLM_CLIENTS:
            log.warning(
                "Duplicate LLM client type '%s' found. The older class will be replaced by the newly imported one.",
                subclass.name,
                old_llm_client_name=subclass.name,
                old_llm_client_class=_LLM_CLIENTS[subclass.name],
                new_llm_client_class=subclass,
            )
        _LLM_CLIENTS[subclass.name] = subclass

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

    # Handoff to the subclass' init
    return _LLM_CLIENTS[llm_client_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 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
class Tool(ABC):
    """
    Tools baseclass for Intentional.
    """

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

    def __repr__(self) -> str:
        return (
            f"<{self.__class__.__name__} id={self.id}, description={self.description}, "
            f"parameters={repr(self.parameters)}>"
        )

    @abstractmethod
    async def run(self, params: Optional[Dict[str, Any]] = None) -> Any:
        """
        Run the tool.
        """
run(params=None) abstractmethod async #

Run the tool.

Source code in intentional-core/src/intentional_core/tools.py 
55
56
57
58
59
@abstractmethod
async def run(self, params: Optional[Dict[str, Any]] = None) -> 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
31
32
33
34
35
36
@dataclass
class ToolParameter:
    """
    A parameter for an Intentional tool.
    """

    name: str
    description: str
    type: Any
    required: bool
    default: Any

    def __repr__(self) -> str:
        return (
            f"<{self.__class__.__name__} name={self.name}, description={self.description}, type={self.type}, "
            f"required={self.required}, default={self.default}>"
        )
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 
 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
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
    subclasses: Set[Tool] = inheritors(Tool)
    log.debug("Collected tool classes", tool_classes=subclasses)
    for subclass in subclasses:
        if not subclass.id:
            log.error(
                "Tool class '%s' does not have an id. This tool will not be usable.",
                subclass.__name__,
                tool_class=subclass,
            )
            continue

        if subclass.id in _TOOL_CLASSES:
            log.debug(
                "Duplicate tool '%s' found. The older class will be replaced by the newly imported one.",
                subclass.id,
                old_tool_id=subclass.id,
                old_tool_class=_TOOL_CLASSES[subclass.id],
                new_tool_class=subclass,
            )
        _TOOL_CLASSES[subclass.id] = subclass

    # Initialize the tools
    tools = {}
    for tool_config in config:
        tool_id = tool_config.pop("id", None)
        if not tool_id:
            raise ValueError("Tool definitions must have an 'id' field.")
        log.debug("Creating tool", tool_id=tool_id)
        if tool_id not in _TOOL_CLASSES:
            raise ValueError(
                f"Unknown tool '{tool_id}'. Available tools: {list(_TOOL_CLASSES)}. "
                "Did you forget to install a plugin?"
            )
        tool_instance: Tool = _TOOL_CLASSES[tool_id](**tool_config)
        if getattr(tool_instance, "name", None) is None:
            raise ValueError(f"Tool '{tool_id}' must have a name.")
        if getattr(tool_instance, "description", None) is None:
            raise ValueError(f"Tool '{tool_id}' must have a description.")
        if getattr(tool_instance, "parameters", None) is None:
            raise ValueError(f"Tool '{tool_id}' must have parameters.")
        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 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
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"):
            log.debug("'_path' not found in '%s', ignoring", dist, dist=dist)
        path = dist._path  # pylint: disable=protected-access
        if path.name.startswith("intentional_"):
            if os.path.exists(path / "top_level.txt"):
                with open(path / "top_level.txt", encoding="utf-8") as file:
                    for name in file.read().splitlines():
                        import_plugin(name)
            else:
                name = path.name.split("-", 1)[0]
                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
34
def import_plugin(name: str):
    """
    Imports the specified package. It does NOT check if this is an Intentional package or not.
    """
    try:
        log.debug("Importing module", module_name=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):
                log.debug("Class found in module", module_name=name, found_class=obj)
                class_found = True
        if not class_found:
            log.debug(
                "No classes found in module: are they imported in the top-level __init__ file of the plugin?",
                module_name=name,
            )
    except ModuleNotFoundError:
        log.exception("Module '%s' not found for import, is it installed?", name, module_name=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):
                    log.debug(
                        "Skipping abstract class from inheritor's list.",
                        abstract_class=child,
                        abstract_methods=list(child.__abstractmethods__),
                    )
                else:
                    subclasses.add(child)

    return subclasses