Build a basic LLM chat app with Solara

In this post, we will build a basic LLM chat app with Solara. Large Language Models (LLMs) have become increasingly popular and Solara provides several components to work with them. Let’s dive in.

First things first, let’s install Solara.

$ pip install solara

Now, let’s start by creating an app.py that sends a simple message with the content “Hello!” as a user. To do that we use the ChatBox and ChatMessage components.

import solara
@solara.component
def Page():
    with solara.lab.ChatBox():
        with solara.lab.ChatMessage(user=True, name="User"):
            solara.Markdown("Hello!")
Page()

You can modify the user name and/or the message as you please.

import solara
@solara.component
def Page():
    with solara.lab.ChatBox():
        with solara.lab.ChatMessage(user=True, name="Morpheus"):
            solara.Markdown("Wake up, Neo...")
Page()

You can also send a message as an assistant.

import solara
@solara.component
def Page():
    with solara.lab.ChatBox():
        with solara.lab.ChatMessage(user=False, name="Assistant",):
            solara.Markdown("Hello! How can I assist you today?")
Page()

To have a conversation, we create a reactive variable messages where we will store the messages. To do that we create a list of dictionaries where we will save the roles (for example, user and assistant) and the messages contents.

import solara
from typing import List
from typing_extensions import TypedDict

class MessageDict(TypedDict):
    role: str
    content: str

messages: solara.Reactive[List[MessageDict]] = solara.reactive([])

We can generate a conversation by adding messages to the reactive variable messages that we previously created and displaying each message one by one.

@solara.component
def Page():
    messages.value = [
        {"role": "user", "content": "Hello!"}, 
        {"role": "assistant",  "content": "Hello! How can I assist you today?"},
    ]
    with solara.lab.ChatBox():
        for item in messages.value:
            with solara.lab.ChatMessage(
                user=item["role"] == "user",
                name="User" if item["role"] == "user" else "Assistant"
            ):
                solara.Markdown(item["content"])
Page()

Let’s now add the possibility to receive messages from the user by adding the ChatInput component and a send function that adds the message to the conversation.

messages: solara.Reactive[List[MessageDict]] = solara.reactive([])
@solara.component
def Page():
    def send(message):
        messages.value = [*messages.value, {"role": "user", "content": message}]
    with solara.lab.ChatBox():
        for item in messages.value:
            with solara.lab.ChatMessage(
                user=item["role"] == "user",
                name="User" if item["role"] == "user" else "Assistant"
            ):
                solara.Markdown(item["content"])
    solara.lab.ChatInput(send_callback=send)
Page()

Try it out by sending a message.

EchoBot

Up to now we are only displaying the message the user sent. Let’s first simulate a conversation by replying exactly the same message we receive from the user. To do that we need to add a response function and a result function that will reply the last message (which will be the one sent by the user) and it will be activated once every time the counter user_message_count changes.

messages: solara.Reactive[List[MessageDict]] = solara.reactive([])
@solara.component
def Page():
    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": message}]
    def result():
        if messages.value != []:
            response(messages.value[-1]["content"])
    result = solara.lab.use_task(result, dependencies=[user_message_count])
    with solara.lab.ChatBox():
        for item in messages.value:
            with solara.lab.ChatMessage(
                user=item["role"] == "user",
                name="User" if item["role"] == "user" else "EchoBot"
            ):
                solara.Markdown(item["content"])
    solara.lab.ChatInput(send_callback=send)
Page()

The complete code can be found below.

Show the code

import solara
from typing import List
from typing_extensions import TypedDict

class MessageDict(TypedDict):
    role: str
    content: str

messages: solara.Reactive[List[MessageDict]] = solara.reactive([])
@solara.component
def Page():
    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": message}]
    def result():
        if messages.value != []:
            response(messages.value[-1]["content"])
    result = solara.lab.use_task(result, dependencies=[user_message_count])
    with solara.lab.ChatBox():
        for item in messages.value:
            with solara.lab.ChatMessage(
                user=item["role"] == "user",
                name="User" if item["role"] == "user" else "EchoBot"
            ):
                solara.Markdown(item["content"])
    solara.lab.ChatInput(send_callback=send)
Page()

Up to now, our EchoBot application looks like this. Try it out!

StreamBot

Let’s now build a Bot that will stream a response message. Let’s first emulate a streamed response with a function that we call response_generator.

