streaming chat messages

[patrickwasp]

Question

I'm trying to create a streaming chat interface where messages update in real-time as they come in. However, the message only shows "Thinking..." and doesn't update until the full response is received.

Expected Behavior

• Message should appear immediately with "Thinking..."
• Message should replace "Thinking..." and update word by word as the stream comes in
• Terminal shows the streaming working correctly

Actual Behavior

• Message appears with "Thinking..."
• Stays as "Thinking..." forever
• Terminal shows the streaming working correctly

Question

How can I get the message to update in real-time as the stream comes in?

import asyncio

from nicegui import ui


class ChatDemo:
    def __init__(self):
        self.container = ui.column()
        self.current_message = None
        self.setup_ui()

    def setup_ui(self):
        with self.container:
            ui.button("Send message", on_click=self.send)

    @ui.refreshable
    def update_message(self, content: str):
        if self.current_message:
            with self.current_message:
                ui.html(content)

    async def mock_stream(self):
        """Simulate streaming response"""
        response = "Hello! This is a test message that should stream word by word."
        for word in response.split():
            yield word + " "
            await asyncio.sleep(0.5)

    async def send(self):
        # Create message immediately with loading text
        with self.container:
            self.current_message = ui.chat_message(
                name="Bot",
                text_html=True,
            )
            with self.current_message:
                ui.html("Thinking...")

        # Process stream
        full_response = ""
        async for chunk in self.mock_stream():
            full_response += chunk
            print(chunk, end="", flush=True)  # Shows up in terminal
            self.update_message.refresh(full_response)  # Doesn't update UI


demo = ChatDemo()
ui.run()

[rodja]

There is no need for the refreshable. Simply update the html element which represents the content of the message:

class ChatDemo:
    def __init__(self):
        self.container = ui.column()
        self.current_message = None
        self.setup_ui()

    def setup_ui(self):
        with self.container:
            ui.button("Send message", on_click=self.send)

    async def mock_stream(self):
        """Simulate streaming response"""
        response = "Hello! This is a test message that should stream word by word."
        for word in response.split():
            yield word + " "
            await asyncio.sleep(0.5)

    async def send(self):
        # Create message immediately with loading text
        with self.container:
            self.current_message = ui.chat_message(
                name="Bot",
                text_html=True,
            )
            with self.current_message:
                part = ui.html("Thinking...")

        # Process stream
        full_response = ""
        async for chunk in self.mock_stream():
            full_response += chunk
            print(chunk, end="", flush=True)  # Shows up in terminal
            part.set_content(full_response)

[patrickwasp]

example with spinner and stop

import asyncio
from typing import AsyncGenerator

from nicegui import ui


class ChatDemo:
    def __init__(self) -> None:
        self.container = ui.column().classes("w-full max-w-2xl")
        self.should_stop = False
        self.setup_ui()

    def setup_ui(self) -> None:
        with self.container:
            self.button = ui.button("Send message", on_click=self.handle_click)

    def handle_click(self) -> None:
        if self.button.text == "Send message":
            self.should_stop = False
            self.button.text = "Stop"
            asyncio.create_task(self.send())
        else:
            self.should_stop = True
            self.button.text = "Send message"

    async def mock_stream(self) -> AsyncGenerator[str, None]:
        """Simulate streaming response"""
        response = "Hello! This is a test message that should stream word by word."
        for word in response.split():
            yield word + " "
            await asyncio.sleep(0.1)

    async def send(self) -> None:
        with self.container:
            thinking_row = ui.row().classes("items-center")
            with thinking_row:
                ui.spinner("dots", size="lg", color="primary")
                ui.label("Thinking...")

        await asyncio.sleep(2)
        if self.should_stop:
            thinking_row.clear()
            return

        thinking_row.clear()

        with self.container:
            message = ui.chat_message(name="Bot", text_html=True)
            with message:
                content = ui.html("")

        response = ""
        async for chunk in self.mock_stream():
            if self.should_stop:
                break
            response += chunk
            content.set_content(response)

        self.button.text = "Send message"


app = ChatDemo()
ui.run()