2574 Modules 349b6200

LLM Gateway Module Decomposition Design

Module Overview

The project is split into the following independent modules, each of which can be developed in parallel by different developers.

┌─────────────────────────────────────────────────────────────────────────────┐
│                              Module Dependency Graph                          │
└─────────────────────────────────────────────────────────────────────────────┘

                    ┌───────────────────┐
                    │   M1: Infrastructure│
                    │  (DB/Config/Common)│
                    └─────────┬─────────┘
                              │
          ┌───────────────────┼───────────────────┐
          │                   │                   │
          ▼                   ▼                   ▼
┌─────────────────┐  ┌─────────────────┐  ┌─────────────────┐
│ M2: Data Access  │  │ M3: Rule Engine │  │ M4: Upstream    │
│  Layer          │  │ (Rule Engine)   │  │ Adapters        │
│  (Repository)   │  │                 │  │ (Providers)     │
└────────┬────────┘  └────────┬────────┘  └────────┬────────┘
         │                    │                    │
         └────────────────────┼────────────────────┘
                              │
                              ▼
                    ┌─────────────────────┐
                    │   M5: Business      │
                    │    Service Layer    │
                    │    (Services)       │
                    └─────────┬───────────┘
                              │
              ┌───────────────┴───────────────┐
              │                               │
              ▼                               ▼
    ┌─────────────────┐             ┌─────────────────┐
    │ M6: Proxy API   │             │ M7: Admin API   │
    │ (Proxy Routes)  │             │ (Admin Routes)  │
    └─────────────────┘             └─────────────────┘

                              │
                              ▼
    ┌─────────────────────────────────────────────────────┐
    │                  M8: Frontend Admin Dashboard         │
    │  ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐│
    │  │ Provider │ │  Model   │ │ API Key  │ │ Log      ││
    │  │ Mgmt     │ │  Mgmt    │ │ Mgmt     │ │ Query    ││
    │  └──────────┘ └──────────┘ └──────────┘ └──────────┘│
    └─────────────────────────────────────────────────────┘

M1: Infrastructure Module

Module Responsibilities

• Database connection and session management
• Configuration management (Environment variables, multi-database switching)
• Common utility functions (Sanitization, Token counting, Timer, Error handling, etc.)

File Structure

backend/app/
├── config.py                 # Configuration management
├── db/
│   ├── __init__.py
│   ├── session.py            # Database session management
│   └── models.py             # ORM Model definitions
└── common/
    ├── __init__.py
    ├── http_client.py        # HTTP Client wrapper
    ├── token_counter.py      # Token Counter
    ├── sanitizer.py          # Data Sanitizer
    ├── errors.py             # Error Definitions
    ├── timer.py              # Timer
    └── utils.py              # Utility functions

Interface Definition

config.py

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    # Database Config
    DATABASE_TYPE: str = "sqlite"  # sqlite | postgresql
    DATABASE_URL: str = "sqlite:///./llm_gateway.db"
    
    # App Config
    APP_NAME: str = "LLM Gateway"
    DEBUG: bool = False
    
    # Retry Config
    RETRY_MAX_ATTEMPTS: int = 3
    RETRY_DELAY_MS: int = 1000

    class Config:
        env_file = ".env"

common/sanitizer.py

def sanitize_authorization(value: str) -> str:
    """Sanitize authorization field"""
    # Bearer sk-xxx...xxx -> Bearer sk-***...***
    pass

def sanitize_headers(headers: dict) -> dict:
    """Sanitize request headers"""
    pass

common/token_counter.py

from abc import ABC, abstractmethod

class TokenCounter(ABC):
    @abstractmethod
    def count_tokens(self, text: str, model: str) -> int:
        pass

class OpenAITokenCounter(TokenCounter):
    def count_tokens(self, text: str, model: str) -> int:
        # Use tiktoken for calculation
        pass

class AnthropicTokenCounter(TokenCounter):
    def count_tokens(self, text: str, model: str) -> int:
        pass

Test Points

• Configuration loading correctness (Environment variable priority)
• Database connection (SQLite/PostgreSQL switching)
• Sanitization functions (authorization field masking)
• Token counting accuracy

Estimated Effort

2-3 Days

M2: Data Access Layer Module

Module Responsibilities

• Define Repository abstract interfaces
• Implement SQLAlchemy concrete implementation
• Support SQLite and PostgreSQL

File Structure

