Spaces:
Running
Running
import solara | |
import time | |
import random | |
from typing import List | |
from typing_extensions import TypedDict | |
# Streamed response emulator | |
def response_generator(): | |
response = random.choice( | |
[ | |
"Hello! How can I assist you today?", | |
"Hey there! If you have any questions or need help with something, feel free to ask.", | |
] | |
) | |
for word in response.split(): | |
yield word + " " | |
time.sleep(0.05) | |
class MessageDict(TypedDict): | |
role: str | |
content: str | |
messages: solara.Reactive[List[MessageDict]] = solara.reactive([]) | |
def add_chunk_to_ai_message(chunk: str): | |
messages.value = [ | |
*messages.value[:-1], | |
{ | |
"role": "assistant", | |
"content": messages.value[-1]["content"] + chunk, | |
}, | |
] | |
def Page(): | |
with solara.Column(style={"padding": "30px"}): | |
solara.Title("StreamBot") | |
solara.Markdown("#StreamBot") | |
user_message_count = len([m for m in messages.value if m["role"] == "user"]) | |
def send(message): | |
messages.value = [ | |
*messages.value, | |
{"role": "user", "content": message}, | |
] | |
def response(message): | |
messages.value = [*messages.value, {"role": "assistant", "content": ""}] | |
for chunk in response_generator(): | |
add_chunk_to_ai_message(chunk) | |
def result(): | |
if messages.value !=[]: response(messages.value[-1]["content"]) | |
result = solara.lab.use_task(result, dependencies=[user_message_count]) # type: ignore | |
with solara.Column(style={"width": "70%"}): | |
with solara.lab.ChatBox(): | |
for item in messages.value: | |
with solara.lab.ChatMessage( | |
user=item["role"] == "user", | |
name="Echobot" if item["role"] == "assistant" else "User", | |
avatar_background_color="#33cccc" if item["role"] == "assistant" else "#ff991f", | |
border_radius="20px", | |
): | |
solara.Markdown(item["content"]) | |
solara.lab.ChatInput(send_callback=send) | |