Chainlit: Building Production-Ready Chat UI for LLM Applications
Building a powerful LLM application is not enough. Users need an intuitive and professional chat interface to interact with AI models. However, building a chat UI from scratch requires significant time and effort, from WebSocket handling, state management, to file handling.
Chainlit is a Python framework that allows you to build production-ready chat UIs in minutes. With Chainlit, you can focus on AI logic while getting a beautiful interface with streaming responses, file uploads, and many other features automatically.What Is Chainlit?
Chainlit is an open-source Python framework for building conversational AI chat interfaces. Its key features include:
- Production-Ready Chat UI: Modern interface with markdown support, code highlighting, and file preview
- Streaming Response: Display LLM responses in real-time token by token
- Multi-Step Reasoning: Visualize AI thinking process with step-by-step display
- File Upload & Download: Built-in file handling for documents, images, and more
- User Authentication: Integrated login system with various providers
- Chat History: Storage and recall of previous conversations
- LangChain/LlamaIndex Integration: Native support for popular AI frameworks
- Custom Actions: Interactive buttons and custom actions within chat
- Theming: Customize appearance to match your brand
Installation
Basic Installation
pip install chainlit
Installation with Additional Dependencies
# With LangChain support
pip install chainlit langchain langchain-openai
With LlamaIndex support
pip install chainlit llama-index
Verify Installation
chainlit --version
Running Your First Application
Create a file app.py:
import chainlit as cl
@cl.onmessage
async def main(message: cl.Message):
# Send a simple response
await cl.Message(
content=f"You sent: {message.content}"
).send()
Run with:
chainlit run app.py
Open your browser at http://localhost:8000 and you will see a professional chat interface.
Core Concepts: Decorators and Messages
@cl.onmessage - Main Message Handler
The @cl.onmessage decorator is the main entry point. This function is called every time a user sends a message.
import chainlit as cl
@cl.onmessage
async def handlemessage(message: cl.Message):
# Access message content
userinput = message.content
# Access uploaded files (if any)
files = message.elements
# Process and send response
response = f"Message received: {userinput}"
await cl.Message(content=response).send()
@cl.onchatstart - Chat Initialization
This decorator is called when a new chat session starts. Use it to set up context, models, or greet the user.
import chainlit as cl
@cl.onchatstart
async def start():
# Store state in session
cl.usersession.set("history", [])
# Send welcome message
await cl.Message(
content="Hello! I am an AI assistant. How can I help you?"
).send()
@cl.onmessage
async def main(message: cl.Message):
# Get history from session
history = cl.usersession.get("history")
history.append({"role": "user", "content": message.content})
# Process message...
response = "Response from AI"
history.append({"role": "assistant", "content": response})
cl.usersession.set("history", history)
await cl.Message(content=response).send()
Message Types: Text, Images, and Files
Text Messages with Markdown
@cl.onmessage
async def main(message: cl.Message):
# Chainlit supports full markdown
response = """
Analysis Results
Here are your data analysis results:
| Metric | Value |
|--------|-------|
| Accuracy | 95.2% |
| Precision | 93.8% |
| Recall | 96.1% |
Conclusion: The model shows excellent performance.
"""
await cl.Message(content=response).send()
Sending Images
@cl.onmessage
async def main(message: cl.Message):
# Send image as element
image = cl.Image(
path="./chart.png",
name="analysis
chart",
display="inline"
)
await cl.Message(
content="Here is the analysis visualization:",
elements=[image]
).send()
Sending Downloadable Files
@cl.onmessage
async def main(message: cl.Message):
# Send file as attachment
file = cl.File(
path="./report.pdf",
name="analysis
report.pdf",
display="inline"
)
await cl.Message(
content="The report has been generated. Please download:",
elements=[file]
).send()
Streaming Responses
Streaming allows responses to be displayed in real-time, providing a better user experience.
Basic Streaming
import chainlit as cl
from openai import AsyncOpenAI
client = AsyncOpenAI()
@cl.onmessage
async def main(message: cl.Message):
msg = cl.Message(content="")
# Stream from OpenAI
stream = await client.chat.completions.create(
model="gpt-4",
messages=[
{"role": "user", "content": message.content}
],
stream=True
)
async for chunk in stream:
if chunk.choices[0].delta.content:
await msg.streamtoken(chunk.choices[0].delta.content)
# Finalize message
await msg.send()
Streaming with Token Tracking
@cl.onmessage
async def main(message: cl.Message):
msg = cl.Message(content="")
total
tokens = 0
stream = await client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": message.content}],
stream=True,
streamoptions={"includeusage": True}
)
async for chunk in stream:
if chunk.choices and chunk.choices[0].delta.content:
await msg.streamtoken(chunk.choices[0].delta.content)
if chunk.usage:
totaltokens = chunk.usage.totaltokens
await msg.send()
# Display token info
await cl.Message(
content=f"Total tokens used: {totaltokens}"
).send()
Multi-Step Reasoning Display
Chainlit can display the AI thinking process step by step using Steps.
import chainlit as cl
@cl.onmessage
async def main(message: cl.Message):
# Step 1: Analyze question
async with cl.Step(name="Analyzing Question") as step:
step.input = message.content
analysis = "Question about machine learning classification"
step.output = analysis
# Step 2: Document search
async with cl.Step(name="Searching Relevant Documents") as step:
step.input = analysis
docs = "Found 5 relevant documents"
step.output = docs
# Step 3: Generate answer
async with cl.Step(name="Generating Answer") as step:
step.input = f"Query: {analysis}\nDocuments: {docs}"
answer = "Based on the documents found..."
step.output = answer
# Send final answer
await cl.Message(content=answer).send()
File Upload Handling
Requesting Files from Users
import chainlit as cl
@cl.onchatstart
async def start():
files = await cl.AskFileMessage(
content="Please upload a CSV file for analysis:",
accept=["text/csv"],
maxsizemb=10,
maxfiles=1
).send()
if files:
file = files[0]
# Read file content
import pandas as pd
df = pd.readcsv(file.path)
await cl.Message(
content=f"File {file.name} uploaded successfully!\n\n"
f"- Rows: {len(df)}\n"
f"- Columns: {len(df.columns)}\n"
f"- Column names: {', '.join(df.columns.tolist())}"
).send()
cl.usersession.set("dataframe", df)
Handling Multiple File Types
@cl.onchatstart
async def start():
files = await cl.AskFileMessage(
content="Upload your documents (PDF, TXT, or DOCX):",
accept={
"application/pdf": [".pdf"],
"text/plain": [".txt"],
"application/vnd.openxmlformats-officedocument.wordprocessingml.document": [".docx"]
},
maxsizemb=20,
maxfiles=5
).send()
if files:
fileinfo = []
for f in files:
fileinfo.append(f"- {f.name} ({f.type}, {f.size} bytes)")
await cl.Message(
content=f"Successfully uploaded {len(files)} files:\n" +
"\n".join(fileinfo)
).send()
User Authentication
Password Authentication
Create a .chainlit/config.toml file:
[project]
name = "My AI Assistant"
enabletelemetry = false
[UI]
name = "AI Assistant"
Implement authentication:
import chainlit as cl
@cl.passwordauthcallback
def authcallback(username: str, password: str):
# Validate credentials (use a database in production)
if username == "admin" and password == "secret123":
return cl.User(
identifier="admin",
metadata={"role": "admin", "name": "Administrator"}
)
return None
@cl.onchatstart
async def start():
user = cl.usersession.get("user")
await cl.Message(
content=f"Welcome, {user.metadata['name']}!"
).send()
Header Authentication (for Reverse Proxy)
@cl.headerauthcallback
def header
auth(headers: dict):
# Authentication via header (e.g., from reverse proxy)
if headers.get("X-User-Email"):
return cl.User(
identifier=headers["X-User-Email"],
metadata={"provider": "header"}
)
return None
Chat History and Persistence
Saving Chat History
import chainlit as cl
import json
from pathlib import Path
HISTORYDIR = Path("chathistory")
HISTORYDIR.mkdir(existok=True)
@cl.onchatstart
async def start():
cl.usersession.set("history", [])
@cl.onmessage
async def main(message: cl.Message):
history = cl.usersession.get("history")
history.append({
"role": "user",
"content": message.content
})
# Process with LLM
response = await generateresponse(history)
history.append({
"role": "assistant",
"content": response
})
cl.usersession.set("history", history)
await cl.Message(content=response).send()
@cl.onchatend
async def end():
history = cl.usersession.get("history")
if history:
# Save to file
sessionid = cl.usersession.get("id")
filepath = HISTORYDIR / f"{sessionid}.json"
with open(filepath, "w") as f:
json.dump(history, f, indent=2)
Using Chainlit Data Layer
Chainlit provides a data layer abstraction for persisting conversations.
import chainlit.data as cldata
from chainlit.data.sql import SQLAlchemyDataLayer
Configure data layer with PostgreSQL
cl
data.init(
SQLAlchemyDataLayer(
conninfo="postgresql://user:pass@localhost/chainlitdb"
)
)
Integration with LangChain
import chainlit as cl
from langchainopenai import ChatOpenAI
from langchain.schema import HumanMessage, AIMessage, SystemMessage
from langchain.schema.runnable.config import RunnableConfig
@cl.onchatstart
async def start():
model = ChatOpenAI(
model="gpt-4",
temperature=0.7,
streaming=True
)
cl.usersession.set("model", model)
cl.usersession.set("history", [
SystemMessage(content="You are a helpful AI assistant.")
])
@cl.onmessage
async def main(message: cl.Message):
model = cl.usersession.get("model")
history = cl.usersession.get("history")
history.append(HumanMessage(content=message.content))
msg = cl.Message(content="")
async for chunk in model.astream(
history,
config=RunnableConfig(callbacks=[cl.LangchainCallbackHandler()])
):
await msg.streamtoken(chunk.content)
history.append(AIMessage(content=msg.content))
cl.usersession.set("history", history)
await msg.send()
Integration with LlamaIndex
import chainlit as cl
from llamaindex.core import (
VectorStoreIndex,
SimpleDirectoryReader,
Settings
)
from llamaindex.llms.openai import OpenAI
@cl.onchatstart
async def start():
Settings.llm = OpenAI(model="gpt-4", temperature=0.1)
# Load and index documents
await cl.Message(content="Loading documents...").send()
documents = SimpleDirectoryReader("./data").loaddata()
index = VectorStoreIndex.fromdocuments(documents)
queryengine = index.asqueryengine(streaming=True)
cl.usersession.set("queryengine", queryengine)
await cl.Message(
content="Documents loaded successfully! Feel free to ask questions."
).send()
@cl.onmessage
async def main(message: cl.Message):
queryengine = cl.usersession.get("queryengine")
msg = cl.Message(content="")
response = await cl.makeasync(queryengine.query)(message.content)
# Display sources
sources = []
for node in response.sourcenodes:
sources.append(f"- {node.metadata.get('filename', 'Unknown')} "
f"(score: {node.score:.3f})")
sourcetext = "\n\nSources:\n" + "\n".join(sources)
await cl.Message(
content=str(response) + sourcetext
).send()
Custom Actions and Buttons
import chainlit as cl
@cl.onchatstart
async def start():
# Send message with action buttons
actions = [
cl.Action(
name="summarize",
payload={"action": "summarize"},
label="Summarize Text",
description="Summarize the given text"
),
cl.Action(
name="translate",
payload={"action": "translate"},
label="Translate",
description="Translate to another language"
),
cl.Action(
name="analyze",
payload={"action": "analyze"},
label="Sentiment Analysis",
description="Analyze text sentiment"
)
]
await cl.Message(
content="Choose an action to perform:",
actions=actions
).send()
@cl.actioncallback("summarize")
async def onsummarize(action: cl.Action):
await cl.Message(content="Please send the text you want to summarize.").send()
cl.usersession.set("mode", "summarize")
@cl.actioncallback("translate")
async def ontranslate(action: cl.Action):
await cl.Message(content="Please send the text you want to translate.").send()
cl.usersession.set("mode", "translate")
@cl.actioncallback("analyze")
async def onanalyze(action: cl.Action):
await cl.Message(content="Please send the text for sentiment analysis.").send()
cl.usersession.set("mode", "analyze")
Theming and Customization
Theme Configuration
Create a .chainlit/config.toml file:
[project]
name = "AI Assistant Pro"
enabletelemetry = false
[UI]
name = "AI Assistant Pro"
description = "Smart AI assistant for your needs"
defaulttheme = "dark"
[UI.theme]
fontfamily = "Inter, sans-serif"
[UI.theme.light]
background = "#FFFFFF"
paper = "#F5F5F5"
[UI.theme.dark]
background = "#1A1A2E"
paper = "#16213E"
Custom Logo and Favicon
Place files in the public/ folder:
public/
logodark.png
logolight.png
favicon.png
Custom CSS
Create a public/stylesheet.css file:
/ Customize chat appearance /
.message-content {
font-size: 15px;
line-height: 1.6;
}
/ Customize code blocks /
pre code {
font-size: 13px;
border-radius: 8px;
}
Deployment
Using Docker
Create a Dockerfile:
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
EXPOSE 8000
CMD ["chainlit", "run", "app.py", "--host", "0.0.0.0", "--port", "8000"]
Create a docker-compose.yml:
version: '3.8'
services:
chainlit-app:
build: .
ports:
- "8000:8000"
environment:
- OPENAIAPIKEY=${OPENAIAPIKEY}
volumes:
- ./data:/app/data
- ./chathistory:/app/chathistory
restart: unless-stopped
Deploy to Cloud
# Build and run
docker build -t chainlit-app .
docker run -d -p 8000:8000 \
-e OPENAIAPIKEY=$OPENAIAPIKEY \
chainlit-app
Practical Example: RAG Chatbot with File Upload and Source Citations
Here is a complete example of building a RAG chatbot that supports file upload and displays source citations.
import chainlit as cl
from langchainopenai import ChatOpenAI, OpenAIEmbeddings
from langchain.textsplitter import RecursiveCharacterTextSplitter
from langchaincommunity.vectorstores import FAISS
from langchain.chains import ConversationalRetrievalChain
from langchain.memory import ConversationBufferWindowMemory
from langchaincommunity.documentloaders import (
PyPDFLoader,
TextLoader,
CSVLoader
)
import tempfile
import os
Configuration
CHUNKSIZE = 1000
CHUNKOVERLAP = 200
def loaddocument(filepath: str, filetype: str):
"""Load document based on file type."""
if filetype == "application/pdf":
loader = PyPDFLoader(filepath)
elif filetype == "text/csv":
loader = CSVLoader(filepath)
else:
loader = TextLoader(filepath)
return loader.load()
@cl.onchatstart
async def start():
# Ask user to upload documents
files = await cl.AskFileMessage(
content="Welcome to RAG Chatbot!\n\n"
"Please upload the documents you want to ask about "
"(PDF, TXT, or CSV):",
accept={
"application/pdf": [".pdf"],
"text/plain": [".txt"],
"text/csv": [".csv"]
},
maxsizemb=20,
maxfiles=5
).send()
if not files:
return
# Process documents
processingmsg = cl.Message(content="Processing your documents...")
await processingmsg.send()
alldocs = []
for file in files:
docs = loaddocument(file.path, file.type)
for doc in docs:
doc.metadata["source"] = file.name
alldocs.extend(docs)
# Split documents
async with cl.Step(name="Splitting Documents") as step:
textsplitter = RecursiveCharacterTextSplitter(
chunksize=CHUNKSIZE,
chunkoverlap=CHUNKOVERLAP
)
splits = textsplitter.splitdocuments(alldocs)
step.output = f"Document split into {len(splits)} chunks"
# Create vector store
async with cl.Step(name="Creating Vector Index") as step:
embeddings = OpenAIEmbeddings()
vectorstore = FAISS.fromdocuments(splits, embeddings)
step.output = "Vector index created successfully"
# Setup retrieval chain
memory = ConversationBufferWindowMemory(
memorykey="chathistory",
returnmessages=True,
outputkey="answer",
k=5
)
llm = ChatOpenAI(
model="gpt-4",
temperature=0.3,
streaming=True
)
chain = ConversationalRetrievalChain.fromllm(
llm=llm,
retriever=vectorstore.asretriever(searchkwargs={"k": 4}),
memory=memory,
returnsourcedocuments=True,
verbose=False
)
# Store in session
cl.usersession.set("chain", chain)
cl.usersession.set("vectorstore", vectorstore)
# Success notification
filenames = ", ".join([f.name for f in files])
await cl.Message(
content=f"Documents processed successfully: {filenames}\n\n"
f"Total {len(splits)} text chunks have been indexed. "
f"Feel free to ask questions about your documents!"
).send()
@cl.onmessage
async def main(message: cl.Message):
chain = cl.usersession.get("chain")
if not chain:
await cl.Message(
content="Please upload documents first."
).send()
return
# Run query with steps
async with cl.Step(name="Searching Relevant Context") as step:
step.input = message.content
result = await cl.makeasync(chain.invoke)(
{"question": message.content}
)
step.output = f"Found {len(result['sourcedocuments'])} sources"
# Format source citations
sources = []
seen = set()
for doc in result["sourcedocuments"]:
sourcename = doc.metadata.get("source", "Unknown")
page = doc.metadata.get("page", "N/A")
key = f"{sourcename}-{page}"
if key not in seen:
seen.add(key)
snippet = doc.pagecontent[:100].replace("\n", " ")
sources.append(
f"- {sourcename} (p. {page}): \"{snippet}...\""
)
# Send answer with sources
answer = result["answer"]
if sources:
answer += "\n\nSource Citations:\n" + "\n".join(sources)
await cl.Message(content=answer).send()
Running the RAG Chatbot
# Install dependencies
pip install chainlit langchain langchain-openai langchain-community \
faiss-cpu pypdf
Set API key
export OPENAIAPIKEY="sk-your-api-key"
Run
chainlit run app.py
Tips and Best Practices
cl.usersession since each user has a separate sessioncl.Step to show behind-the-scenes processesConclusion
Chainlit is an excellent choice for building production-ready AI chat interfaces. With its simple yet powerful API, you can focus on AI logic without spending time building UI from scratch. Native integration with LangChain and LlamaIndex makes it even easier to build complete RAG applications with enterprise features like authentication, chat history, and easy deployment.