在实际 AI 应用开发中，直接调用官方大模型 API 往往面临地域限制、账号风控、价格波动和模型覆盖不全等问题。很多开发者会选择使用第三方 API 中转站来统一管理多个大模型的调用，但市面上的中转站质量参差不齐，存在稳定性、计费透明度和模型纯度等方面的风险。

更好的解决方案是自己编写一个 skills 系统，将 API 中转能力集成到开发环境中，实现自主控制、成本透明和灵活扩展。本文将基于常见的 AI 开发工具链，介绍如何从零构建一个可管理多模型 API 中转的 skills 模块，涵盖设计思路、环境准备、核心实现、配置管理和生产级优化。

1. 理解 API 中转站的核心价值与风险

1.1 为什么需要 API 中转站

大模型 API 调用在实际项目中会遇到几个典型问题：

• 地域限制：部分官方 API 对特定地区 IP 有访问限制
• 账号风控：高频调用或异常模式可能导致账号被封禁
• 模型碎片化：不同项目可能需要切换 GPT、Claude、DeepSeek 等多种模型
• 成本控制：需要统一监控各模型的 token 消耗和费用情况
• 故障隔离：当某个模型服务不可用时能快速切换到备用方案

API 中转站本质上是一个代理层，它封装了多个上游模型供应商的接口，对外提供统一的调用规范。

1.2 第三方中转站的常见风险

从实际使用经验看，第三方中转站存在以下风险点：

1.3 自建 skills 系统的优势

自己开发 API 中转 skills 可以解决上述问题：

• 完全可控：自主决定路由策略、重试机制和降级方案
• 成本透明：直接对接官方 API，避免中间商加价
• 数据安全：敏感数据不经过第三方服务器
• 灵活扩展：可根据业务需求定制特殊功能
• 技术积累：掌握核心集成能力，不依赖外部服务

2. 环境准备与项目结构设计

2.1 技术栈选择

基于常见的 AI 开发场景，推荐以下技术组合：

# 技术栈说明 核心语言: Python 3.9+ Web框架: FastAPI (轻量级，适合API服务) HTTP客户端: httpx (支持异步，比requests更现代) 配置管理: pydantic-settings (类型安全的配置管理) 缓存: redis (用于token计数和频率限制) 监控: prometheus-client (指标收集) 日志: structlog (结构化日志)

2.2 开发环境准备

# 创建项目目录 mkdir ai-relay-skills && cd ai-relay-skills # 创建虚拟环境 python -m venv venv source venv/bin/activate # Linux/Mac # venv\Scripts\activate # Windows # 安装核心依赖 pip install fastapi uvicorn httpx redis pydantic-settings pip install structlog prometheus-client # 开发工具 pip install black isort mypy pytest

2.3 项目结构设计

ai-relay-skills/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI应用入口 │ ├── config/ │ │ ├── __init__.py │ │ └── settings.py # 配置管理 │ ├── models/ │ │ ├── __init__.py │ │ ├── api_models.py # 请求响应模型 │ │ └── data_models.py # 数据模型 │ ├── services/ │ │ ├── __init__.py │ │ ├── relay_service.py # 中转核心逻辑 │ │ ├── cache_service.py # 缓存管理 │ │ └── metrics_service.py # 监控指标 │ ├── routers/ │ │ ├── __init__.py │ │ └── api.py # API路由 │ └── utils/ │ ├── __init__.py │ ├── logger.py # 日志配置 │ └── exceptions.py # 异常处理 ├── tests/ │ ├── __init__.py │ ├── test_relay.py │ └── conftest.py ├── requirements.txt ├── requirements-dev.txt └── README.md

3. 核心实现：多模型 API 中转服务

3.1 配置管理设计

首先定义类型安全的配置模型：

# app/config/settings.py from pydantic_settings import BaseSettings from typing import Dict, List, Optional from enum import Enum class ModelProvider(str, Enum): OPENAI = "openai" ANTHROPIC = "anthropic" DEEPSEEK = "deepseek" GEMINI = "gemini" class ProviderConfig(BaseSettings): api_key: str base_url: str max_retries: int = 3 timeout: int = 30 rate_limit: int = 100 # 每分钟最大请求数 class RelaySettings(BaseSettings): # 各提供商配置 providers: Dict[ModelProvider, ProviderConfig] # 默认模型映射 default_models: Dict[str, ModelProvider] = { "gpt-4": ModelProvider.OPENAI, "claude-3-opus": ModelProvider.ANTHROPIC, "deepseek-chat": ModelProvider.DEEPSEEK, "gemini-pro": ModelProvider.GEMINI } # 缓存配置 redis_url: str = "redis://localhost:6379" cache_ttl: int = 300 # 缓存5分钟 # 监控配置 enable_metrics: bool = True metrics_port: int = 9090 class Config: env_file = ".env" case_sensitive = False settings = RelaySettings()

