INTRODUCTION
Building a small yet powerful Large Language Model chatbot requires careful consideration of multiple architectural components. The term "smallest" refers to minimizing dependencies, memory footprint, and code complexity while "powerful" means supporting multiple hardware backends, both local and remote model inference, and production-grade reliability. This article explores every constituent part of such a system, from hardware abstraction to conversation management.
The fundamental challenge lies in creating an abstraction layer that works seamlessly across different GPU architectures including Nvidia CUDA, AMD ROCm, Apple Metal Performance Shaders, and Intel architectures, while also supporting remote API-based models. The system must be flexible enough to handle various use cases without becoming bloated with unnecessary features.
CORE ARCHITECTURAL COMPONENTS
A minimal yet powerful LLM chatbot consists of several key components that work together. The GPU acceleration layer provides hardware abstraction. The model loading system handles both local model files and remote API endpoints. The inference engine manages token generation and sampling. The conversation manager maintains context and history. The configuration system provides flexible setup options. Finally, the API interface exposes functionality to end users.
Each component must be designed with clean architecture principles in mind. Dependencies should flow inward, with core business logic independent of external frameworks. The system should be testable, maintainable, and extensible without requiring major refactoring.
GPU ACCELERATION LAYER
The GPU acceleration layer is the foundation that enables efficient inference across different hardware platforms. Each GPU vendor provides different libraries and APIs. Nvidia uses CUDA, AMD uses ROCm, Apple uses Metal Performance Shaders, and Intel uses oneAPI. The abstraction layer must detect available hardware and configure the appropriate backend.
Here is how we detect and configure the GPU backend:
import torch
import platform
import subprocess
class GPUBackend:
def __init__(self):
self.device = None
self.device_name = None
self.backend_type = None
self._detect_backend()
def _detect_backend(self):
# Check for CUDA (Nvidia)
if torch.cuda.is_available():
self.device = torch.device("cuda")
self.device_name = torch.cuda.get_device_name(0)
self.backend_type = "cuda"
torch.backends.cudnn.benchmark = True
return
# Check for MPS (Apple Silicon)
if hasattr(torch.backends, 'mps') and torch.backends.mps.is_available():
self.device = torch.device("mps")
self.device_name = "Apple Silicon GPU"
self.backend_type = "mps"
return
# Check for ROCm (AMD)
if torch.version.hip is not None:
self.device = torch.device("cuda") # ROCm uses cuda device string
self.device_name = "AMD ROCm GPU"
self.backend_type = "rocm"
return
# Fallback to CPU
self.device = torch.device("cpu")
self.device_name = "CPU"
self.backend_type = "cpu"
The detection logic first checks for CUDA availability since it is the most common GPU backend. The PyTorch library provides a simple boolean check through torch.cuda.is_available(). If CUDA is present, we enable cuDNN benchmarking for optimized convolution algorithms.
For Apple Silicon, we check if the MPS backend exists in the PyTorch installation and whether it is available on the current system. MPS provides significant acceleration on M1, M2, and M3 chips compared to CPU inference.
AMD ROCm detection is more subtle because ROCm-enabled PyTorch uses the same "cuda" device string but exposes a different version string through torch.version.hip. When this attribute is not None, we know ROCm is being used.
The CPU fallback ensures the system always has a functional backend even when no GPU is available. This is critical for development, testing, and deployment on systems without dedicated graphics hardware.
MODEL LOADING SYSTEM
The model loading system must handle two fundamentally different scenarios. Local models are loaded from disk and run on the available hardware. Remote models are accessed through API endpoints and run on external infrastructure. The abstraction must make both scenarios look identical to higher-level code.
For local models, we need to handle model weights, tokenizers, and configuration files. Modern LLMs use the Hugging Face transformers library format, which provides a standardized structure. Here is the local model loader:
from transformers import AutoModelForCausalLM, AutoTokenizer
import os
class LocalModelLoader:
def __init__(self, gpu_backend):
self.gpu_backend = gpu_backend
self.model = None
self.tokenizer = None
self.model_path = None
def load_model(self, model_path, precision="float16"):
self.model_path = model_path
# Determine dtype based on backend and precision
dtype = self._get_dtype(precision)
# Load tokenizer
self.tokenizer = AutoTokenizer.from_pretrained(
model_path,
trust_remote_code=True,
use_fast=True
)
# Load model with appropriate settings
load_kwargs = {
"pretrained_model_name_or_path": model_path,
"torch_dtype": dtype,
"trust_remote_code": True,
"low_cpu_mem_usage": True
}
# Add device map for multi-GPU or specific placement
if self.gpu_backend.backend_type in ["cuda", "rocm"]:
load_kwargs["device_map"] = "auto"
self.model = AutoModelForCausalLM.from_pretrained(**load_kwargs)
# Move to device if not using device_map
if self.gpu_backend.backend_type == "mps":
self.model = self.model.to(self.gpu_backend.device)
# Set to evaluation mode
self.model.eval()
return self.model, self.tokenizer
def _get_dtype(self, precision):
if precision == "float16":
if self.gpu_backend.backend_type == "mps":
return torch.float32 # MPS has limited float16 support
return torch.float16
elif precision == "bfloat16":
return torch.bfloat16
else:
return torch.float32
The local model loader handles several important considerations. First, it determines the appropriate data type based on the requested precision and hardware capabilities. Apple MPS has limited float16 support, so we fall back to float32 for compatibility. Nvidia and AMD GPUs generally support float16 well, which reduces memory usage by half compared to float32.
The device_map parameter enables automatic model sharding across multiple GPUs when available. This is particularly useful for large models that do not fit in a single GPU's memory. The transformers library handles the complexity of splitting layers across devices.
For remote models, we create a different loader that communicates with API endpoints:
import requests
import json
class RemoteModelLoader:
def __init__(self, api_endpoint, api_key=None):
self.api_endpoint = api_endpoint
self.api_key = api_key
self.headers = {}
if api_key:
self.headers["Authorization"] = f"Bearer {api_key}"
self.headers["Content-Type"] = "application/json"
def generate(self, prompt, max_tokens=512, temperature=0.7, top_p=0.9):
payload = {
"prompt": prompt,
"max_tokens": max_tokens,
"temperature": temperature,
"top_p": top_p,
"stream": False
}
response = requests.post(
self.api_endpoint,
headers=self.headers,
json=payload,
timeout=60
)
response.raise_for_status()
result = response.json()
return result.get("text", result.get("choices", [{}])[0].get("text", ""))
The remote model loader abstracts away the HTTP communication details. It constructs appropriate request payloads, handles authentication through API keys, and parses responses. Different API providers use slightly different response formats, so the code checks multiple possible locations for the generated text.
INFERENCE ENGINE
The inference engine is responsible for generating tokens from the model. This involves encoding the input prompt, running the model forward pass, sampling from the output distribution, and decoding tokens back to text. Efficient inference requires careful attention to memory management and computational efficiency.
Here is the core inference engine for local models:
import torch
from typing import Iterator
class InferenceEngine:
def __init__(self, model, tokenizer, gpu_backend):
self.model = model
self.tokenizer = tokenizer
self.gpu_backend = gpu_backend
def generate(self, prompt, max_tokens=512, temperature=0.7,
top_p=0.9, top_k=50, stream=False):
# Encode the prompt
inputs = self.tokenizer(
prompt,
return_tensors="pt",
padding=True,
truncation=True,
max_length=2048
)
# Move inputs to device
input_ids = inputs["input_ids"].to(self.gpu_backend.device)
attention_mask = inputs["attention_mask"].to(self.gpu_backend.device)
if stream:
return self._generate_stream(
input_ids, attention_mask, max_tokens,
temperature, top_p, top_k
)
else:
return self._generate_complete(
input_ids, attention_mask, max_tokens,
temperature, top_p, top_k
)
def _generate_complete(self, input_ids, attention_mask, max_tokens,
temperature, top_p, top_k):
with torch.no_grad():
outputs = self.model.generate(
input_ids=input_ids,
attention_mask=attention_mask,
max_new_tokens=max_tokens,
temperature=temperature,
top_p=top_p,
top_k=top_k,
do_sample=True,
pad_token_id=self.tokenizer.pad_token_id,
eos_token_id=self.tokenizer.eos_token_id
)
# Decode only the new tokens
generated_tokens = outputs[0][input_ids.shape[1]:]
response = self.tokenizer.decode(
generated_tokens,
skip_special_tokens=True
)
return response
def _generate_stream(self, input_ids, attention_mask, max_tokens,
temperature, top_p, top_k) -> Iterator[str]:
past_key_values = None
current_input_ids = input_ids
current_attention_mask = attention_mask
for _ in range(max_tokens):
with torch.no_grad():
outputs = self.model(
input_ids=current_input_ids,
attention_mask=current_attention_mask,
past_key_values=past_key_values,
use_cache=True
)
past_key_values = outputs.past_key_values
logits = outputs.logits[:, -1, :]
# Apply temperature
logits = logits / temperature
# Apply top-k filtering
if top_k > 0:
indices_to_remove = logits < torch.topk(logits, top_k)[0][..., -1, None]
logits[indices_to_remove] = float('-inf')
# Apply top-p (nucleus) filtering
if top_p < 1.0:
sorted_logits, sorted_indices = torch.sort(logits, descending=True)
cumulative_probs = torch.cumsum(
torch.softmax(sorted_logits, dim=-1), dim=-1
)
sorted_indices_to_remove = cumulative_probs > top_p
sorted_indices_to_remove[..., 1:] = sorted_indices_to_remove[..., :-1].clone()
sorted_indices_to_remove[..., 0] = 0
indices_to_remove = sorted_indices_to_remove.scatter(
1, sorted_indices, sorted_indices_to_remove
)
logits[indices_to_remove] = float('-inf')
# Sample next token
probs = torch.softmax(logits, dim=-1)
next_token = torch.multinomial(probs, num_samples=1)
# Check for end of sequence
if next_token.item() == self.tokenizer.eos_token_id:
break
# Decode and yield the token
token_text = self.tokenizer.decode(next_token[0], skip_special_tokens=True)
yield token_text
# Prepare for next iteration
current_input_ids = next_token
current_attention_mask = torch.cat([
current_attention_mask,
torch.ones((1, 1), device=self.gpu_backend.device)
], dim=1)
The inference engine provides both complete generation and streaming generation. Complete generation uses the model's built-in generate method, which is highly optimized. Streaming generation manually implements the generation loop to yield tokens as they are produced.
The streaming implementation uses key-value caching through the past_key_values parameter. This optimization avoids recomputing attention for previously generated tokens, significantly improving performance. Each iteration only processes the most recent token while reusing cached computations from earlier tokens.
Temperature scaling controls the randomness of the output. Lower temperatures make the model more deterministic, while higher temperatures increase diversity. Top-k filtering limits sampling to the k most likely tokens. Top-p (nucleus) filtering dynamically adjusts the sampling pool based on cumulative probability, providing better quality than fixed top-k in many cases.
CONVERSATION MANAGEMENT
Conversation management maintains the context and history of interactions. A chatbot needs to remember previous messages to provide coherent responses. However, LLMs have finite context windows, so we must carefully manage what information to retain.
Here is the conversation manager implementation:
from collections import deque
from typing import List, Dict
import json
class ConversationManager:
def __init__(self, max_history=10, max_context_tokens=2048):
self.max_history = max_history
self.max_context_tokens = max_context_tokens
self.messages = deque(maxlen=max_history)
self.system_prompt = None
def set_system_prompt(self, prompt):
self.system_prompt = prompt
def add_message(self, role, content):
message = {"role": role, "content": content}
self.messages.append(message)
def get_formatted_prompt(self, tokenizer):
# Build the conversation history
conversation = []
if self.system_prompt:
conversation.append({"role": "system", "content": self.system_prompt})
conversation.extend(self.messages)
# Format using the tokenizer's chat template if available
if hasattr(tokenizer, 'apply_chat_template'):
prompt = tokenizer.apply_chat_template(
conversation,
tokenize=False,
add_generation_prompt=True
)
else:
# Fallback to simple formatting
prompt = self._format_simple(conversation)
# Truncate if necessary
tokens = tokenizer.encode(prompt)
if len(tokens) > self.max_context_tokens:
# Remove oldest messages until we fit
while len(tokens) > self.max_context_tokens and len(conversation) > 1:
if conversation[0]["role"] == "system":
conversation.pop(1) # Keep system prompt
else:
conversation.pop(0)
if hasattr(tokenizer, 'apply_chat_template'):
prompt = tokenizer.apply_chat_template(
conversation,
tokenize=False,
add_generation_prompt=True
)
else:
prompt = self._format_simple(conversation)
tokens = tokenizer.encode(prompt)
return prompt
def _format_simple(self, conversation):
formatted = ""
for msg in conversation:
role = msg["role"].capitalize()
content = msg["content"]
formatted += f"{role}: {content}\n"
formatted += "Assistant: "
return formatted
def clear_history(self):
self.messages.clear()
def save_to_file(self, filepath):
data = {
"system_prompt": self.system_prompt,
"messages": list(self.messages)
}
with open(filepath, 'w', encoding='utf-8') as f:
json.dump(data, f, indent=2, ensure_ascii=False)
def load_from_file(self, filepath):
with open(filepath, 'r', encoding='utf-8') as f:
data = json.load(f)
self.system_prompt = data.get("system_prompt")
self.messages.clear()
for msg in data.get("messages", []):
self.messages.append(msg)
The conversation manager uses a deque with a maximum length to automatically limit history size. This prevents unbounded memory growth in long conversations. The max_history parameter controls how many message pairs to retain.
The get_formatted_prompt method constructs the full prompt from the conversation history. Modern models often have specific chat templates that format messages in a particular way. The apply_chat_template method handles this automatically when available. For models without a chat template, we fall back to a simple format with role labels.
Token-based truncation ensures the prompt fits within the model's context window. When the conversation exceeds the maximum token count, we remove the oldest messages while preserving the system prompt. This maintains the model's instructions while making room for recent context.
Persistence methods allow saving and loading conversations to disk. This enables resuming conversations across application restarts or sharing conversation histories between different components.
CONFIGURATION SYSTEM
A flexible configuration system allows users to customize the chatbot's behavior without modifying code. Configuration should support multiple sources including files, environment variables, and programmatic settings.
Here is the configuration manager:
import os
import yaml
from typing import Any, Dict
class ConfigurationManager:
def __init__(self, config_path=None):
self.config = self._load_defaults()
if config_path and os.path.exists(config_path):
self._load_from_file(config_path)
self._load_from_environment()
def _load_defaults(self) -> Dict[str, Any]:
return {
"model": {
"type": "local", # or "remote"
"path": None,
"api_endpoint": None,
"api_key": None,
"precision": "float16"
},
"generation": {
"max_tokens": 512,
"temperature": 0.7,
"top_p": 0.9,
"top_k": 50,
"stream": False
},
"conversation": {
"max_history": 10,
"max_context_tokens": 2048,
"system_prompt": "You are a helpful AI assistant."
},
"server": {
"host": "0.0.0.0",
"port": 8000,
"workers": 1
}
}
def _load_from_file(self, config_path):
with open(config_path, 'r', encoding='utf-8') as f:
file_config = yaml.safe_load(f)
self._deep_update(self.config, file_config)
def _load_from_environment(self):
# Model configuration
if os.getenv("LLM_MODEL_TYPE"):
self.config["model"]["type"] = os.getenv("LLM_MODEL_TYPE")
if os.getenv("LLM_MODEL_PATH"):
self.config["model"]["path"] = os.getenv("LLM_MODEL_PATH")
if os.getenv("LLM_API_ENDPOINT"):
self.config["model"]["api_endpoint"] = os.getenv("LLM_API_ENDPOINT")
if os.getenv("LLM_API_KEY"):
self.config["model"]["api_key"] = os.getenv("LLM_API_KEY")
# Generation configuration
if os.getenv("LLM_MAX_TOKENS"):
self.config["generation"]["max_tokens"] = int(os.getenv("LLM_MAX_TOKENS"))
if os.getenv("LLM_TEMPERATURE"):
self.config["generation"]["temperature"] = float(os.getenv("LLM_TEMPERATURE"))
# Server configuration
if os.getenv("LLM_SERVER_PORT"):
self.config["server"]["port"] = int(os.getenv("LLM_SERVER_PORT"))
def _deep_update(self, base_dict, update_dict):
for key, value in update_dict.items():
if key in base_dict and isinstance(base_dict[key], dict) and isinstance(value, dict):
self._deep_update(base_dict[key], value)
else:
base_dict[key] = value
def get(self, key_path, default=None):
keys = key_path.split('.')
value = self.config
for key in keys:
if isinstance(value, dict) and key in value:
value = value[key]
else:
return default
return value
def set(self, key_path, value):
keys = key_path.split('.')
config = self.config
for key in keys[:-1]:
if key not in config:
config[key] = {}
config = config[key]
config[keys[-1]] = value
The configuration manager loads settings from multiple sources with a clear precedence order. Default values are defined first. File-based configuration overrides defaults. Environment variables override file configuration. This allows flexible deployment scenarios where sensitive values like API keys come from environment variables while general settings come from files.
The deep update method recursively merges nested dictionaries, preserving values that are not explicitly overridden. This allows partial configuration files that only specify changed values.
The dot-notation access pattern through the get and set methods provides a clean interface for accessing nested configuration values. For example, config.get("model.path") retrieves the model path without requiring multiple dictionary accesses.
API INTERFACE
The API interface exposes the chatbot functionality through a REST API. This allows integration with web applications, mobile apps, and other services. We use FastAPI for its performance, automatic documentation, and type safety.
Here is the API implementation:
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field
from typing import Optional, List
import asyncio
import uvicorn
class ChatMessage(BaseModel):
role: str = Field(..., description="Role of the message sender")
content: str = Field(..., description="Content of the message")
class ChatRequest(BaseModel):
messages: List[ChatMessage] = Field(..., description="Conversation history")
max_tokens: Optional[int] = Field(None, description="Maximum tokens to generate")
temperature: Optional[float] = Field(None, description="Sampling temperature")
top_p: Optional[float] = Field(None, description="Nucleus sampling parameter")
stream: Optional[bool] = Field(False, description="Enable streaming response")
class ChatResponse(BaseModel):
message: ChatMessage
model: str
usage: dict
class ChatbotAPI:
def __init__(self, chatbot_instance, config):
self.chatbot = chatbot_instance
self.config = config
self.app = FastAPI(
title="LLM Chatbot API",
description="Minimal yet powerful LLM chatbot API",
version="1.0.0"
)
self._setup_routes()
def _setup_routes(self):
@self.app.post("/v1/chat/completions", response_model=ChatResponse)
async def chat_completion(request: ChatRequest):
try:
# Clear and rebuild conversation from request
self.chatbot.conversation.clear_history()
for msg in request.messages:
self.chatbot.conversation.add_message(msg.role, msg.content)
# Get generation parameters
max_tokens = request.max_tokens or self.config.get("generation.max_tokens")
temperature = request.temperature or self.config.get("generation.temperature")
top_p = request.top_p or self.config.get("generation.top_p")
if request.stream:
return StreamingResponse(
self._generate_stream(max_tokens, temperature, top_p),
media_type="text/event-stream"
)
else:
response = self.chatbot.generate(
max_tokens=max_tokens,
temperature=temperature,
top_p=top_p,
stream=False
)
return ChatResponse(
message=ChatMessage(role="assistant", content=response),
model=self.chatbot.model_name,
usage={
"prompt_tokens": 0, # Would need tokenizer to calculate
"completion_tokens": 0,
"total_tokens": 0
}
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@self.app.get("/health")
async def health_check():
return {"status": "healthy", "model_loaded": self.chatbot.is_loaded()}
@self.app.get("/models")
async def list_models():
return {
"models": [
{
"id": self.chatbot.model_name,
"type": self.config.get("model.type"),
"backend": self.chatbot.gpu_backend.backend_type
}
]
}
async def _generate_stream(self, max_tokens, temperature, top_p):
for token in self.chatbot.generate(
max_tokens=max_tokens,
temperature=temperature,
top_p=top_p,
stream=True
):
yield f"data: {token}\n\n"
await asyncio.sleep(0) # Allow other tasks to run
yield "data: [DONE]\n\n"
def run(self, host=None, port=None, workers=None):
host = host or self.config.get("server.host")
port = port or self.config.get("server.port")
workers = workers or self.config.get("server.workers")
uvicorn.run(
self.app,
host=host,
port=port,
workers=workers
)
The API interface uses Pydantic models for request and response validation. This provides automatic type checking and generates OpenAPI documentation. The ChatRequest model accepts a list of messages along with optional generation parameters.
The streaming endpoint returns a StreamingResponse with server-sent events. Each token is sent as a separate event, allowing clients to display responses progressively. The asyncio.sleep(0) call yields control to the event loop, preventing blocking.
Health check and model listing endpoints provide operational visibility. The health endpoint allows load balancers to verify the service is running. The models endpoint returns information about the loaded model and hardware backend.
UNIFIED CHATBOT CLASS
The unified chatbot class brings all components together into a cohesive interface. It handles initialization, model loading, and generation while abstracting away the complexity of different backends.
Here is the main chatbot class:
class LLMChatbot:
def __init__(self, config_manager):
self.config = config_manager
self.gpu_backend = GPUBackend()
self.conversation = ConversationManager(
max_history=self.config.get("conversation.max_history"),
max_context_tokens=self.config.get("conversation.max_context_tokens")
)
system_prompt = self.config.get("conversation.system_prompt")
if system_prompt:
self.conversation.set_system_prompt(system_prompt)
self.model_type = self.config.get("model.type")
self.model_name = None
self.model_loader = None
self.inference_engine = None
self._initialize_model()
def _initialize_model(self):
if self.model_type == "local":
model_path = self.config.get("model.path")
if not model_path:
raise ValueError("Local model path not specified in configuration")
self.model_loader = LocalModelLoader(self.gpu_backend)
model, tokenizer = self.model_loader.load_model(
model_path,
precision=self.config.get("model.precision")
)
self.inference_engine = InferenceEngine(
model, tokenizer, self.gpu_backend
)
self.model_name = os.path.basename(model_path)
elif self.model_type == "remote":
api_endpoint = self.config.get("model.api_endpoint")
api_key = self.config.get("model.api_key")
if not api_endpoint:
raise ValueError("Remote API endpoint not specified in configuration")
self.model_loader = RemoteModelLoader(api_endpoint, api_key)
self.model_name = "remote-model"
else:
raise ValueError(f"Unknown model type: {self.model_type}")
def generate(self, max_tokens=None, temperature=None, top_p=None, stream=False):
max_tokens = max_tokens or self.config.get("generation.max_tokens")
temperature = temperature or self.config.get("generation.temperature")
top_p = top_p or self.config.get("generation.top_p")
if self.model_type == "local":
prompt = self.conversation.get_formatted_prompt(
self.model_loader.tokenizer
)
response = self.inference_engine.generate(
prompt=prompt,
max_tokens=max_tokens,
temperature=temperature,
top_p=top_p,
top_k=self.config.get("generation.top_k"),
stream=stream
)
if not stream:
self.conversation.add_message("assistant", response)
return response
elif self.model_type == "remote":
# For remote, we need to format the conversation ourselves
messages = []
if self.conversation.system_prompt:
messages.append({
"role": "system",
"content": self.conversation.system_prompt
})
messages.extend(list(self.conversation.messages))
# Convert to simple prompt format
prompt = ""
for msg in messages:
prompt += f"{msg['role'].capitalize()}: {msg['content']}\n"
prompt += "Assistant: "
response = self.model_loader.generate(
prompt=prompt,
max_tokens=max_tokens,
temperature=temperature,
top_p=top_p
)
self.conversation.add_message("assistant", response)
return response
def chat(self, user_message):
self.conversation.add_message("user", user_message)
return self.generate()
def is_loaded(self):
return self.model_loader is not None
def get_info(self):
return {
"model_name": self.model_name,
"model_type": self.model_type,
"backend": self.gpu_backend.backend_type,
"device": str(self.gpu_backend.device),
"device_name": self.gpu_backend.device_name
}
The unified chatbot class provides a simple interface for common operations. The chat method accepts a user message, adds it to the conversation history, generates a response, and returns the result. This single method call handles all the complexity of prompt formatting, model inference, and history management.
The generate method provides lower-level access for custom use cases. It allows overriding generation parameters and supports streaming. The method automatically handles differences between local and remote models.
The get_info method returns diagnostic information about the loaded model and hardware backend. This is useful for debugging and monitoring.
COMMAND LINE INTERFACE
A command line interface provides an easy way to interact with the chatbot during development and testing. It demonstrates the core functionality in a simple interactive loop.
Here is the CLI implementation:
import sys
import argparse
class ChatbotCLI:
def __init__(self, chatbot):
self.chatbot = chatbot
self.running = False
def print_banner(self):
info = self.chatbot.get_info()
print("=" * 60)
print("LLM Chatbot - Interactive Mode")
print("=" * 60)
print(f"Model: {info['model_name']}")
print(f"Type: {info['model_type']}")
print(f"Backend: {info['backend']}")
print(f"Device: {info['device_name']}")
print("=" * 60)
print("Commands:")
print(" /clear - Clear conversation history")
print(" /save - Save conversation to file")
print(" /load - Load conversation from file")
print(" /info - Display model information")
print(" /quit - Exit the chatbot")
print("=" * 60)
print()
def run(self):
self.print_banner()
self.running = True
while self.running:
try:
user_input = input("You: ").strip()
if not user_input:
continue
if user_input.startswith('/'):
self._handle_command(user_input)
else:
response = self.chatbot.chat(user_input)
print(f"Assistant: {response}\n")
except KeyboardInterrupt:
print("\n\nExiting...")
self.running = False
except Exception as e:
print(f"Error: {e}\n")
def _handle_command(self, command):
cmd = command.lower().split()[0]
if cmd == '/quit' or cmd == '/exit':
self.running = False
print("Goodbye!")
elif cmd == '/clear':
self.chatbot.conversation.clear_history()
print("Conversation history cleared.\n")
elif cmd == '/save':
filename = input("Enter filename: ").strip()
if filename:
self.chatbot.conversation.save_to_file(filename)
print(f"Conversation saved to {filename}\n")
elif cmd == '/load':
filename = input("Enter filename: ").strip()
if filename:
self.chatbot.conversation.load_from_file(filename)
print(f"Conversation loaded from {filename}\n")
elif cmd == '/info':
info = self.chatbot.get_info()
print("\nModel Information:")
for key, value in info.items():
print(f" {key}: {value}")
print()
else:
print(f"Unknown command: {cmd}\n")
The CLI provides an interactive loop where users can type messages and receive responses. Special commands starting with a forward slash provide additional functionality like clearing history or saving conversations.
Error handling ensures the CLI remains responsive even when exceptions occur. Keyboard interrupts are caught gracefully to allow clean exits.
PRODUCTION READY RUNNING EXAMPLE
The following is a complete, production-ready implementation that integrates all the components discussed above. This code can be deployed directly and supports all the features described in the article.
#!/usr/bin/env python3
"""
Minimal Yet Powerful LLM Chatbot
A production-ready chatbot supporting local and remote LLMs across multiple GPU architectures.
"""
import torch
import platform
import subprocess
import os
import sys
import json
import yaml
import argparse
import requests
from collections import deque
from typing import Iterator, List, Dict, Any, Optional
from transformers import AutoModelForCausalLM, AutoTokenizer
from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field
import asyncio
import uvicorn
# ============================================================================
# GPU BACKEND DETECTION AND CONFIGURATION
# ============================================================================
class GPUBackend:
"""
Detects and configures the appropriate GPU backend for the system.
Supports CUDA (Nvidia), ROCm (AMD), MPS (Apple Silicon), and CPU fallback.
"""
def __init__(self):
self.device = None
self.device_name = None
self.backend_type = None
self._detect_backend()
def _detect_backend(self):
"""Detect available GPU backend and configure accordingly."""
# Check for CUDA (Nvidia)
if torch.cuda.is_available():
self.device = torch.device("cuda")
self.device_name = torch.cuda.get_device_name(0)
self.backend_type = "cuda"
torch.backends.cudnn.benchmark = True
print(f"[GPU Backend] Using CUDA: {self.device_name}")
return
# Check for MPS (Apple Silicon)
if hasattr(torch.backends, 'mps') and torch.backends.mps.is_available():
self.device = torch.device("mps")
self.device_name = "Apple Silicon GPU"
self.backend_type = "mps"
print(f"[GPU Backend] Using MPS: {self.device_name}")
return
# Check for ROCm (AMD)
if torch.version.hip is not None:
self.device = torch.device("cuda") # ROCm uses cuda device string
self.device_name = "AMD ROCm GPU"
self.backend_type = "rocm"
print(f"[GPU Backend] Using ROCm: {self.device_name}")
return
# Fallback to CPU
self.device = torch.device("cpu")
self.device_name = "CPU"
self.backend_type = "cpu"
print(f"[GPU Backend] Using CPU (no GPU detected)")
# ============================================================================
# LOCAL MODEL LOADER
# ============================================================================
class LocalModelLoader:
"""
Loads and manages local LLM models from disk.
Handles model weights, tokenizers, and device placement.
"""
def __init__(self, gpu_backend):
self.gpu_backend = gpu_backend
self.model = None
self.tokenizer = None
self.model_path = None
def load_model(self, model_path, precision="float16"):
"""
Load a model from the specified path with the given precision.
Args:
model_path: Path to the model directory or Hugging Face model ID
precision: Data type precision (float16, bfloat16, float32)
Returns:
Tuple of (model, tokenizer)
"""
self.model_path = model_path
print(f"[Local Model] Loading model from {model_path}...")
# Determine dtype based on backend and precision
dtype = self._get_dtype(precision)
# Load tokenizer
print("[Local Model] Loading tokenizer...")
self.tokenizer = AutoTokenizer.from_pretrained(
model_path,
trust_remote_code=True,
use_fast=True
)
# Ensure pad token is set
if self.tokenizer.pad_token is None:
self.tokenizer.pad_token = self.tokenizer.eos_token
# Load model with appropriate settings
print(f"[Local Model] Loading model weights (dtype: {dtype})...")
load_kwargs = {
"pretrained_model_name_or_path": model_path,
"torch_dtype": dtype,
"trust_remote_code": True,
"low_cpu_mem_usage": True
}
# Add device map for multi-GPU or specific placement
if self.gpu_backend.backend_type in ["cuda", "rocm"]:
load_kwargs["device_map"] = "auto"
self.model = AutoModelForCausalLM.from_pretrained(**load_kwargs)
# Move to device if not using device_map
if self.gpu_backend.backend_type == "mps":
print("[Local Model] Moving model to MPS device...")
self.model = self.model.to(self.gpu_backend.device)
# Set to evaluation mode
self.model.eval()
print("[Local Model] Model loaded successfully!")
return self.model, self.tokenizer
def _get_dtype(self, precision):
"""Determine the appropriate PyTorch dtype based on precision and backend."""
if precision == "float16":
if self.gpu_backend.backend_type == "mps":
# MPS has limited float16 support, use float32
return torch.float32
return torch.float16
elif precision == "bfloat16":
return torch.bfloat16
else:
return torch.float32
# ============================================================================
# REMOTE MODEL LOADER
# ============================================================================
class RemoteModelLoader:
"""
Communicates with remote LLM APIs for inference.
Supports various API providers with authentication.
"""
def __init__(self, api_endpoint, api_key=None):
self.api_endpoint = api_endpoint
self.api_key = api_key
self.headers = {}
if api_key:
self.headers["Authorization"] = f"Bearer {api_key}"
self.headers["Content-Type"] = "application/json"
print(f"[Remote Model] Configured endpoint: {api_endpoint}")
def generate(self, prompt, max_tokens=512, temperature=0.7, top_p=0.9):
"""
Generate text using the remote API.
Args:
prompt: Input prompt text
max_tokens: Maximum tokens to generate
temperature: Sampling temperature
top_p: Nucleus sampling parameter
Returns:
Generated text string
"""
payload = {
"prompt": prompt,
"max_tokens": max_tokens,
"temperature": temperature,
"top_p": top_p,
"stream": False
}
try:
response = requests.post(
self.api_endpoint,
headers=self.headers,
json=payload,
timeout=60
)
response.raise_for_status()
result = response.json()
# Try different response formats
if "text" in result:
return result["text"]
elif "choices" in result and len(result["choices"]) > 0:
return result["choices"][0].get("text", "")
else:
return str(result)
except requests.exceptions.RequestException as e:
raise RuntimeError(f"Remote API error: {e}")
# ============================================================================
# INFERENCE ENGINE
# ============================================================================
class InferenceEngine:
"""
Handles token generation and sampling for local models.
Supports both complete and streaming generation.
"""
def __init__(self, model, tokenizer, gpu_backend):
self.model = model
self.tokenizer = tokenizer
self.gpu_backend = gpu_backend
def generate(self, prompt, max_tokens=512, temperature=0.7,
top_p=0.9, top_k=50, stream=False):
"""
Generate text from the given prompt.
Args:
prompt: Input prompt text
max_tokens: Maximum tokens to generate
temperature: Sampling temperature
top_p: Nucleus sampling parameter
top_k: Top-k sampling parameter
stream: Whether to stream tokens as they are generated
Returns:
Generated text string or iterator of token strings
"""
# Encode the prompt
inputs = self.tokenizer(
prompt,
return_tensors="pt",
padding=True,
truncation=True,
max_length=2048
)
# Move inputs to device
input_ids = inputs["input_ids"].to(self.gpu_backend.device)
attention_mask = inputs["attention_mask"].to(self.gpu_backend.device)
if stream:
return self._generate_stream(
input_ids, attention_mask, max_tokens,
temperature, top_p, top_k
)
else:
return self._generate_complete(
input_ids, attention_mask, max_tokens,
temperature, top_p, top_k
)
def _generate_complete(self, input_ids, attention_mask, max_tokens,
temperature, top_p, top_k):
"""Generate complete response using model's built-in generation."""
with torch.no_grad():
outputs = self.model.generate(
input_ids=input_ids,
attention_mask=attention_mask,
max_new_tokens=max_tokens,
temperature=temperature,
top_p=top_p,
top_k=top_k,
do_sample=True,
pad_token_id=self.tokenizer.pad_token_id,
eos_token_id=self.tokenizer.eos_token_id
)
# Decode only the new tokens
generated_tokens = outputs[0][input_ids.shape[1]:]
response = self.tokenizer.decode(
generated_tokens,
skip_special_tokens=True
)
return response
def _generate_stream(self, input_ids, attention_mask, max_tokens,
temperature, top_p, top_k) -> Iterator[str]:
"""Generate response token by token with streaming."""
past_key_values = None
current_input_ids = input_ids
current_attention_mask = attention_mask
for _ in range(max_tokens):
with torch.no_grad():
outputs = self.model(
input_ids=current_input_ids,
attention_mask=current_attention_mask,
past_key_values=past_key_values,
use_cache=True
)
past_key_values = outputs.past_key_values
logits = outputs.logits[:, -1, :]
# Apply temperature
logits = logits / temperature
# Apply top-k filtering
if top_k > 0:
indices_to_remove = logits < torch.topk(logits, top_k)[0][..., -1, None]
logits[indices_to_remove] = float('-inf')
# Apply top-p (nucleus) filtering
if top_p < 1.0:
sorted_logits, sorted_indices = torch.sort(logits, descending=True)
cumulative_probs = torch.cumsum(
torch.softmax(sorted_logits, dim=-1), dim=-1
)
sorted_indices_to_remove = cumulative_probs > top_p
sorted_indices_to_remove[..., 1:] = sorted_indices_to_remove[..., :-1].clone()
sorted_indices_to_remove[..., 0] = 0
indices_to_remove = sorted_indices_to_remove.scatter(
1, sorted_indices, sorted_indices_to_remove
)
logits[indices_to_remove] = float('-inf')
# Sample next token
probs = torch.softmax(logits, dim=-1)
next_token = torch.multinomial(probs, num_samples=1)
# Check for end of sequence
if next_token.item() == self.tokenizer.eos_token_id:
break
# Decode and yield the token
token_text = self.tokenizer.decode(next_token[0], skip_special_tokens=True)
yield token_text
# Prepare for next iteration
current_input_ids = next_token
current_attention_mask = torch.cat([
current_attention_mask,
torch.ones((1, 1), device=self.gpu_backend.device)
], dim=1)
# ============================================================================
# CONVERSATION MANAGER
# ============================================================================
class ConversationManager:
"""
Manages conversation history and context.
Handles message storage, formatting, and persistence.
"""
def __init__(self, max_history=10, max_context_tokens=2048):
self.max_history = max_history
self.max_context_tokens = max_context_tokens
self.messages = deque(maxlen=max_history)
self.system_prompt = None
def set_system_prompt(self, prompt):
"""Set the system prompt that defines the assistant's behavior."""
self.system_prompt = prompt
def add_message(self, role, content):
"""Add a message to the conversation history."""
message = {"role": role, "content": content}
self.messages.append(message)
def get_formatted_prompt(self, tokenizer):
"""
Format the conversation history into a prompt string.
Handles truncation if the conversation exceeds token limits.
"""
# Build the conversation history
conversation = []
if self.system_prompt:
conversation.append({"role": "system", "content": self.system_prompt})
conversation.extend(self.messages)
# Format using the tokenizer's chat template if available
if hasattr(tokenizer, 'apply_chat_template'):
prompt = tokenizer.apply_chat_template(
conversation,
tokenize=False,
add_generation_prompt=True
)
else:
# Fallback to simple formatting
prompt = self._format_simple(conversation)
# Truncate if necessary
tokens = tokenizer.encode(prompt)
if len(tokens) > self.max_context_tokens:
# Remove oldest messages until we fit
while len(tokens) > self.max_context_tokens and len(conversation) > 1:
if conversation[0]["role"] == "system":
conversation.pop(1) # Keep system prompt
else:
conversation.pop(0)
if hasattr(tokenizer, 'apply_chat_template'):
prompt = tokenizer.apply_chat_template(
conversation,
tokenize=False,
add_generation_prompt=True
)
else:
prompt = self._format_simple(conversation)
tokens = tokenizer.encode(prompt)
return prompt
def _format_simple(self, conversation):
"""Simple fallback formatting when chat template is not available."""
formatted = ""
for msg in conversation:
role = msg["role"].capitalize()
content = msg["content"]
formatted += f"{role}: {content}\n"
formatted += "Assistant: "
return formatted
def clear_history(self):
"""Clear all conversation history."""
self.messages.clear()
def save_to_file(self, filepath):
"""Save conversation to a JSON file."""
data = {
"system_prompt": self.system_prompt,
"messages": list(self.messages)
}
with open(filepath, 'w', encoding='utf-8') as f:
json.dump(data, f, indent=2, ensure_ascii=False)
def load_from_file(self, filepath):
"""Load conversation from a JSON file."""
with open(filepath, 'r', encoding='utf-8') as f:
data = json.load(f)
self.system_prompt = data.get("system_prompt")
self.messages.clear()
for msg in data.get("messages", []):
self.messages.append(msg)
# ============================================================================
# CONFIGURATION MANAGER
# ============================================================================
class ConfigurationManager:
"""
Manages application configuration from multiple sources.
Supports defaults, file-based config, and environment variables.
"""
def __init__(self, config_path=None):
self.config = self._load_defaults()
if config_path and os.path.exists(config_path):
self._load_from_file(config_path)
self._load_from_environment()
def _load_defaults(self) -> Dict[str, Any]:
"""Load default configuration values."""
return {
"model": {
"type": "local", # or "remote"
"path": None,
"api_endpoint": None,
"api_key": None,
"precision": "float16"
},
"generation": {
"max_tokens": 512,
"temperature": 0.7,
"top_p": 0.9,
"top_k": 50,
"stream": False
},
"conversation": {
"max_history": 10,
"max_context_tokens": 2048,
"system_prompt": "You are a helpful AI assistant."
},
"server": {
"host": "0.0.0.0",
"port": 8000,
"workers": 1
}
}
def _load_from_file(self, config_path):
"""Load configuration from a YAML file."""
print(f"[Config] Loading configuration from {config_path}")
with open(config_path, 'r', encoding='utf-8') as f:
file_config = yaml.safe_load(f)
self._deep_update(self.config, file_config)
def _load_from_environment(self):
"""Load configuration from environment variables."""
# Model configuration
if os.getenv("LLM_MODEL_TYPE"):
self.config["model"]["type"] = os.getenv("LLM_MODEL_TYPE")
if os.getenv("LLM_MODEL_PATH"):
self.config["model"]["path"] = os.getenv("LLM_MODEL_PATH")
if os.getenv("LLM_API_ENDPOINT"):
self.config["model"]["api_endpoint"] = os.getenv("LLM_API_ENDPOINT")
if os.getenv("LLM_API_KEY"):
self.config["model"]["api_key"] = os.getenv("LLM_API_KEY")
# Generation configuration
if os.getenv("LLM_MAX_TOKENS"):
self.config["generation"]["max_tokens"] = int(os.getenv("LLM_MAX_TOKENS"))
if os.getenv("LLM_TEMPERATURE"):
self.config["generation"]["temperature"] = float(os.getenv("LLM_TEMPERATURE"))
# Server configuration
if os.getenv("LLM_SERVER_PORT"):
self.config["server"]["port"] = int(os.getenv("LLM_SERVER_PORT"))
def _deep_update(self, base_dict, update_dict):
"""Recursively update nested dictionaries."""
for key, value in update_dict.items():
if key in base_dict and isinstance(base_dict[key], dict) and isinstance(value, dict):
self._deep_update(base_dict[key], value)
else:
base_dict[key] = value
def get(self, key_path, default=None):
"""Get a configuration value using dot notation."""
keys = key_path.split('.')
value = self.config
for key in keys:
if isinstance(value, dict) and key in value:
value = value[key]
else:
return default
return value
def set(self, key_path, value):
"""Set a configuration value using dot notation."""
keys = key_path.split('.')
config = self.config
for key in keys[:-1]:
if key not in config:
config[key] = {}
config = config[key]
config[keys[-1]] = value
# ============================================================================
# UNIFIED CHATBOT CLASS
# ============================================================================
class LLMChatbot:
"""
Main chatbot class that integrates all components.
Provides a unified interface for both local and remote models.
"""
def __init__(self, config_manager):
self.config = config_manager
self.gpu_backend = GPUBackend()
self.conversation = ConversationManager(
max_history=self.config.get("conversation.max_history"),
max_context_tokens=self.config.get("conversation.max_context_tokens")
)
system_prompt = self.config.get("conversation.system_prompt")
if system_prompt:
self.conversation.set_system_prompt(system_prompt)
self.model_type = self.config.get("model.type")
self.model_name = None
self.model_loader = None
self.inference_engine = None
self._initialize_model()
def _initialize_model(self):
"""Initialize the appropriate model loader based on configuration."""
if self.model_type == "local":
model_path = self.config.get("model.path")
if not model_path:
raise ValueError("Local model path not specified in configuration")
self.model_loader = LocalModelLoader(self.gpu_backend)
model, tokenizer = self.model_loader.load_model(
model_path,
precision=self.config.get("model.precision")
)
self.inference_engine = InferenceEngine(
model, tokenizer, self.gpu_backend
)
self.model_name = os.path.basename(model_path)
elif self.model_type == "remote":
api_endpoint = self.config.get("model.api_endpoint")
api_key = self.config.get("model.api_key")
if not api_endpoint:
raise ValueError("Remote API endpoint not specified in configuration")
self.model_loader = RemoteModelLoader(api_endpoint, api_key)
self.model_name = "remote-model"
else:
raise ValueError(f"Unknown model type: {self.model_type}")
def generate(self, max_tokens=None, temperature=None, top_p=None, stream=False):
"""
Generate a response based on the current conversation history.
Args:
max_tokens: Maximum tokens to generate (uses config default if None)
temperature: Sampling temperature (uses config default if None)
top_p: Nucleus sampling parameter (uses config default if None)
stream: Whether to stream the response
Returns:
Generated text string or iterator of token strings
"""
max_tokens = max_tokens or self.config.get("generation.max_tokens")
temperature = temperature or self.config.get("generation.temperature")
top_p = top_p or self.config.get("generation.top_p")
if self.model_type == "local":
prompt = self.conversation.get_formatted_prompt(
self.model_loader.tokenizer
)
response = self.inference_engine.generate(
prompt=prompt,
max_tokens=max_tokens,
temperature=temperature,
top_p=top_p,
top_k=self.config.get("generation.top_k"),
stream=stream
)
if not stream:
self.conversation.add_message("assistant", response)
return response
elif self.model_type == "remote":
# For remote, we need to format the conversation ourselves
messages = []
if self.conversation.system_prompt:
messages.append({
"role": "system",
"content": self.conversation.system_prompt
})
messages.extend(list(self.conversation.messages))
# Convert to simple prompt format
prompt = ""
for msg in messages:
prompt += f"{msg['role'].capitalize()}: {msg['content']}\n"
prompt += "Assistant: "
response = self.model_loader.generate(
prompt=prompt,
max_tokens=max_tokens,
temperature=temperature,
top_p=top_p
)
self.conversation.add_message("assistant", response)
return response
def chat(self, user_message):
"""
Simple chat interface that handles a single user message.
Args:
user_message: The user's input message
Returns:
The assistant's response
"""
self.conversation.add_message("user", user_message)
return self.generate()
def is_loaded(self):
"""Check if a model is loaded and ready."""
return self.model_loader is not None
def get_info(self):
"""Get information about the loaded model and system."""
return {
"model_name": self.model_name,
"model_type": self.model_type,
"backend": self.gpu_backend.backend_type,
"device": str(self.gpu_backend.device),
"device_name": self.gpu_backend.device_name
}
# ============================================================================
# REST API INTERFACE
# ============================================================================
class ChatMessage(BaseModel):
"""Pydantic model for chat messages."""
role: str = Field(..., description="Role of the message sender")
content: str = Field(..., description="Content of the message")
class ChatRequest(BaseModel):
"""Pydantic model for chat completion requests."""
messages: List[ChatMessage] = Field(..., description="Conversation history")
max_tokens: Optional[int] = Field(None, description="Maximum tokens to generate")
temperature: Optional[float] = Field(None, description="Sampling temperature")
top_p: Optional[float] = Field(None, description="Nucleus sampling parameter")
stream: Optional[bool] = Field(False, description="Enable streaming response")
class ChatResponse(BaseModel):
"""Pydantic model for chat completion responses."""
message: ChatMessage
model: str
usage: dict
class ChatbotAPI:
"""
FastAPI-based REST API for the chatbot.
Provides endpoints for chat completions, health checks, and model information.
"""
def __init__(self, chatbot_instance, config):
self.chatbot = chatbot_instance
self.config = config
self.app = FastAPI(
title="LLM Chatbot API",
description="Minimal yet powerful LLM chatbot API",
version="1.0.0"
)
self._setup_routes()
def _setup_routes(self):
"""Configure API routes."""
@self.app.post("/v1/chat/completions", response_model=ChatResponse)
async def chat_completion(request: ChatRequest):
"""
Generate a chat completion based on the provided messages.
Supports both streaming and non-streaming responses.
"""
try:
# Clear and rebuild conversation from request
self.chatbot.conversation.clear_history()
for msg in request.messages:
self.chatbot.conversation.add_message(msg.role, msg.content)
# Get generation parameters
max_tokens = request.max_tokens or self.config.get("generation.max_tokens")
temperature = request.temperature or self.config.get("generation.temperature")
top_p = request.top_p or self.config.get("generation.top_p")
if request.stream:
return StreamingResponse(
self._generate_stream(max_tokens, temperature, top_p),
media_type="text/event-stream"
)
else:
response = self.chatbot.generate(
max_tokens=max_tokens,
temperature=temperature,
top_p=top_p,
stream=False
)
return ChatResponse(
message=ChatMessage(role="assistant", content=response),
model=self.chatbot.model_name,
usage={
"prompt_tokens": 0,
"completion_tokens": 0,
"total_tokens": 0
}
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@self.app.get("/health")
async def health_check():
"""Health check endpoint for monitoring."""
return {"status": "healthy", "model_loaded": self.chatbot.is_loaded()}
@self.app.get("/models")
async def list_models():
"""List available models and system information."""
return {
"models": [
{
"id": self.chatbot.model_name,
"type": self.config.get("model.type"),
"backend": self.chatbot.gpu_backend.backend_type
}
]
}
async def _generate_stream(self, max_tokens, temperature, top_p):
"""Generate streaming response using server-sent events."""
for token in self.chatbot.generate(
max_tokens=max_tokens,
temperature=temperature,
top_p=top_p,
stream=True
):
yield f"data: {token}\n\n"
await asyncio.sleep(0) # Allow other tasks to run
yield "data: [DONE]\n\n"
def run(self, host=None, port=None, workers=None):
"""Start the API server."""
host = host or self.config.get("server.host")
port = port or self.config.get("server.port")
workers = workers or self.config.get("server.workers")
print(f"[API Server] Starting on {host}:{port}")
uvicorn.run(
self.app,
host=host,
port=port,
workers=workers
)
# ============================================================================
# COMMAND LINE INTERFACE
# ============================================================================
class ChatbotCLI:
"""
Interactive command-line interface for the chatbot.
Provides a simple way to chat and manage conversations.
"""
def __init__(self, chatbot):
self.chatbot = chatbot
self.running = False
def print_banner(self):
"""Display welcome banner with system information."""
info = self.chatbot.get_info()
print("=" * 60)
print("LLM Chatbot - Interactive Mode")
print("=" * 60)
print(f"Model: {info['model_name']}")
print(f"Type: {info['model_type']}")
print(f"Backend: {info['backend']}")
print(f"Device: {info['device_name']}")
print("=" * 60)
print("Commands:")
print(" /clear - Clear conversation history")
print(" /save - Save conversation to file")
print(" /load - Load conversation from file")
print(" /info - Display model information")
print(" /quit - Exit the chatbot")
print("=" * 60)
print()
def run(self):
"""Run the interactive chat loop."""
self.print_banner()
self.running = True
while self.running:
try:
user_input = input("You: ").strip()
if not user_input:
continue
if user_input.startswith('/'):
self._handle_command(user_input)
else:
response = self.chatbot.chat(user_input)
print(f"Assistant: {response}\n")
except KeyboardInterrupt:
print("\n\nExiting...")
self.running = False
except Exception as e:
print(f"Error: {e}\n")
def _handle_command(self, command):
"""Handle special commands starting with /."""
cmd = command.lower().split()[0]
if cmd == '/quit' or cmd == '/exit':
self.running = False
print("Goodbye!")
elif cmd == '/clear':
self.chatbot.conversation.clear_history()
print("Conversation history cleared.\n")
elif cmd == '/save':
filename = input("Enter filename: ").strip()
if filename:
self.chatbot.conversation.save_to_file(filename)
print(f"Conversation saved to {filename}\n")
elif cmd == '/load':
filename = input("Enter filename: ").strip()
if filename:
self.chatbot.conversation.load_from_file(filename)
print(f"Conversation loaded from {filename}\n")
elif cmd == '/info':
info = self.chatbot.get_info()
print("\nModel Information:")
for key, value in info.items():
print(f" {key}: {value}")
print()
else:
print(f"Unknown command: {cmd}\n")
# ============================================================================
# MAIN ENTRY POINT
# ============================================================================
def main():
"""Main entry point for the application."""
parser = argparse.ArgumentParser(
description="Minimal Yet Powerful LLM Chatbot"
)
parser.add_argument(
"--config",
type=str,
help="Path to configuration file (YAML)"
)
parser.add_argument(
"--mode",
type=str,
choices=["cli", "api"],
default="cli",
help="Run mode: cli for interactive chat, api for REST server"
)
parser.add_argument(
"--model-path",
type=str,
help="Path to local model (overrides config)"
)
parser.add_argument(
"--model-type",
type=str,
choices=["local", "remote"],
help="Model type (overrides config)"
)
parser.add_argument(
"--api-endpoint",
type=str,
help="Remote API endpoint (overrides config)"
)
parser.add_argument(
"--api-key",
type=str,
help="API key for remote endpoint (overrides config)"
)
args = parser.parse_args()
# Load configuration
config = ConfigurationManager(args.config)
# Apply command-line overrides
if args.model_path:
config.set("model.path", args.model_path)
if args.model_type:
config.set("model.type", args.model_type)
if args.api_endpoint:
config.set("model.api_endpoint", args.api_endpoint)
if args.api_key:
config.set("model.api_key", args.api_key)
# Initialize chatbot
try:
chatbot = LLMChatbot(config)
except Exception as e:
print(f"Error initializing chatbot: {e}")
sys.exit(1)
# Run in the specified mode
if args.mode == "cli":
cli = ChatbotCLI(chatbot)
cli.run()
elif args.mode == "api":
api = ChatbotAPI(chatbot, config)
api.run()
if __name__ == "__main__":
main()
This complete implementation provides a production-ready LLM chatbot system. The code supports both local and remote models, automatically detects and configures GPU backends across Nvidia CUDA, AMD ROCm, Apple MPS, and Intel architectures, manages conversation history with intelligent truncation, provides both command-line and REST API interfaces, and includes comprehensive configuration management.
To use this system with a local model, create a configuration file named config.yaml with the following content:
model: type: local path: /path/to/your/model precision: float16
generation: max_tokens: 512 temperature: 0.7 top_p: 0.9
conversation: max_history: 10 system_prompt: You are a helpful AI assistant.
Then run the chatbot in CLI mode with the command:
python chatbot.py --config config.yaml --mode cli
For remote API usage, configure the endpoint:
model: type: remote api_endpoint: https://api.example.com/v1/completions api_key: your-api-key-here
The system automatically detects available GPU hardware and configures the appropriate backend. On systems with Nvidia GPUs, it uses CUDA with cuDNN optimizations. On Apple Silicon Macs, it uses Metal Performance Shaders. On AMD systems with ROCm, it uses the ROCm backend. When no GPU is available, it falls back to CPU inference.
The inference engine implements both complete generation using the model's optimized generate method and streaming generation with manual token-by-token processing. Streaming uses key-value caching to avoid recomputing attention for previous tokens, significantly improving performance.
The conversation manager maintains context across multiple turns while respecting token limits. When conversations exceed the maximum context length, the system automatically removes the oldest messages while preserving the system prompt. This ensures the model always has the most recent and relevant context.
The REST API provides OpenAPI-compliant endpoints compatible with standard chat completion APIs. The streaming endpoint uses server-sent events to deliver tokens as they are generated, enabling real-time response display in client applications.
The configuration system supports multiple deployment scenarios through layered configuration sources. Default values ensure the system works out of the box. File-based configuration allows persistent settings. Environment variables enable secure handling of sensitive values like API keys in containerized deployments.
This architecture demonstrates how to build a minimal yet powerful LLM chatbot that works across different hardware platforms and deployment scenarios while maintaining clean code organization and production-grade reliability.
