Files

tech 26738973bd feat: implement AI model API proxy service

2026-03-11 18:22:16 +08:00

42 KiB

Raw Permalink Blame History

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: 添加新依赖

[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

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: 编写数据库连接模块

"""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

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

"""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

"""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

"""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

"""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

"""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

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: 创建配置模块

"""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

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

"""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

"""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

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

"""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

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

"""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

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

"""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

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

"""Provider adapters."""
from .litellm_wrapper import LiteLLMWrapper

__all__ = ["LiteLLMWrapper"]

Step 2: 创建 litellm_wrapper.py

"""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

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

"""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

"""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

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

"""Middleware components."""
from .auth import AuthMiddleware, get_current_client

__all__ = ["AuthMiddleware", "get_current_client"]

Step 2: 创建 auth.py

"""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

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

"""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

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

"""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

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

"""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

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

"""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

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

"""Server entry point."""
from app.main import run_server

if __name__ == "__main__":
    run_server()

Step 2: Commit

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

"""zzrouter backend application."""
from .main import app

__all__ = ["app"]

Step 2: Commit

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

# 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 <client_key>

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

-- 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:

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:

Chunk 1: 项目初始化与数据库层 (Tasks 1-3)
Chunk 2: 核心服务层 (Tasks 4-8)
Chunk 3: 提供商适配器与路由 (Tasks 9-14)
Chunk 4: 应用入口与集成 (Tasks 15-19)

42 KiB Raw Permalink Blame History Unescape Escape

zzrouter 模型代理服务实现计划

Chunk 1: 项目初始化与数据库层

Task 1: 更新依赖配置

Task 2: 数据库连接模块

Task 3: 数据库模型

Chunk 2: 核心服务层

Task 4: 配置模块

Task 5: 认证服务

Task 6: Key 轮询选择器

Task 7: 模型路由服务

Task 8: 异步日志服务

Chunk 3: 提供商适配器与路由

Task 9: LiteLLM 封装器

Task 10: 健康检查路由

Task 11: 认证中间件

Task 12: OpenAI 兼容聊天路由

Task 13: OpenAI 原生路由

Task 14: Anthropic 原生路由

Chunk 4: 应用入口与集成

Task 15: FastAPI 应用主入口

Task 16: 更新根入口

Task 17: 更新 app/init.py

Task 18: 更新 CLAUDE.md

API Endpoints

Authentication

Database

Adding Configuration

Task 19: 最终集成测试

Summary

42 KiB

Raw Permalink Blame History