BUILDING A LOCAL LLM SERVER WITH OPENAI-COMPATIBLE API INTERFACE

 

INTRODUCTION TO LOCAL LLM SERVING

The landscape of artificial intelligence has been transformed by large language models, but many organizations face challenges when using cloud-based services. Privacy concerns, data sovereignty requirements, cost considerations, and the need for offline operation drive the demand for locally hosted LLM solutions. However, most modern applications and development tools are built around the OpenAI API specification, which has become the industry standard interface for interacting with language models.

This article presents a comprehensive guide to building a server that runs local LLM models while providing an OpenAI-compatible API interface. We will explore two distinct approaches. The first approach leverages existing tools like Ollama to quickly deploy a functional server with minimal configuration. The second approach involves building a complete custom server from the ground up, incorporating user authentication, email verification, API key management, rate limiting, and persistent data storage. The custom solution provides enterprise-grade features including multi-user support, access control, and the ability to serve multiple models simultaneously through a thread-based architecture.

The motivation for building such a system stems from several real-world scenarios. Organizations handling sensitive data often cannot send information to external APIs due to compliance requirements such as GDPR, HIPAA, or industry-specific regulations. Research institutions may need to run experiments with consistent model behavior without worrying about API changes or rate limits imposed by cloud providers. Developers working in environments with limited or unreliable internet connectivity require local solutions that function offline. Cost-conscious teams processing large volumes of requests may find that the cumulative API costs exceed the investment in local hardware and infrastructure.

Beyond these practical considerations, there is significant educational value in understanding how LLM serving infrastructure works at a fundamental level. By building a complete server from scratch, developers gain insights into model loading, inference optimization, request queuing, authentication flows, and API design patterns. This knowledge transfers to other domains and enables more informed decisions when evaluating commercial solutions or designing custom AI infrastructure.

UNDERSTANDING THE OPENAI API INTERFACE

Before diving into implementation details, it is essential to understand what makes an API OpenAI-compatible and why this compatibility matters. The OpenAI API has become the de facto standard for interacting with language models, similar to how the S3 API became the standard for object storage. This standardization creates a powerful ecosystem effect where libraries, tools, and applications can work with any compatible service without modification.

The OpenAI API uses a RESTful architecture with JSON payloads for communication. REST, or Representational State Transfer, is an architectural style that uses standard HTTP methods and status codes to create predictable, stateless interactions. The primary endpoint for chat completions is typically accessed via POST requests to a path like /v1/chat/completions. The /v1 prefix indicates the API version, allowing for future evolution while maintaining backward compatibility with existing clients.

The request body, as shown in the following example, contains several key fields that control the model's behavior. The model field specifies which language model to use, allowing a single server to host multiple models and let clients choose based on their needs. The messages array represents the conversation history, with each message having a role (system, user, or assistant) and content. The system role sets the overall behavior and personality of the assistant, the user role represents input from the human, and the assistant role contains previous responses from the model. This structure enables multi-turn conversations where context is preserved across exchanges.

{

  "model": "gpt-3.5-turbo",

  "messages": [

    {"role": "system", "content": "You are a helpful assistant."},

    {"role": "user", "content": "What is the capital of France?"}

  ],

  "temperature": 0.7,

  "max_tokens": 150

}

Optional parameters provide fine-grained control over generation behavior. The temperature parameter, typically ranging from 0 to 2, controls randomness in the output. Lower values like 0.1 make the model more deterministic and focused, while higher values like 1.5 increase creativity and diversity at the cost of coherence. The max_tokens parameter limits the length of the generated response, which is crucial for controlling costs and preventing excessively long outputs. Other common parameters include top_p for nucleus sampling, frequency_penalty to discourage repetition, and presence_penalty to encourage topic diversity.

The response from the API includes the generated text within a structured format that provides metadata about the completion. This metadata includes a unique identifier for the request, a timestamp, the model used, and token usage statistics. Token usage information is particularly important for billing and monitoring purposes, as it breaks down the number of tokens in the prompt, the completion, and the total. Understanding token consumption helps optimize prompts and manage costs effectively.

By adhering to this format, any server we build can seamlessly integrate with existing applications, libraries, and tools that were designed for the OpenAI API. This compatibility is crucial because it allows developers to switch between cloud-based and local models without modifying their application code. Popular libraries like LangChain, LlamaIndex, and the official OpenAI Python client all work with any OpenAI-compatible endpoint simply by changing the base URL. This interoperability dramatically reduces the friction of adopting local LLM solutions.

APPROACH ONE: DEPLOYING WITH OLLAMA

Ollama represents the quickest path to running local LLMs with an OpenAI-compatible interface. Ollama is an open-source project that packages popular language models and provides a simple command-line interface for downloading, running, and serving these models. The tool handles all the complexity of model loading, memory management, and API serving, making it ideal for developers who need a working solution immediately without deep knowledge of model internals or server architecture.

The philosophy behind Ollama is similar to Docker for containers—it provides a standardized way to package and run models regardless of their underlying implementation. Just as Docker abstracts away the complexities of process isolation and dependency management, Ollama abstracts away the complexities of model quantization, memory allocation, and inference optimization. This abstraction makes it accessible to developers who want to use LLMs without becoming experts in machine learning infrastructure.

Installing Ollama on a Linux system is straightforward. You can download and install it using a single command that fetches the installation script and executes it. On macOS and Windows, Ollama provides native installers that integrate with the operating system. The installation process sets up Ollama as a system service that can start automatically on boot, ensuring that your models are always available. Once installed, Ollama runs as a background service that listens for API requests on localhost port 11434 by default. This port can be configured through environment variables if it conflicts with other services.

After installation, you can download a model using the Ollama command-line tool:

ollama pull llama2

 The pull command downloads the model files and stores them in Ollama's model directory, typically located in the user's home directory. The download may take some time depending on your internet connection speed, as these models can be several gigabytes in size. For example, Llama 2 7B in 4-bit quantization is approximately 3.8 GB, while the 13B variant is around 7.3 GB. Ollama uses a content-addressable storage system similar to Git, which means that common layers between models are shared, reducing total disk usage when you have multiple models installed.

Once downloaded, you can start serving the model by running the command shown:

ollama serve

The serve command starts the Ollama server in the foreground, displaying logs and status information. If you want it to run in the background, you can use system service managers like systemd on Linux or run it in a detached screen session. On macOS and Windows, the Ollama desktop application handles this automatically, running the server as a background process that starts with the operating system.

With the server running, you can now make API requests that are compatible with the OpenAI format. The following code presents how to interact with an Ollama server using the OpenAI Python client library. 

from openai import OpenAI

# Configure the client to point to your local Ollama server

client = OpenAI(

    base_url='http://localhost:11434/v1',

    api_key='ollama'

)

# Make a chat completion request

response = client.chat.completions.create(

    model='llama2',

    messages=[

        {'role': 'system', 'content': 'You are a helpful assistant.'},

        {'role': 'user', 'content': 'Explain quantum computing.'}

    ],

    temperature=0.7,

    max_tokens=200

)

# Extract and print the response

print(response.choices[0].message.content)

This code illustrates the key advantage of OpenAI compatibility—the same code that works with OpenAI's cloud service works with your local Ollama server by simply changing the base_url parameter. The api_key parameter is required by the OpenAI client library for authentication, but Ollama does not actually validate it in the default configuration, so any non-empty string will work.

The client configuration points to localhost:11434/v1, which is where Ollama exposes its OpenAI-compatible API. The /v1 path prefix matches OpenAI's versioning scheme, ensuring compatibility with clients that expect this structure. When you call client.chat.completions.create(), the library serializes your request into JSON, sends it via HTTP POST to the Ollama server, and deserializes the response into a Python object with convenient attribute access.

The messages parameter follows the OpenAI conversation format, with a system message setting the assistant's behavior and a user message containing the actual query. The system message is particularly important as it establishes the context and constraints for the model's responses. For example, you might instruct the model to "respond in a professional tone suitable for business communication" or "explain concepts as if teaching a beginner." These instructions significantly influence the style and content of the generated text.

Temperature and max_tokens provide additional control over the generation process. A temperature of 0.7 is a balanced middle ground that produces coherent yet somewhat varied responses. For tasks requiring factual accuracy and consistency, like code generation or data extraction, lower temperatures around 0.1-0.3 work better. For creative tasks like story writing or brainstorming, higher temperatures around 0.8-1.2 encourage more diverse outputs.

Ollama supports multiple models simultaneously, which is one of its most powerful features. You can download several models using the pull command and switch between them by changing the model parameter in your API requests. For instance, you might have Llama 2 7B for general tasks, CodeLlama for programming questions, and Mistral 7B for creative writing. The server intelligently manages memory, loading and unloading models as needed based on usage patterns. When you request a model that isn't currently loaded, Ollama loads it into memory, which may take a few seconds. If memory becomes constrained, Ollama unloads the least recently used model to make room.

This automatic management makes Ollama suitable for scenarios where you want to experiment with different models without manual intervention. However, it also means that the first request to a model after it's been unloaded will have higher latency due to the loading time. For production scenarios with predictable usage patterns, you might want to implement model warming strategies that keep frequently used models loaded.

However, Ollama's simplicity comes with limitations that become apparent in multi-user or production environments. The default installation does not include user authentication, API key validation, or multi-user support. Every request to the Ollama server is processed without access control, which is acceptable for personal use or trusted internal networks but unsuitable for production environments with multiple users or external access. There's no way to track which user made which request, no ability to set different rate limits for different users, and no mechanism to prevent abuse or manage resource allocation.

Additionally, Ollama doesn't provide built-in monitoring, logging, or analytics capabilities that are essential for production deployments. You can't easily answer questions like "How many requests did we process today?" or "Which model is most popular?" or "Is user X consuming excessive resources?" These limitations are by design—Ollama prioritizes simplicity and ease of use over enterprise features. This is where building a custom server becomes necessary for organizations with more sophisticated requirements.

APPROACH TWO: BUILDING A CUSTOM SERVER WITH AUTHENTICATION

Building a custom server provides complete control over authentication, authorization, and user management. This approach is more complex but essential for production deployments where you need to track usage, control access, provide a secure multi-user environment, and implement business logic specific to your organization. Our custom server will implement user registration with email verification, API key generation and validation, rate limiting to prevent abuse, persistent storage of user data, and a thread-based architecture for serving multiple models in parallel.

The decision to build a custom server should be based on specific requirements that off-the-shelf solutions cannot meet. If you need to integrate LLM capabilities into an existing application with its own user management system, a custom server allows seamless integration. If you need to implement custom billing logic, usage quotas, or access controls based on user roles or departments, a custom server provides the flexibility to implement these features exactly as needed. If you need to add custom preprocessing or postprocessing of requests and responses, such as content filtering, logging, or analytics, a custom server gives you complete control over the request pipeline.

