Skip to content

TUIAgent

autogen.beta.textual.TUIAgent #

TUIAgent(agent)

Bases: App

Source code in autogen/beta/textual.py 
36
37
38
39
40
41
def __init__(self, agent: Agent):
    super().__init__()

    self.stream = MemoryStream()
    self.agent = agent
    self.conversation: AgentReply | None = None

stream instance-attribute #

stream = MemoryStream()

agent instance-attribute #

agent = agent

conversation instance-attribute #

conversation = None

on_mount #

on_mount()
Source code in autogen/beta/textual.py 
43
44
45
def on_mount(self) -> None:
    self.title = f"AG2 TUI of `{self.agent.name}` agent"
    self.query_one("#input", Input).focus()

compose #

compose()
Source code in autogen/beta/textual.py 
47
48
49
50
51
52
53
def compose(self) -> "ComposeResult":
    yield Header(name=self.agent.name)
    yield ScrollableContainer(id="chat_history")
    yield Input(
        placeholder="Type message to your agent",
        id="input",
    )

input_submitter async #

input_submitter(event)
Source code in autogen/beta/textual.py 
 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
@on(Input.Submitted)
async def input_submitter(self, event: Input.Submitted) -> None:
    if text := event.value:
        inp = event.input
        inp.disabled = True
        inp.clear()

        chat_container = self.query_one("#chat_history", ScrollableContainer)

        # Add user message to history
        user_message = Markdown(f"**You:** {text}", classes="user-message")
        await chat_container.mount(user_message)
        user_message.scroll_visible(immediate=True)

        # Create a buffer block for assistant streaming
        thinking_block = Markdown(f"**{self.agent.name}:** *Thinking...* ", classes="assistant-message")
        await chat_container.mount(thinking_block)
        thinking_block.scroll_visible(immediate=True)

        async def put_reasoning_chunk(event: ModelReasoning, context: Context) -> None:
            await thinking_block.append(event.content)

        # Create agent message block
        assistant_message = Markdown(f"**{self.agent.name}:** ", classes="assistant-message")
        text_mounted = False

        async def put_message_chunk(event: ModelMessageChunk, context: Context) -> None:
            nonlocal text_mounted
            if not text_mounted:
                await thinking_block.remove()
                await chat_container.mount(assistant_message)
                text_mounted = True

            await assistant_message.append(event.content)
            assistant_message.scroll_visible(immediate=True)

        try:
            with (
                self.stream.where(ModelMessageChunk).sub_scope(put_message_chunk),
                self.stream.where(ModelReasoning).sub_scope(put_reasoning_chunk),
            ):
                c = self.conversation = await (
                    self.conversation.ask(text) if self.conversation else self.agent.ask(text, stream=self.stream)
                )
                if not text_mounted:
                    await thinking_block.remove()
                    await chat_container.mount(assistant_message)

                final_content = c.content
                await assistant_message.update(f"**{self.agent.name}:** {final_content}")
                assistant_message.scroll_visible(immediate=True)

        finally:
            inp.disabled = False
            inp.focus()