Skip to content

API Reference - Textual TUI#

src #

intentional_textual_ui #

Init file for the intentional_textual_ui package.

__about__ #

Package descriptors for intentional-textual-ui.

audio_stream_ui #

Textual UI for audio stream bots.

AudioStreamInterface #

Bases: App

The main interface class for the audio stream bot UI.

Source code in plugins/intentional-textual-ui/src/intentional_textual_ui/audio_stream_ui.py 
 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
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
class AudioStreamInterface(App):
    """
    The main interface class for the audio stream bot UI.
    """

    CSS_PATH = "example.tcss"

    def __init__(self, bot: BotStructure, audio_output_handler: "AudioHandler"):
        super().__init__()
        self.bot = bot
        self.audio_handler = audio_output_handler
        self.bot.add_event_handler("on_audio_message_from_llm", self.handle_audio_messages)
        self.bot.add_event_handler("on_llm_speech_transcribed", self.handle_transcript)
        self.bot.add_event_handler("on_user_speech_transcribed", self.handle_transcript)
        self.bot.add_event_handler("on_user_speech_started", self.handle_start_user_response)
        self.bot.add_event_handler("on_user_speech_ended", self.handle_finish_user_response)
        self.bot.add_event_handler("on_system_prompt_updated", self.handle_system_prompt_updated)
        self.bot.add_event_handler("on_conversation_ended", self.handle_conversation_end)

        self.conversation = ""

    def compose(self) -> ComposeResult:
        """
        Layout of the UI.
        """
        yield Horizontal(
            Vertical(
                Markdown("# Chat History"),
                ScrollableContainer(ChatHistory()),
                UserStatus("# User is silent..."),
                classes="column bordered chat",
            ),
            Vertical(
                Markdown("# System Prompt"),
                SystemPrompt(),
                classes="bordered column",
            ),
        )

    def on_mount(self) -> None:
        """
        Operations to be performed at mount time.
        """
        self.query_one(SystemPrompt).update(self.bot.llm.system_prompt)

    async def handle_transcript(self, event: Dict[str, Any]) -> None:
        """
        Prints the transcripts in the chat history.
        """
        if event["type"] == "on_user_speech_transcribed":
            self.conversation += f"\n**User:** {event['transcript']}\n"
        elif event["type"] == "on_llm_speech_transcribed":
            self.conversation += f"\n**Assistant:** {event['transcript']}\n"
        else:
            log.debug("Unknown event with transcript received.", event_name=event["type"])
            self.conversation += f"\n**{event['type']}:** {event['transcript']}\n"
        self.query_one(ChatHistory).update(self.conversation)

    async def handle_system_prompt_updated(self, event: Dict[str, Any]) -> None:
        """
        Prints to the console any text message from the bot.

        Args:
            event: The event dictionary containing the message.
        """
        self.query_one(SystemPrompt).update(event["system_prompt"])

    async def handle_start_user_response(self, _) -> None:
        """
        Updates the user status when they start speaking.
        """
        self.query_one(UserStatus).update("# User is speaking...")

    async def handle_finish_user_response(self, _) -> None:
        """
        Updates the user status when they stop speaking.
        """
        self.query_one(UserStatus).update("# User is silent...")

    async def handle_audio_messages(self, event: Dict[str, Any]) -> None:
        """
        Plays audio responses from the bot and updates the bot status line.

        Args:
            event: The event dictionary containing the audio message.
        """
        # self.query_one(BotStatus).update("# Bot is speaking...")
        if event["delta"]:
            self.audio_handler.play_audio(event["delta"])

    async def handle_conversation_end(self, _) -> None:
        """
        At the end of the conversation, closes the UI.
        """
        self.exit(0)
        self.audio_handler.stop_streaming()
        self.audio_handler.cleanup()
compose() #

Layout of the UI.

Source code in plugins/intentional-textual-ui/src/intentional_textual_ui/audio_stream_ui.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
def compose(self) -> ComposeResult:
    """
    Layout of the UI.
    """
    yield Horizontal(
        Vertical(
            Markdown("# Chat History"),
            ScrollableContainer(ChatHistory()),
            UserStatus("# User is silent..."),
            classes="column bordered chat",
        ),
        Vertical(
            Markdown("# System Prompt"),
            SystemPrompt(),
            classes="bordered column",
        ),
    )
