Repository: Avaiga/taipy
File: doc/gui/examples/controls/chat-streaming.py (Python)

Code before:

# add an image if one is present in the wikipedia returned json
def add_image_to_message(state: State, idx: int, text: str, image_url: str):
    msg_content: str = state.messages[idx][1]
    pos = msg_content.find(text)
    if pos > -1:
        msg_content = msg_content[: pos + len(text)] + f"\n\n![{text}]({image_url})" + msg_content[pos + len(text) :]
        set_message(state, msg_content, idx)


# invoked by update_message through a thread
def update_message_with_image(gui: Gui, state_id: str, message_idx: int, text: str, image: dict):
    if src := image.get("source"):
        time.sleep(0.2) # to show the streaming effect
        invoke_callback(
            gui,
            state_id,
            add_image_to_message,
            [message_idx, text, src],
        )


# invoked by request_wikipedia
def update_message(state: State, json, event_type: str, for_date: str, idx: int):
    if isinstance(json, dict):
        # response initial message
        set_message(state, f"{event_type} for {for_date}: \n", idx)

        for event in json.get(event_type, []):
            time.sleep(0.2) # to show the streaming effect
            # update response
            append_to_message(state, f"\n* {event.get('year', '')}: {event.get('text', '')}", idx)
            # invoke update_message_with_image in a thread
            invoke_long_callback(
                state=state,
                user_function=update_message_with_image,
                user_function_args=[
                    state.get_gui(),
                    get_state_id(state),
                    idx,
                    event.get("text", ""),
                    pages[0].get("thumbnail", {}) if (pages := event.get("pages", [])) and len(pages) else {},
                ],
            )


# set or append a message returning its index in the list
def set_message(state: State, message: str, idx: t.Optional[int] = None):
    if idx is not None and idx < len(state.messages):
        msg = state.messages[idx]
        state.messages[idx] = (msg[0], message, msg[2])

Review comment (suggestion):
```suggestion
        # Initial response content
```

Code after:

# Add the image if there is one in the Wikipedia returned data
def add_image_to_message(state: State, idx: int, text: str, image_url: str):
    msg_content: str = state.messages[idx][1]
    if (pos := msg_content.find(text)) > -1:
        msg_content = msg_content[: pos + len(text)] + f"\n\n![{text}]({image_url})" + msg_content[pos + len(text) :]
        set_message(state, msg_content, idx)


# Invoked by update_message through a thread
def update_message_with_image(gui: Gui, state_id: str, message_idx: int, text: str, image: dict):
    if src := image.get("source"):
        time.sleep(0.2) # Apply the typewriter effect
        invoke_callback(
            gui,
            state_id,
            add_image_to_message,
            [message_idx, text, src],
        )


# Invoked by query_wikipedia()
def update_message(state: State, json, event_type: str, for_date: str, idx: int):
    if isinstance(json, dict):
        # Initial response content
        set_message(state, f"{event_type} for {for_date}:\n", idx)

        for event in json.get(event_type, []):
            time.sleep(0.2) # Apply the typewriter effect
            # Update response text
            append_to_message(state, f"\n* {event.get('year', '')}: {event.get('text', '')}", idx)
            # Invoke update_message_with_image() in a separated thread
            invoke_long_callback(
                state=state,
                user_function=update_message_with_image,
                user_function_args=[
                    state.get_gui(),
                    get_state_id(state),
                    idx,
                    event.get("text", ""),
                    pages[0].get("thumbnail", {}) if (pages := event.get("pages", [])) and len(pages) else {},
                ],
            )


# Set a new message or append to an existing message.
# Return the message index in the list.
def set_message(state: State, message: str, idx: t.Optional[int] = None):
    if idx is not None and idx < len(state.messages):
        msg = state.messages[idx]
        state.messages[idx] = (msg[0], message, msg[2])