Speedy Prototyping of Chatbots with Streamlit and Chainlit

0
1404
Speedy Prototyping of Chatbots with Streamlit and Chainlit


means of constructing — and accumulating common consumer suggestions on — easy variations of a product to rapidly validate necessary assumptions and hypotheses, and assess key dangers. This method is carefully aligned with the observe of agile software program growth, and the “build-measure-learn” course of within the Lean Startup methodology, and may considerably scale back growth prices and shorten the time-to-market. Speedy prototyping is particularly helpful for transport profitable AI merchandise, given the early-stage nature of associated applied sciences, use instances, and consumer expectations.

To this finish, Streamlit was launched in 2019 as a Python framework that simplifies the method of prototyping AI apps that require consumer interfaces (UIs). Information scientists and engineers can concentrate on the backend elements (e.g., coaching an ML mannequin and exposing a prediction endpoint through an API), and with just a few strains of Python code, Streamlit can spin up a user-friendly, customizable UI. Chainlit, additionally a Python framework, was launched extra not too long ago in 2023 to particularly deal with ache factors in prototyping conversational AI functions (i.e., chatbots). Whereas Streamlit and Chainlit are related in some methods, there are additionally necessary variations. On this article, we’ll look at the professionals and cons of each frameworks by constructing end-to-end demo chatbot functions, and supply sensible suggestions.

Observe: All figures within the following sections have been created by the creator of this text.

Finish-to-Finish Chatbot Demos

Native Setup

For simplicity, we’ll construct the demo functions in order that they will simply be examined in an area surroundings utilizing open-source giant language fashions (LLMs) accessed through Ollama, a device for downloading, managing, and interacting with open-source LLMs in a user-friendly approach on one’s native machine.

After all, the demos can later be modified to be used in manufacturing, e.g., by leveraging the most recent LLMs supplied by the likes of OpenAI or Google, and by deploying the chatbot on a generally used hyperscaler resembling AWS, Azure, or GCP. All implementation steps beneath have been examined on macOS Sequoia 15.6.1, and must be roughly related on Linux and Home windows.

Go right here to obtain and set up Ollama. Examine that the set up was profitable by operating this command within the Terminal:

ollama --version

We are going to use Google’s light-weight Gemma 2 mannequin with 2B parameters, which could be downloaded with this command:

ollama pull gemma:2b

The mannequin file dimension is round 1.7 GB, so the obtain may take a couple of minutes relying in your web connection. Confirm that the mannequin has been downloaded utilizing this command:

ollama listing

It will present all of the fashions which have been downloaded through Ollama thus far.

Subsequent, we’ll arrange the venture listing utilizing uv, a quick and user-friendly venture administration device for Python. Comply with the directions right here to put in uv, and confirm the set up utilizing this command:

uv --version

Initialize a venture listing referred to as chatbot-demos at an appropriate location in your native machine like this:

uv init --bare chatbot-demos

With out specifying the --bare choice, uv would have created some customary artifacts throughout initialization, resembling fundamental.py, README.md, and a Python model pin file, however these will not be wanted for our demos. The minimal course of solely creates a pyproject.toml file.

Within the chatbot-demos venture listing, create a necessities.txt file with the next dependencies:

chainlit==2.7.2
ollama==0.5.3
streamlit==1.49.1

Now create a digital Python 3.12 surroundings contained in the venture listing, activate the surroundings, and set up the dependencies:

uv venv --python=3.12 
supply .venv/bin/activate
uv add -r necessities.txt

Examine that the dependencies have been put in:

uv pip listing

We are going to implement a category referred to as LLMClient for backend performance that may be decoupled from the UI-centric performance, which is the important thing differentiator of frameworks like Streamlit and Chainlit. For instance, LLMClient might handle duties resembling selecting between LLM suppliers, executing LLM calls, interacting with exterior databases for retrieval-augmented era (RAG), and logging the dialog historical past for later evaluation. Right here is an instance implementation of LLMClient, saved in a file referred to as llm_client.py:

import logging
import time
from datetime import datetime, timezone
from typing import Record, Dict, Non-compulsory, Callable, Any, Generator
import os
import ollama

LOG_FILE = os.path.be part of(os.path.dirname(__file__), "conversation_history.log")

logger = logging.getLogger("conversation_logger")
logger.setLevel(logging.INFO)