handle_audio_messages(event) async #

Plays audio responses from the bot and updates the bot status line.

Parameters:

Name Type Description Default
event Dict[str, Any]

The event dictionary containing the audio message.

 required
Source code in plugins/intentional-textual-ui/src/intentional_textual_ui/audio_stream_ui.py 
117
118
119
120
121
122
123
124
125
126
async def handle_audio_messages(self, event: Dict[str, Any]) -> None:
    """
    Plays audio responses from the bot and updates the bot status line.

    Args:
        event: The event dictionary containing the audio message.
    """
    # self.query_one(BotStatus).update("# Bot is speaking...")
    if event["delta"]:
        self.audio_handler.play_audio(event["delta"])
handle_conversation_end(_) async #

At the end of the conversation, closes the UI.

Source code in plugins/intentional-textual-ui/src/intentional_textual_ui/audio_stream_ui.py 
128
129
130
131
132
133
134
async def handle_conversation_end(self, _) -> None:
    """
    At the end of the conversation, closes the UI.
    """
    self.exit(0)
    self.audio_handler.stop_streaming()
    self.audio_handler.cleanup()
handle_finish_user_response(_) async #

Updates the user status when they stop speaking.

Source code in plugins/intentional-textual-ui/src/intentional_textual_ui/audio_stream_ui.py 
111
112
113
114
115
async def handle_finish_user_response(self, _) -> None:
    """
    Updates the user status when they stop speaking.
    """
    self.query_one(UserStatus).update("# User is silent...")
handle_start_user_response(_) async #

Updates the user status when they start speaking.

Source code in plugins/intentional-textual-ui/src/intentional_textual_ui/audio_stream_ui.py 
105
106
107
108
109
async def handle_start_user_response(self, _) -> None:
    """
    Updates the user status when they start speaking.
    """
    self.query_one(UserStatus).update("# User is speaking...")
handle_system_prompt_updated(event) async #

Prints to the console any text message from the bot.

Parameters:

Name Type Description Default
event Dict[str, Any]

The event dictionary containing the message.

 required
Source code in plugins/intentional-textual-ui/src/intentional_textual_ui/audio_stream_ui.py 
 96
 97
 98
 99
100
101
102
103
async def handle_system_prompt_updated(self, event: Dict[str, Any]) -> None:
    """
    Prints to the console any text message from the bot.

    Args:
        event: The event dictionary containing the message.
    """
    self.query_one(SystemPrompt).update(event["system_prompt"])
handle_transcript(event) async #

Prints the transcripts in the chat history.

Source code in plugins/intentional-textual-ui/src/intentional_textual_ui/audio_stream_ui.py 
83
84
85
86
87
88
89
90
91
92
93
94
async def handle_transcript(self, event: Dict[str, Any]) -> None:
    """
    Prints the transcripts in the chat history.
    """
    if event["type"] == "on_user_speech_transcribed":
        self.conversation += f"\n**User:** {event['transcript']}\n"
    elif event["type"] == "on_llm_speech_transcribed":
        self.conversation += f"\n**Assistant:** {event['transcript']}\n"
    else:
        log.debug("Unknown event with transcript received.", event_name=event["type"])
        self.conversation += f"\n**{event['type']}:** {event['transcript']}\n"
    self.query_one(ChatHistory).update(self.conversation)
on_mount() #

Operations to be performed at mount time.

Source code in plugins/intentional-textual-ui/src/intentional_textual_ui/audio_stream_ui.py 
77
78
79
80
81
def on_mount(self) -> None:
    """
    Operations to be performed at mount time.
    """
    self.query_one(SystemPrompt).update(self.bot.llm.system_prompt)
ChatHistory #

Bases: Markdown

A markdown widget that displays the chat history.

Source code in plugins/intentional-textual-ui/src/intentional_textual_ui/audio_stream_ui.py 
20
21
22
23
class ChatHistory(Markdown):
    """
    A markdown widget that displays the chat history.
    """
SystemPrompt #

Bases: Markdown

A markdown widget that displays the system prompt.

Source code in plugins/intentional-textual-ui/src/intentional_textual_ui/audio_stream_ui.py 
32
33
34
35
class SystemPrompt(Markdown):
    """
    A markdown widget that displays the system prompt.
    """
