A retrieval-augmented generation (RAG) pipeline can answer questions, but colleagues still need a usable way to ask them. Chainlit is an open-source Python framework for putting a conversational interface in front of that pipeline without building a separate frontend. It provides chat UI features such as streaming responses, conversation history, source display, feedback collection, file uploads, and session management. section 12.3
Keep the application logic separate from the chat interface:
- Install Chainlit with `pip install chainlit`. Put retrieval, model initialization, and query processing in a reusable Python component rather than in UI callbacks. section 12.3
- Have the core return structured answers, source information, document counts, and errors. For streaming, emit source information before successive response tokens. section 12.3
- Use Chainlit’s chat-start and message handlers to connect the interface to the core. Check initialization, validate each user query, and report failures in the conversation rather than silently dropping them. section 12.3
Listing 12.17 Chainlit initialization with DakkaBot startup
import chainlit as cl #1
from dakka_bot_core import DakkaBotCore
from guardrail_helpers import validate_user_inputs #2
# Global bot instance
dakka_bot = DakkaBotCore() #3
@cl.on_chat_start #4
async def start():
"""Initialize DakkaBot when chat starts"""
await cl.Message(
content=(
"
**DakkaBot** is starting up...\n\n"
Initializing DataKrypt knowledge base "
"and AI systems..."),
author="System" #5
).send()
try:
await dakka_bot.initialize() #6
await cl.Message(
content="""
**DakkaBot Ready!**
I'm your DataKrypt developer assistant. I can help you with:
**Encryption & Security**
- Key rotation procedures
- HSM management
- Compliance setup
**APIs & Integration**
- Authentication flows
- Service configurations
- Troubleshooting guides
**Operations & Monitoring**
- Deployment procedures
- Performance optimization
- Incident response
**Compliance & Governance**
- HIPAA, PCI DSS, SOX setup
- Audit procedures
- Data protection
**Ask me anything about DataKrypt!** 
*Example: "How do I rotate encryption keys?"*""",
author="DakkaBot"
).send()
except Exception as e:
await cl.Message(
content=(
f"
**Error starting DakkaBot**: {str(e)}\n\n"
"Please check your configuration and try again."),
author="System"
).send()
The startup handler shows where Chainlit takes over: `@cl.on_chat_start` sends a status message, awaits RAG initialization, and then announces that the assistant is ready. section 12.3
Listing 12.18 Main message handler with streaming and validation
@cl.on_message #1
async def main(message: cl.Message):
"""Handle user messages with streaming response"""
if not dakka_bot.initialized: #2
await cl.Message(
content=(
" Please refresh the page."),
author="System"
).send()
return
user_query = message.content
user_query, is_valid = validate_user_inputs( #3
user_query) #3
if not is_valid:
await cl.Message(
content=user_query, # Contains error message
author="System"
).send()
return
thinking_msg = cl.Message(
content="
Searching DataKrypt documentation...", #4
author="DakkaBot"
)
await thinking_msg.send()
try:
response_msg = cl.Message( #5
content="", author="DakkaBot") #5
sources_info = None
async for chunk in
dakka_bot.process_query_stream( #6
user_query): #6
if chunk["type"] == "sources":
thinking_msg.content = ( #7
f"
Found {chunk['doc_count']} " #7
"relevant documents. " #7
" Generating response..."
await thinking_msg.update()
sources_info = chunk
elif chunk["type"] == "token":
if thinking_msg:
await thinking_msg.remove() #8
thinking_msg = None
await response_msg.stream_token( #9
chunk["content"]) #9
elif chunk["type"] == "done":
await response_msg.send() #10
break
elif chunk["type"] == "error":
if thinking_msg:
await thinking_msg.remove()
await cl.Message(
content=(
f" **Error processing your question**: "
f"{chunk['content']}\n\nPlease try "
f"rephrasing your question or check "
f"if the DataKrypt documentation is "
f"properly loaded."
),
author="DakkaBot"
).send()
return
except Exception as e:
if thinking_msg:
await thinking_msg.remove()
await cl.Message(
content=f" **Unexpected error**: {str(e)}\n\nPlease try again.",
author="DakkaBot"
).send()
The message handler checks that initialization succeeded, validates `message.content`, and sends a searching status before preparing a response. These checks sit at the UI boundary, not inside the retrieval code. section 12.3
This split lets the same RAG core serve a Chainlit chat, an API, or a test wrapper without duplicating retrieval logic. Chainlit is the chat-facing part of the deployment; it does not replace the underlying embeddings, vector store, or language model. section 12.3 section 2.1.5