ARCHITECTURAL OVERVIEW OF THE CUSTOM SERVER

The custom server follows a layered architecture that separates concerns and promotes maintainability. This separation of concerns is a fundamental principle of software engineering that makes the codebase easier to understand, test, and modify. Each layer has a specific responsibility and communicates with adjacent layers through well-defined interfaces.

At the foundation is the database layer, which stores user accounts, email verification tokens, and API keys. We use SQLite for development and small deployments due to its simplicity and zero-configuration setup. SQLite is a self-contained, serverless database engine that stores the entire database in a single file, making it easy to back up, version control, and deploy. For production deployments with higher concurrency requirements, PostgreSQL or MySQL would be more appropriate, offering better performance under concurrent load, advanced features like full-text search and JSON operators, and robust replication and backup tools.

Above the database sits the authentication layer, which handles user registration, email verification, and API key validation. This layer implements the security-critical functions that protect the system from unauthorized access. User registration involves validating email addresses, checking for duplicates, and securely hashing passwords. Email verification ensures that users control the email addresses they register with, preventing spam and abuse. API key validation checks that each request includes a valid, active key associated with a verified user account.

The business logic layer manages model loading, inference requests, and response formatting. This is where the core functionality of the LLM server resides. Model loading involves initializing the inference engine with the appropriate model files and configuration. Inference requests are processed by converting the OpenAI-format messages into prompts suitable for the model, running the inference, and converting the output back to OpenAI format. Response formatting ensures that all responses match the OpenAI API specification exactly, including proper status codes, error messages, and metadata.

Finally, the API layer exposes HTTP endpoints that match the OpenAI specification. This layer handles HTTP-specific concerns like request parsing, header validation, content negotiation, and response serialization. By keeping HTTP concerns separate from business logic, we make the code more testable and easier to adapt if we need to support additional protocols in the future.

The server uses a thread-based model loading strategy that allows multiple models to process requests simultaneously without blocking each other. When the server starts, it creates a dedicated thread for each available LLM model. Each thread initializes an inference engine for its assigned model and then enters a loop where it waits for incoming requests. This architecture provides several benefits: multiple models can process requests concurrently, each model has dedicated resources and doesn't interfere with others, and requests to one model don't block requests to other models.

A request router examines the model parameter in each API request and forwards the request to the appropriate thread for processing. The router uses a queue-based communication pattern where requests are placed in a queue specific to the target model, and the model thread pulls requests from its queue. This decoupling between request submission and processing provides natural backpressure—if a model is busy, requests queue up rather than failing immediately, and the queue depth can be monitored to detect overload conditions.

User authentication follows a standard flow that balances security with user experience. New users register by providing an email address and password. The server validates the email format using a regular expression, checks that the email is not already registered by querying the database, and hashes the password securely using a salted hash function. The server then generates a unique verification token using cryptographically secure random number generation and sends an email containing a verification link to the provided address.

When the user clicks the link in the email, their browser makes a GET request to the verification endpoint with the token as a query parameter. The server validates the token by checking that it exists in the database, hasn't been used before, hasn't expired, and is associated with an unverified account. If all checks pass, the server marks the account as verified and the token as used. This verification process ensures that users control the email addresses they register with, which is important for account recovery and preventing spam.

Only verified users can request API keys. The API key generation endpoint requires the user to authenticate with their email and password. After verifying the credentials and checking that the account is verified, the server generates a cryptographically secure API key and stores it in the database associated with the user's account. The API key is returned to the user, who must store it securely and include it in the Authorization header of all subsequent API requests.

All subsequent API requests must include a valid API key in the Authorization header using the Bearer authentication scheme. The server extracts the key from the header, validates it against the database, checks that it's active and associated with a verified account, and updates the last_used_at timestamp. This validation happens for every request, ensuring that revoked or invalid keys are immediately rejected.

TECHNOLOGY STACK SELECTION

For implementing the custom server, we need to select appropriate technologies for each component. These choices significantly impact development velocity, performance, maintainability, and operational characteristics. The technology stack should align with team expertise, project requirements, and long-term maintenance considerations.

Python is an excellent choice for the server implementation because it has mature libraries for web serving, database access, and machine learning. Python's extensive ecosystem means that most functionality we need is already available in well-tested libraries, reducing development time and bug risk. The language's readability and expressiveness make the codebase easier to understand and maintain, which is particularly valuable for teams with varying levels of experience. Python's interpreted nature and dynamic typing enable rapid prototyping and iteration, though they come with performance trade-offs compared to compiled languages.

The Flask framework provides a lightweight and flexible foundation for building REST APIs. Flask's minimalist philosophy means it includes only essential features, allowing developers to add exactly what they need without carrying unnecessary baggage. This contrasts with more opinionated frameworks like Django, which include everything from ORM to admin interfaces. Flask's simplicity makes it easier to understand the entire request lifecycle, which is valuable for debugging and optimization. The framework's decorator-based routing is intuitive and keeps route definitions close to their handler functions.

For the database, SQLite offers simplicity for development and small deployments. SQLite requires no separate server process, no configuration files, and no administrative overhead. The entire database is a single file that can be easily backed up, copied, or version controlled. SQLite supports most SQL features including transactions, foreign keys, and indexes, making it suitable for complex queries and data integrity constraints. However, SQLite has limitations in concurrent write scenarios—only one writer can access the database at a time, which can become a bottleneck under heavy load.

PostgreSQL provides better performance and features for production use. PostgreSQL handles concurrent connections efficiently, supports advanced features like full-text search and JSON operators, offers robust replication and backup tools, and has excellent monitoring and diagnostic capabilities. The migration from SQLite to PostgreSQL is straightforward since both use SQL, though some syntax differences exist. For production deployments expecting significant load, PostgreSQL or MySQL would be the recommended choice.

Email sending requires an SMTP server for delivering verification messages. For development, you can use services like Mailtrap or MailHog that capture emails without actually sending them, allowing you to test the verification flow without needing real email infrastructure. These tools provide web interfaces where you can view captured emails and verify that the content and formatting are correct. For production, you might use transactional email services like SendGrid, Amazon SES, or Mailgun, which handle deliverability, spam filtering, and bounce management. Alternatively, you can configure your own SMTP server using software like Postfix, though this requires more operational expertise.

The email verification mechanism requires generating secure random tokens that cannot be guessed or predicted. Python's secrets module provides cryptographically strong random number generation suitable for security-sensitive applications. The secrets.token_urlsafe() function generates random tokens that are safe to include in URLs, which is perfect for email verification links. Never use the random module for security purposes, as it uses a predictable pseudo-random number generator that can be exploited by attackers.

For loading and running LLM models, we can use libraries like llama-cpp-python for GGUF format models, or transformers from Hugging Face for a wider variety of models. The choice depends on which models you want to support and their format. The llama-cpp-python library is particularly efficient for running quantized models on CPU, using optimized C++ code with SIMD instructions for fast inference. It supports various quantization formats like Q4_K_M and Q5_K_S that balance model size and quality. The library also supports GPU acceleration via CUDA or Metal, though CPU inference is often sufficient for moderate loads.

The transformers library from Hugging Face provides broader model support but may require GPU acceleration for acceptable performance. Transformers supports thousands of models including BERT, GPT, T5, and many others. The library handles tokenization, model loading, and generation with a consistent API across different model architectures. 

However, transformers models are typically larger and slower than quantized GGUF models, making them more suitable for GPU deployment. The choice between llama-cpp-python and transformers depends on your specific models, hardware, and performance requirements.

DATABASE SCHEMA DESIGN

The database schema must store user information, verification tokens, and API keys in a way that maintains data integrity, supports efficient queries, and enables future extensions. Good schema design is crucial for long-term maintainability and performance. We need three primary tables: users, verification_tokens, and api_keys, each with carefully chosen fields and constraints.

The users table, as shown in the code below contains the core account information including email address, password hash, and verification status. 

CREATE TABLE users (

    id INTEGER PRIMARY KEY AUTOINCREMENT,

    email TEXT UNIQUE NOT NULL,

    password_hash TEXT NOT NULL,

    is_verified INTEGER DEFAULT 0,

    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP

);

CREATE TABLE verification_tokens (

    id INTEGER PRIMARY KEY AUTOINCREMENT,

    user_id INTEGER NOT NULL,

    token TEXT UNIQUE NOT NULL,

    expires_at TIMESTAMP NOT NULL,

    used INTEGER DEFAULT 0,

    FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE

);

CREATE TABLE api_keys (

    id INTEGER PRIMARY KEY AUTOINCREMENT,

    user_id INTEGER NOT NULL,

    key_value TEXT UNIQUE NOT NULL,

    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,

    last_used_at TIMESTAMP,

    is_active INTEGER DEFAULT 1,

    FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE

);

The id field is an auto-incrementing integer primary key that uniquely identifies each user. Using an integer primary key provides efficient indexing and foreign key references. The email field is marked UNIQUE to prevent duplicate registrations and NOT NULL to ensure every user has an email address. The uniqueness constraint is enforced at the database level, preventing race conditions where two simultaneous registrations with the same email could both succeed.

The password_hash field stores the hashed password, never the plaintext password. Storing plaintext passwords is a critical security vulnerability that exposes users if the database is compromised. The hash is generated using a salted hash function, where a random salt is combined with the password before hashing. The salt prevents rainbow table attacks and ensures that two users with the same password have different hashes. The stored hash includes both the salt and the hash value in a format like "salt$hash", allowing the verification function to extract the salt and recompute the hash for comparison.

The is_verified field tracks whether the user has clicked the verification link in their email. This field uses an integer (0 or 1) rather than a boolean because SQLite does not have a native boolean type. The default value of 0 means new users are unverified until they complete the verification process. This field is checked before allowing API key generation, ensuring that only users who control their email addresses can access the service.

The created_at field records when the account was created using a timestamp. This field is useful for analytics, debugging, and implementing time-based policies like "delete unverified accounts older than 30 days." The DEFAULT CURRENT_TIMESTAMP clause automatically sets this field to the current time when a row is inserted, eliminating the need to specify it explicitly in INSERT statements.

The verification_tokens table stores tokens sent via email along with their expiration times and usage status. Each token is associated with a user via the user_id foreign key, which references the id field in the users table. The ON DELETE CASCADE clause ensures that when a user is deleted, all their verification tokens are automatically deleted as well, maintaining referential integrity and preventing orphaned records.

The token field stores the actual verification token as a unique string. The uniqueness constraint prevents token collisions, though with cryptographically secure random generation, collisions are astronomically unlikely. The token should be long enough to resist brute-force attacks—32 bytes of random data encoded in URL-safe base64 provides approximately 256 bits of entropy, making guessing infeasible.