UserStatus #

Bases: Markdown

A markdown widget that displays the user status (speaking/silent).

Source code in plugins/intentional-textual-ui/src/intentional_textual_ui/audio_stream_ui.py 
26
27
28
29
class UserStatus(Markdown):
    """
    A markdown widget that displays the user status (speaking/silent).
    """

bot_interface #

Local bot interface for Intentional.

TextualUIBotInterface #

Bases: BotInterface

Bot that uses a Textual UI command line interface to interact with the user.

Source code in plugins/intentional-textual-ui/src/intentional_textual_ui/bot_interface.py 
 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
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
class TextualUIBotInterface(BotInterface):
    """
    Bot that uses a Textual UI command line interface to interact with the user.
    """

    name = "textual_ui"

    def __init__(self, config: Dict[str, Any], intent_router: IntentRouter):
        # Init the structure
        bot_structure_config = config.pop("bot", None)
        if not bot_structure_config:
            raise ValueError(
                f"{self.__class__.__name__} requires a 'bot' configuration key to know how to structure the bot."
            )
        log.debug("Creating bot structure", bot_structure_config=bot_structure_config)
        self.bot: BotStructure = load_bot_structure_from_dict(intent_router=intent_router, config=bot_structure_config)

        # Check the modality
        self.modality = config.pop("modality")
        log.debug("Setting modality for bot structure", modality=self.modality)

        # Handlers
        self.audio_handler = None
        self.input_handler = None
        self.app = None

    async def run(self) -> None:
        """
        Chooses the specific loop to use for this combination of bot and modality and kicks it off.
        """
        if self.modality == "audio_stream":
            await self._run_audio_stream(self.bot)

        elif self.modality == "text_messages":
            await self._run_text_messages(self.bot)

        else:
            raise ValueError(
                f"Modality '{self.modality}' is not yet supported."
                "These are the supported modalities: 'audio_stream', 'text_messages'."
            )

    async def _run_text_messages(self, bot: BotStructure) -> None:
        """
        Runs the CLI interface for the text turns modality.
        """
        log.debug("Running in text turns mode.")
        self.app = TextChatInterface(bot=bot)
        await bot.connect()
        await self._launch_ui()

    async def _run_audio_stream(self, bot: BotStructure) -> None:
        """
        Runs the CLI interface for the continuous audio streaming modality.
        """
        log.debug("Running in continuous audio streaming mode.")

        try:
            self.audio_handler = AudioHandler()
            self.app = AudioStreamInterface(bot=bot, audio_output_handler=self.audio_handler)
            await bot.connect()
            await self._launch_ui(gather=[self.bot.run()])
            await self.audio_handler.start_streaming(bot.send)

        except Exception as e:  # pylint: disable=broad-except
            raise e
        finally:
            self.audio_handler.stop_streaming()
            self.audio_handler.cleanup()
            await bot.disconnect()
            print("Chat is finished. Bye!")

    async def _launch_ui(self, gather: List[Callable] = None):
        """
        Launches the Textual UI interface. If there's any async task that should be run in parallel, it can be passed
        as a list of callables to the `gather` parameter.

        Args:
            gather: A list of callables that should be run in parallel with the UI.
        """
        self.app._loop = asyncio.get_running_loop()  # pylint: disable=protected-access
        self.app._thread_id = threading.get_ident()  # pylint: disable=protected-access
        with self.app._context():  # pylint: disable=protected-access
            try:
                if not gather:
                    await self.app.run_async(
                        headless=False,
                        inline=False,
                        inline_no_clear=False,
                        mouse=True,
                        size=None,
                        auto_pilot=None,
                    )
                else:
                    asyncio.gather(
                        self.app.run_async(
                            headless=False,
                            inline=False,
                            inline_no_clear=False,
                            mouse=True,
                            size=None,
                            auto_pilot=None,
                        ),
                        *gather,
                    )
            finally:
                self.app._loop = None  # pylint: disable=protected-access
                self.app._thread_id = 0  # pylint: disable=protected-access
run() async #

Chooses the specific loop to use for this combination of bot and modality and kicks it off.