对应的环境配置文件：

# .env OPENAI_API_KEY=sk-your-openai-key OPENAI_BASE_URL=https://api.openai.com/v1 OPENAI_MAX_RETRIES=3 ANTHROPIC_API_KEY=your-anthropic-key ANTHROPIC_BASE_URL=https://api.anthropic.com ANTHROPIC_TIMEOUT=30 DEEPSEEK_API_KEY=your-deepseek-key DEEPSEEK_BASE_URL=https://api.deepseek.com DEEPSEEK_RATE_LIMIT=200 REDIS_URL=redis://localhost:6379 ENABLE_METRICS=true

3.2 请求响应模型定义

# app/models/api_models.py from pydantic import BaseModel from typing import List, Dict, Any, Optional from enum import Enum class MessageRole(str, Enum): USER = "user" ASSISTANT = "assistant" SYSTEM = "system" class ChatMessage(BaseModel): role: MessageRole content: str class ChatRequest(BaseModel): model: str messages: List[ChatMessage] temperature: Optional[float] = 0.7 max_tokens: Optional[int] = 1000 stream: Optional[bool] = False class ChatResponse(BaseModel): id: str model: str choices: List[Dict[str, Any]] usage: Dict[str, int] provider: str class ErrorResponse(BaseModel): error: str code: int details: Optional[Dict[str, Any]] = None

3.3 中转服务核心逻辑