if not logger.handlers:
    fh = logging.FileHandler(LOG_FILE, encoding="utf-8")
    fmt = logging.Formatter("%(asctime)s - %(message)s")
    fh.setFormatter(fmt)
    logger.addHandler(fh)

class LLMClient:
    def __init__(
        self,
        supplier: str = "ollama",
        mannequin: str = "gemma:2b",
        temperature: float = 0.2,
        retriever: Non-compulsory[Callable[[str], Record[str]]] = None,
        feedback_handler: Non-compulsory[Callable[[Dict[str, Any]], None]] = None,
        logger: Non-compulsory[Callable[[Dict[str, Any]], None]] = None
    ):
        self.supplier = supplier
        self.mannequin = mannequin
        self.temperature = temperature
        self.retriever = retriever
        self.feedback_handler = feedback_handler
        self.logger = logger or self.default_logger

    def default_logger(self, information: Dict[str, Any]):
        logging.data(f"[LLMClient] {information}")

    def _format_messages(self, messages: Record[Dict[str, str]]) -> str:
        return "n".be part of(f"{m['role'].capitalize()}: {m['content']}" for m in messages)

    def _stream_provider(self, immediate: str, temperature: float) -> Generator[str, None, None]:
        if self.supplier == "ollama":
            for chunk in ollama.generate(
                mannequin=self.mannequin,
                immediate=immediate,
                stream=True,
                choices={"temperature": temperature}
            ):
                yield chunk.get("response", "")
        else:
            elevate ValueError(f"Streaming not applied for supplier: {self.supplier}")

    def stream_generate(
        self,
        messages: Record[Dict[str, str]],
        on_token: Callable[[str], None],
        temperature: Non-compulsory[float] = None
    ) -> Dict[str, Any]:
        start_time = time.time()

        if self.retriever:
            question = messages[-1]["content"]
            docs = self.retriever(question)
            if docs:
                context_str = "n".be part of(docs)
                messages = [{"role": "system", "content": f"Use this context:n{context_str}"}] + messages

        immediate = self._format_messages(messages)
        assembled_text = ""
        temp_to_use = temperature if temperature isn't None else self.temperature

        strive:
            for token in self._stream_provider(immediate, temp_to_use):
                assembled_text += token
                on_token(token)
        besides Exception as e:
            assembled_text = f"Error: {e}"

        latency = time.time() - start_time

        outcome = {
            "textual content": assembled_text,
            "timestamp": datetime.now(timezone.utc),
            "latency": latency,
            "supplier": self.supplier,
            "mannequin": self.mannequin,
            "temperature": temp_to_use,
            "messages": messages
        }

        self.logger({
            "occasion": "llm_stream_call",
            "supplier": self.supplier,
            "mannequin": self.mannequin,
            "temperature": temp_to_use,
            "latency": latency,
            "immediate": immediate,
            "response": assembled_text
        })

        return outcome

    def record_feedback(self, suggestions: Dict[str, Any]):
        if self.feedback_handler:
            self.feedback_handler(suggestions)
        else:
            self.logger({"occasion": "suggestions", **suggestions})
    
    def log_interaction(self, function: str, content material: str):
        logger.data(f"{function.higher()}: {content material}")

Primary Streamlit Demo

Create a file referred to as st_app_basic.py within the venture listing and paste within the following code:

import streamlit as st
from llm_client import LLMClient

MAX_HISTORY = 5
llm_client = LLMClient(supplier="ollama", mannequin="gemma:2b")

st.set_page_config(page_title="Streamlit Primary Chatbot", format="centered")
st.title("Streamlit Primary Chatbot")

if "messages" not in st.session_state:
    st.session_state.messages = []

# Show chat historical past
for msg in st.session_state.messages:
    with st.chat_message(msg["role"]):
        st.markdown(msg["content"])

# Person enter
if immediate := st.chat_input("Sort your message..."):
    st.session_state.messages.append({"function": "consumer", "content material": immediate})
    st.session_state.messages = st.session_state.messages[-MAX_HISTORY:]
    llm_client.log_interaction("consumer", immediate)

    with st.chat_message("assistant"):
        response_container = st.empty()
        state = {"full_response": ""}

        def on_token(token):
            state["full_response"] += token
            response_container.markdown(state["full_response"])

        outcome = llm_client.stream_generate(st.session_state.messages, on_token)
        st.session_state.messages.append({"function": "assistant", "content material": outcome["text"]})
        llm_client.log_interaction("assistant", outcome["text"])