Source code in plugins/intentional-textual-ui/src/intentional_textual_ui/bot_interface.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
async def run(self) -> None:
    """
    Chooses the specific loop to use for this combination of bot and modality and kicks it off.
    """
    if self.modality == "audio_stream":
        await self._run_audio_stream(self.bot)

    elif self.modality == "text_messages":
        await self._run_text_messages(self.bot)

    else:
        raise ValueError(
            f"Modality '{self.modality}' is not yet supported."
            "These are the supported modalities: 'audio_stream', 'text_messages'."
        )

text_chat_ui #

Textual UI for text-based bots.

ChatHistory #

Bases: Markdown

A markdown widget that displays the chat history.

Source code in plugins/intentional-textual-ui/src/intentional_textual_ui/text_chat_ui.py 
16
17
18
19
class ChatHistory(Markdown):
    """
    A markdown widget that displays the chat history.
    """
MessageBox #

Bases: Input

An input widget that allows the user to type a message.

Source code in plugins/intentional-textual-ui/src/intentional_textual_ui/text_chat_ui.py 
22
23
24
25
26
27
class MessageBox(Input):
    """
    An input widget that allows the user to type a message.
    """

    placeholder = "Message..."
SystemPrompt #

Bases: Markdown

A markdown widget that displays the system prompt.

Source code in plugins/intentional-textual-ui/src/intentional_textual_ui/text_chat_ui.py 
30
31
32
33
class SystemPrompt(Markdown):
    """
    A markdown widget that displays the system prompt.
    """
TextChatInterface #

Bases: App

The main interface class for the text-based bot UI.

Source code in plugins/intentional-textual-ui/src/intentional_textual_ui/text_chat_ui.py 
 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
114
115
116
117
118
119
120
121
122
123
124
125
class TextChatInterface(App):
    """
    The main interface class for the text-based bot UI.
    """

    CSS_PATH = "example.tcss"

    def __init__(self, bot: BotStructure):
        super().__init__()
        self.bot = bot
        self.bot.add_event_handler("on_text_message_from_llm", self.handle_text_messages)
        self.bot.add_event_handler("on_llm_starts_generating_response", self.handle_start_text_response)
        self.bot.add_event_handler("on_llm_stops_generating_response", self.handle_finish_text_response)
        self.bot.add_event_handler("on_system_prompt_updated", self.handle_system_prompt_updated)

        self.conversation = ""
        self.generating_response = False

    def compose(self) -> ComposeResult:
        """
        Layout for the text-based bot UI.
        """
        yield Horizontal(
            Vertical(
                Markdown("# Chat History"),
                ScrollableContainer(ChatHistory()),
                MessageBox(placeholder="Message..."),
                classes="column bordered chat",
            ),
            Vertical(
                Markdown("# System Prompt"),
                ScrollableContainer(SystemPrompt()),
                classes="column bordered",
            ),
        )

    def on_mount(self) -> None:
        """
        Operations to perform when the UI is mounted.
        """
        self.query_one(SystemPrompt).update(self.bot.llm.system_prompt)
        self.query_one(MessageBox).focus()

    @on(MessageBox.Submitted)
    async def send_message(self, event: MessageBox.Changed) -> None:
        """
        Sends a message to the bot when the user presses enter.

        Args:
            event: The event containing the message to send.
        """
        self.conversation += "\n\n**User**: " + event.value
        self.query_one(MessageBox).clear()
        self.query_one(ChatHistory).update(self.conversation)
        await self.bot.send({"text_message": {"role": "user", "content": event.value}})

    async def handle_start_text_response(self, _) -> None:
        """
        Prints to the console when the bot starts generating a text response.
        """
        if not self.generating_response:  # To avoid the duplication due to function calls.
            self.generating_response = True
            self.conversation += "\n\n**Assistant:** "
            self.query_one(ChatHistory).update(self.conversation)

    async def handle_finish_text_response(self, _) -> None:
        """
        Prints to the console when the bot stops generating a text response.
        """
        self.generating_response = False

    async def handle_text_messages(self, event: Dict[str, Any]) -> None:
        """
        Prints to the console any text message from the bot. It is usually a chunk as the output is being streamed out.

        Args:
            event: The event dictionary containing the message chunk.
        """
        if event["delta"]:
            self.conversation += event["delta"]
            self.query_one(ChatHistory).update(self.conversation)

    async def handle_system_prompt_updated(self, event: Dict[str, Any]) -> None:
        """
        Prints to the console any text message from the bot.

        Args:
            event: The event dictionary containing the message.
        """
        self.query_one(SystemPrompt).update(event["system_prompt"])  # self.bot.llm.system_prompt)
