A natural language interface lets colleagues query an internal documentation system by sending conversational messages instead of calling an API directly. For a retrieval-augmented generation (RAG) application, the interface turns document search and answer generation into a familiar chat workflow. section 12.3 section 12.2.1
Chainlit supplies the frontend for this workflow without requiring separate React, JavaScript, or CSS development. Its chat-oriented features include streaming responses, conversation history, source-document display, file uploads, session management, and feedback collection. Keep retrieval and generation in a reusable core so the same logic can also serve an API or a test harness. section 12.3
A practical message flow is:
- Initialize the RAG core when the chat starts, and tell the user whether the assistant is starting or ready. section 12.3
- On each message, check initialization and validate the user’s input before processing the query; return an error message if either check fails. section 12.3
- Retrieve relevant documents, then stream generated answer chunks to the interface. Pass source information alongside the response so users can inspect its basis. 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 Chainlit message handler shows where readiness checks, input validation, progress messages, and the streamed response connect to the chat UI. section 12.3
The UI is only one layer of the system. Reusable query logic makes it easier to test and maintain, while observability remains important because generated answers can vary and production LLM calls incur token-based costs. section 12.3 section 12.2.1