The expires_at field specifies when the token becomes invalid. Verification tokens should have a limited lifetime, typically 24 hours, to reduce the window of opportunity for token theft or misuse. When validating a token, the server checks that the current time is before the expiration time. Expired tokens are rejected even if they haven't been used, forcing users to request a new verification email if they wait too long.

The used field tracks whether the token has been consumed. Once a token is used to verify an account, it should not be usable again. This prevents replay attacks where an attacker who obtains a used token tries to use it again. The verification endpoint checks this field and rejects tokens that have already been used.

The api_keys table associates keys with users and tracks when they were created and last used. Like verification tokens, API keys have a user_id foreign key with ON DELETE CASCADE to maintain referential integrity. The key_value field stores the actual API key as a unique string. API keys should follow a recognizable format, such as starting with "sk-" (for "secret key"), which makes them identifiable in logs and helps prevent accidental exposure.

The created_at field records when the key was generated, which is useful for security auditing and implementing key rotation policies. The last_used_at field is updated every time the key is used to make an API request. This field helps identify inactive keys that can be revoked and detect unusual usage patterns that might indicate key compromise. For example, if a key that's normally used from a specific IP address suddenly appears from a different country, that might warrant investigation.

The is_active field allows administrators to revoke keys without deleting them. Setting this field to 0 immediately prevents the key from being used, while preserving the record for audit purposes. This is preferable to deletion because it maintains a complete history of all keys ever issued, which is important for compliance and security investigations. Inactive keys are rejected during validation, even if they're otherwise valid.

IMPLEMENTING USER REGISTRATION

The user registration process begins when a client sends a POST request to the registration endpoint with an email and password. The server must validate the email format, check that the email is not already registered, hash the password securely, create the user record, generate a verification token, and send the verification email. Each of these steps is critical for security and user experience.

The next example shows the imports and Flask app initialization. 

from flask import Flask, request, jsonify

import sqlite3

import hashlib

import secrets

import smtplib

from email.mime.text import MIMEText

from email.mime.multipart import MIMEMultipart

from datetime import datetime, timedelta

import re

app = Flask(__name__)

These imports provide all the functionality needed for the server: Flask for web serving, sqlite3 for database access, hashlib for password hashing, secrets for secure random generation, smtplib for email sending, email.mime for constructing email messages, datetime for timestamp handling, and re for regular expressions. The Flask app is initialized with __name__, which tells Flask where to find resources like templates and static files.

The next code example implements the database connection function. 

def get_db_connection():

    """

    Establish and return a connection to the SQLite database.

    Sets row_factory to sqlite3.Row for dict-like access to columns.

    """

    conn = sqlite3.connect('llm_server.db')

    conn.row_factory = sqlite3.Row

    return conn

This function creates a connection to the SQLite database file and sets the row_factory to sqlite3.Row. The row factory determines how query results are returned. By default, SQLite returns results as tuples, which requires accessing columns by numeric index. Setting row_factory to sqlite3.Row allows accessing columns by name, making the code more readable and maintainable. For example, you can write row['email'] instead of row[0], which is clearer and less error-prone.

Now, we implement email validation using a regular expression. 

def is_valid_email(email):

    """

    Validate email format using a regular expression.

    Returns True if email matches standard email pattern.

    """

    pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'

    return re.match(pattern, email) is not None

Email validation is surprisingly complex because the email specification is very permissive, allowing many unusual formats that are rarely used in practice. This regex implements a simplified validation that accepts most common email formats while rejecting obviously invalid inputs. It checks for a local part (before the @), an @ symbol, a domain name, and a top-level domain. This validation is not perfect—it will accept some invalid emails and reject some valid ones—but it's sufficient for most practical purposes. For more robust validation, consider using a dedicated email validation library or sending a verification email and only accepting addresses that can receive it.

The application implements password hashing using SHA-256 with a random salt. The salt is a random 16-byte value encoded as hexadecimal, providing 128 bits of entropy. The password and salt are concatenated and hashed together, and the result is stored in the format "salt$hash". This format allows the verification function to extract the salt and recompute the hash. 

def hash_password(password):

    """

    Hash a password using SHA-256 with a random salt.

    In production, use bcrypt or argon2 for better security.

    Returns a string in format: salt$hash

    """

    salt = secrets.token_hex(16)

    password_hash = hashlib.sha256((password + salt).encode()).hexdigest()

    return f"{salt}${password_hash}"

While this implementation works, production systems should use dedicated password hashing algorithms like bcrypt, scrypt, or Argon2. These algorithms are specifically designed for password hashing and include features like configurable work factors that make brute-force attacks more expensive. They also handle salting automatically and use key stretching to slow down hash computation, making offline attacks much harder.

We need to implement password verification by extracting the salt from the stored hash, recomputing the hash with the provided password, and comparing the result to the expected hash. 

def verify_password(password, stored_hash):

    """

    Verify a password against a stored hash.

    Extracts the salt from stored_hash and recomputes the hash.

    Returns True if password matches.

    """

    try:

        salt, expected_hash = stored_hash.split('$')

        password_hash = hashlib.sha256((password + salt).encode()).hexdigest()

        return password_hash == expected_hash

    except ValueError:

        return False

The try-except block handles the case where the stored hash is not in the expected format, returning False rather than raising an exception. This defensive programming prevents crashes from malformed data. The comparison should ideally use a constant-time comparison function to prevent timing attacks, though for password verification the risk is minimal since the hash computation dominates the timing.

The next code snippet implements email sending using SMTP. This function constructs a MIME multipart message with both plain text and HTML versions of the verification email. 

def send_verification_email(email, token):

    """

    Send a verification email to the user with a clickable link.

    Returns True if email was sent successfully, False otherwise.

    """

    # Configure your SMTP settings here

    smtp_server = 'smtp.example.com'

    smtp_port = 587

    smtp_username = 'your_email@example.com'

    smtp_password = 'your_password'

    

    verification_link = f"http://localhost:5000/verify?token={token}"

    

    message = MIMEMultipart('alternative')

    message['Subject'] = 'Verify your email address'

    message['From'] = smtp_username

    message['To'] = email

    

    text_content = f"""

    Please verify your email address by clicking the following link:

    {verification_link}

    

    This link will expire in 24 hours.

    """

    

    html_content = f"""

    <html>

      <body>

        <p>Please verify your email address by clicking the link below:</p>

        <p><a href="{verification_link}">Verify Email Address</a></p>

        <p>This link will expire in 24 hours.</p>

      </body>

    </html>

    """

    

    part1 = MIMEText(text_content, 'plain')

    part2 = MIMEText(html_content, 'html')

    message.attach(part1)

    message.attach(part2)

    

    try:

        server = smtplib.SMTP(smtp_server, smtp_port)

        server.starttls()

        server.login(smtp_username, smtp_password)

        server.sendmail(smtp_username, email, message.as_string())

        server.quit()

        return True

    except Exception as e:

        print(f"Failed to send email: {e}")

        return False

Providing both versions ensures compatibility with all email clients—those that support HTML will display the formatted version, while those that don't will display the plain text version. The verification link includes the token as a query parameter, allowing the user to verify their account by simply clicking the link.

The SMTP configuration includes server address, port, username, and password. For production use, these should be stored in environment variables or a configuration file rather than hardcoded. The function uses STARTTLS to encrypt the connection, protecting the credentials and email content from eavesdropping. The try-except block handles connection errors, authentication failures, and other issues that might occur during email sending, returning False to indicate failure rather than crashing the server.

Let us implement the registration endpoint that ties all these pieces together. 

@app.route('/register', methods=['POST'])

def register():

    """

    Handle user registration requests.

    Expects JSON with 'email' and 'password' fields.

    """

    data = request.get_json()

    

    if not data or 'email' not in data or 'password' not in data:

        return jsonify({'error': 'Email and password required'}), 400

    

    email = data['email'].lower().strip()

    password = data['password']

    

    # Validate email format

    if not is_valid_email(email):

        return jsonify({'error': 'Invalid email format'}), 400

    

    # Validate password strength

    if len(password) < 8:

        return jsonify({'error': 'Password must be at least 8 characters'}), 400

    

    conn = get_db_connection()

    cursor = conn.cursor()

    

    # Check if email already exists

    cursor.execute('SELECT id FROM users WHERE email = ?', (email,))

    if cursor.fetchone():

        conn.close()

        return jsonify({'error': 'Email already registered'}), 409

    

    # Hash password and create user

    password_hash = hash_password(password)

    cursor.execute(

        'INSERT INTO users (email, password_hash) VALUES (?, ?)',

        (email, password_hash)

    )

    user_id = cursor.lastrowid

    

    # Generate verification token

    token = secrets.token_urlsafe(32)

    expires_at = datetime.now() + timedelta(hours=24)

    

    cursor.execute(

        'INSERT INTO verification_tokens (user_id, token, expires_at) VALUES (?, ?, ?)',

        (user_id, token, expires_at)

    )

    

    conn.commit()

    conn.close()

    

    # Send verification email

    if send_verification_email(email, token):

        return jsonify({

            'message': 'Registration successful. Please check your email to verify your account.'

        }), 201

    else:

        return jsonify({

            'message': 'Registration successful but failed to send verification email.'

        }), 201

The endpoint expects a JSON request body with email and password fields. It validates that both fields are present, normalizes the email address by converting to lowercase and stripping whitespace, validates the email format, and checks password strength. The password strength check is minimal—only requiring 8 characters—but production systems should enforce stronger requirements like requiring mixed case, numbers, and special characters.

The endpoint then checks if the email is already registered by querying the database. If a user with that email exists, it returns a 409 Conflict status code. This check prevents duplicate registrations and provides clear feedback to the user. The endpoint then hashes the password and inserts a new user record. The cursor.lastrowid attribute provides the auto-generated user ID, which is needed to create the verification token.

The verification token is generated using secrets.token_urlsafe(32), which produces a URL-safe random string with approximately 256 bits of entropy. The expiration time is set to 24 hours from now using datetime.now() + timedelta(hours=24). The token and expiration are inserted into the verification_tokens table, associated with the new user.

Finally, the endpoint sends the verification email and returns an appropriate response. If email sending succeeds, it returns a 201 Created status with a message instructing the user to check their email. If email sending fails, it still returns 201 because the account was created successfully, but includes a message indicating the email failure. This allows the user to request a new verification email later.

IMPLEMENTING EMAIL VERIFICATION ENDPOINT

The next listing shows the implementation of  the email verification endpoint that processes verification links. 

@app.route('/verify', methods=['GET'])