compose() #

Layout for the text-based bot UI.

Source code in plugins/intentional-textual-ui/src/intentional_textual_ui/text_chat_ui.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
def compose(self) -> ComposeResult:
    """
    Layout for the text-based bot UI.
    """
    yield Horizontal(
        Vertical(
            Markdown("# Chat History"),
            ScrollableContainer(ChatHistory()),
            MessageBox(placeholder="Message..."),
            classes="column bordered chat",
        ),
        Vertical(
            Markdown("# System Prompt"),
            ScrollableContainer(SystemPrompt()),
            classes="column bordered",
        ),
    )
handle_finish_text_response(_) async #

Prints to the console when the bot stops generating a text response.

Source code in plugins/intentional-textual-ui/src/intentional_textual_ui/text_chat_ui.py 
101
102
103
104
105
async def handle_finish_text_response(self, _) -> None:
    """
    Prints to the console when the bot stops generating a text response.
    """
    self.generating_response = False
handle_start_text_response(_) async #

Prints to the console when the bot starts generating a text response.

Source code in plugins/intentional-textual-ui/src/intentional_textual_ui/text_chat_ui.py 
92
93
94
95
96
97
98
99
async def handle_start_text_response(self, _) -> None:
    """
    Prints to the console when the bot starts generating a text response.
    """
    if not self.generating_response:  # To avoid the duplication due to function calls.
        self.generating_response = True
        self.conversation += "\n\n**Assistant:** "
        self.query_one(ChatHistory).update(self.conversation)
handle_system_prompt_updated(event) async #

Prints to the console any text message from the bot.

Parameters:

Name Type Description Default
event Dict[str, Any]

The event dictionary containing the message.

 required
Source code in plugins/intentional-textual-ui/src/intentional_textual_ui/text_chat_ui.py 
118
119
120
121
122
123
124
125
async def handle_system_prompt_updated(self, event: Dict[str, Any]) -> None:
    """
    Prints to the console any text message from the bot.

    Args:
        event: The event dictionary containing the message.
    """
    self.query_one(SystemPrompt).update(event["system_prompt"])  # self.bot.llm.system_prompt)
handle_text_messages(event) async #

Prints to the console any text message from the bot. It is usually a chunk as the output is being streamed out.

Parameters:

Name Type Description Default
event Dict[str, Any]

The event dictionary containing the message chunk.

 required
Source code in plugins/intentional-textual-ui/src/intentional_textual_ui/text_chat_ui.py 
107
108
109
110
111
112
113
114
115
116
async def handle_text_messages(self, event: Dict[str, Any]) -> None:
    """
    Prints to the console any text message from the bot. It is usually a chunk as the output is being streamed out.

    Args:
        event: The event dictionary containing the message chunk.
    """
    if event["delta"]:
        self.conversation += event["delta"]
        self.query_one(ChatHistory).update(self.conversation)
on_mount() #

Operations to perform when the UI is mounted.

Source code in plugins/intentional-textual-ui/src/intentional_textual_ui/text_chat_ui.py 
72
73
74
75
76
77
def on_mount(self) -> None:
    """
    Operations to perform when the UI is mounted.
    """
    self.query_one(SystemPrompt).update(self.bot.llm.system_prompt)
    self.query_one(MessageBox).focus()
send_message(event) async #

Sends a message to the bot when the user presses enter.

Parameters:

Name Type Description Default
event Changed

The event containing the message to send.

 required
Source code in plugins/intentional-textual-ui/src/intentional_textual_ui/text_chat_ui.py 
79
80
81
82
83
84
85
86
87
88
89
90
@on(MessageBox.Submitted)
async def send_message(self, event: MessageBox.Changed) -> None:
    """
    Sends a message to the bot when the user presses enter.

    Args:
        event: The event containing the message to send.
    """
    self.conversation += "\n\n**User**: " + event.value
    self.query_one(MessageBox).clear()
    self.query_one(ChatHistory).update(self.conversation)
    await self.bot.send({"text_message": {"role": "user", "content": event.value}})