# app/services/relay_service.py import httpx import json from typing import Dict, List, Optional from app.config.settings import settings, ModelProvider from app.models.api_models import ChatRequest, ChatResponse, ErrorResponse from app.utils.logger import logger from app.services.cache_service import CacheService class RelayService: def __init__(self): self.cache = CacheService() self.client = httpx.AsyncClient(timeout=30) # 提供商特定的请求适配器 self.provider_adapters = { ModelProvider.OPENAI: self._adapt_to_openai, ModelProvider.ANTHROPIC: self._adapt_to_anthropic, ModelProvider.DEEPSEEK: self._adapt_to_deepseek, ModelProvider.GEMINI: self._adapt_to_gemini } # 提供商特定的响应解析器 self.response_parsers = { ModelProvider.OPENAI: self._parse_openai_response, ModelProvider.ANTHROPIC: self._parse_anthropic_response, ModelProvider.DEEPSEEK: self._parse_deepseek_response, ModelProvider.GEMINI: self._parse_gemini_response } async def chat_completion(self, request: ChatRequest) -> ChatResponse: """统一聊天补全接口""" # 1. 确定模型对应的提供商 provider = self._resolve_provider(request.model) if not provider: raise ValueError(f"Unsupported model: {request.model}") # 2. 检查缓存 cache_key = self._generate_cache_key(request) cached_response = await self.cache.get(cache_key) if cached_response: logger.info("Cache hit", model=request.model, cache_key=cache_key) return ChatResponse(**cached_response) # 3. 适配请求格式 provider_config = settings.providers[provider] adapted_request = self.provider_adapters[provider](request) # 4. 发送请求 try: response = await self._make_request( provider, provider_config, adapted_request ) # 5. 解析响应 parsed_response = self.response_parsers[provider](response) parsed_response.provider = provider.value # 6. 缓存结果 await self.cache.set(cache_key, parsed_response.dict()) return parsed_response except httpx.HTTPStatusError as e: logger.error("HTTP error", provider=provider, status_code=e.response.status_code) raise self._handle_http_error(e) except Exception as e: logger.error("Unexpected error", provider=provider, error=str(e)) raise def _resolve_provider(self, model: str) -> Optional[ModelProvider]: """根据模型名称解析提供商""" # 先检查显式映射 if model in settings.default_models: return settings.default_models[model] # 基于模型名称前缀推断 model_lower = model.lower() if model_lower.startswith('gpt-'): return ModelProvider.OPENAI elif model_lower.startswith('claude-'): return ModelProvider.ANTHROPIC elif 'deepseek' in model_lower: return ModelProvider.DEEPSEEK elif 'gemini' in model_lower: return ModelProvider.GEMINI return None def _generate_cache_key(self, request: ChatRequest) -> str: """生成缓存键""" content = json.dumps([msg.dict() for msg in request.messages], sort_keys=True) return f"chat:{request.model}:{hash(content)}" async def _make_request(self, provider: ModelProvider, config, adapted_request: Dict): """向具体提供商发送请求""" headers = { "Authorization": f"Bearer {config.api_key}", "Content-Type": "application/json" } if provider == ModelProvider.ANTHROPIC: headers["anthropic-version"] = "2023-06-01" response = await self.client.post( f"{config.base_url}/chat/completions", json=adapted_request, headers=headers, timeout=config.timeout ) response.raise_for_status() return response.json() def _adapt_to_openai(self, request: ChatRequest) -> Dict: """适配到 OpenAI 格式""" return { "model": request.model, "messages": [msg.dict() for msg in request.messages], "temperature": request.temperature, "max_tokens": request.max_tokens, "stream": request.stream } def _adapt_to_anthropic(self, request: ChatRequest) -> Dict: """适配到 Anthropic 格式""" return { "model": request.model, "messages": [msg.dict() for msg in request.messages], "max_tokens": request.max_tokens, "temperature": request.temperature } def _parse_openai_response(self, response: Dict) -> ChatResponse: """解析 OpenAI 响应""" return ChatResponse( id=response["id"], model=response["model"], choices=response["choices"], usage=response.get("usage", {}) ) def _parse_anthropic_response(self, response: Dict) -> ChatResponse: """解析 Anthropic 响应""" return ChatResponse( id=response.get("id", ""), model=response.get("model", ""), choices=[{ "message": { "role": "assistant", "content": response["content"][0]["text"] } }], usage={ "prompt_tokens": response.get("usage", {}).get("input_tokens", 0), "completion_tokens": response.get("usage", {}).get("output_tokens", 0) } ) def _handle_http_error(self, error: httpx.HTTPStatusError) -> Exception: """处理 HTTP 错误""" error_mapping = { 400: ValueError("Bad request - check your parameters"), 401: ValueError("Invalid API key"), 429: RuntimeError("Rate limit exceeded"), 500: RuntimeError("Provider server error"), 503: RuntimeError("Service unavailable") } return error_mapping.get(error.response.status_code, RuntimeError(f"HTTP error: {error.response.status_code}")) # 提供商特定的适配器和解析器实现（简化版） def _adapt_to_deepseek(self, request: ChatRequest) -> Dict: """适配到 DeepSeek 格式""" return self._adapt_to_openai(request) # DeepSeek 兼容 OpenAI 格式 def _adapt_to_gemini(self, request: ChatRequest) -> Dict: """适配到 Gemini 格式""" return { "contents": [ { "parts": [{"text": msg.content} for msg in request.messages if msg.role in ["user", "system"]] } ], "generationConfig": { "temperature": request.temperature, "maxOutputTokens": request.max_tokens } } def _parse_deepseek_response(self, response: Dict) -> ChatResponse: """解析 DeepSeek 响应""" return self._parse_openai_response(response) def _parse_gemini_response(self, response: Dict) -> ChatResponse: """解析 Gemini 响应""" return ChatResponse( id=response.get("candidates", [{}])[0].get("id", ""), model=request.model, # 需要从上下文获取 choices=[{ "message": { "role": "assistant", "content": response.get("candidates", [{}])[0].get("content", {}).get("parts", [{}])[0].get("text", "") } }], usage={} # Gemini 使用信息需要额外处理 )

3.4 缓存服务实现

# app/services/cache_service.py import redis.asyncio as redis import json from typing import Optional, Any from app.config.settings import settings from app.utils.logger import logger class CacheService: def __init__(self): self.redis_client = None async def _get_client(self): """获取 Redis 客户端（懒加载）""" if self.redis_client is None: self.redis_client = redis.from_url( settings.redis_url, encoding="utf-8", decode_responses=True ) return self.redis_client async def get(self, key: str) -> Optional[Any]: """获取缓存值""" try: client = await self._get_client() value = await client.get(key) if value: return json.loads(value) except Exception as e: logger.warning("Cache get failed", key=key, error=str(e)) return None async def set(self, key: str, value: Any, ttl: int = None) -> bool: """设置缓存值""" try: client = await self._get_client() ttl = ttl or settings.cache_ttl await client.setex(key, ttl, json.dumps(value)) return True except Exception as e: logger.warning("Cache set failed", key=key, error=str(e)) return False async def close(self): """关闭 Redis 连接""" if self.redis_client: await self.redis_client.close()

