适配器系统已完全集成到所有 OpenAI 兼容接口!
所有本地模型调用现在都通过统一的 _handle_local_model_request() 方法,该方法会:
已集成的接口:
chat_completions - 对话补全embeddings - 文本嵌入image_generations - 文生图image_edits - 图生图audio_speech - 文字转语音 (TTS)audio_transcriptions - 语音转文字 (STT)video_generations - 文生视频image_to_video_generations - 图生视频模型适配器系统提供了统一的接口,让平台能够兼容各种不同的 AI 模型提供商,无需修改业务代码。
系统会根据模型的 base_url 自动识别提供商:
# 硅基流动
base_url = "https://api.siliconflow.cn/v1" → ModelProvider.SILICONFLOW
# 阿里云百炼
base_url = "https://dashscope.aliyuncs.com" → ModelProvider.DASHSCOPE
# 智谱AI
base_url = "https://open.bigmodel.cn/api" → ModelProvider.ZHIPU
# 月之暗面
base_url = "https://api.moonshot.cn/v1" → ModelProvider.MOONSHOT
# DeepSeek
base_url = "https://api.deepseek.com/v1" → ModelProvider.DEEPSEEK
# 通用 OpenAI 兼容
其他 URL → ModelProvider.GENERIC
输入:统一的 OpenAI 格式参数 输出:提供商特定格式的参数
# OpenAI 格式(用户输入)
{
"model": "dall-e-3",
"prompt": "a beautiful sunset",
"size": "1024x1024",
"n": 1
}
# 自动转换为阿里云格式
{
"model": "wanx-v1",
"input": {
"prompt": "a beautiful sunset"
},
"parameters": {
"size": "1024*1024",
"n": 1
}
}
输入:提供商特定格式的响应 输出:统一的 OpenAI 格式响应
# 阿里云原始响应
{
"request_id": "xxx",
"output": {
"results": [
{"url": "https://..."}
]
}
}
# 自动转换为 OpenAI 格式
{
"created": "xxx",
"data": [
{"url": "https://...", "revised_prompt": null}
]
}
特殊处理:
ModelProvider 枚举中添加class ModelProvider(str, Enum):
# ... 现有提供商
NEW_PROVIDER = "new_provider" # 新提供商
detect_provider 中添加检测规则@staticmethod
def detect_provider(base_url: str, model_name: str) -> ModelProvider:
base_url_lower = base_url.lower()
# ... 现有规则
if "newprovider" in base_url_lower:
return ModelProvider.NEW_PROVIDER
class ImageAdapter:
@staticmethod
def adapt_request(provider: ModelProvider, openai_request: Dict[str, Any]) -> Dict[str, Any]:
# ... 现有逻辑
if provider == ModelProvider.NEW_PROVIDER:
# 添加新提供商的请求转换逻辑
adapted = {
"custom_field": openai_request.get("prompt"),
# ...
}
return adapted
return openai_request
@staticmethod
def adapt_response(provider: ModelProvider, provider_response: Dict[str, Any]) -> Dict[str, Any]:
# ... 现有逻辑
if provider == ModelProvider.NEW_PROVIDER:
# 添加新提供商的响应转换逻辑
return {
"created": ...,
"data": [...]
}
return provider_response
from app.services.model_adapters import BaseAdapter, get_adapter
# 1. 检测提供商
provider = BaseAdapter.detect_provider(base_url, model_name)
# 2. 获取适配器
adapter = get_adapter("image") # "chat", "image", "audio_tts", etc.
# 3. 适配请求
adapted_request = adapter.adapt_request(provider, openai_format_request)
# 4. 发送请求
response = await client.post(api_url, json=adapted_request)
# 5. 适配响应
openai_format_response = adapter.adapt_response(provider, response.json())
解决:检查 base_url 是否包含正确的关键词,或在 detect_provider 中添加更精确的规则
解决:检查适配器中的 adapt_request 方法,确保所有必需字段都已映射
解决:检查适配器中的 adapt_response 方法,确保响应格式正确转换