在实际 AI 应用开发中直接调用官方大模型 API 往往面临地域限制、账号风控、价格波动和模型覆盖不全等问题。很多开发者会选择使用第三方 API 中转站来统一管理多个大模型的调用但市面上的中转站质量参差不齐存在稳定性、计费透明度和模型纯度等方面的风险。更好的解决方案是自己编写一个 skills 系统将 API 中转能力集成到开发环境中实现自主控制、成本透明和灵活扩展。本文将基于常见的 AI 开发工具链介绍如何从零构建一个可管理多模型 API 中转的 skills 模块涵盖设计思路、环境准备、核心实现、配置管理和生产级优化。1. 理解 API 中转站的核心价值与风险1.1 为什么需要 API 中转站大模型 API 调用在实际项目中会遇到几个典型问题地域限制部分官方 API 对特定地区 IP 有访问限制账号风控高频调用或异常模式可能导致账号被封禁模型碎片化不同项目可能需要切换 GPT、Claude、DeepSeek 等多种模型成本控制需要统一监控各模型的 token 消耗和费用情况故障隔离当某个模型服务不可用时能快速切换到备用方案API 中转站本质上是一个代理层它封装了多个上游模型供应商的接口对外提供统一的调用规范。1.2 第三方中转站的常见风险从实际使用经验看第三方中转站存在以下风险点风险类型具体表现影响程度稳定性风险服务频繁宕机、响应延迟高高 - 直接影响业务连续性计费不透明Token 计数不准确、隐藏费用中 - 造成成本不可控模型掺水用低价模型冒充高价模型高 - 影响输出质量数据安全请求数据可能被留存或滥用极高 - 涉及商业机密服务跑路站点突然关闭、余额无法提现中 - 造成短期业务中断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 pytest2.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.md3. 核心实现多模型 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_KEYsk-your-openai-key OPENAI_BASE_URLhttps://api.openai.com/v1 OPENAI_MAX_RETRIES3 ANTHROPIC_API_KEYyour-anthropic-key ANTHROPIC_BASE_URLhttps://api.anthropic.com ANTHROPIC_TIMEOUT30 DEEPSEEK_API_KEYyour-deepseek-key DEEPSEEK_BASE_URLhttps://api.deepseek.com DEEPSEEK_RATE_LIMIT200 REDIS_URLredis://localhost:6379 ENABLE_METRICStrue3.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]] None3.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(timeout30) # 提供商特定的请求适配器 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(fUnsupported 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, modelrequest.model, cache_keycache_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, providerprovider, status_codee.response.status_code) raise self._handle_http_error(e) except Exception as e: logger.error(Unexpected error, providerprovider, errorstr(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_keysTrue) return fchat:{request.model}:{hash(content)} async def _make_request(self, provider: ModelProvider, config, adapted_request: Dict): 向具体提供商发送请求 headers { Authorization: fBearer {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, jsonadapted_request, headersheaders, timeoutconfig.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( idresponse[id], modelresponse[model], choicesresponse[choices], usageresponse.get(usage, {}) ) def _parse_anthropic_response(self, response: Dict) - ChatResponse: 解析 Anthropic 响应 return ChatResponse( idresponse.get(id, ), modelresponse.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(fHTTP 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( idresponse.get(candidates, [{}])[0].get(id, ), modelrequest.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, encodingutf-8, decode_responsesTrue ) 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, keykey, errorstr(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, keykey, errorstr(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_modelChatResponse) async def chat_completion(request: ChatRequest): 统一的聊天补全接口 try: logger.info(Chat request received, modelrequest.model, message_countlen(request.messages)) response await relay_service.chat_completion(request) # 记录使用指标 logger.info(Chat request completed, modelrequest.model, token_usageresponse.usage) return response except ValueError as e: logger.warning(Invalid request, errorstr(e)) raise HTTPException(status_code400, detailstr(e)) except RuntimeError as e: logger.error(Service error, errorstr(e)) raise HTTPException(status_code503, detailstr(e)) except Exception as e: logger.error(Unexpected error, errorstr(e)) raise HTTPException(status_code500, detailInternal 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( titleAI Relay Skills, description自主管理的多模型 API 中转服务, version1.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, host0.0.0.0, port8000, reloadTrue # 开发模式热重载 )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( methodrequest.method, endpointrequest.url.path, statusresponse.status_code ).inc() REQUEST_DURATION.labels( endpointrequest.url.path ).observe(duration) return response app.get(/metrics) async def metrics(): Prometheus 指标端点 return Response( generate_latest(), media_typetext/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:80005.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, jsonrequest_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_value5.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_URLredis://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_KEYsk-prod-xxx ANTHROPIC_API_KEYsk-ant-prod-xxx REDIS_URLredis://redis-prod:6379 CACHE_TTL600 ENABLE_METRICStrue # 开发环境配置 .env.development OPENAI_API_KEYsk-dev-xxx ANTHROPIC_API_KEYsk-ant-dev-xxx REDIS_URLredis://localhost:6379 CACHE_TTL60 ENABLE_METRICSfalse安全配置检查清单# 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(fMissing 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 issues6.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 .env7. 常见问题排查与解决方案7.1 API 调用问题排查表问题现象可能原因检查方式解决方案401 UnauthorizedAPI key 无效或过期检查环境变量配置更新有效的 API key429 Rate Limit请求频率超限查看提供商速率限制降低请求频率或升级套餐500 Internal Error服务端问题检查提供商状态页面等待服务恢复或切换提供商响应缓慢网络延迟或提供商负载高测试网络连接和延迟优化网络或使用 CDN缓存不生效Redis 连接问题或配置错误检查 Redis 连接和日志修复 Redis 配置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(fmtiso), structlog.processors.StackInfoRenderer(), structlog.processors.format_exc_info, structlog.processors.UnicodeDecoder(), structlog.processors.JSONRenderer() ], context_classdict, logger_factorystructlog.stdlib.LoggerFactory(), wrapper_classstructlog.stdlib.BoundLogger, cache_logger_on_first_useTrue, ) # 使用示例 logger structlog.get_logger() logger.info(API request, provideropenai, modelgpt-4, duration0.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_providerprimary_provider, fallback_providerprovider) return response except Exception as e: logger.warning(Provider failed, providerprovider, errorstr(e)) continue raise RuntimeError(fAll 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(daysdays))) 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,