def verify_email():

    """

    Handle email verification via token in query parameter.

    """

    token = request.args.get('token')

    

    if not token:

        return jsonify({'error': 'Verification token required'}), 400

    

    conn = get_db_connection()

    cursor = conn.cursor()

    

    # Find the token

    cursor.execute(

        '''SELECT vt.id, vt.user_id, vt.expires_at, vt.used, u.is_verified

           FROM verification_tokens vt

           JOIN users u ON vt.user_id = u.id

           WHERE vt.token = ?''',

        (token,)

    )

    result = cursor.fetchone()

    

    if not result:

        conn.close()

        return jsonify({'error': 'Invalid verification token'}), 404

    

    token_id, user_id, expires_at, used, is_verified = result

    

    # Check if already used

    if used:

        conn.close()

        return jsonify({'error': 'Verification token already used'}), 400

    

    # Check if already verified

    if is_verified:

        conn.close()

        return jsonify({'message': 'Email already verified'}), 200

    

    # Check if expired

    expires_at_dt = datetime.fromisoformat(expires_at)

    if datetime.now() > expires_at_dt:

        conn.close()

        return jsonify({'error': 'Verification token expired'}), 400

    

    # Mark token as used and user as verified

    cursor.execute('UPDATE verification_tokens SET used = 1 WHERE id = ?', (token_id,))

    cursor.execute('UPDATE users SET is_verified = 1 WHERE id = ?', (user_id,))

    

    conn.commit()

    conn.close()

    

    return jsonify({'message': 'Email verified successfully'}), 200

This endpoint expects a GET request with the token as a query parameter. It validates that the token is present, queries the database to find the token, and performs several checks before marking the account as verified.

The database query joins the verification_tokens and users tables to retrieve all necessary information in a single query. This is more efficient than making separate queries and ensures consistency. The query returns the token ID, user ID, expiration time, used status, and verification status. If no matching token is found, the endpoint returns a 404 Not Found status.

The endpoint then checks if the token has already been used. This prevents replay attacks where someone tries to use the same token multiple times. If the token is already used, the endpoint returns a 400 Bad Request status. The endpoint also checks if the account is already verified, returning a 200 OK status with a message indicating that verification is not needed. This handles the case where a user clicks the verification link multiple times.

The expiration check compares the current time to the stored expiration time. SQLite stores timestamps as strings in ISO format, so they must be parsed using datetime.fromisoformat() before comparison. If the token has expired, the endpoint returns a 400 Bad Request status. This enforces the 24-hour lifetime for verification tokens.

If all checks pass, the endpoint marks the token as used and the user as verified by updating both tables. These updates are performed within a transaction (implicitly, since we commit at the end), ensuring that either both updates succeed or neither does. This maintains consistency even if the server crashes between the updates. The endpoint returns a 200 OK status with a success message.

IMPLEMENTING API KEY GENERATION

The next code example implements the API key generation endpoint that allows verified users to obtain API keys for accessing the service. 

@app.route('/generate-api-key', methods=['POST'])

def generate_api_key():

    """

    Generate a new API key for a verified user.

    Expects JSON with 'email' and 'password' for authentication.

    """

    data = request.get_json()

    

    if not data or 'email' not in data or 'password' not in data:

        return jsonify({'error': 'Email and password required'}), 400

    

    email = data['email'].lower().strip()

    password = data['password']

    

    conn = get_db_connection()

    cursor = conn.cursor()

    

    # Find user

    cursor.execute(

        'SELECT id, password_hash, is_verified FROM users WHERE email = ?',

        (email,)

    )

    result = cursor.fetchone()

    

    if not result:

        conn.close()

        return jsonify({'error': 'Invalid credentials'}), 401

    

    user_id, password_hash, is_verified = result

    

    # Verify password

    if not verify_password(password, password_hash):

        conn.close()

        return jsonify({'error': 'Invalid credentials'}), 401

    

    # Check if email is verified

    if not is_verified:

        conn.close()

        return jsonify({'error': 'Email not verified'}), 403

    

    # Generate API key

    api_key = f"sk-{secrets.token_urlsafe(32)}"

    

    cursor.execute(

        'INSERT INTO api_keys (user_id, key_value) VALUES (?, ?)',

        (user_id, api_key)

    )

    

    conn.commit()

    conn.close()

    

    return jsonify({'api_key': api_key}), 201

This endpoint requires authentication with email and password, ensuring that only the account owner can generate keys. The endpoint expects a JSON request body with email and password fields.

The endpoint queries the database to find the user by email, retrieving the user ID, password hash, and verification status. If no user is found, it returns a 401 Unauthorized status with a generic error message. Using a generic message prevents attackers from determining which email addresses are registered, reducing the information available for targeted attacks.

The endpoint verifies the password using the verify_password function. If the password is incorrect, it returns 401 Unauthorized with the same generic error message. This consistency prevents timing attacks that might distinguish between "user not found" and "wrong password" based on response time.

The endpoint checks if the email is verified. Unverified accounts cannot generate API keys, ensuring that only users who control their email addresses can access the service. If the account is unverified, the endpoint returns a 403 Forbidden status with a message explaining the requirement.

If all checks pass, the endpoint generates a new API key using secrets.token_urlsafe(32) and prefixes it with "sk-" to make it recognizable. The key is inserted into the api_keys table, associated with the user's account. The endpoint returns a 201 Created status with the API key in the response body. The user must store this key securely, as it will be required for all API requests.

IMPLEMENTING API KEY VALIDATION

The next code example implements the API key validation function that checks whether a provided key is valid and returns the associated user ID. This function is called by every protected endpoint to authenticate requests. It takes an API key as input and returns the user ID if the key is valid, or None if it's invalid.

def validate_api_key(api_key):

    """

    Validate an API key and return the associated user_id.

    Returns user_id if valid, None otherwise.

    Also updates the last_used_at timestamp.

    """

    conn = get_db_connection()

    cursor = conn.cursor()

    

    cursor.execute(

        '''SELECT ak.user_id, ak.is_active, u.is_verified

           FROM api_keys ak

           JOIN users u ON ak.user_id = u.id

           WHERE ak.key_value = ?''',

        (api_key,)

    )

    result = cursor.fetchone()

    

    if not result:

        conn.close()

        return None

    

    user_id, is_active, is_verified = result

    

    if not is_active or not is_verified:

        conn.close()

        return None

    

    # Update last used timestamp

    cursor.execute(

        'UPDATE api_keys SET last_used_at = ? WHERE key_value = ?',

        (datetime.now(), api_key)

    )

    

    conn.commit()

    conn.close()

    

    return user_id

The function queries the database, joining the api_keys and users tables to retrieve the user ID, active status, and verification status. This join ensures that the key is associated with a verified account, preventing revoked or unverified accounts from accessing the service. If no matching key is found, the function returns None.

The function checks that the key is active and the associated account is verified. Inactive keys are rejected even if they exist in the database, allowing administrators to revoke keys without deleting them. Unverified accounts are also rejected, though in practice this should not happen since API keys can only be generated for verified accounts.

If the key is valid, the function updates the last_used_at timestamp to record when the key was used. This information is valuable for security monitoring and identifying inactive keys. The update is committed to the database before returning the user ID. This function is called frequently, so it should be optimized for performance. Consider adding an index on the key_value column to speed up lookups.

IMPLEMENTING MODEL LOADING WITH THREADING

The next code implements the ModelThread class that loads a model in a dedicated thread and processes inference requests. This architecture allows multiple models to run concurrently without blocking each other. Each ModelThread instance manages a single model, loading it during initialization and processing requests in a loop.

import threading

import queue

from llama_cpp import Llama

class ModelThread:

    """

    A thread that loads a model and processes inference requests.

    """

    

    def __init__(self, model_name, model_path):

        """

        Initialize the model thread.

        

        Args:

            model_name: Name identifier for the model

            model_path: Path to the model file

        """

        self.model_name = model_name

        self.model_path = model_path

        self.request_queue = queue.Queue()

        self.model = None

        self.thread = threading.Thread(target=self._run, daemon=True)

        self.thread.start()

    

    def _run(self):

        """

        Main thread loop. Loads model and processes requests.

        """

        print(f"Loading model {self.model_name} from {self.model_path}")

        

        # Load the model

        self.model = Llama(

            model_path=self.model_path,

            n_ctx=2048,

            n_threads=4

        )

        

        print(f"Model {self.model_name} loaded successfully")

        

        # Process requests

        while True:

            request_data, response_queue = self.request_queue.get()

            

            try:

                result = self._process_request(request_data)

                response_queue.put(('success', result))

            except Exception as e:

                response_queue.put(('error', str(e)))

    

    def _process_request(self, request_data):

        """

        Process a single inference request.

        

        Args:

            request_data: Dictionary containing messages, temperature, max_tokens, etc.

        

        Returns:

            Dictionary with completion result

        """

        messages = request_data.get('messages', [])

        temperature = request_data.get('temperature', 0.7)

        max_tokens = request_data.get('max_tokens', 150)

        

        # Convert messages to prompt

        prompt = self._messages_to_prompt(messages)

        

        # Generate completion

        output = self.model(

            prompt,

            max_tokens=max_tokens,

            temperature=temperature,

            stop=["</s>", "User:", "Assistant:"]

        )

        

        return {

            'text': output['choices'][0]['text'],

            'tokens_used': output['usage']['total_tokens']

        }

    

    def _messages_to_prompt(self, messages):

        """

        Convert OpenAI-style messages to a prompt string.

        

        Args:

            messages: List of message dictionaries

        

        Returns:

            Formatted prompt string

        """

        prompt = ""

        

        for msg in messages:

            role = msg.get('role', 'user')

            content = msg.get('content', '')

            

            if role == 'system':

                prompt += f"System: {content}\n\n"

            elif role == 'user':

                prompt += f"User: {content}\n\n"

            elif role == 'assistant':

                prompt += f"Assistant: {content}\n\n"

        

        prompt += "Assistant: "

        return prompt

    

    def submit_request(self, request_data):

        """

        Submit a request to this model thread.

        

        Args:

            request_data: Dictionary containing request parameters

        

        Returns:

            Tuple of (status, result) where status is 'success' or 'error'

        """

        response_queue = queue.Queue()

        self.request_queue.put((request_data, response_queue))

        return response_queue.get(timeout=60)

The __init__ method stores the model name and path, creates a request queue for incoming requests, and starts a background thread that runs the _run method. The daemon=True flag ensures that the thread will be terminated when the main program exits, preventing the program from hanging if the thread is still running. The thread starts immediately, beginning the model loading process.

The _run method is the main loop that runs in the background thread. It first loads the model using the Llama class from llama-cpp-python. The model_path parameter specifies the file containing the model weights. The n_ctx parameter sets the context window size, which determines how much text the model can consider when generating responses. Larger context windows allow longer conversations but require more memory. The n_threads parameter controls how many CPU threads are used for inference, with higher values providing faster inference on multi-core systems.