backend/app/
├── domain/                        # Domain Models/DTO
│   ├── __init__.py
│   ├── provider.py
│   ├── model.py
│   ├── api_key.py
│   └── log.py
└── repositories/
    ├── __init__.py
    ├── base.py                    # Base Repository Interface
    ├── provider_repo.py           # Provider Repository Interface
    ├── model_repo.py              # Model Repository Interface
    ├── api_key_repo.py            # API Key Repository Interface
    ├── log_repo.py                # Log Repository Interface
    └── sqlalchemy/                # SQLAlchemy Implementation
        ├── __init__.py
        ├── provider_repo.py
        ├── model_repo.py
        ├── api_key_repo.py
        └── log_repo.py

Interface Definition

repositories/provider_repo.py

from abc import ABC, abstractmethod
from typing import Optional, List
from app.domain.provider import Provider, ProviderCreate, ProviderUpdate

class ProviderRepository(ABC):
    @abstractmethod
    async def create(self, data: ProviderCreate) -> Provider:
        pass
    
    @abstractmethod
    async def get_by_id(self, id: int) -> Optional[Provider]:
        pass
    
    @abstractmethod
    async def get_all(self, is_active: Optional[bool] = None) -> List[Provider]:
        pass
    
    @abstractmethod
    async def update(self, id: int, data: ProviderUpdate) -> Optional[Provider]:
        pass
    
    @abstractmethod
    async def delete(self, id: int) -> bool:
        pass

repositories/model_repo.py

from abc import ABC, abstractmethod
from typing import Optional, List
from app.domain.model import ModelMapping, ModelMappingProvider

class ModelRepository(ABC):
    @abstractmethod
    async def create_mapping(self, data: ModelMappingCreate) -> ModelMapping:
        pass
    
    @abstractmethod
    async def get_mapping(self, requested_model: str) -> Optional[ModelMapping]:
        pass
    
    @abstractmethod
    async def get_all_mappings(self) -> List[ModelMapping]:
        pass
    
    @abstractmethod
    async def add_provider_mapping(self, data: ModelMappingProviderCreate) -> ModelMappingProvider:
        pass
    
    @abstractmethod
    async def get_provider_mappings(self, requested_model: str) -> List[ModelMappingProvider]:
        pass

repositories/log_repo.py

from abc import ABC, abstractmethod
from typing import List, Optional
from datetime import datetime

class LogRepository(ABC):
    @abstractmethod
    async def create(self, data: LogCreate) -> RequestLog:
        pass
    
    @abstractmethod
    async def get_by_id(self, id: int) -> Optional[RequestLog]:
        pass
    
    @abstractmethod
    async def query(
        self,
        start_time: Optional[datetime] = None,
        end_time: Optional[datetime] = None,
        requested_model: Optional[str] = None,
        provider_id: Optional[int] = None,
        status_min: Optional[int] = None,
        status_max: Optional[int] = None,
        has_error: Optional[bool] = None,
        api_key_id: Optional[int] = None,
        page: int = 1,
        page_size: int = 20
    ) -> tuple[List[RequestLog], int]:
        pass

Test Points

• CRUD operations correctness
• Pagination query
• Multi-condition filtering
• Foreign key constraints
• SQLite and PostgreSQL compatibility

Estimated Effort

3-4 Days

M3: Rule Engine Module

Module Responsibilities

• Define rule context structure
• Implement rule evaluation logic
• Output candidate providers and their target models

File Structure

backend/app/rules/
├── __init__.py
├── engine.py              # Rule Engine Core
├── context.py             # Rule Context
├── evaluator.py           # Rule Evaluator
└── models.py              # Rule Model Definition

Interface Definition

rules/context.py

from dataclasses import dataclass
from typing import Dict, Any

@dataclass
class RuleContext:
    """Rule Engine Context"""
    current_model: str              # requested_model
    headers: Dict[str, str]         # Request headers
    request_body: Dict[str, Any]    # Request body
    token_usage: TokenUsage         # Token consumption

@dataclass
class TokenUsage:
    input_tokens: int
    output_tokens: int = 0

rules/models.py

from dataclasses import dataclass
from typing import List, Optional, Any

@dataclass
class Rule:
    """Rule Definition"""
    field: str              # Matching field (model, headers.x-custom, body.temperature)
    operator: str           # Operator (eq, ne, gt, lt, gte, lte, contains, regex)
    value: Any              # Matching value
    
@dataclass
class RuleSet:
    """Rule Set (AND Logic)"""
    rules: List[Rule]
    
@dataclass
class CandidateProvider:
    """Candidate Provider"""
    provider_id: int
    provider_name: str
    target_model: str
    priority: int
    weight: int

rules/engine.py

from typing import List
from app.rules.context import RuleContext
from app.rules.models import CandidateProvider

