How-To: Set up an API Server with UI for a chatbot in Ragbits#
This guide shows you how to set up a Ragbits API server with a web UI for your chatbot application, covering all response types, customization options, and advanced features.
Quick Start#
1. Create a Basic Chat Implementation#
First, create a chat implementation by subclassing ChatInterface. Here's a minimal example:
from collections.abc import AsyncGenerator
from ragbits.chat.interface import ChatInterface
from ragbits.chat.interface.types import ChatResponse
from ragbits.core.prompt import ChatFormat
from ragbits.core.llms import LiteLLM
class MyChat(ChatInterface):
def __init__(self) -> None:
self.llm = LiteLLM(model_name="gpt-4o-mini")
async def chat(
self,
message: str,
history: ChatFormat,
context: ChatContext,
) -> AsyncGenerator[ChatResponse, None]:
async for chunk in self.llm.generate_streaming([*history, {"role": "user", "content": message}]):
yield self.create_text_response(chunk)
Save this to a file, for example chat.py.
2. Start the User Interface#
Launch the API server with the built-in web UI using the Ragbits CLI:
Note:
path.to.your.moduleshould be the dotted Python module path without the.pyextension.
The server will start on port 8000 by default. Open your browser and navigate to:
You'll see the chat interface where you can interact with your chatbot immediately.
Response Types#
Ragbits Chat supports multiple response types that can be yielded from your chat method to create rich, interactive experiences.
Text Responses#
Text responses are the primary way to stream content to users. Use create_text_response() to yield text chunks:
async def chat(self, message: str, history: ChatFormat, context: ChatContext) -> AsyncGenerator[ChatResponse, None]:
# Stream response from LLM
async for chunk in self.llm.generate_streaming([*history, {"role": "user", "content": message}]):
yield self.create_text_response(chunk)
References#
References allow you to cite sources, documents, or external links that support your response:
async def chat(self, message: str, history: ChatFormat, context: ChatContext) -> AsyncGenerator[ChatResponse, None]:
# Add a reference
yield self.create_reference(
title="Example Reference",
content="This is an example reference document that might be relevant to your query.",
url="https://example.com/reference1",
)
Images#
You can include images in your responses using create_image_response():
import uuid
async def chat(self, message: str, history: ChatFormat, context: ChatContext) -> AsyncGenerator[ChatResponse, None]:
# Add an image to the response
yield self.create_image_response(
str(uuid.uuid4()), # Unique identifier for the image
"https://example.com/image.jpg" # Image URL
)
Follow-up Messages#
Provide suggested follow-up questions to guide the conversation:
async def chat(self, message: str, history: ChatFormat, context: ChatContext) -> AsyncGenerator[ChatResponse, None]:
# Main response...
async for chunk in self.llm.generate_streaming([*history, {"role": "user", "content": message}]):
yield self.create_text_response(chunk)
# Add follow-up suggestions
yield self.create_followup_messages([
"Tell me more about this topic",
"Can you provide another example?",
"How does this relate to X?"
])
Note: Follow-up messages will be displayed as buttons which will be sent to the server as a message after the user clicks on the button.
Live Updates#
Live updates show real-time progress for long-running operations (like web searches, API calls, or data processing):
import asyncio
from ragbits.chat.interface.types import LiveUpdateType
async def chat(self, message: str, history: ChatFormat, context: ChatContext) -> AsyncGenerator[ChatResponse, None]:
# Start a live update
yield self.create_live_update(
"search_task", # Unique task ID
LiveUpdateType.START,
"Searching the web for information..."
)
# Simulate some work
await asyncio.sleep(2)
# Update the live task
yield self.create_live_update(
"search_task",
LiveUpdateType.FINISH,
"Web search completed",
"Found 5 relevant results." # Optional description
)
# You can have multiple concurrent live updates
yield self.create_live_update(
"analysis_task",
LiveUpdateType.FINISH,
"Analysis completed",
"Processed 3 documents."
)
# Continue with text response...
async for chunk in self.llm.generate_streaming([*history, {"role": "user", "content": message}]):
yield self.create_text_response(chunk)
State Management#
Ragbits Chat provides secure state management to maintain conversation context across requests. State data is automatically signed using HMAC to prevent tampering.
Storing State#
Use create_state_update() to store state information that persists across conversation turns:
from ragbits.chat.interface.types import ChatContext
from ragbits.core.prompt import ChatFormat
async def chat(
self,
message: str,
history: ChatFormat,
context: ChatContext
) -> AsyncGenerator[ChatResponse, None]:
# Access existing state from context
current_state = context.state if context else {}
# Update state with new information
new_state = {
**current_state,
"user_preference": "example_value",
"conversation_count": current_state.get("conversation_count", 0) + 1,
"last_topic": "extracted from current message"
}
# Store the updated state
yield self.create_state_update(new_state)
Security Considerations#
- State data is automatically signed with HMAC-SHA256 to prevent client-side tampering
- The secret key is obtained from the
RAGBITS_SECRET_KEYenvironment variable - If no environment variable is set, a random key is generated (with a warning)
- For production, always set
RAGBITS_SECRET_KEYto a strong, unique value - State signatures are verified on each request - tampering results in a 400 error
User Interface Configuration#
Configure User Forms#
Ragbits Chat supports two types of user forms: feedback forms and user settings forms.
Feedback Forms#
Configure feedback forms to collect user ratings and comments:
from typing import Literal
from pydantic import BaseModel, ConfigDict, Field
from ragbits.chat.interface.forms import FeedbackConfig
class LikeFormExample(BaseModel):
"""Form shown when user likes a response."""
model_config = ConfigDict(
title="Like Form",
json_schema_serialization_defaults_required=True,
)
like_reason: str = Field(
description="Why do you like this response?",
min_length=1,
)
class DislikeFormExample(BaseModel):
"""Form shown when user dislikes a response."""
model_config = ConfigDict(
title="Dislike Form",
json_schema_serialization_defaults_required=True
)
issue_type: Literal["Incorrect information", "Not helpful", "Unclear", "Other"] = Field(
description="What was the issue?"
)
feedback: str = Field(
description="Please provide more details",
min_length=1
)
class MyChat(ChatInterface):
feedback_config = FeedbackConfig(
like_enabled=True,
like_form=LikeFormExample,
dislike_enabled=True,
dislike_form=DislikeFormExample,
)
# ... rest of your implementation
User Settings Forms#
Configure user settings forms to collect user preferences:
from ragbits.chat.interface.forms import UserSettings
class UserSettingsFormExample(BaseModel):
"""Form for user preferences and settings."""
model_config = ConfigDict(
title="User Settings",
json_schema_serialization_defaults_required=True
)
language: Literal["English", "Spanish", "French", "German"] = Field(
description="Preferred language",
default="English"
)
response_style: Literal["Concise", "Detailed", "Technical"] = Field(
description="How would you like responses formatted?",
default="Detailed"
)
class MyChat(ChatInterface):
user_settings = UserSettings(form=UserSettingsFormExample)
# ... rest of your implementation
Customize UI Appearance#
Configure the chat interface appearance with custom icons, headers, and messages:
from ragbits.chat.interface.ui_customization import (
UICustomization,
HeaderCustomization,
PageMetaCustomization
)
class MyChat(ChatInterface):
ui_customization = UICustomization(
# Header customization
header=HeaderCustomization(
title="My AI Assistant",
subtitle="Powered by Ragbits",
logo="🤖"
),
# Welcome message shown when chat starts
welcome_message=(
"Hello! I'm your AI assistant.\n\n"
"How can I help you today? You can ask me **anything**! "
"I can provide information, answer questions, and assist with various tasks."
),
# Page metadata
meta=PageMetaCustomization(
favicon="🤖",
page_title="My AI Assistant"
)
)
# ... rest of your implementation
Enable Conversation History#
To enable conversation history persistence across sessions, set the conversation_history attribute:
class MyChat(ChatInterface):
conversation_history = True # Enable conversation history
# ... rest of your implementation
When enabled, the chat interface will automatically maintain conversation history and show it in side panel.
API Endpoints#
The API server exposes the following endpoints:
GET /: Serves the web UIGET /api/config: Returns UI configuration including feedback formsPOST /api/chat: Accepts chat messages and returns streaming responsesPOST /api/feedback: Accepts feedback from the user
Server Configuration#
Launch the API Server#
You can start the API server using the Ragbits CLI:
Note:
path.to.your.moduleshould be the dotted Python module path without the.pyextension.
Custom UI#
To use a custom UI build, use the --ui-build-dir option:
CORS Configuration#
To allow cross-origin requests, use the --cors-origin option (can be specified multiple times):
ragbits api run path.to.your.module:MyChat --cors-origin http://localhost:3000 --cors-origin https://your-domain.com
Custom Host and Port#
To run on a different host or port:
Enable Debug Mode#
To enable debug mode for detailed logging and error information, use the --debug flag. It will enable button in UI to toggle debug side panel which will show you all the internal state of the chat:
Complete Example#
Here's a comprehensive example demonstrating all features of a Ragbits Chat implementation:
import asyncio
import uuid
from collections.abc import AsyncGenerator
from typing import Literal
from pydantic import BaseModel, ConfigDict, Field
from ragbits.chat.interface import ChatInterface
from ragbits.chat.interface.forms import FeedbackConfig, UserSettings
from ragbits.chat.interface.types import ChatContext, ChatResponse, LiveUpdateType
from ragbits.core.prompt import ChatFormat
from ragbits.chat.interface.ui_customization import HeaderCustomization, PageMetaCustomization, UICustomization
from ragbits.core.llms import LiteLLM
class LikeFormExample(BaseModel):
"""Form shown when user likes a response."""
model_config = ConfigDict(
title="Like Form",
json_schema_serialization_defaults_required=True,
)
like_reason: str = Field(
description="Why do you like this response?",
min_length=1,
)
class DislikeFormExample(BaseModel):
"""Form shown when user dislikes a response."""
model_config = ConfigDict(
title="Dislike Form",
json_schema_serialization_defaults_required=True
)
issue_type: Literal["Incorrect information", "Not helpful", "Unclear", "Other"] = Field(
description="What was the issue?"
)
feedback: str = Field(
description="Please provide more details",
min_length=1
)
class UserSettingsFormExample(BaseModel):
"""Form for user preferences and settings."""
model_config = ConfigDict(
title="User Settings",
json_schema_serialization_defaults_required=True
)
language: Literal["English", "Spanish", "French"] = Field(
description="Preferred language",
default="English"
)
response_style: Literal["Concise", "Detailed", "Technical"] = Field(
description="Response style preference",
default="Detailed"
)
class MyChat(ChatInterface):
"""A comprehensive example implementation of the ChatInterface with all features."""
# Configure feedback forms
feedback_config = FeedbackConfig(
like_enabled=True,
like_form=LikeFormExample,
dislike_enabled=True,
dislike_form=DislikeFormExample,
)
# Configure user settings
user_settings = UserSettings(form=UserSettingsFormExample)
# Customize UI appearance
ui_customization = UICustomization(
header=HeaderCustomization(
title="My AI Assistant",
subtitle="Powered by Ragbits",
logo="🤖"
),
welcome_message=(
"Hello! I'm your AI assistant.\n\n"
"How can I help you today? You can ask me **anything**! "
"I can provide information, answer questions, and assist with various tasks."
),
meta=PageMetaCustomization(
favicon="🤖",
page_title="My AI Assistant"
),
)
# Enable conversation history
conversation_history = True
def __init__(self) -> None:
self.llm = LiteLLM(model_name="gpt-4o-mini")
async def chat(
self,
message: str,
history: ChatFormat,
context: ChatContext,
) -> AsyncGenerator[ChatResponse, None]:
"""
Comprehensive chat implementation demonstrating all response types.
Args:
message: The current user message
history: Optional list of previous messages in the conversation
context: Optional context including state and user settings
Yields:
ChatResponse objects with various content types
"""
# Access and update state
current_state = context.state if context else {}
conversation_count = current_state.get("conversation_count", 0) + 1
updated_state = {
**current_state,
"conversation_count": conversation_count,
"last_message_length": len(message),
"user_preference": "example_value"
}
yield self.create_state_update(updated_state)
# Add reference documents
yield self.create_reference(
title="Example Reference Document",
content="This is an example reference that might be relevant to your query.",
url="https://example.com/reference1",
)
# Add an example image
yield self.create_image_response(
str(uuid.uuid4()),
"https://picsum.photos/400/300"
)
# Demonstrate live updates for long-running operations
example_live_updates = [
self.create_live_update(
"search_task",
LiveUpdateType.START,
"Searching for information..."
),
self.create_live_update(
"search_task",
LiveUpdateType.FINISH,
"Search completed",
f"Found {conversation_count * 3} relevant results."
),
self.create_live_update(
"analysis_task",
LiveUpdateType.FINISH,
"Analysis completed",
"Processed and analyzed the search results."
),
]
for live_update in example_live_updates:
yield live_update
await asyncio.sleep(1) # Simulate processing time
# Personalize response based on conversation count
if conversation_count == 1:
intro_text = "Welcome! This is our first conversation. "
elif conversation_count > 5:
intro_text = f"We've been chatting quite a bit ({conversation_count} messages)! "
else:
intro_text = ""
# Stream the main response from the LLM
full_prompt = [
*history,
{"role": "user", "content": f"{intro_text}{message}"}
]
async for chunk in self.llm.generate_streaming(full_prompt):
yield self.create_text_response(chunk)
# Add follow-up suggestions
yield self.create_followup_messages([
"Tell me more about this topic",
"Can you provide another example?",
"How does this relate to other concepts?"
])
Save this as my_chat.py and run it with:
Then open http://127.0.0.1:8000 in your browser to see your fully-featured chat interface!