After loading the model, the thread enters an infinite loop where it waits for requests from the queue. The queue.get() method blocks until a request is available, allowing the thread to sleep when idle and wake up immediately when work arrives. Each request is a tuple containing the request data and a response queue. The thread processes the request by calling _process_request, catches any exceptions that occur, and puts the result (either success or error) into the response queue. This pattern decouples request submission from processing, allowing the caller to submit a request and wait for the response without blocking other operations.

The _process_request method extracts parameters from the request data, converts the OpenAI-format messages to a prompt string, runs the inference, and returns the result. The messages_to_prompt conversion is model-specific—different models expect different prompt formats. For example, Llama 2 uses a specific format with [INST] tags, while other models might use different conventions. The implementation shown uses a simple format with "System:", "User:", and "Assistant:" prefixes, which works reasonably well for most models.

The model() call performs the actual inference, generating text based on the prompt. The max_tokens parameter limits the length of the generated text, preventing excessively long responses. The temperature parameter controls randomness, with lower values producing more deterministic outputs. The stop parameter specifies strings that should terminate generation, such as end-of-sequence tokens or role markers. This prevents the model from continuing to generate text beyond the intended response.

The _messages_to_prompt method converts the OpenAI message format to a prompt string. It iterates through the messages, formatting each one according to its role. System messages are prefixed with "System:", user messages with "User:", and assistant messages with "Assistant:". The final prompt ends with "Assistant: " to prompt the model to generate a response. This simple format works for many models, but production systems should use model-specific prompt templates for best results.

The submit_request method provides a convenient interface for submitting requests to the model thread. It creates a response queue, puts the request and response queue into the request queue, and waits for the response with a timeout. The timeout prevents the caller from waiting indefinitely if the model thread crashes or hangs. If the timeout expires, a queue Empty exception is raised, which the caller can handle appropriately.

IMPLEMENTING MODEL MANAGER

We need to implement the ModelManager class that manages multiple model threads and routes requests to them. This class provides a centralized interface for adding models, retrieving models by name, and listing available models. It maintains a dictionary mapping model names to ModelThread instances.

class ModelManager:

    """

    Manages multiple model threads and routes requests to them.

    """

    

    def __init__(self):

        """

        Initialize the model manager with an empty models dictionary.

        """

        self.models = {}

    

    def add_model(self, model_name, model_path):

        """

        Add a new model to the manager.

        

        Args:

            model_name: Name identifier for the model

            model_path: Path to the model file

        """

        if model_name in self.models:

            print(f"Model {model_name} already loaded")

            return

        

        self.models[model_name] = ModelThread(model_name, model_path)

    

    def get_model(self, model_name):

        """

        Get a model thread by name.

        

        Args:

            model_name: Name of the model

        

        Returns:

            ModelThread instance or None if not found

        """

        return self.models.get(model_name)

    

    def list_models(self):

        """

        Get a list of all available model names.

        

        Returns:

            List of model name strings

        """

        return list(self.models.keys())

# Global model manager instance

model_manager = ModelManager()

The add_model method creates a new ModelThread for the specified model and adds it to the dictionary. It checks if the model is already loaded to prevent duplicate loading. The method is called during server startup to initialize all available models. Each model runs in its own thread, allowing concurrent processing of requests to different models.

The get_model method retrieves a ModelThread by name, returning None if the model doesn't exist. This method is used by the API endpoints to route requests to the appropriate model. The list_models method returns a list of all available model names, which is used by the /v1/models endpoint to inform clients which models are available.

The global model_manager instance is created at module level and used throughout the application. This singleton pattern ensures that all parts of the application share the same model manager, preventing duplicate model loading and ensuring consistent state.

IMPLEMENTING CHAT COMPLETIONS ENDPOINT

Let us now implement the /v1/chat/completions endpoint, which is the main endpoint for generating text. This endpoint matches the OpenAI API specification, allowing clients to use the same code for both OpenAI and local models. The endpoint requires authentication via API key, validates the request, routes it to the appropriate model, and returns the response in OpenAI format.

@app.route('/v1/chat/completions', methods=['POST'])

def chat_completions():

    """

    OpenAI-compatible chat completions endpoint.

    Requires valid API key in Authorization header.

    """

    # Extract and validate API key

    auth_header = request.headers.get('Authorization')

    

    if not auth_header or not auth_header.startswith('Bearer '):

        return jsonify({'error': 'Missing or invalid Authorization header'}), 401

    

    api_key = auth_header.replace('Bearer ', '')

    user_id = validate_api_key(api_key)

    

    if not user_id:

        return jsonify({'error': 'Invalid API key'}), 401

    

    # Parse request data

    data = request.get_json()

    

    if not data:

        return jsonify({'error': 'Request body required'}), 400

    

    model_name = data.get('model')

    messages = data.get('messages', [])

    temperature = data.get('temperature', 0.7)

    max_tokens = data.get('max_tokens', 150)

    

    if not model_name:

        return jsonify({'error': 'Model name required'}), 400

    

    if not messages:

        return jsonify({'error': 'Messages required'}), 400

    

    # Get the model

    model_thread = model_manager.get_model(model_name)

    

    if not model_thread:

        return jsonify({

            'error': f'Model {model_name} not found',

            'available_models': model_manager.list_models()

        }), 404

    

    # Submit request to model thread

    try:

        status, result = model_thread.submit_request({

            'messages': messages,

            'temperature': temperature,

            'max_tokens': max_tokens

        })

        

        if status == 'error':

            return jsonify({'error': result}), 500

        

        # Format response in OpenAI format

        response = {

            'id': f"chatcmpl-{secrets.token_hex(12)}",

            'object': 'chat.completion',

            'created': int(datetime.now().timestamp()),

            'model': model_name,

            'choices': [

                {

                    'index': 0,

                    'message': {

                        'role': 'assistant',

                        'content': result['text']

                    },

                    'finish_reason': 'stop'

                }

            ],

            'usage': {

                'prompt_tokens': 0,

                'completion_tokens': result['tokens_used'],

                'total_tokens': result['tokens_used']

            }

        }

        

        return jsonify(response), 200

        

    except queue.Empty:

        return jsonify({'error': 'Request timeout'}), 504

    except Exception as e:

        return jsonify({'error': str(e)}), 500

The endpoint first extracts the API key from the Authorization header. The header should be in the format "Bearer ", where is the API key. The endpoint checks that the header is present and starts with "Bearer ", then extracts the key by removing the prefix. If the header is missing or malformed, the endpoint returns a 401 Unauthorized status.

The endpoint validates the API key by calling the validate_api_key function. This function checks that the key exists, is active, and is associated with a verified account. If validation fails, the endpoint returns a 401 Unauthorized status. This authentication check ensures that only authorized users can access the service.

The endpoint parses the request body as JSON and extracts the required parameters. The model parameter specifies which model to use, the messages parameter contains the conversation history, and optional parameters like temperature and max_tokens control generation behavior. If the model or messages are missing, the endpoint returns a 400 Bad Request status.

The endpoint retrieves the ModelThread for the specified model using the model manager. If the model doesn't exist, the endpoint returns a 404 Not Found status with a list of available models. This helps clients discover which models are available and correct typos in model names.

The endpoint submits the request to the model thread and waits for the response. The submit_request method blocks until the model processes the request or the timeout expires. If the timeout expires, a queue.Empty exception is raised, and the endpoint returns a 504 Gateway Timeout status. If the model returns an error, the endpoint returns a 500 Internal Server Error status with the error message.

If the request succeeds, the endpoint formats the response in OpenAI format. The response includes a unique ID generated using secrets.token_hex(), an object type of "chat.completion", a timestamp, the model name, the generated text in the choices array, and token usage statistics. This format exactly matches the OpenAI API, ensuring compatibility with existing clients.

IMPLEMENTING MODELS LIST ENDPOINT

It is time to create the /v1/models endpoint, which returns a list of available models. This endpoint matches the OpenAI API specification and allows clients to discover which models are available. Like the chat completions endpoint, it requires authentication via API key.

@app.route('/v1/models', methods=['GET'])

def list_models():

    """

    OpenAI-compatible endpoint to list available models.

    Requires valid API key in Authorization header.

    """

    # Extract and validate API key

    auth_header = request.headers.get('Authorization')

    

    if not auth_header or not auth_header.startswith('Bearer '):

        return jsonify({'error': 'Missing or invalid Authorization header'}), 401

    

    api_key = auth_header.replace('Bearer ', '')

    user_id = validate_api_key(api_key)

    

    if not user_id:

        return jsonify({'error': 'Invalid API key'}), 401

    

    # Get list of models

    model_names = model_manager.list_models()

    

    # Format in OpenAI format

    models_data = []

    for name in model_names:

        models_data.append({

            'id': name,

            'object': 'model',

            'created': int(datetime.now().timestamp()),

            'owned_by': 'local'

        })

    

    response = {

        'object': 'list',

        'data': models_data

    }

    

    return jsonify(response), 200

The endpoint extracts and validates the API key using the same logic as the chat completions endpoint. If authentication fails, it returns a 401 Unauthorized status. The endpoint then retrieves the list of model names from the model manager and formats them in OpenAI format.

Each model is represented as a dictionary with an ID (the model name), an object type of "model", a creation timestamp, and an owner of "local". The response wraps these model dictionaries in a list object, matching the OpenAI API format. This allows clients to use the same code to list models from both OpenAI and local servers.

IMPLEMENTING DATABASE INITIALIZATION

How to  implement  the database initialization function, that creates the required tables if they don't exist, is shown below. This function should be called during server startup to ensure the database schema is set up correctly. It uses CREATE TABLE IF NOT EXISTS statements, which create the tables only if they don't already exist, making the function safe to call multiple times.

def init_database():

    """

    Initialize the database with required tables.

    """

    conn = sqlite3.connect('llm_server.db')

    cursor = conn.cursor()

    

    # Create users table

    cursor.execute('''

        CREATE TABLE IF NOT EXISTS users (

            id INTEGER PRIMARY KEY AUTOINCREMENT,

            email TEXT UNIQUE NOT NULL,

            password_hash TEXT NOT NULL,

            is_verified INTEGER DEFAULT 0,

            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP

        )

    ''')

    

    # Create verification_tokens table

    cursor.execute('''

        CREATE TABLE IF NOT EXISTS verification_tokens (

            id INTEGER PRIMARY KEY AUTOINCREMENT,

            user_id INTEGER NOT NULL,

            token TEXT UNIQUE NOT NULL,

            expires_at TIMESTAMP NOT NULL,

            used INTEGER DEFAULT 0,

            FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE

        )

    ''')

    

    # Create api_keys table

    cursor.execute('''

        CREATE TABLE IF NOT EXISTS api_keys (

            id INTEGER PRIMARY KEY AUTOINCREMENT,

            user_id INTEGER NOT NULL,

            key_value TEXT UNIQUE NOT NULL,

            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,

            last_used_at TIMESTAMP,

            is_active INTEGER DEFAULT 1,

            FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE

        )

    ''')

    

    conn.commit()

    conn.close()

    

    print("Database initialized successfully")