class RuleEngine:
    """Rule Engine"""
    
    async def evaluate(
        self,
        context: RuleContext,
        model_mapping: ModelMapping,
        provider_mappings: List[ModelMappingProvider]
    ) -> List[CandidateProvider]:
        """
        Evaluate all rules, return list of candidate providers
        
        Process:
        1. Check model-level rules (model_mapping.matching_rules)
        2. Check provider-level rules for each provider (provider_mapping.provider_rules)
        3. Return all passed providers and their target_model
        """
        pass

Rule Format Example

{
  "rules": [
    {"field": "headers.x-priority", "operator": "eq", "value": "high"},
    {"field": "body.temperature", "operator": "lte", "value": 0.5}
  ],
  "logic": "AND"
}

Test Points

• Various operators (eq, ne, gt, lt, contains, regex)
• Nested field access (headers.x-custom, body.messages[0].role)
• Multi-rule AND/OR combinations
• Empty rule handling (Default pass)
• No matching provider handling

Estimated Effort

2-3 Days

M4: Upstream Provider Adapter Module

Module Responsibilities

• Encapsulate upstream API calls
• Support OpenAI and Anthropic protocols
• Handle streaming responses

File Structure

backend/app/providers/
├── __init__.py
├── base.py                # Base Adapter Interface
├── openai_client.py       # OpenAI Client
└── anthropic_client.py    # Anthropic Client

Interface Definition

providers/base.py

from abc import ABC, abstractmethod
from typing import AsyncGenerator, Any
from dataclasses import dataclass

@dataclass
class ProviderResponse:
    status_code: int
    headers: dict
    body: Any
    first_byte_delay_ms: int
    total_time_ms: int

class ProviderClient(ABC):
    """Base class for upstream provider client"""
    
    @abstractmethod
    async def forward(
        self,
        base_url: str,
        api_key: str,
        path: str,
        method: str,
        headers: dict,
        body: dict,
        target_model: str
    ) -> ProviderResponse:
        """Forward request to upstream provider"""
        pass
    
    @abstractmethod
    async def forward_stream(
        self,
        base_url: str,
        api_key: str,
        path: str,
        method: str,
        headers: dict,
        body: dict,
        target_model: str
    ) -> AsyncGenerator[bytes, None]:
        """Forward streaming request"""
        pass

providers/openai_client.py

class OpenAIClient(ProviderClient):
    """OpenAI Protocol Client"""
    
    async def forward(self, ...) -> ProviderResponse:
        # 1. Copy body, replace model field
        # 2. Forward to base_url + path
        # 3. Record latency metrics
        pass

Test Points

• Request forwarding correctness
• Only modify model field verification
• Streaming response handling
• Error response handling
• Timeout handling

Estimated Effort

2-3 Days

M5: Business Service Layer Module

Module Responsibilities

• Proxy core logic orchestration
• Retry and failover
• Round Robin strategy implementation
• Log recording service

File Structure

backend/app/services/
├── __init__.py
├── proxy_service.py       # Proxy Core Service
├── provider_service.py    # Provider Management Service
├── model_service.py       # Model Management Service
├── api_key_service.py     # API Key Service
├── log_service.py         # Log Service
├── retry_handler.py       # Retry Handler
└── strategy.py            # Strategy Service (Round Robin)

Interface Definition

services/proxy_service.py

class ProxyService:
    """Proxy Core Service"""
    
    def __init__(
        self,
        model_repo: ModelRepository,
        provider_repo: ProviderRepository,
        log_repo: LogRepository,
        rule_engine: RuleEngine,
        strategy: SelectionStrategy,
        retry_handler: RetryHandler,
        token_counter: TokenCounter
    ):
        pass
    
    async def process_request(
        self,
        api_key_id: int,
        api_key_name: str,
        protocol: str,
        path: str,
        method: str,
        headers: dict,
        body: dict
    ) -> ProxyResponse:
        """
        Process proxy request
        
        Process:
        1. Extract requested_model
        2. Calculate Input Token
        3. Build Rule Context
        4. Rule Engine Match -> Candidate Provider List
        5. Round Robin Strategy selects Provider
        6. Forward Request (with Retry/Failover logic)
        7. Calculate Output Token
        8. Log Request
        9. Return Response
        """
        pass

services/retry_handler.py

class RetryHandler:
    """Retry and Failover Handler"""
    
    async def execute_with_retry(
        self,
        candidates: List[CandidateProvider],
        forward_fn: Callable,
        **kwargs
    ) -> tuple[ProviderResponse, int, CandidateProvider]:
        """
        Execute request with retry
        
        Logic:
        - status >= 500: Retry on same provider 3 times, 1s interval
        - status < 500: Switch directly to next provider
        - All failed: Return last error
        
        Returns:
            (Response, Retry Count, Final Provider Used)
        """
        pass