4. API 路由与监控集成

4.1 FastAPI 路由实现

# app/routers/api.py from fastapi import APIRouter, HTTPException from app.models.api_models import ChatRequest, ChatResponse, ErrorResponse from app.services.relay_service import RelayService from app.utils.logger import logger router = APIRouter() relay_service = RelayService() @router.post("/chat/completions", response_model=ChatResponse) async def chat_completion(request: ChatRequest): """统一的聊天补全接口""" try: logger.info("Chat request received", model=request.model, message_count=len(request.messages)) response = await relay_service.chat_completion(request) # 记录使用指标 logger.info("Chat request completed", model=request.model, token_usage=response.usage) return response except ValueError as e: logger.warning("Invalid request", error=str(e)) raise HTTPException(status_code=400, detail=str(e)) except RuntimeError as e: logger.error("Service error", error=str(e)) raise HTTPException(status_code=503, detail=str(e)) except Exception as e: logger.error("Unexpected error", error=str(e)) raise HTTPException(status_code=500, detail="Internal server error") @router.get("/health") async def health_check(): """健康检查端点""" return {"status": "healthy", "service": "ai-relay-skills"} @router.get("/models") async def list_models(): """列出支持的模型""" from app.config.settings import settings return { "models": list(settings.default_models.keys()), "providers": [provider.value for provider in settings.providers.keys()] }

4.2 应用入口点

# app/main.py from fastapi import FastAPI from app.routers import api from app.utils.logger import logger from app.services.metrics_service import setup_metrics import uvicorn def create_app() -> FastAPI: app = FastAPI( title="AI Relay Skills", description="自主管理的多模型 API 中转服务", version="1.0.0" ) # 注册路由 app.include_router(api.router, prefix="/api/v1") # 设置监控 if settings.enable_metrics: setup_metrics(app) @app.on_event("startup") async def startup_event(): logger.info("AI Relay Skills starting up") @app.on_event("shutdown") async def shutdown_event(): logger.info("AI Relay Skills shutting down") # 清理资源 await api.relay_service.cache.close() return app app = create_app() if __name__ == "__main__": uvicorn.run( "app.main:app", host="0.0.0.0", port=8000, reload=True # 开发模式热重载 )

4.3 监控指标收集

# app/services/metrics_service.py from prometheus_client import Counter, Histogram, generate_latest from fastapi import Response import time # 定义指标 REQUEST_COUNT = Counter( 'relay_requests_total', 'Total API requests', ['method', 'endpoint', 'status'] ) REQUEST_DURATION = Histogram( 'relay_request_duration_seconds', 'Request duration in seconds', ['endpoint'] ) TOKEN_USAGE = Counter( 'relay_tokens_total', 'Total tokens used', ['model', 'type'] # type: prompt or completion ) def setup_metrics(app): """设置监控中间件""" @app.middleware("http") async def metrics_middleware(request, call_next): start_time = time.time() response = await call_next(request) # 记录请求指标 duration = time.time() - start_time REQUEST_COUNT.labels( method=request.method, endpoint=request.url.path, status=response.status_code ).inc() REQUEST_DURATION.labels( endpoint=request.url.path ).observe(duration) return response @app.get("/metrics") async def metrics(): """Prometheus 指标端点""" return Response( generate_latest(), media_type="text/plain" )

5. 运行验证与测试

5.1 启动服务

# 开发模式启动 uvicorn app.main:app --reload --host 0.0.0.0 --port 8000 # 生产模式启动（使用gunicorn） gunicorn -w 4 -k uvicorn.workers.UvicornWorker app.main:app --bind 0.0.0.0:8000

5.2 测试 API 接口

# tests/test_relay.py import pytest import httpx from app.main import app from fastapi.testclient import TestClient client = TestClient(app) def test_health_check(): response = client.get("/health") assert response.status_code == 200 assert response.json()["status"] == "healthy" def test_list_models(): response = client.get("/api/v1/models") assert response.status_code == 200 data = response.json() assert "models" in data assert "providers" in data def test_chat_completion(): # 测试请求格式 request_data = { "model": "gpt-4", "messages": [ {"role": "user", "content": "Hello, world!"} ], "temperature": 0.7, "max_tokens": 100 } response = client.post("/api/v1/chat/completions", json=request_data) # 注意：实际测试需要配置有效的 API key assert response.status_code in [200, 400, 401] # 根据配置情况 @pytest.mark.asyncio async def test_cache_integration(): """测试缓存集成""" from app.services.cache_service import CacheService cache = CacheService() test_key = "test_key" test_value = {"test": "value"} # 测试设置缓存 result = await cache.set(test_key, test_value) assert result is True # 测试获取缓存 cached = await cache.get(test_key) assert cached == test_value