The function creates the three tables defined in the schema: users, verification_tokens, and api_keys. Each CREATE TABLE statement includes all the columns, constraints, and foreign keys defined in the schema. The function commits the changes and closes the connection, ensuring that the schema is persisted to disk.

IMPLEMENTING RATE LIMITING

To prevent abuse and ensure fair resource allocation, we need to implement rate limiting that restricts how many requests each user can make within a time window. Rate limiting is essential for production deployments to prevent individual users from overwhelming the server and degrading service for others.

We'll implement a token bucket algorithm, which is a common and effective rate limiting strategy. Each user has a bucket that holds a certain number of tokens. Each request consumes one token. Tokens are replenished at a constant rate. If a user's bucket is empty, their requests are rejected until tokens are replenished.

First, we need to add a rate_limit table to track token buckets for each user:

CREATE TABLE rate_limits (

    user_id INTEGER PRIMARY KEY,

    tokens REAL NOT NULL,

    last_refill TIMESTAMP NOT NULL,

    FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE

);

Now we implement the rate limiting logic:

import time

# Rate limiting configuration

RATE_LIMIT_TOKENS = 100  # Maximum tokens in bucket

RATE_LIMIT_REFILL_RATE = 10  # Tokens per minute

RATE_LIMIT_COST = 1  # Tokens consumed per request

def check_rate_limit(user_id):

    """

    Check if a user has available rate limit tokens.

    Implements token bucket algorithm.

    

    Args:

        user_id: The user ID to check

    

    Returns:

        Tuple of (allowed, tokens_remaining)

    """

    conn = get_db_connection()

    cursor = conn.cursor()

    

    # Get current rate limit state

    cursor.execute(

        'SELECT tokens, last_refill FROM rate_limits WHERE user_id = ?',

        (user_id,)

    )

    result = cursor.fetchone()

    

    current_time = datetime.now()

    

    if result is None:

        # First request from this user - initialize bucket

        tokens = RATE_LIMIT_TOKENS - RATE_LIMIT_COST

        cursor.execute(

            'INSERT INTO rate_limits (user_id, tokens, last_refill) VALUES (?, ?, ?)',

            (user_id, tokens, current_time)

        )

        conn.commit()

        conn.close()

        return (True, tokens)

    

    tokens, last_refill = result

    last_refill_dt = datetime.fromisoformat(last_refill)

    

    # Calculate tokens to add based on time elapsed

    time_elapsed = (current_time - last_refill_dt).total_seconds() / 60.0  # minutes

    tokens_to_add = time_elapsed * RATE_LIMIT_REFILL_RATE

    tokens = min(RATE_LIMIT_TOKENS, tokens + tokens_to_add)

    

    # Check if user has enough tokens

    if tokens >= RATE_LIMIT_COST:

        tokens -= RATE_LIMIT_COST

        cursor.execute(

            'UPDATE rate_limits SET tokens = ?, last_refill = ? WHERE user_id = ?',

            (tokens, current_time, user_id)

        )

        conn.commit()

        conn.close()

        return (True, tokens)

    else:

        conn.close()

        return (False, tokens)

This rate limiting implementation needs to be integrated into the API endpoints. We modify the chat completions endpoint to check rate limits before processing requests:

@app.route('/v1/chat/completions', methods=['POST'])

def chat_completions():

    """

    OpenAI-compatible chat completions endpoint with rate limiting.

    Requires valid API key in Authorization header.

    """

    # Extract and validate API key

    auth_header = request.headers.get('Authorization')

    

    if not auth_header or not auth_header.startswith('Bearer '):

        return jsonify({'error': 'Missing or invalid Authorization header'}), 401

    

    api_key = auth_header.replace('Bearer ', '')

    user_id = validate_api_key(api_key)

    

    if not user_id:

        return jsonify({'error': 'Invalid API key'}), 401

    

    # Check rate limit

    allowed, tokens_remaining = check_rate_limit(user_id)

    

    if not allowed:

        return jsonify({

            'error': 'Rate limit exceeded',

            'tokens_remaining': tokens_remaining,

            'retry_after': 60 / RATE_LIMIT_REFILL_RATE  # seconds until next token

        }), 429

    

    # Parse request data

    data = request.get_json()

    

    if not data:

        return jsonify({'error': 'Request body required'}), 400

    

    model_name = data.get('model')

    messages = data.get('messages', [])

    temperature = data.get('temperature', 0.7)

    max_tokens = data.get('max_tokens', 150)

    

    if not model_name:

        return jsonify({'error': 'Model name required'}), 400

    

    if not messages:

        return jsonify({'error': 'Messages required'}), 400

    

    # Get the model

    model_thread = model_manager.get_model(model_name)

    

    if not model_thread:

        return jsonify({

            'error': f'Model {model_name} not found',

            'available_models': model_manager.list_models()

        }), 404

    

    # Submit request to model thread

    try:

        status, result = model_thread.submit_request({

            'messages': messages,

            'temperature': temperature,

            'max_tokens': max_tokens

        })

        

        if status == 'error':

            return jsonify({'error': result}), 500

        

        # Format response in OpenAI format

        response = {

            'id': f"chatcmpl-{secrets.token_hex(12)}",

            'object': 'chat.completion',

            'created': int(datetime.now().timestamp()),

            'model': model_name,

            'choices': [

                {

                    'index': 0,

                    'message': {

                        'role': 'assistant',

                        'content': result['text']

                    },

                    'finish_reason': 'stop'

                }

            ],

            'usage': {

                'prompt_tokens': 0,

                'completion_tokens': result['tokens_used'],

                'total_tokens': result['tokens_used']

            }

        }

        

        # Add rate limit headers

        response_headers = {

            'X-RateLimit-Limit': str(RATE_LIMIT_TOKENS),

            'X-RateLimit-Remaining': str(int(tokens_remaining)),

            'X-RateLimit-Reset': str(int(current_time.timestamp() + (60 / RATE_LIMIT_REFILL_RATE)))

        }

        

        return jsonify(response), 200, response_headers

        

    except queue.Empty:

        return jsonify({'error': 'Request timeout'}), 504

    except Exception as e:

        return jsonify({'error': str(e)}), 500

IMPLEMENTING SERVER STARTUP

The following code fragment implements the server startup logic that initializes the database and loads models. This code runs when the script is executed directly (not imported as a module). It first calls init_database() to ensure the database schema exists, then adds models to the model manager, and finally starts the Flask server.

if __name__ == '__main__':

    # Initialize database

    init_database()

    

    # Load models

    print("Loading models...")

    

    # Add your models here

    model_manager.add_model(

        'llama2-7b',

        '/path/to/llama-2-7b.Q4_K_M.gguf'

    )

    

    model_manager.add_model(

        'mistral-7b',

        '/path/to/mistral-7b-instruct.Q4_K_M.gguf'

    )

    

    print("All models loaded")

    print(f"Available models: {model_manager.list_models()}")

    

    # Start Flask server

    print("Starting server on http://localhost:5000")

    app.run(host='0.0.0.0', port=5000, debug=False)

The model loading section should be customized with your actual model paths. Each add_model call specifies a model name and the path to the model file. The model name is what clients will use in their API requests, so choose names that are clear and consistent with your naming conventions.

The Flask server is started with host='0.0.0.0' to listen on all network interfaces, making it accessible from other machines. The port is set to 5000, which is Flask's default. The debug=False setting disables debug mode, which should always be disabled in production for security and performance reasons.

TESTING THE SERVER

Let me  provide a comprehensive testing script that demonstrates the complete user flow from registration through making API requests. This script is valuable for verifying that all components work together correctly and for understanding how clients should interact with the server.

import requests

import json

BASE_URL = 'http://localhost:5000'

# Step 1: Register a new user

print("Step 1: Registering user...")

response = requests.post(

    f'{BASE_URL}/register',

    json={

        'email': 'test@example.com',

        'password': 'SecurePassword123'

    }

)

print(f"Registration: {response.status_code} - {response.json()}")

# Step 2: Verify email (you need to get the token from your email or database)

print("\nStep 2: Verifying email...")

# In a real scenario, you'd click the link in the email

# For testing, you can get the token from the database

token = 'your_verification_token_here'

response = requests.get(f'{BASE_URL}/verify?token={token}')

print(f"Verification: {response.status_code} - {response.json()}")

# Step 3: Generate API key

print("\nStep 3: Generating API key...")

response = requests.post(

    f'{BASE_URL}/generate-api-key',

    json={

        'email': 'test@example.com',

        'password': 'SecurePassword123'

    }

)

print(f"API Key Generation: {response.status_code} - {response.json()}")

api_key = response.json().get('api_key')

# Step 4: List available models

print("\nStep 4: Listing models...")

response = requests.get(

    f'{BASE_URL}/v1/models',

    headers={'Authorization': f'Bearer {api_key}'}

)

print(f"Models: {response.status_code} - {json.dumps(response.json(), indent=2)}")

# Step 5: Make a chat completion request

print("\nStep 5: Making chat completion request...")

response = requests.post(

    f'{BASE_URL}/v1/chat/completions',

    headers={'Authorization': f'Bearer {api_key}'},

    json={

        'model': 'llama2-7b',

        'messages': [

            {'role': 'system', 'content': 'You are a helpful assistant.'},

            {'role': 'user', 'content': 'What is the capital of France?'}

        ],

        'temperature': 0.7,

        'max_tokens': 100

    }

)

print(f"Completion: {response.status_code} - {json.dumps(response.json(), indent=2)}")