services/strategy.py

from abc import ABC, abstractmethod

class SelectionStrategy(ABC):
    """Provider Selection Strategy"""
    
    @abstractmethod
    async def select(
        self,
        candidates: List[CandidateProvider],
        requested_model: str
    ) -> CandidateProvider:
        pass

class RoundRobinStrategy(SelectionStrategy):
    """Round Robin Strategy"""
    
    async def select(self, candidates, requested_model) -> CandidateProvider:
        # Use atomic counter to implement concurrency-safe round robin
        pass

Test Points

• Complete proxy flow
• Retry logic (>=500 Retry on same provider 3 times)
• Switch logic (<500 Switch directly)
• Round Robin strategy correctness and concurrency safety
• Log recording completeness

Estimated Effort

4-5 Days

M6: Proxy API Module

Module Responsibilities

• OpenAI Compatible Interface
• Anthropic Compatible Interface
• API Key Authentication

File Structure

backend/app/api/
├── __init__.py
├── deps.py                # Dependency Injection
└── proxy/
    ├── __init__.py
    ├── openai.py          # OpenAI Compatible Interface
    └── anthropic.py       # Anthropic Compatible Interface

Interface Definition

api/proxy/openai.py

from fastapi import APIRouter, Depends, Request, Header
from app.api.deps import get_api_key, get_proxy_service

router = APIRouter()

@router.post("/v1/chat/completions")
async def chat_completions(
    request: Request,
    authorization: str = Header(...),
    proxy_service: ProxyService = Depends(get_proxy_service)
):
    """OpenAI Chat Completions Proxy Interface"""
    pass

@router.post("/v1/completions")
async def completions(request: Request, ...):
    """OpenAI Completions Proxy Interface"""
    pass

@router.post("/v1/embeddings")
async def embeddings(request: Request, ...):
    """OpenAI Embeddings Proxy Interface"""
    pass

api/proxy/anthropic.py

router = APIRouter()

@router.post("/v1/messages")
async def messages(request: Request, ...):
    """Anthropic Messages Proxy Interface"""
    pass

Test Points

• API Key Authentication
• Request Parsing
• Response Format Correctness
• Streaming Response
• Error Handling

Estimated Effort

2-3 Days

M7: Admin API Module

Module Responsibilities

• Provider CRUD API
• Model Mapping CRUD API
• API Key CRUD API
• Log Query API

File Structure

backend/app/api/admin/
├── __init__.py
├── providers.py           # Provider Management
├── models.py              # Model Management
├── api_keys.py            # API Key Management
└── logs.py                # Log Query

API Details see API Documentation below

Test Points

• CRUD Operations
• Parameter Validation
• Pagination and Filtering
• Error Handling

Estimated Effort

2-3 Days

M8: Frontend Admin Dashboard Module

Sub-module Breakdown

M8.1: Basic Framework and Common Components

• Project Initialization (Next.js + TypeScript)
• UI Component Library Integration (shadcn/ui)
• Common Component Development
• API Client Wrapper

Estimated Effort: 2-3 Days

M8.2: Provider Management Page

• Provider List
• Add/Edit Form
• Delete Confirmation

Estimated Effort: 1-2 Days

M8.3: Model Management Page

• Model Mapping List
• Model-Provider Mapping Configuration
• Rule Editor

Estimated Effort: 3-4 Days

M8.4: API Key Management Page

• API Key List
• Add (key_value generated by backend)
• State Management

Estimated Effort: 1-2 Days

M8.5: Log Query Page

• Log List (Pagination, Sorting)
• Multi-condition Filter
• Log Detail Page

Estimated Effort: 2-3 Days

Development Sequence Suggestion

Phase 1 (Parallel):
├── M1: Infrastructure (Developer A)
├── M3: Rule Engine (Developer B)
└── M8.1: Frontend Basic Framework (Developer C)

Phase 2 (Parallel, Depends on M1):
├── M2: Data Access Layer (Developer A)
├── M4: Upstream Adapter (Developer B)
└── M8.2: Provider Management Page (Developer C)

Phase 3 (Parallel, Depends on M2, M3, M4):
├── M5: Business Service Layer (Developer A)
├── M7: Admin API (Developer B)
└── M8.3: Model Management Page (Developer C)

Phase 4 (Parallel, Depends on M5):
├── M6: Proxy API (Developer A)
├── M8.4: API Key Management Page (Developer B)
└── M8.5: Log Query Page (Developer C)

Phase 5:
└── Integration Test and Fixes

Total Estimated Effort

Parallel Development (3 people): Approx 12-18 Days