5.3 使用 curl 测试

# 健康检查 curl http://localhost:8000/health # 列出模型 curl http://localhost:8000/api/v1/models # 发送聊天请求 curl -X POST http://localhost:8000/api/v1/chat/completions \ -H "Content-Type: application/json" \ -d '{ "model": "gpt-4", "messages": [{"role": "user", "content": "Hello!"}], "temperature": 0.7 }'

6. 生产环境部署与优化

6.1 Docker 容器化部署

# Dockerfile FROM python:3.11-slim WORKDIR /app # 安装系统依赖 RUN apt-get update && apt-get install -y \ gcc \ && rm -rf /var/lib/apt/lists/* # 复制依赖文件 COPY requirements.txt . # 安装 Python 依赖 RUN pip install --no-cache-dir -r requirements.txt # 复制应用代码 COPY app/ ./app/ # 创建非root用户 RUN useradd --create-home --shell /bin/bash app USER app # 暴露端口 EXPOSE 8000 # 启动命令 CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

对应的 Docker Compose 配置：

# docker-compose.yml version: '3.8' services: ai-relay: build: . ports: - "8000:8000" environment: - REDIS_URL=redis://redis:6379 - OPENAI_API_KEY=${OPENAI_API_KEY} - ANTHROPIC_API_KEY=${ANTHROPIC_API_KEY} depends_on: - redis restart: unless-stopped redis: image: redis:7-alpine ports: - "6379:6379" volumes: - redis_data:/data restart: unless-stopped volumes: redis_data:

6.2 配置管理与安全最佳实践

环境配置分离：

# 生产环境配置 .env.production OPENAI_API_KEY=sk-prod-xxx ANTHROPIC_API_KEY=sk-ant-prod-xxx REDIS_URL=redis://redis-prod:6379 CACHE_TTL=600 ENABLE_METRICS=true # 开发环境配置 .env.development OPENAI_API_KEY=sk-dev-xxx ANTHROPIC_API_KEY=sk-ant-dev-xxx REDIS_URL=redis://localhost:6379 CACHE_TTL=60 ENABLE_METRICS=false

安全配置检查清单：

# app/utils/security.py import os from typing import List def validate_environment() -> List[str]: """验证环境配置安全性""" issues = [] required_vars = [ 'OPENAI_API_KEY', 'ANTHROPIC_API_KEY', 'REDIS_URL' ] for var in required_vars: if not os.getenv(var): issues.append(f"Missing required environment variable: {var}") # 检查 API key 格式 openai_key = os.getenv('OPENAI_API_KEY', '') if openai_key and not openai_key.startswith('sk-'): issues.append("OpenAI API key format appears invalid") # 检查 Redis 连接安全性 redis_url = os.getenv('REDIS_URL', '') if redis_url and 'password' not in redis_url and 'localhost' not in redis_url: issues.append("Redis connection may be insecure") return issues

6.3 性能优化配置

# 优化后的配置示例 class OptimizedSettings(BaseSettings): # 连接池配置 http_pool_size: int = 100 http_max_keepalive: int = 20 http_timeout: int = 30 # 重试策略 max_retries: int = 3 retry_backoff: float = 0.5 # 指数退避基数 # 缓存优化 cache_ttl: int = 300 cache_max_size: int = 10000 # 限流配置 rate_limit_per_minute: int = 1000 burst_limit: int = 100 class Config: env_file = ".env"

7. 常见问题排查与解决方案

7.1 API 调用问题排查表

7.2 日志分析与监控

配置结构化日志便于排查：

# app/utils/logger.py import structlog import logging def setup_logging(): structlog.configure( processors=[ structlog.stdlib.filter_by_level, structlog.stdlib.add_logger_name, structlog.stdlib.add_log_level, structlog.stdlib.PositionalArgumentsFormatter(), structlog.processors.TimeStamper(fmt="iso"), structlog.processors.StackInfoRenderer(), structlog.processors.format_exc_info, structlog.processors.UnicodeDecoder(), structlog.processors.JSONRenderer() ], context_class=dict, logger_factory=structlog.stdlib.LoggerFactory(), wrapper_class=structlog.stdlib.BoundLogger, cache_logger_on_first_use=True, ) # 使用示例 logger = structlog.get_logger() logger.info("API request", provider="openai", model="gpt-4", duration=0.45)