The script performs five main steps: registering a new user, verifying the email (in practice, you'd get the token from your email), generating an API key, listing available models, and making a chat completion request. Each step includes error handling and prints the response for inspection.

For the email verification step, you'll need to retrieve the token from your email or directly from the database during testing. In a production environment, users would click the link in their email, but for automated testing, you can query the database to get the token.

The chat completion request demonstrates how to use the API with proper authentication. The Authorization header includes the API key in Bearer format, and the request body matches the OpenAI specification. The response is formatted as JSON and includes all the fields expected by OpenAI-compatible clients.

COMPLETE SERVER SOURCE CODE

Here is the complete, production-ready source code for the custom LLM server with all features integrated:

from flask import Flask, request, jsonify

import sqlite3

import hashlib

import secrets

import smtplib

from email.mime.text import MIMEText

from email.mime.multipart import MIMEMultipart

from datetime import datetime, timedelta

import re

import threading

import queue

from llama_cpp import Llama

import time

app = Flask(__name__)

# Configuration

SMTP_SERVER = 'smtp.example.com'

SMTP_PORT = 587

SMTP_USERNAME = 'your_email@example.com'

SMTP_PASSWORD = 'your_password'

SERVER_URL = 'http://localhost:5000'

# Rate limiting configuration

RATE_LIMIT_TOKENS = 100  # Maximum tokens in bucket

RATE_LIMIT_REFILL_RATE = 10  # Tokens per minute

RATE_LIMIT_COST = 1  # Tokens consumed per request

# Database functions

def get_db_connection():

    """

    Establish and return a connection to the SQLite database.

    Sets row_factory to sqlite3.Row for dict-like access to columns.

    """

    conn = sqlite3.connect('llm_server.db')

    conn.row_factory = sqlite3.Row

    return conn

def init_database():

    """

    Initialize the database with required tables.

    """

    conn = sqlite3.connect('llm_server.db')

    cursor = conn.cursor()

    

    # Create users table

    cursor.execute('''

        CREATE TABLE IF NOT EXISTS users (

            id INTEGER PRIMARY KEY AUTOINCREMENT,

            email TEXT UNIQUE NOT NULL,

            password_hash TEXT NOT NULL,

            is_verified INTEGER DEFAULT 0,

            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP

        )

    ''')

    

    # Create verification_tokens table

    cursor.execute('''

        CREATE TABLE IF NOT EXISTS verification_tokens (

            id INTEGER PRIMARY KEY AUTOINCREMENT,

            user_id INTEGER NOT NULL,

            token TEXT UNIQUE NOT NULL,

            expires_at TIMESTAMP NOT NULL,

            used INTEGER DEFAULT 0,

            FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE

        )

    ''')

    

    # Create api_keys table

    cursor.execute('''

        CREATE TABLE IF NOT EXISTS api_keys (

            id INTEGER PRIMARY KEY AUTOINCREMENT,

            user_id INTEGER NOT NULL,

            key_value TEXT UNIQUE NOT NULL,

            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,

            last_used_at TIMESTAMP,

            is_active INTEGER DEFAULT 1,

            FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE

        )

    ''')

    

    # Create rate_limits table

    cursor.execute('''

        CREATE TABLE IF NOT EXISTS rate_limits (

            user_id INTEGER PRIMARY KEY,

            tokens REAL NOT NULL,

            last_refill TIMESTAMP NOT NULL,

            FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE

        )

    ''')

    

    conn.commit()

    conn.close()

    

    print("Database initialized successfully")

# Validation functions

def is_valid_email(email):

    """

    Validate email format using a regular expression.

    Returns True if email matches standard email pattern.

    """

    pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'

    return re.match(pattern, email) is not None

# Password functions

def hash_password(password):

    """

    Hash a password using SHA-256 with a random salt.

    In production, use bcrypt or argon2 for better security.

    Returns a string in format: salt$hash

    """

    salt = secrets.token_hex(16)

    password_hash = hashlib.sha256((password + salt).encode()).hexdigest()

    return f"{salt}${password_hash}"

def verify_password(password, stored_hash):

    """

    Verify a password against a stored hash.

    Extracts the salt from stored_hash and recomputes the hash.

    Returns True if password matches.

    """

    try:

        salt, expected_hash = stored_hash.split('$')

        password_hash = hashlib.sha256((password + salt).encode()).hexdigest()

        return password_hash == expected_hash

    except ValueError:

        return False

# Email functions

def send_verification_email(email, token):

    """

    Send a verification email to the user with a clickable link.

    Returns True if email was sent successfully, False otherwise.

    """

    verification_link = f"{SERVER_URL}/verify?token={token}"

    

    message = MIMEMultipart('alternative')

    message['Subject'] = 'Verify your email address'

    message['From'] = SMTP_USERNAME

    message['To'] = email

    

    text_content = f"""

    Please verify your email address by clicking the following link:

    {verification_link}

    

    This link will expire in 24 hours.

    """

    

    html_content = f"""

    <html>

      <body>

        <p>Please verify your email address by clicking the link below:</p>

        <p><a href="{verification_link}">Verify Email Address</a></p>

        <p>This link will expire in 24 hours.</p>

      </body>

    </html>

    """

    

    part1 = MIMEText(text_content, 'plain')

    part2 = MIMEText(html_content, 'html')

    message.attach(part1)

    message.attach(part2)

    

    try:

        server = smtplib.SMTP(SMTP_SERVER, SMTP_PORT)

        server.starttls()

        server.login(SMTP_USERNAME, SMTP_PASSWORD)

        server.sendmail(SMTP_USERNAME, email, message.as_string())

        server.quit()

        return True

    except Exception as e:

        print(f"Failed to send email: {e}")

        return False

# API key validation

def validate_api_key(api_key):

    """

    Validate an API key and return the associated user_id.

    Returns user_id if valid, None otherwise.

    Also updates the last_used_at timestamp.

    """

    conn = get_db_connection()

    cursor = conn.cursor()

    

    cursor.execute(

        '''SELECT ak.user_id, ak.is_active, u.is_verified

           FROM api_keys ak

           JOIN users u ON ak.user_id = u.id

           WHERE ak.key_value = ?''',

        (api_key,)

    )

    result = cursor.fetchone()

    

    if not result:

        conn.close()

        return None

    

    user_id, is_active, is_verified = result

    

    if not is_active or not is_verified:

        conn.close()

        return None

    

    # Update last used timestamp

    cursor.execute(

        'UPDATE api_keys SET last_used_at = ? WHERE key_value = ?',

        (datetime.now(), api_key)

    )

    

    conn.commit()

    conn.close()

    

    return user_id

# Rate limiting

def check_rate_limit(user_id):

    """

    Check if a user has available rate limit tokens.

    Implements token bucket algorithm.

    

    Args:

        user_id: The user ID to check

    

    Returns:

        Tuple of (allowed, tokens_remaining)

    """

    conn = get_db_connection()

    cursor = conn.cursor()

    

    # Get current rate limit state

    cursor.execute(

        'SELECT tokens, last_refill FROM rate_limits WHERE user_id = ?',

        (user_id,)

    )

    result = cursor.fetchone()

    

    current_time = datetime.now()

    

    if result is None:

        # First request from this user - initialize bucket

        tokens = RATE_LIMIT_TOKENS - RATE_LIMIT_COST

        cursor.execute(

            'INSERT INTO rate_limits (user_id, tokens, last_refill) VALUES (?, ?, ?)',

            (user_id, tokens, current_time)

        )

        conn.commit()

        conn.close()

        return (True, tokens)

    

    tokens, last_refill = result

    last_refill_dt = datetime.fromisoformat(last_refill)

    

    # Calculate tokens to add based on time elapsed

    time_elapsed = (current_time - last_refill_dt).total_seconds() / 60.0  # minutes

    tokens_to_add = time_elapsed * RATE_LIMIT_REFILL_RATE

    tokens = min(RATE_LIMIT_TOKENS, tokens + tokens_to_add)

    

    # Check if user has enough tokens

    if tokens >= RATE_LIMIT_COST:

        tokens -= RATE_LIMIT_COST

        cursor.execute(

            'UPDATE rate_limits SET tokens = ?, last_refill = ? WHERE user_id = ?',

            (tokens, current_time, user_id)

        )

        conn.commit()

        conn.close()

        return (True, tokens)

    else:

        conn.close()

        return (False, tokens)

# Model thread class

class ModelThread:

    """

    A thread that loads a model and processes inference requests.

    """

    

    def __init__(self, model_name, model_path):

        """

        Initialize the model thread.

        

        Args:

            model_name: Name identifier for the model

            model_path: Path to the model file

        """

        self.model_name = model_name

        self.model_path = model_path

        self.request_queue = queue.Queue()

        self.model = None

        self.thread = threading.Thread(target=self._run, daemon=True)

        self.thread.start()

    

    def _run(self):

        """

        Main thread loop. Loads model and processes requests.

        """

        print(f"Loading model {self.model_name} from {self.model_path}")

        

        # Load the model

        self.model = Llama(

            model_path=self.model_path,

            n_ctx=2048,

            n_threads=4

        )

        

        print(f"Model {self.model_name} loaded successfully")

        

        # Process requests

        while True:

            request_data, response_queue = self.request_queue.get()

            

            try:

                result = self._process_request(request_data)

                response_queue.put(('success', result))

            except Exception as e:

                response_queue.put(('error', str(e)))

    

    def _process_request(self, request_data):

        """

        Process a single inference request.

        

        Args:

            request_data: Dictionary containing messages, temperature, max_tokens, etc.

        

        Returns:

            Dictionary with completion result

        """

        messages = request_data.get('messages', [])

        temperature = request_data.get('temperature', 0.7)

        max_tokens = request_data.get('max_tokens', 150)

        

        # Convert messages to prompt

        prompt = self._messages_to_prompt(messages)

        

        # Generate completion

        output = self.model(

            prompt,

            max_tokens=max_tokens,

            temperature=temperature,

            stop=["</s>", "User:", "Assistant:"]

        )

        

        return {

            'text': output['choices'][0]['text'],

            'tokens_used': output['usage']['total_tokens']

        }

    

    def _messages_to_prompt(self, messages):

        """

        Convert OpenAI-style messages to a prompt string.

        

        Args:

            messages: List of message dictionaries

        

        Returns:

            Formatted prompt string

        """

        prompt = ""

        

        for msg in messages:

            role = msg.get('role', 'user')

            content = msg.get('content', '')

            

            if role == 'system':

                prompt += f"System: {content}\n\n"

            elif role == 'user':

                prompt += f"User: {content}\n\n"

            elif role == 'assistant':

                prompt += f"Assistant: {content}\n\n"

        

        prompt += "Assistant: "

        return prompt

    

    def submit_request(self, request_data):

        """

        Submit a request to this model thread.

        

        Args:

            request_data: Dictionary containing request parameters

        

        Returns:

            Tuple of (status, result) where status is 'success' or 'error'

        """

        response_queue = queue.Queue()

        self.request_queue.put((request_data, response_queue))

        return response_queue.get(timeout=60)

# Model manager class

class ModelManager:

    """

    Manages multiple model threads and routes requests to them.

    """

    

    def __init__(self):

        """

        Initialize the model manager with an empty models dictionary.

        """

        self.models = {}

    

    def add_model(self, model_name, model_path):

        """

        Add a new model to the manager.

        

        Args:

            model_name: Name identifier for the model

            model_path: Path to the model file

        """

        if model_name in self.models:

            print(f"Model {model_name} already loaded")

            return

        

        self.models[model_name] = ModelThread(model_name, model_path)

    

    def get_model(self, model_name):

        """

        Get a model thread by name.

        

        Args:

            model_name: Name of the model

        

        Returns:

            ModelThread instance or None if not found

        """

        return self.models.get(model_name)

    

    def list_models(self):

        """

        Get a list of all available model names.

        

        Returns:

            List of model name strings

        """

        return list(self.models.keys())

# Global model manager instance

model_manager = ModelManager()

# API endpoints

@app.route('/register', methods=['POST'])

def register():

    """

    Handle user registration requests.

    Expects JSON with 'email' and 'password' fields.

    """

    data = request.get_json()

    

    if not data or 'email' not in data or 'password' not in data:

        return jsonify({'error': 'Email and password required'}), 400

    

    email = data['email'].lower().strip()

    password = data['password']

    

    # Validate email format

    if not is_valid_email(email):

        return jsonify({'error': 'Invalid email format'}), 400

    

    # Validate password strength

    if len(password) < 8:

        return jsonify({'error': 'Password must be at least 8 characters'}), 400

    

    conn = get_db_connection()

    cursor = conn.cursor()

    

    # Check if email already exists

    cursor.execute('SELECT id FROM users WHERE email = ?', (email,))

    if cursor.fetchone():

        conn.close()

        return jsonify({'error': 'Email already registered'}), 409

    

    # Hash password and create user

    password_hash = hash_password(password)

    cursor.execute(

        'INSERT INTO users (email, password_hash) VALUES (?, ?)',

        (email, password_hash)

    )

    user_id = cursor.lastrowid

    

    # Generate verification token

    token = secrets.token_urlsafe(32)

    expires_at = datetime.now() + timedelta(hours=24)

    

    cursor.execute(

        'INSERT INTO verification_tokens (user_id, token, expires_at) VALUES (?, ?, ?)',

        (user_id, token, expires_at)

    )

    

    conn.commit()

    conn.close()

    

    # Send verification email

    if send_verification_email(email, token):

        return jsonify({

            'message': 'Registration successful. Please check your email to verify your account.'

        }), 201

    else:

        return jsonify({

            'message': 'Registration successful but failed to send verification email.'

        }), 201

@app.route('/verify', methods=['GET'])

def verify_email():

    """

    Handle email verification via token in query parameter.

    """

    token = request.args.get('token')

    

    if not token:

        return jsonify({'error': 'Verification token required'}), 400

    

    conn = get_db_connection()

    cursor = conn.cursor()

    

    # Find the token

    cursor.execute(

        '''SELECT vt.id, vt.user_id, vt.expires_at, vt.used, u.is_verified

           FROM verification_tokens vt

           JOIN users u ON vt.user_id = u.id

           WHERE vt.token = ?''',

        (token,)

    )

    result = cursor.fetchone()

    

    if not result:

        conn.close()

        return jsonify({'error': 'Invalid verification token'}), 404

    

    token_id, user_id, expires_at, used, is_verified = result

    

    # Check if already used

    if used:

        conn.close()

        return jsonify({'error': 'Verification token already used'}), 400

    

    # Check if already verified

    if is_verified:

        conn.close()

        return jsonify({'message': 'Email already verified'}), 200

    

    # Check if expired

    expires_at_dt = datetime.fromisoformat(expires_at)

    if datetime.now() > expires_at_dt:

        conn.close()

        return jsonify({'error': 'Verification token expired'}), 400

    

    # Mark token as used and user as verified

    cursor.execute('UPDATE verification_tokens SET used = 1 WHERE id = ?', (token_id,))

    cursor.execute('UPDATE users SET is_verified = 1 WHERE id = ?', (user_id,))

    

    conn.commit()

    conn.close()

    

    return jsonify({'message': 'Email verified successfully'}), 200

@app.route('/generate-api-key', methods=['POST'])

def generate_api_key():

    """

    Generate a new API key for a verified user.

    Expects JSON with 'email' and 'password' for authentication.

    """

    data = request.get_json()

    

    if not data or 'email' not in data or 'password' not in data:

        return jsonify({'error': 'Email and password required'}), 400

    

    email = data['email'].lower().strip()

    password = data['password']

    

    conn = get_db_connection()

    cursor = conn.cursor()

    

    # Find user

    cursor.execute(

        'SELECT id, password_hash, is_verified FROM users WHERE email = ?',

        (email,)

    )

    result = cursor.fetchone()

    

    if not result:

        conn.close()

        return jsonify({'error': 'Invalid credentials'}), 401

    

    user_id, password_hash, is_verified = result

    

    # Verify password

    if not verify_password(password, password_hash):

        conn.close()

        return jsonify({'error': 'Invalid credentials'}), 401

    

    # Check if email is verified

    if not is_verified:

        conn.close()

        return jsonify({'error': 'Email not verified'}), 403

    

    # Generate API key

    api_key = f"sk-{secrets.token_urlsafe(32)}"

    

    cursor.execute(

        'INSERT INTO api_keys (user_id, key_value) VALUES (?, ?)',

        (user_id, api_key)

    )

    

    conn.commit()

    conn.close()

    

    return jsonify({'api_key': api_key}), 201

@app.route('/v1/chat/completions', methods=['POST'])

def chat_completions():

    """

    OpenAI-compatible chat completions endpoint with rate limiting.

    Requires valid API key in Authorization header.

    """

    # Extract and validate API key

    auth_header = request.headers.get('Authorization')

    

    if not auth_header or not auth_header.startswith('Bearer '):

        return jsonify({'error': 'Missing or invalid Authorization header'}), 401

    

    api_key = auth_header.replace('Bearer ', '')

    user_id = validate_api_key(api_key)

    

    if not user_id:

        return jsonify({'error': 'Invalid API key'}), 401

    

    # Check rate limit

    allowed, tokens_remaining = check_rate_limit(user_id)

    

    if not allowed:

        return jsonify({

            'error': 'Rate limit exceeded',

            'tokens_remaining': tokens_remaining,

            'retry_after': 60 / RATE_LIMIT_REFILL_RATE  # seconds until next token

        }), 429

    

    # Parse request data

    data = request.get_json()

    

    if not data:

        return jsonify({'error': 'Request body required'}), 400

    

    model_name = data.get('model')

    messages = data.get('messages', [])

    temperature = data.get('temperature', 0.7)

    max_tokens = data.get('max_tokens', 150)

    

    if not model_name:

        return jsonify({'error': 'Model name required'}), 400

    

    if not messages:

        return jsonify({'error': 'Messages required'}), 400

    

    # Get the model

    model_thread = model_manager.get_model(model_name)

    

    if not model_thread:

        return jsonify({

            'error': f'Model {model_name} not found',

            'available_models': model_manager.list_models()

        }), 404

    

    # Submit request to model thread

    try:

        status, result = model_thread.submit_request({

            'messages': messages,

            'temperature': temperature,

            'max_tokens': max_tokens

        })

        

        if status == 'error':

            return jsonify({'error': result}), 500

        

        # Format response in OpenAI format

        response = {

            'id': f"chatcmpl-{secrets.token_hex(12)}",

            'object': 'chat.completion',

            'created': int(datetime.now().timestamp()),

            'model': model_name,

            'choices': [

                {

                    'index': 0,

                    'message': {

                        'role': 'assistant',

                        'content': result['text']

                    },

                    'finish_reason': 'stop'

                }

            ],

            'usage': {

                'prompt_tokens': 0,

                'completion_tokens': result['tokens_used'],

                'total_tokens': result['tokens_used']

            }

        }

        

        # Add rate limit headers

        current_time = datetime.now()

        response_headers = {

            'X-RateLimit-Limit': str(RATE_LIMIT_TOKENS),

            'X-RateLimit-Remaining': str(int(tokens_remaining)),

            'X-RateLimit-Reset': str(int(current_time.timestamp() + (60 / RATE_LIMIT_REFILL_RATE)))

        }

        

        return jsonify(response), 200, response_headers

        

    except queue.Empty:

        return jsonify({'error': 'Request timeout'}), 504

    except Exception as e:

        return jsonify({'error': str(e)}), 500

@app.route('/v1/models', methods=['GET'])

def list_models():

    """

    OpenAI-compatible endpoint to list available models.

    Requires valid API key in Authorization header.

    """

    # Extract and validate API key

    auth_header = request.headers.get('Authorization')

    

    if not auth_header or not auth_header.startswith('Bearer '):

        return jsonify({'error': 'Missing or invalid Authorization header'}), 401

    

    api_key = auth_header.replace('Bearer ', '')

    user_id = validate_api_key(api_key)

    

    if not user_id:

        return jsonify({'error': 'Invalid API key'}), 401

    

    # Get list of models

    model_names = model_manager.list_models()

    

    # Format in OpenAI format

    models_data = []

    for name in model_names:

        models_data.append({

            'id': name,

            'object': 'model',

            'created': int(datetime.now().timestamp()),

            'owned_by': 'local'

        })

    

    response = {

        'object': 'list',

        'data': models_data

    }

    

    return jsonify(response), 200

# Server startup

if __name__ == '__main__':

    # Initialize database

    init_database()

    

    # Load models

    print("Loading models...")

    

    # Add your models here - replace with actual paths

    model_manager.add_model(

        'llama2-7b',

        '/path/to/llama-2-7b.Q4_K_M.gguf'

    )

    

    model_manager.add_model(

        'mistral-7b',

        '/path/to/mistral-7b-instruct.Q4_K_M.gguf'

    )

    

    print("All models loaded")

    print(f"Available models: {model_manager.list_models()}")

    

    # Start Flask server

    print("Starting server on http://localhost:5000")

    app.run(host='0.0.0.0', port=5000, debug=False)

CONCLUSION

This guide has presented two approaches to running local LLM servers with OpenAI-compatible APIs. The Ollama approach provides a quick and simple solution for personal use and experimentation, while the custom server implementation offers enterprise-grade features including user authentication, email verification, API key management, rate limiting, and multi-model support through threading.

The custom server architecture demonstrates how to build a production-ready system that can scale to support multiple users and models simultaneously. By maintaining OpenAI API compatibility, both solutions allow seamless integration with existing tools and applications, reducing the friction of adopting local LLM solutions.

Key takeaways include the importance of proper authentication and authorization in multi-user environments, the benefits of thread-based model loading for concurrent request handling, the necessity of rate limiting to prevent abuse and ensure fair resource allocation, and the value of adhering to established API standards for maximum compatibility.

The rate limiting implementation using the token bucket algorithm provides a fair and effective way to control resource usage while allowing burst traffic. Users receive clear feedback through rate limit headers and error messages, enabling them to adjust their usage patterns accordingly.

For production deployments, consider additional enhancements such as migrating from SQLite to PostgreSQL for better concurrency, implementing request logging and analytics for monitoring and debugging, adding support for streaming responses for real-time applications, implementing user roles and permissions for fine-grained access control, adding model-specific rate limits for high-demand models, implementing request queuing with priority levels, adding health check endpoints for monitoring, and implementing graceful shutdown to finish processing requests before stopping.

This server provides a solid foundation that can be extended and customized to meet specific organizational requirements while maintaining compatibility with the broader LLM ecosystem.