Launch the app at localhost:8501 like this:

streamlit run st_app_basic.py

If the app doesn’t open robotically in your default browser, navigate to the URL manually (http://localhost:8501). It is best to see a bare-bones chat interface. Enter the next query within the immediate discipline and hit Enter:

What’s the components to transform Celsius to Fahrenheit?

Determine 1 reveals the outcome:

Determine 1: Preliminary Streamlit Q&A

Now, ask this follow-up query:

Are you able to implement that components in Python?

Since our demo implementation retains observe of the dialog historical past for as much as 5 earlier messages, the chatbot will be capable to affiliate “that components” with the one within the previous immediate, as proven in Determine 2 beneath:

Determine 2: Comply with-Up Streamlit Q&A

Be at liberty to mess around with a number of extra prompts. To shut the app, execute Management + c within the Terminal.

Primary Chainlit Demo

Create a file referred to as cl_app_basic.py within the venture listing and paste within the following code:

import chainlit as cl
from llm_client import LLMClient

MAX_HISTORY = 5
llm_client = LLMClient(supplier="ollama", mannequin="gemma:2b")

@cl.on_chat_start
async def begin():
    await cl.Message(content material="Welcome! Ask me something.").ship()
    cl.user_session.set("messages", [])

@cl.on_message
async def fundamental(message: cl.Message):
    messages = cl.user_session.get("messages")
    messages.append({"function": "consumer", "content material": message.content material})
    messages[:] = messages[-MAX_HISTORY:]
    llm_client.log_interaction("consumer", message.content material)

    state = {"full_response": ""}
    
    def on_token(token):
        state["full_response"] += token

    outcome = llm_client.stream_generate(messages, on_token)
    messages.append({"function": "assistant", "content material": outcome["text"]})
    llm_client.log_interaction("assistant", outcome["text"])

    await cl.Message(content material=outcome["text"]).ship()

Launch the app at localhost:8000 (word the totally different port) like this:

chainlit run cl_app_basic.py

For the sake of comparability, we’ll run the identical two prompts as earlier than. The outcomes are proven in Figures 3 and 4 beneath:

Determine 3: Preliminary Chainlit Q&A
Determine 4: Comply with-Up Chainlit Q&A

As earlier than, after enjoying round with some extra prompts, shut the app by executing Management + c within the Terminal.

Superior Streamlit Demo

We are going to now lengthen the essential Streamlit demo with a persistent sidebar on the left aspect with a slider widget to toggle the temperature parameter of the LLM, a button to obtain chat historical past, and suggestions buttons beneath every chatbot response (“Useful”, “Not Useful”). Customizing the app format and including international widgets could be achieved comparatively simply in Streamlit however could also be cumbersome to copy in Chainlit — readers can provide it a go to expertise the difficulties first-hand.

Right here is the prolonged Streamlit app, saved in a file referred to as st_app_advanced.py:

import streamlit as st
from llm_client import LLMClient
import json

MAX_HISTORY = 5
llm_client = LLMClient(supplier="ollama", mannequin="gemma:2b")

st.set_page_config(page_title="Streamlit Superior Chatbot", format="huge")
st.title("Streamlit Superior Chatbot")

# Sidebar controls
st.sidebar.header("Mannequin Settings")
temperature = st.sidebar.slider("Temperature", 0.0, 1.0, 0.2, 0.1)  # min, max, default, increment dimension
st.sidebar.download_button(
    "Obtain Chat Historical past",
    information=json.dumps(st.session_state.get("messages", []), indent=2),
    file_name="chat_history.json",
    mime="utility/json"
)

if "messages" not in st.session_state:
    st.session_state.messages = []

# Show chat historical past
for msg in st.session_state.messages:
    with st.chat_message(msg["role"]):
        st.markdown(msg["content"])

# Person enter
if immediate := st.chat_input("Sort your message..."):
    st.session_state.messages.append({"function": "consumer", "content material": immediate})
    st.session_state.messages = st.session_state.messages[-MAX_HISTORY:]
    llm_client.log_interaction("consumer", immediate)

    with st.chat_message("assistant"):
        response_container = st.empty()
        state = {"full_response": ""}

        def on_token(token):
            state["full_response"] += token
            response_container.markdown(state["full_response"])

        outcome = llm_client.stream_generate(
            st.session_state.messages,
            on_token,
            temperature=temperature
        )
        llm_client.log_interaction("assistant", outcome["text"])
        st.session_state.messages.append({"function": "assistant", "content material": outcome["text"]})

        # Suggestions buttons
        col1, col2 = st.columns(2)
        if col1.button("Useful"):
            llm_client.record_feedback({"ranking": "up", "remark": "Person preferred the reply"})
        if col2.button("Not Useful"):
            llm_client.record_feedback({"ranking": "down", "remark": "Person disliked the reply"})

Determine 5 reveals an instance screenshot:

Determine 5: Demo of Superior Streamlit Options

Superior Chainlit Demo

Subsequent, we’ll lengthen the essential Chainlit demo with per-message interactive actions and multimodal enter dealing with (textual content and pictures in our case). The chat-native primitives of the Chainlit framework make it simpler to implement most of these options than in Streamlit. Once more, readers are inspired to expertise the distinction by making an attempt to copy the performance utilizing Streamlit.

Right here is the prolonged Chainlit app, saved in a file referred to as cl_app_advanced.py:

import os
import json
from typing import Record, Dict
import chainlit as cl
from llm_client import LLMClient

MAX_HISTORY = 5
DEFAULT_TEMPERATURE = 0.2
SESSIONS_DIR = os.path.be part of(os.path.dirname(__file__), "periods")
os.makedirs(SESSIONS_DIR, exist_ok=True)

llm_client = LLMClient(supplier="ollama", mannequin="gemma:2b", temperature=DEFAULT_TEMPERATURE)

def _session_file(session_name: str) -> str:
    protected = "".be part of(c for c in session_name if c.isalnum() or c in ("-", "_"))
    return os.path.be part of(SESSIONS_DIR, f"{protected or 'default'}.json")

def _save_session(session_name: str, messages: Record[Dict]):
    with open(_session_file(session_name), "w", encoding="utf-8") as f:
        json.dump(messages, f, ensure_ascii=False, indent=2)

def _load_session(session_name: str) -> Record[Dict]:
    path = _session_file(session_name)
    if os.path.exists(path):
        with open(path, "r", encoding="utf-8") as f:
            return json.load(f)
    return []

@cl.on_chat_start
async def begin():
    cl.user_session.set("messages", [])
    cl.user_session.set("session_name", "default")
    cl.user_session.set("last_assistant_idx", None)

    await cl.Message(
        content material=(
            "Welcome! Ask me something."
        ),
        actions=[
            cl.Action(name="set_session_name", label="Set session name", payload={"turn": None}),
            cl.Action(name="save_session", label="Save session", payload={"turn": "save"}),
            cl.Action(name="load_session", label="Load session", payload={"turn": "load"}),
        ],
    ).ship()

@cl.action_callback("set_session_name")
async def set_session_name(motion):
    await cl.Message(content material="Please sort: /identify YOUR_SESSION_NAME").ship()

@cl.action_callback("save_session")
async def save_session(motion):
    session_name = cl.user_session.get("session_name")
    _save_session(session_name, cl.user_session.get("messages", []))
    await cl.Message(content material=f"Session saved as '{session_name}'.").ship()

@cl.action_callback("load_session")
async def load_session(motion):
    session_name = cl.user_session.get("session_name")
    loaded = _load_session(session_name)
    cl.user_session.set("messages", loaded[-MAX_HISTORY:])
    await cl.Message(content material=f"Loaded session '{session_name}' with {len(loaded)} flip(s).").ship()

@cl.on_message
async def fundamental(message: cl.Message):
    if message.content material.strip().startswith("/identify "):
        new_name = message.content material.strip()[6:].strip() or "default"
        cl.user_session.set("session_name", new_name)
        await cl.Message(content material=f"Session identify set to '{new_name}'.").ship()
        return

    messages = cl.user_session.get("messages")

    user_text = message.content material or ""
    if message.components:
        for aspect in message.components:
            if getattr(aspect, "mime", "").startswith("picture/"):
                user_text += f" [Image: {element.name}]"

    messages.append({"function": "consumer", "content material": user_text})
    messages[:] = messages[-MAX_HISTORY:]
    llm_client.log_interaction("consumer", user_text)

    state = {"full_response": ""}
    msg = cl.Message(content material="")

    def on_token(token: str):
        state["full_response"] += token
        cl.run_sync(msg.stream_token(token))

    outcome = llm_client.stream_generate(messages, on_token, temperature=DEFAULT_TEMPERATURE)
    messages.append({"function": "assistant", "content material": outcome["text"]})
    llm_client.log_interaction("assistant", outcome["text"])

    msg.content material = state["full_response"]
    await msg.ship()
    
    turn_idx = len(messages) - 1
    cl.user_session.set("last_assistant_idx", turn_idx)

    await cl.Message(
        content material="Was this beneficial?",
        actions=[
            cl.Action(name="thumbs_up", label="Yes", payload={"turn": turn_idx}),
            cl.Action(name="thumbs_down", label="No", payload={"turn": turn_idx}),
            cl.Action(name="save_session", label="Save session", payload={"turn": "save"}),
        ],
    ).ship()

@cl.action_callback("thumbs_up")
async def thumbs_up(motion):
    flip = motion.payload.get("flip")
    llm_client.record_feedback({"ranking": "up", "flip": flip})
    await cl.Message(content material="Thanks in your suggestions!").ship()

@cl.action_callback("thumbs_down")
async def thumbs_down(motion):
    flip = motion.payload.get("flip")
    llm_client.record_feedback({"ranking": "down", "flip": flip})
    await cl.Message(content material="Thanks in your suggestions.").ship()

Determine 6 reveals an instance screenshot:

Determine 6: Demo of Superior Chainlit Options

Sensible Steering

Because the earlier part demonstrates, it’s attainable to quickly prototype easy chatbot functions with each Streamlit and Chainlit. Within the primary demos that we applied, there have been a number of architectural similarities: the calls to Ollama and dialog logging was abstracted away utilizing the LLMClient class, the context dimension was restricted utilizing a relentless variable referred to as MAX_HISTORY, and the historical past was serialized right into a plaintext chat format. Because the superior demos present, nonetheless, the scope of every framework is considerably totally different, which entails sure execs and cons relying on the use case together with associated sensible suggestions.

Whereas Streamlit is a general-purpose framework for data-centric, interactive net apps, Chainlit is concentrated on constructing and deploying conversational AI apps. Thus, Chainlit might make extra sense to make use of if the chatbot is central to the prototype; because the above code examples illustrate, Chainlit takes care of a number of boilerplate operational particulars (e.g., built-in chat options for native typing indicators, message streaming, and markdown/code rendering). But when the chatbot is embedded in a bigger AI product, Streamlit might be able to higher address the bigger utility scope (e.g., combining the chat interface with information visualizations, dashboards, international widgets, and customized layouts).

Moreover, the conversational components in AI functions might should be dealt with in an asynchronous method to make sure a great consumer expertise (UX), since messages can arrive at any time and should be processed rapidly whereas different duties could also be in progress (e.g., calling one other API or streaming mannequin output). Chainlit makes it straightforward to prototype asynchronous chat logic utilizing Python’s async and await key phrases, guaranteeing that the app can deal with concurrent operations with out blocking the UI. The framework takes care of low-level particulars round managing WebSocket connections and customized polling, in order that every time an occasion is triggered (e.g., message despatched, token streamed, state modified), the occasion dealing with logic of Chainlit robotically triggers UI updates as required. Against this, Streamlit makes use of synchronous communication, which causes the app script to rerun with every consumer interplay; for advanced apps that must juggle a number of concurrent processes, Chainlit might enable for a smoother UX than Streamlit.

Lastly, past the restrictions that include focusing totally on chat-based functions, Chainlit was launched a number of years after Streamlit, so it’s at the moment much less technically mature and has a smaller developer group; e.g., fewer third‑get together extensions, group‑contributed examples, and troubleshooting assets can be found for the time being. Though Chainlit is evolving quickly and gaps are actively being addressed, builders might encounter occasional breaking modifications between variations, much less complete documentation for superior use instances, and restricted integration steering for sure deployment environments. Product groups that also want to prototype chatbot-centric AI functions utilizing Chainlit resulting from potential long-term architectural advantages ought to thus be ready to make some extra short-term investments in customized growth, experimentation, and direct engagement with the framework maintainers and related group boards to resolve points and request extra performance.