7.3 故障转移与降级策略

# app/services/fallback_service.py from typing import List, Optional from app.config.settings import ModelProvider class FallbackService: def __init__(self): self.fallback_chains = { "gpt-4": [ModelProvider.OPENAI, ModelProvider.DEEPSEEK, ModelProvider.GEMINI], "claude-3-opus": [ModelProvider.ANTHROPIC, ModelProvider.OPENAI], "deepseek-chat": [ModelProvider.DEEPSEEK, ModelProvider.OPENAI] } async def try_with_fallback(self, request, primary_provider: ModelProvider): """使用故障转移链尝试请求""" model = request.model fallback_chain = self.fallback_chains.get(model, [primary_provider]) for provider in fallback_chain: try: # 尝试当前提供商 response = await self._make_provider_request(provider, request) logger.info("Fallback success", original_provider=primary_provider, fallback_provider=provider) return response except Exception as e: logger.warning("Provider failed", provider=provider, error=str(e)) continue raise RuntimeError(f"All fallback providers failed for model: {model}")

8. 扩展方向与进阶功能

8.1 成本控制与用量统计

# app/services/billing_service.py from datetime import datetime, timedelta from typing import Dict import sqlite3 class BillingService: def __init__(self, db_path: str = "billing.db"): self.db_path = db_path self._init_db() def _init_db(self): """初始化计费数据库""" with sqlite3.connect(self.db_path) as conn: conn.execute(""" CREATE TABLE IF NOT EXISTS usage_records ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id TEXT NOT NULL, model TEXT NOT NULL, provider TEXT NOT NULL, prompt_tokens INTEGER DEFAULT 0, completion_tokens INTEGER DEFAULT 0, total_tokens INTEGER DEFAULT 0, cost REAL DEFAULT 0.0, timestamp DATETIME DEFAULT CURRENT_TIMESTAMP ) """) async def record_usage(self, user_id: str, model: str, provider: str, usage: Dict, cost: float): """记录使用量和成本""" with sqlite3.connect(self.db_path) as conn: conn.execute(""" INSERT INTO usage_records (user_id, model, provider, prompt_tokens, completion_tokens, total_tokens, cost) VALUES (?, ?, ?, ?, ?, ?, ?) """, (user_id, model, provider, usage.get('prompt_tokens', 0), usage.get('completion_tokens', 0), usage.get('total_tokens', 0), cost)) async def get_user_usage(self, user_id: str, days: int = 30) -> Dict: """获取用户指定时间段内的使用统计""" with sqlite3.connect(self.db_path) as conn: cursor = conn.execute(""" SELECT model, SUM(prompt_tokens), SUM(completion_tokens), SUM(cost) FROM usage_records WHERE user_id = ? AND timestamp >= ? GROUP BY model """, (user_id, datetime.now() - timedelta(days=days))) return { row[0]: { "prompt_tokens": row[1], "completion_tokens": row[2], "total_cost": row[3] } for row in cursor.fetchall() }

8.2 模型质量评估与自动选型

# app/services/model_evaluation.py from typing import List, Dict import asyncio class ModelEvaluationService: def __init__(self): self.evaluation_metrics = {} async def evaluate_model_quality(self, model: str, test_prompts: List[str]) -> Dict: """评估模型质量""" results = {} for prompt in test_prompts: # 使用标准测试提示词评估 response = await self._get_model_response(model, prompt) score = self._calculate_quality_score(response, prompt) results[prompt] = score return { "model": model, "average_score": sum(results.values()) / len(results), "detailed_scores": results } def _calculate_quality_score(self, response: str, prompt: str) -> float: """计算响应质量分数（简化版）""" # 实际实现可以包含语法检查、相关性分析、事实准确性等 score = 0.0 # 响应长度合理性 if 50 <= len(response) <= 1000: score += 0.3 # 响应相关性（简单关键词匹配） prompt_words = set(prompt.lower().split()) response_words = set(response.lower().split()) overlap = len(prompt_words & response_words) / len(prompt_words) if prompt_words else 0 score += min(overlap * 0.7,