# Streamed response emulator
import time
import random
def response_generator():
    response = random.choice(
        [
            "Hello! How can I assist you today?",
            "Hello! 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)

Let’s see that it’s working as expected.

for chunk in response_generator():
    print(chunk)

Hello! 
How 
can 
I 
assist 
you 
today?

It works. Notice that for the moment the response_generator function will give one of the two possible responses at random without considering the user message.

Let’s now create a function that will be adding the chunks successively to the message.

def add_chunk_to_ai_message(chunk: str):
    messages.value = [
        *messages.value[:-1],
        {
            "role": "assistant",
            "content": messages.value[-1]["content"] + chunk,
        },
    ]

We need to modify the EchoBot code to include this functionality as follows.

messages: solara.Reactive[List[MessageDict]] = solara.reactive([])
@solara.component
def Page():
    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])
    with solara.lab.ChatBox():
        for item in messages.value:
            with solara.lab.ChatMessage(
                user=item["role"] == "user",
                name="User" if item["role"] == "user" else "StreamBot"
            ):
                solara.Markdown(item["content"])
    solara.lab.ChatInput(send_callback=send)
Page()

The complete code can be found below.

Show the code

import solara
import time
import random
from typing import List
from typing_extensions import TypedDict

class MessageDict(TypedDict):
    role: str
    content: str

messages: solara.Reactive[List[MessageDict]] = solara.reactive([])

# Streamed response emulator
def response_generator():
    response = random.choice(
        [
            "Hello! How can I assist you today?",
            "Hello! 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)

def add_chunk_to_ai_message(chunk: str):
    messages.value = [
        *messages.value[:-1],
        {
            "role": "assistant",
            "content": messages.value[-1]["content"] + chunk,
        },
    ]
    
messages: solara.Reactive[List[MessageDict]] = solara.reactive([])
@solara.component
def Page():
    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])
    with solara.lab.ChatBox():
        for item in messages.value:
            with solara.lab.ChatMessage(
                user=item["role"] == "user",
                name="User" if item["role"] == "user" else "StreamBot"
            ):
                solara.Markdown(item["content"])
    solara.lab.ChatInput(send_callback=send)
Page()

Our StreamBot application looks like this. Try it out!

ChatGPT bot

The StreamBot application don’t take into account the user message. To reply something coherent, let’s use one of OpenAI models (in this example, gpt-3.5-turbo).

First, obtain an OPENAI_API_KEY=sk-... and replace it below.

import os
import openai
from openai import OpenAI
from dotenv import load_dotenv, find_dotenv
_ = load_dotenv(find_dotenv()) # read local .env file
openai.api_key = os.environ['OPENAI_API_KEY']

client = OpenAI()

Now we can define a new response_generator function that will use OpenAI to give a coherent answer.

def response_generator(message):
    return client.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=[
           {"role": "system", "content": "You are a helpful assistant."},
           {"role": "user", "content": message}
        ],
        stream=True
    )

Let’s see that it works (as you can see in the code, we need to add some cleaning to the chunks and verify they are not None).

for chunk in response_generator("Hello!"):
    if chunk.choices[0].delta.content is not None:
        print(chunk.choices[0].delta.content)


Hello
!
 How
 can
 I
 assist
 you
 today
?

We need to modify the StreamBot code as follows.

messages: solara.Reactive[List[MessageDict]] = solara.reactive([])
@solara.component
def Page():
    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(message):
            if chunk.choices[0].delta.content is not None:
                add_chunk_to_ai_message(chunk.choices[0].delta.content)
    def result():
        if messages.value != []:
            response(messages.value[-1]["content"])
    result = solara.lab.use_task(result, dependencies=[user_message_count])
    with solara.lab.ChatBox():
        for item in messages.value:
            with solara.lab.ChatMessage(
                user=item["role"] == "user",
                name="User" if item["role"] == "user" else "ChatGPT"
            ):
                solara.Markdown(item["content"])
    solara.lab.ChatInput(send_callback=send)
Page()

The complete code can be found below.

Show the code

import solara
from typing import List
from typing_extensions import TypedDict
import os
import openai
from openai import OpenAI

openai.api_key = os.environ['OPENAI_API_KEY']

client = OpenAI()

def response_generator(message):
    return client.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=[
           {"role": "system", "content": "You are a helpful assistant."},
           {"role": "user", "content": message}
        ],
        stream=True
    )

class MessageDict(TypedDict):
    role: str
    content: str

messages: solara.Reactive[List[MessageDict]] = solara.reactive([])
@solara.component
def Page():
    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(message):
            if chunk.choices[0].delta.content is not None:
                add_chunk_to_ai_message(chunk.choices[0].delta.content)
    def result():
        if messages.value != []:
            response(messages.value[-1]["content"])
    result = solara.lab.use_task(result, dependencies=[user_message_count])
    with solara.lab.ChatBox():
        for item in messages.value:
            with solara.lab.ChatMessage(
                user=item["role"] == "user",
                name="User" if item["role"] == "user" else "ChatGPT"
            ):
                solara.Markdown(item["content"])
    solara.lab.ChatInput(send_callback=send)
Page()