# zzrouter 模型代理服务实现计划 > **For agentic workers:** REQUIRED: Use superpowers:subagent-driven-development (if subagents available) or superpowers:executing-plans to implement this plan. Steps use checkbox (`- [ ]`) syntax for tracking. **Goal:** 构建一个 AI 模型 API 代理服务,支持 OpenAI 和 Anthropic 提供商,提供统一 API 接口、Key 轮换和请求日志。 **Architecture:** FastAPI + SQLite + LiteLLM。客户端通过 Bearer Token 认证,服务根据模型名路由到对应提供商,轮询选择 API Key,异步批量记录请求日志。 **Tech Stack:** Python 3.13, FastAPI, LiteLLM, aiosqlite, uvicorn --- ## Chunk 1: 项目初始化与数据库层 ### Task 1: 更新依赖配置 **Files:** - Modify: `backend/pyproject.toml` - [ ] **Step 1: 添加新依赖** ```toml [project] name = "zzrouter-backend" version = "0.1.0" description = "AI Model API Proxy Service" readme = "README.md" requires-python = ">=3.13" dependencies = [ "fastapi>=0.135.1", "uvicorn[standard]>=0.41.0", "litellm>=1.50.0", "aiosqlite>=0.20.0", "pydantic>=2.0.0", ] [project.scripts] zzrouter = "app.main:run_server" ``` - [ ] **Step 2: 安装依赖** Run: `cd backend && uv sync` Expected: 依赖安装成功 - [ ] **Step 3: Commit** ```bash cd backend && git add pyproject.toml uv.lock && git commit -m "chore: add litellm and aiosqlite dependencies" ``` --- ### Task 2: 数据库连接模块 **Files:** - Create: `backend/app/database.py` - [ ] **Step 1: 编写数据库连接模块** ```python """Database connection and initialization.""" import aiosqlite from pathlib import Path from contextlib import asynccontextmanager DB_PATH = Path(__file__).parent.parent / "data" / "zzrouter.db" async def get_db() -> aiosqlite.Connection: """Get database connection.""" DB_PATH.parent.mkdir(parents=True, exist_ok=True) conn = await aiosqlite.connect(DB_PATH) conn.row_factory = aiosqlite.Row return conn @asynccontextmanager async def db_connection(): """Context manager for database connection.""" conn = await get_db() try: yield conn finally: await conn.close() async def init_db(): """Initialize database schema.""" async with db_connection() as db: await db.executescript(""" CREATE TABLE IF NOT EXISTS client_keys ( id INTEGER PRIMARY KEY AUTOINCREMENT, key TEXT UNIQUE NOT NULL, name TEXT, is_active BOOLEAN DEFAULT TRUE, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS providers ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT UNIQUE NOT NULL, base_url TEXT NOT NULL, api_type TEXT NOT NULL, is_active BOOLEAN DEFAULT TRUE ); CREATE TABLE IF NOT EXISTS provider_keys ( id INTEGER PRIMARY KEY AUTOINCREMENT, provider_id INTEGER NOT NULL REFERENCES providers(id), key TEXT NOT NULL, is_active BOOLEAN DEFAULT TRUE, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS request_logs ( id INTEGER PRIMARY KEY AUTOINCREMENT, client_key_id INTEGER REFERENCES client_keys(id), provider_id INTEGER REFERENCES providers(id), model TEXT, prompt_tokens INTEGER, completion_tokens INTEGER, latency_ms INTEGER, success BOOLEAN, error_message TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); CREATE INDEX IF NOT EXISTS idx_request_logs_created_at ON request_logs(created_at); CREATE INDEX IF NOT EXISTS idx_request_logs_client_key_id ON request_logs(client_key_id); """) await db.commit() async def seed_default_providers(): """Seed default providers if not exist.""" async with db_connection() as db: await db.executemany( """INSERT OR IGNORE INTO providers (name, base_url, api_type) VALUES (?, ?, ?)""", [ ("openai", "https://api.openai.com/v1", "openai"), ("anthropic", "https://api.anthropic.com/v1", "anthropic"), ] ) await db.commit() ``` - [ ] **Step 2: Commit** ```bash cd backend && git add app/database.py && git commit -m "feat: add database connection and schema initialization" ``` --- ### Task 3: 数据库模型 **Files:** - Create: `backend/app/models/__init__.py` - Create: `backend/app/models/client_key.py` - Create: `backend/app/models/provider.py` - Create: `backend/app/models/provider_key.py` - Create: `backend/app/models/request_log.py` - [ ] **Step 1: 创建 models/__init__.py** ```python """Database models.""" from .client_key import ClientKey from .provider import Provider from .provider_key import ProviderKey from .request_log import RequestLog __all__ = ["ClientKey", "Provider", "ProviderKey", "RequestLog"] ``` - [ ] **Step 2: 创建 client_key.py** ```python """Client API Key model.""" from dataclasses import dataclass from typing import Optional @dataclass class ClientKey: """Client API key for authentication.""" id: int key: str name: Optional[str] is_active: bool created_at: str ``` - [ ] **Step 3: 创建 provider.py** ```python """Provider model.""" from dataclasses import dataclass from typing import Optional @dataclass class Provider: """Model provider configuration.""" id: int name: str base_url: str api_type: str is_active: bool ``` - [ ] **Step 4: 创建 provider_key.py** ```python """Provider API Key model.""" from dataclasses import dataclass from typing import Optional @dataclass class ProviderKey: """API key for a provider.""" id: int provider_id: int key: str is_active: bool created_at: str ``` - [ ] **Step 5: 创建 request_log.py** ```python """Request log model.""" from dataclasses import dataclass from typing import Optional @dataclass class RequestLog: """Request log entry.""" id: Optional[int] client_key_id: int provider_id: int model: str prompt_tokens: Optional[int] completion_tokens: Optional[int] latency_ms: Optional[int] success: bool error_message: Optional[str] created_at: Optional[str] ``` - [ ] **Step 6: Commit** ```bash cd backend && git add app/models/ && git commit -m "feat: add database models" ``` --- ## Chunk 2: 核心服务层 ### Task 4: 配置模块 **Files:** - Create: `backend/app/config.py` - [ ] **Step 1: 创建配置模块** ```python """Application configuration.""" from pathlib import Path # Database DB_PATH = Path(__file__).parent.parent / "data" / "zzrouter.db" # Server HOST = "0.0.0.0" PORT = 8000 # Logging LOG_BATCH_SIZE = 100 LOG_FLUSH_INTERVAL = 1.0 # seconds # Model routing MODEL_PROVIDER_MAP = { "gpt": "openai", "o1": "openai", "o3": "openai", "claude": "anthropic", } ``` - [ ] **Step 2: Commit** ```bash cd backend && git add app/config.py && git commit -m "feat: add configuration module" ``` --- ### Task 5: 认证服务 **Files:** - Create: `backend/app/services/__init__.py` - Create: `backend/app/services/auth.py` - [ ] **Step 1: 创建 services/__init__.py** ```python """Business logic services.""" from .auth import AuthService from .key_selector import KeySelector from .router import ModelRouter from .logger import AsyncLogger __all__ = ["AuthService", "KeySelector", "ModelRouter", "AsyncLogger"] ``` - [ ] **Step 2: 创建 auth.py** ```python """Authentication service.""" import aiosqlite from typing import Optional from ..database import db_connection from ..models import ClientKey class AuthService: """Service for client key authentication.""" @staticmethod async def validate_key(api_key: str) -> Optional[ClientKey]: """Validate a client API key. Args: api_key: The API key to validate Returns: ClientKey if valid and active, None otherwise """ async with db_connection() as db: cursor = await db.execute( """SELECT id, key, name, is_active, created_at FROM client_keys WHERE key = ?""", (api_key,) ) row = await cursor.fetchone() if row and row["is_active"]: return ClientKey( id=row["id"], key=row["key"], name=row["name"], is_active=row["is_active"], created_at=row["created_at"] ) return None ``` - [ ] **Step 3: Commit** ```bash cd backend && git add app/services/ && git commit -m "feat: add authentication service" ``` --- ### Task 6: Key 轮询选择器 **Files:** - Create: `backend/app/services/key_selector.py` - [ ] **Step 1: 创建 key_selector.py** ```python """Provider key selector with round-robin strategy.""" import asyncio from typing import Optional, Dict from ..database import db_connection from ..models import ProviderKey class KeySelector: """Round-robin key selector for provider API keys.""" def __init__(self): self._indices: Dict[int, int] = {} # provider_id -> current index self._lock = asyncio.Lock() async def get_next_key(self, provider_id: int) -> Optional[ProviderKey]: """Get the next available API key for a provider. Args: provider_id: The provider ID Returns: ProviderKey if available, None if no keys configured """ async with self._lock: async with db_connection() as db: cursor = await db.execute( """SELECT id, provider_id, key, is_active, created_at FROM provider_keys WHERE provider_id = ? AND is_active = TRUE""", (provider_id,) ) keys = await cursor.fetchall() if not keys: return None # Get or initialize index if provider_id not in self._indices: self._indices[provider_id] = 0 # Round-robin selection index = self._indices[provider_id] % len(keys) self._indices[provider_id] = index + 1 row = keys[index] return ProviderKey( id=row["id"], provider_id=row["provider_id"], key=row["key"], is_active=row["is_active"], created_at=row["created_at"] ) def reset_index(self, provider_id: int): """Reset the round-robin index for a provider.""" self._indices.pop(provider_id, None) # Global instance key_selector = KeySelector() ``` - [ ] **Step 2: Commit** ```bash cd backend && git add app/services/key_selector.py && git commit -m "feat: add round-robin key selector" ``` --- ### Task 7: 模型路由服务 **Files:** - Create: `backend/app/services/router.py` - [ ] **Step 1: 创建 router.py** ```python """Model to provider router.""" from typing import Optional, Tuple from ..database import db_connection from ..models import Provider from ..config import MODEL_PROVIDER_MAP class ModelRouter: """Routes model names to providers.""" @staticmethod def get_provider_name(model: str) -> Optional[str]: """Determine provider name from model name. Args: model: The model name (e.g., 'gpt-4o', 'claude-3-opus') Returns: Provider name or None if unknown """ model_lower = model.lower() for prefix, provider in MODEL_PROVIDER_MAP.items(): if model_lower.startswith(prefix): return provider return None @staticmethod async def get_provider(provider_name: str) -> Optional[Provider]: """Get provider configuration by name. Args: provider_name: The provider name (e.g., 'openai', 'anthropic') Returns: Provider if found and active, None otherwise """ async with db_connection() as db: cursor = await db.execute( """SELECT id, name, base_url, api_type, is_active FROM providers WHERE name = ? AND is_active = TRUE""", (provider_name,) ) row = await cursor.fetchone() if row: return Provider( id=row["id"], name=row["name"], base_url=row["base_url"], api_type=row["api_type"], is_active=row["is_active"] ) return None @staticmethod async def route_model(model: str) -> Tuple[Optional[Provider], Optional[str]]: """Route a model to its provider. Args: model: The model name Returns: Tuple of (Provider, error_message) """ provider_name = ModelRouter.get_provider_name(model) if not provider_name: return None, f"Unsupported model: {model}" provider = await ModelRouter.get_provider(provider_name) if not provider: return None, f"Provider not found: {provider_name}" return provider, None # Global instance model_router = ModelRouter() ``` - [ ] **Step 2: Commit** ```bash cd backend && git add app/services/router.py && git commit -m "feat: add model routing service" ``` --- ### Task 8: 异步日志服务 **Files:** - Create: `backend/app/services/logger.py` - [ ] **Step 1: 创建 logger.py** ```python """Async batch logger for request logs.""" import asyncio from typing import Optional, List, Dict, Any from dataclasses import dataclass, asdict from ..database import db_connection from ..config import LOG_BATCH_SIZE, LOG_FLUSH_INTERVAL @dataclass class LogEntry: """A single log entry.""" client_key_id: int provider_id: int model: str prompt_tokens: Optional[int] = None completion_tokens: Optional[int] = None latency_ms: Optional[int] = None success: bool = True error_message: Optional[str] = None class AsyncLogger: """Asynchronous batch logger for request logs.""" def __init__(self): self._queue: asyncio.Queue[LogEntry] = asyncio.Queue() self._task: Optional[asyncio.Task] = None async def start(self): """Start the background flush task.""" if self._task is None: self._task = asyncio.create_task(self._flush_loop()) async def stop(self): """Stop the background flush task.""" if self._task: self._task.cancel() try: await self._task except asyncio.CancelledError: pass self._task = None # Flush remaining logs await self._flush_all() async def log(self, entry: LogEntry): """Queue a log entry.""" await self._queue.put(entry) async def _flush_loop(self): """Background loop to flush logs periodically.""" while True: try: await asyncio.sleep(LOG_FLUSH_INTERVAL) await self._flush_batch() except asyncio.CancelledError: raise async def _flush_batch(self): """Flush a batch of logs to database.""" entries: List[LogEntry] = [] # Collect up to batch_size entries while len(entries) < LOG_BATCH_SIZE: try: entry = self._queue.get_nowait() entries.append(entry) except asyncio.QueueEmpty: break if not entries: return # Insert batch async with db_connection() as db: await db.executemany( """INSERT INTO request_logs (client_key_id, provider_id, model, prompt_tokens, completion_tokens, latency_ms, success, error_message) VALUES (?, ?, ?, ?, ?, ?, ?, ?)""", [ ( e.client_key_id, e.provider_id, e.model, e.prompt_tokens, e.completion_tokens, e.latency_ms, e.success, e.error_message ) for e in entries ] ) await db.commit() async def _flush_all(self): """Flush all remaining logs.""" while not self._queue.empty(): await self._flush_batch() # Global instance async_logger = AsyncLogger() ``` - [ ] **Step 2: Commit** ```bash cd backend && git add app/services/logger.py && git commit -m "feat: add async batch logger" ``` --- ## Chunk 3: 提供商适配器与路由 ### Task 9: LiteLLM 封装器 **Files:** - Create: `backend/app/providers/__init__.py` - Create: `backend/app/providers/litellm_wrapper.py` - [ ] **Step 1: 创建 providers/__init__.py** ```python """Provider adapters.""" from .litellm_wrapper import LiteLLMWrapper __all__ = ["LiteLLMWrapper"] ``` - [ ] **Step 2: 创建 litellm_wrapper.py** ```python """LiteLLM wrapper for unified model access.""" import time from typing import AsyncIterator, Optional, Dict, Any, List import litellm from litellm import acompletion from ..models import Provider from ..services.logger import async_logger, LogEntry class LiteLLMWrapper: """Wrapper around LiteLLM for proxy functionality.""" @staticmethod def _build_model_string(provider: Provider, model: str) -> str: """Build LiteLLM model string. Args: provider: The provider configuration model: The model name Returns: LiteLLM-compatible model string """ if provider.api_type == "anthropic": return f"anthropic/{model}" else: return f"openai/{model}" @staticmethod async def chat_completion( provider: Provider, api_key: str, model: str, messages: List[Dict[str, Any]], stream: bool = False, **kwargs ) -> tuple[Any, int, int]: """Execute a chat completion request. Args: provider: The provider configuration api_key: The API key to use model: The model name messages: The chat messages stream: Whether to stream the response **kwargs: Additional parameters Returns: Tuple of (response, prompt_tokens, completion_tokens) """ model_str = LiteLLMWrapper._build_model_string(provider, model) # Set API key for this request api_base = provider.base_url if provider.api_type == "openai" else None start_time = time.time() try: response = await acompletion( model=model_str, messages=messages, api_key=api_key, api_base=api_base, stream=stream, **kwargs ) if stream: # For streaming, return the async iterator # Token counts will be collected during iteration return response, 0, 0 else: # Extract token usage from response usage = getattr(response, "usage", None) or {} prompt_tokens = getattr(usage, "prompt_tokens", 0) or 0 completion_tokens = getattr(usage, "completion_tokens", 0) or 0 return response, prompt_tokens, completion_tokens except Exception as e: # Re-raise with context raise @staticmethod async def stream_response( stream_iterator: AsyncIterator, client_key_id: int, provider_id: int, model: str, start_time: float ) -> AsyncIterator[str]: """Process streaming response and yield SSE-formatted data. Args: stream_iterator: The LiteLLM stream iterator client_key_id: Client key ID for logging provider_id: Provider ID for logging model: The model name start_time: Request start time Yields: SSE-formatted strings """ prompt_tokens = 0 completion_tokens = 0 try: async for chunk in stream_iterator: # Extract token counts if available if hasattr(chunk, "usage") and chunk.usage: prompt_tokens = getattr(chunk.usage, "prompt_tokens", 0) or 0 completion_tokens = getattr(chunk.usage, "completion_tokens", 0) or 0 # Convert to SSE format if hasattr(chunk, "model_dump"): chunk_dict = chunk.model_dump() else: chunk_dict = chunk import json yield f"data: {json.dumps(chunk_dict)}\n\n" yield "data: [DONE]\n\n" # Log successful request latency_ms = int((time.time() - start_time) * 1000) await async_logger.log(LogEntry( client_key_id=client_key_id, provider_id=provider_id, model=model, prompt_tokens=prompt_tokens, completion_tokens=completion_tokens, latency_ms=latency_ms, success=True )) except Exception as e: # Log failed request latency_ms = int((time.time() - start_time) * 1000) await async_logger.log(LogEntry( client_key_id=client_key_id, provider_id=provider_id, model=model, prompt_tokens=prompt_tokens, completion_tokens=completion_tokens, latency_ms=latency_ms, success=False, error_message=str(e) )) # Send error as SSE import json yield f"data: {json.dumps({'error': str(e)})}\n\n" ``` - [ ] **Step 3: Commit** ```bash cd backend && git add app/providers/ && git commit -m "feat: add LiteLLM wrapper for unified model access" ``` --- ### Task 10: 健康检查路由 **Files:** - Create: `backend/app/routes/__init__.py` - Create: `backend/app/routes/health.py` - [ ] **Step 1: 创建 routes/__init__.py** ```python """API routes.""" from fastapi import APIRouter from .health import router as health_router from .chat import router as chat_router from .openai import router as openai_router from .anthropic import router as anthropic_router __all__ = ["health_router", "chat_router", "openai_router", "anthropic_router"] ``` - [ ] **Step 2: 创建 health.py** ```python """Health check endpoint.""" from fastapi import APIRouter router = APIRouter(tags=["health"]) @router.get("/health") async def health_check(): """Health check endpoint.""" return {"status": "ok"} ``` - [ ] **Step 3: Commit** ```bash cd backend && git add app/routes/health.py app/routes/__init__.py && git commit -m "feat: add health check endpoint" ``` --- ### Task 11: 认证中间件 **Files:** - Create: `backend/app/middleware/__init__.py` - Create: `backend/app/middleware/auth.py` - [ ] **Step 1: 创建 middleware/__init__.py** ```python """Middleware components.""" from .auth import AuthMiddleware, get_current_client __all__ = ["AuthMiddleware", "get_current_client"] ``` - [ ] **Step 2: 创建 auth.py** ```python """Authentication middleware.""" from typing import Optional from fastapi import Depends, HTTPException, status from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials from ..models import ClientKey from ..services.auth import AuthService security = HTTPBearer(auto_error=False) async def get_current_client( credentials: Optional[HTTPAuthorizationCredentials] = Depends(security) ) -> ClientKey: """Dependency to get and validate the current client. Raises: HTTPException: If authentication fails Returns: The validated ClientKey """ if not credentials: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail={"error": "Missing API key"} ) client_key = await AuthService.validate_key(credentials.credentials) if not client_key: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail={"error": "Invalid API key"} ) if not client_key.is_active: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail={"error": "API key disabled"} ) return client_key class AuthMiddleware: """Authentication middleware for request processing.""" def __init__(self, exclude_paths: list[str] = None): self.exclude_paths = exclude_paths or ["/health", "/docs", "/openapi.json"] async def __call__(self, request, call_next): # Middleware is handled via Depends in routes return await call_next(request) ``` - [ ] **Step 3: Commit** ```bash cd backend && git add app/middleware/ && git commit -m "feat: add authentication middleware" ``` --- ### Task 12: OpenAI 兼容聊天路由 **Files:** - Create: `backend/app/routes/chat.py` - [ ] **Step 1: 创建 chat.py** ```python """OpenAI-compatible chat completion endpoint.""" import time from typing import Optional, List, Dict, Any from fastapi import APIRouter, Depends, HTTPException from fastapi.responses import StreamingResponse from pydantic import BaseModel from ..models import ClientKey, Provider from ..middleware.auth import get_current_client from ..services.router import model_router from ..services.key_selector import key_selector from ..services.logger import async_logger, LogEntry from ..providers.litellm_wrapper import LiteLLMWrapper router = APIRouter(prefix="/v1", tags=["chat"]) class ChatMessage(BaseModel): """Chat message.""" role: str content: str class ChatCompletionRequest(BaseModel): """Chat completion request.""" model: str messages: List[ChatMessage] stream: bool = False temperature: Optional[float] = None max_tokens: Optional[int] = None top_p: Optional[float] = None class ChatCompletionResponse(BaseModel): """Chat completion response.""" id: str object: str = "chat.completion" created: int model: str choices: List[Dict[str, Any]] usage: Dict[str, int] @router.post("/chat/completions") async def chat_completions( request: ChatCompletionRequest, client: ClientKey = Depends(get_current_client) ): """OpenAI-compatible chat completions endpoint.""" start_time = time.time() # Route model to provider provider, error = await model_router.route_model(request.model) if error: raise HTTPException(status_code=400, detail={"error": error}) # Get next API key provider_key = await key_selector.get_next_key(provider.id) if not provider_key: raise HTTPException( status_code=503, detail={"error": f"No available API key for provider: {provider.name}"} ) # Build messages messages = [{"role": m.role, "content": m.content} for m in request.messages] # Build kwargs kwargs = {} if request.temperature is not None: kwargs["temperature"] = request.temperature if request.max_tokens is not None: kwargs["max_tokens"] = request.max_tokens if request.top_p is not None: kwargs["top_p"] = request.top_p try: response, prompt_tokens, completion_tokens = await LiteLLMWrapper.chat_completion( provider=provider, api_key=provider_key.key, model=request.model, messages=messages, stream=request.stream, **kwargs ) if request.stream: return StreamingResponse( LiteLLMWrapper.stream_response( response, client.id, provider.id, request.model, start_time ), media_type="text/event-stream" ) else: # Log successful request latency_ms = int((time.time() - start_time) * 1000) await async_logger.log(LogEntry( client_key_id=client.id, provider_id=provider.id, model=request.model, prompt_tokens=prompt_tokens, completion_tokens=completion_tokens, latency_ms=latency_ms, success=True )) # Return response return response except HTTPException: raise except Exception as e: # Log failed request latency_ms = int((time.time() - start_time) * 1000) await async_logger.log(LogEntry( client_key_id=client.id, provider_id=provider.id, model=request.model, latency_ms=latency_ms, success=False, error_message=str(e) )) # Handle specific errors error_str = str(e).lower() if "rate" in error_str or "limit" in error_str: raise HTTPException(status_code=429, detail={"error": str(e)}) raise HTTPException(status_code=500, detail={"error": str(e)}) @router.get("/models") async def list_models(client: ClientKey = Depends(get_current_client)): """List available models.""" return { "object": "list", "data": [ {"id": "gpt-4o", "object": "model", "owned_by": "openai"}, {"id": "gpt-4o-mini", "object": "model", "owned_by": "openai"}, {"id": "o1", "object": "model", "owned_by": "openai"}, {"id": "o1-mini", "object": "model", "owned_by": "openai"}, {"id": "claude-3-opus", "object": "model", "owned_by": "anthropic"}, {"id": "claude-3-sonnet", "object": "model", "owned_by": "anthropic"}, {"id": "claude-3-haiku", "object": "model", "owned_by": "anthropic"}, ] } ``` - [ ] **Step 2: Commit** ```bash cd backend && git add app/routes/chat.py && git commit -m "feat: add OpenAI-compatible chat completions endpoint" ``` --- ### Task 13: OpenAI 原生路由 **Files:** - Create: `backend/app/routes/openai.py` - [ ] **Step 1: 创建 openai.py** ```python """OpenAI native endpoint (passthrough).""" import time from typing import Optional, List from fastapi import APIRouter, Depends, HTTPException from fastapi.responses import StreamingResponse from pydantic import BaseModel from ..models import ClientKey from ..middleware.auth import get_current_client from ..services.router import ModelRouter from ..services.key_selector import key_selector from ..services.logger import async_logger, LogEntry from ..providers.litellm_wrapper import LiteLLMWrapper router = APIRouter(prefix="/v1/openai", tags=["openai"]) class ChatMessage(BaseModel): role: str content: str class ChatRequest(BaseModel): model: str messages: List[ChatMessage] stream: bool = False temperature: Optional[float] = None max_tokens: Optional[int] = None @router.post("/chat/completions") async def openai_chat( request: ChatRequest, client: ClientKey = Depends(get_current_client) ): """OpenAI native chat completions (passthrough).""" start_time = time.time() # Get OpenAI provider provider = await ModelRouter.get_provider("openai") if not provider: raise HTTPException(status_code=503, detail={"error": "OpenAI provider not configured"}) # Get next API key provider_key = await key_selector.get_next_key(provider.id) if not provider_key: raise HTTPException(status_code=503, detail={"error": "No available OpenAI API key"}) messages = [{"role": m.role, "content": m.content} for m in request.messages] kwargs = {} if request.temperature is not None: kwargs["temperature"] = request.temperature if request.max_tokens is not None: kwargs["max_tokens"] = request.max_tokens try: response, prompt_tokens, completion_tokens = await LiteLLMWrapper.chat_completion( provider=provider, api_key=provider_key.key, model=request.model, messages=messages, stream=request.stream, **kwargs ) if request.stream: return StreamingResponse( LiteLLMWrapper.stream_response( response, client.id, provider.id, request.model, start_time ), media_type="text/event-stream" ) else: latency_ms = int((time.time() - start_time) * 1000) await async_logger.log(LogEntry( client_key_id=client.id, provider_id=provider.id, model=request.model, prompt_tokens=prompt_tokens, completion_tokens=completion_tokens, latency_ms=latency_ms, success=True )) return response except Exception as e: latency_ms = int((time.time() - start_time) * 1000) await async_logger.log(LogEntry( client_key_id=client.id, provider_id=provider.id, model=request.model, latency_ms=latency_ms, success=False, error_message=str(e) )) raise HTTPException(status_code=500, detail={"error": str(e)}) @router.get("/models") async def openai_models(client: ClientKey = Depends(get_current_client)): """List OpenAI models.""" return { "object": "list", "data": [ {"id": "gpt-4o", "object": "model", "owned_by": "openai"}, {"id": "gpt-4o-mini", "object": "model", "owned_by": "openai"}, {"id": "o1", "object": "model", "owned_by": "openai"}, {"id": "o1-mini", "object": "model", "owned_by": "openai"}, ] } ``` - [ ] **Step 2: Commit** ```bash cd backend && git add app/routes/openai.py && git commit -m "feat: add OpenAI native endpoint" ``` --- ### Task 14: Anthropic 原生路由 **Files:** - Create: `backend/app/routes/anthropic.py` - [ ] **Step 1: 创建 anthropic.py** ```python """Anthropic native endpoint (passthrough).""" import time from typing import Optional, List from fastapi import APIRouter, Depends, HTTPException from fastapi.responses import StreamingResponse from pydantic import BaseModel from ..models import ClientKey from ..middleware.auth import get_current_client from ..services.router import ModelRouter from ..services.key_selector import key_selector from ..services.logger import async_logger, LogEntry from ..providers.litellm_wrapper import LiteLLMWrapper router = APIRouter(prefix="/v1/anthropic", tags=["anthropic"]) class Message(BaseModel): role: str content: str class AnthropicRequest(BaseModel): model: str messages: List[Message] max_tokens: int = 1024 stream: bool = False temperature: Optional[float] = None @router.post("/messages") async def anthropic_messages( request: AnthropicRequest, client: ClientKey = Depends(get_current_client) ): """Anthropic native messages endpoint.""" start_time = time.time() # Get Anthropic provider provider = await ModelRouter.get_provider("anthropic") if not provider: raise HTTPException(status_code=503, detail={"error": "Anthropic provider not configured"}) # Get next API key provider_key = await key_selector.get_next_key(provider.id) if not provider_key: raise HTTPException(status_code=503, detail={"error": "No available Anthropic API key"}) messages = [{"role": m.role, "content": m.content} for m in request.messages] kwargs = {"max_tokens": request.max_tokens} if request.temperature is not None: kwargs["temperature"] = request.temperature try: response, prompt_tokens, completion_tokens = await LiteLLMWrapper.chat_completion( provider=provider, api_key=provider_key.key, model=request.model, messages=messages, stream=request.stream, **kwargs ) if request.stream: return StreamingResponse( LiteLLMWrapper.stream_response( response, client.id, provider.id, request.model, start_time ), media_type="text/event-stream" ) else: latency_ms = int((time.time() - start_time) * 1000) await async_logger.log(LogEntry( client_key_id=client.id, provider_id=provider.id, model=request.model, prompt_tokens=prompt_tokens, completion_tokens=completion_tokens, latency_ms=latency_ms, success=True )) return response except Exception as e: latency_ms = int((time.time() - start_time) * 1000) await async_logger.log(LogEntry( client_key_id=client.id, provider_id=provider.id, model=request.model, latency_ms=latency_ms, success=False, error_message=str(e) )) raise HTTPException(status_code=500, detail={"error": str(e)}) @router.get("/models") async def anthropic_models(client: ClientKey = Depends(get_current_client)): """List Anthropic models.""" return { "object": "list", "data": [ {"id": "claude-3-opus-20240229", "object": "model", "owned_by": "anthropic"}, {"id": "claude-3-sonnet-20240229", "object": "model", "owned_by": "anthropic"}, {"id": "claude-3-haiku-20240307", "object": "model", "owned_by": "anthropic"}, ] } ``` - [ ] **Step 2: Commit** ```bash cd backend && git add app/routes/anthropic.py && git commit -m "feat: add Anthropic native endpoint" ``` --- ## Chunk 4: 应用入口与集成 ### Task 15: FastAPI 应用主入口 **Files:** - Modify: `backend/app/main.py` - [ ] **Step 1: 更新 app/main.py** ```python """FastAPI application entry point.""" from contextlib import asynccontextmanager from fastapi import FastAPI from .database import init_db, seed_default_providers from .services.logger import async_logger from .routes import health_router, chat_router, openai_router, anthropic_router @asynccontextmanager async def lifespan(app: FastAPI): """Application lifespan handler.""" # Startup await init_db() await seed_default_providers() await async_logger.start() yield # Shutdown await async_logger.stop() app = FastAPI( title="zzrouter", description="AI Model API Proxy Service", version="0.1.0", lifespan=lifespan ) # Register routers app.include_router(health_router) app.include_router(chat_router) app.include_router(openai_router) app.include_router(anthropic_router) def run_server(): """Run the server (entry point for script).""" import uvicorn from .config import HOST, PORT uvicorn.run(app, host=HOST, port=PORT) if __name__ == "__main__": run_server() ``` - [ ] **Step 2: Commit** ```bash cd backend && git add app/main.py && git commit -m "feat: add FastAPI application with lifespan management" ``` --- ### Task 16: 更新根入口 **Files:** - Modify: `backend/main.py` - [ ] **Step 1: 更新根 main.py** ```python """Server entry point.""" from app.main import run_server if __name__ == "__main__": run_server() ``` - [ ] **Step 2: Commit** ```bash cd backend && git add main.py && git commit -m "chore: update root entry point" ``` --- ### Task 17: 更新 app/__init__.py **Files:** - Modify: `backend/app/__init__.py` - [ ] **Step 1: 更新 __init__.py** ```python """zzrouter backend application.""" from .main import app __all__ = ["app"] ``` - [ ] **Step 2: Commit** ```bash cd backend && git add app/__init__.py && git commit -m "chore: update app __init__" ``` --- ### Task 18: 更新 CLAUDE.md **Files:** - Modify: `CLAUDE.md` - [ ] **Step 1: 更新 CLAUDE.md** ```markdown # CLAUDE.md This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository. ## Project Overview zzrouter is an AI model API proxy service with OpenAI-compatible and Anthropic support, API key rotation, and request logging. ## Project Structure ``` zzrouter/ ├── backend/ # Python FastAPI proxy server └── frontend/ # (placeholder) ``` ## Backend Commands ```bash cd backend # Install dependencies uv sync # Run development server uv run uvicorn app.main:app --reload --port 8000 # Run directly uv run python main.py ``` ## API Endpoints | Endpoint | Description | |----------|-------------| | `POST /v1/chat/completions` | OpenAI-compatible chat (auto-routes by model) | | `POST /v1/openai/chat/completions` | OpenAI native passthrough | | `POST /v1/anthropic/messages` | Anthropic native passthrough | | `GET /health` | Health check | ## Authentication All API endpoints require Bearer token authentication: ``` Authorization: Bearer ``` ## Database SQLite database stored at `backend/data/zzrouter.db`. Key tables: - `client_keys` - Client API keys - `providers` - Model provider configs - `provider_keys` - Provider API keys (supports multiple per provider) - `request_logs` - Request statistics ## Adding Configuration ```sql -- Add a client key INSERT INTO client_keys (key, name) VALUES ('my-secret-key', 'my-app'); -- Add provider API keys INSERT INTO provider_keys (provider_id, key) VALUES (1, 'sk-openai-key'); INSERT INTO provider_keys (provider_id, key) VALUES (2, 'sk-ant-anthropic-key'); ``` ``` - [ ] **Step 2: Commit** ```bash git add CLAUDE.md && git commit -m "docs: update CLAUDE.md with API documentation" ``` --- ### Task 19: 最终集成测试 - [ ] **Step 1: 启动服务** Run: `cd backend && uv run uvicorn app.main:app --reload --port 8000` Expected: 服务启动成功,显示 "Uvicorn running on http://0.0.0.0:8000" - [ ] **Step 2: 测试健康检查** Run: `curl http://localhost:8000/health` Expected: `{"status": "ok"}` - [ ] **Step 3: 添加测试客户端 Key** Run: ```bash cd backend && sqlite3 data/zzrouter.db "INSERT INTO client_keys (key, name) VALUES ('test-key-123', 'test-client');" ``` - [ ] **Step 4: 测试认证失败** Run: `curl http://localhost:8000/v1/models` Expected: 401 `{"detail":{"error":"Missing API key"}}` - [ ] **Step 5: 测试认证成功** Run: `curl -H "Authorization: Bearer test-key-123" http://localhost:8000/v1/models` Expected: 返回模型列表 --- ## Summary 实现计划完成。按顺序执行以下 chunk: 1. **Chunk 1**: 项目初始化与数据库层 (Tasks 1-3) 2. **Chunk 2**: 核心服务层 (Tasks 4-8) 3. **Chunk 3**: 提供商适配器与路由 (Tasks 9-14) 4. **Chunk 4**: 应用入口与集成 (Tasks 15-19)