主要变更: - 项目名称统一为"秒思智能审核平台"(替换 SmartAudit) - 完善 TDD 实施评估与计划 (featuredoc/tdd_plan.md V2.0) - 新增项目现状诊断与可行性分析 - 新增前后端测试策略与工具链配置模板 - 新增 CI/CD 集成方案与 Codecov 配置说明 - 标注所有待创建模板文件 - 新增 GitHub 配置脚本 (scripts/setup-github.sh) - 自动配置分支保护规则 - 验证 GitHub CLI 登录状态 - 更新 TASK-005-C 包含分支保护与 Codecov 配置 - 同步更新 F-51/F-52 功能至所有相关文档 - UI 设计 Logo 统一为"秒思" Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
26 KiB
26 KiB
AIProviderConfig.md - AI 服务配置架构设计
| 文档类型 | Technical Design (技术设计文档) |
|---|---|
| 项目名称 | 秒思智能审核平台 (AI 营销内容合规审核平台) |
| 版本号 | V2.1 |
| 日期 | 2026-02-03 |
| 侧重 | AI 服务动态配置、多租户隔离、模型选择 |
版本历史 (Version History)
| 版本 | 日期 | 作者 | 变更说明 |
|---|---|---|---|
| V1.0 | 2026-02-02 | Claude | 初稿:AI 厂商动态配置架构设计 |
| V2.0 | 2026-02-02 | Claude | 重构:简化为统一提供商+三模型配置方案 |
| V2.1 | 2026-02-03 | Claude | 文档一致性修订:明确单提供商模式与可切换原则 |
1. 设计背景与目标
1.1 业务需求
秒思智能审核平台 系统需要调用三类 AI 服务完成视频审核:
| 服务类型 | 用途 | 示例模型 |
|---|---|---|
| 文字处理模型 | Brief 解析、违禁词检测、语义分析、舆情分析 | claude-opus-4-5-20251101, deepseek-chat |
| 视频分析模型 | 画面理解、场景分析、产品识别 | Doubao-Seed-1.6-thinking, qwen-vl-max |
| 音频解析模型 | 视频口播转文字 (ASR) | whisper-large-v3, paraformer-v2 |
1.2 设计目标
| 目标 | 描述 |
|---|---|
| 灵活配置 | 品牌方可在后台自由选择 AI 提供商和模型 |
| 单一提供商 | 每租户仅保留一套提供商配置,必要时手动切换 |
| 统一接入 | 支持 OneAPI/OpenRouter 中转,一套配置调用多种模型 |
| 直连支持 | 也支持直连 Anthropic、OpenAI、DeepSeek 等厂商 |
| 多租户隔离 | 不同品牌方使用独立的 AI 配置和配额 |
| 动态模型列表 | 根据 API Key 自动获取可用模型 |
| 连接测试 | 保存前可测试三个模型的连通性 |
1.3 使用流程
┌─────────────────────────────────────────────────────────────────────────────┐
│ AI 配置使用流程 │
├─────────────────────────────────────────────────────────────────────────────┤
│ │
│ 🛡️ 品牌方配置 AI 服务 │
│ ┌──────────────────────────────────────────────────────────────────────┐ │
│ │ • 选择 AI 提供商 (OneAPI/Anthropic/OpenAI/...) │ │
│ │ • 填写 Base URL 和 API Key │ │
│ │ • 选择三个模型 (文字处理/视频分析/音频解析) │ │
│ │ • 配置参数 (Temperature/Max Tokens) │ │
│ └──────────────────────────────────────────────────────────────────────┘ │
│ │ │
│ │ 自动继承 │
│ ▼ │
│ 👥 代理商 / 👤 达人 │
│ ┌──────────────────────────────────────────────────────────────────────┐ │
│ │ 上传视频 → 系统自动调用品牌方配置的 AI 服务 → 获得审核结果 │ │
│ │ (用户无感知,不知道也不需要关心使用的是哪个 AI) │ │
│ └──────────────────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────────────┘
2. 系统架构
2.1 架构概览
┌─────────────────────────────────────────────────────────────────────────┐
│ 品牌方管理后台 (Brand Admin) │
│ ┌──────────────────────────────────────────────────────────────────┐ │
│ │ 系统设置 → AI 服务配置 │ │
│ │ • 选择提供商 │ │
│ │ • 配置连接信息 │ │
│ │ • 选择模型 │ │
│ │ • 测试连接 │ │
│ └──────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────┐
│ API 层 (FastAPI) │
│ ┌──────────────────────────────────────────────────────────────────┐ │
│ │ GET /api/v1/ai-config - 获取当前配置 │ │
│ │ PUT /api/v1/ai-config - 更新配置 │ │
│ │ POST /api/v1/ai-config/models - 获取可用模型列表 │ │
│ │ POST /api/v1/ai-config/test - 测试连接 (三个模型) │ │
│ └──────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────┐
│ AI 服务路由器 (AIServiceRouter) │
│ ┌──────────────────────────────────────────────────────────────────┐ │
│ │ • 根据租户 ID 获取对应的 AI 配置 │ │
│ │ • 根据任务类型选择对应的模型 │ │
│ │ • 创建 AI 客户端并调用 │ │
│ └──────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────┘
│
┌───────────────┼───────────────┐
▼ ▼ ▼
┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ 文字处理 │ │ 视频分析 │ │ 音频解析 │
│ Claude │ │ 豆包 VL │ │ Whisper │
└─────────────┘ └─────────────┘ └─────────────┘
2.2 核心组件
| 组件 | 职责 |
|---|---|
| AIConfig | 数据模型,存储品牌方的 AI 配置 |
| AIServiceRouter | 路由器,根据租户和任务类型选择模型 |
| AIClientFactory | 工厂类,创建 OpenAI 兼容客户端 |
| ModelRegistry | 模型注册表,缓存可用模型列表 |
| SecretsManager | 加密存储和解密 API Key |
3. 数据模型设计
3.1 AI 配置表 (ai_configs)
CREATE TABLE ai_configs (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
-- 租户
tenant_id UUID NOT NULL UNIQUE, -- 品牌方 ID,一个品牌方只有一条配置
-- 提供商
provider VARCHAR(50) NOT NULL, -- 提供商类型
-- 连接配置
base_url VARCHAR(500) NOT NULL, -- API Base URL
api_key_encrypted BYTEA NOT NULL, -- 加密后的 API Key
-- 模型配置
text_model VARCHAR(100) NOT NULL, -- 文字处理模型
vision_model VARCHAR(100) NOT NULL, -- 视频分析模型
audio_model VARCHAR(100) NOT NULL, -- 音频解析模型
-- 参数配置
temperature DECIMAL(3,2) DEFAULT 0.7, -- 温度参数
max_tokens INT DEFAULT 2000, -- 最大 Token 数
-- 缓存的可用模型列表
available_models JSONB DEFAULT '{}', -- {"text": [...], "vision": [...], "audio": [...]}
models_updated_at TIMESTAMPTZ, -- 模型列表更新时间
-- 状态
is_configured BOOLEAN DEFAULT false, -- 是否已完成配置
last_test_at TIMESTAMPTZ, -- 最后测试时间
last_test_result JSONB, -- 最后测试结果
-- 元数据
created_at TIMESTAMPTZ DEFAULT NOW(),
updated_at TIMESTAMPTZ DEFAULT NOW(),
updated_by UUID,
-- 外键
CONSTRAINT fk_tenant FOREIGN KEY (tenant_id) REFERENCES brands(id)
);
-- 索引
CREATE INDEX idx_ai_config_tenant ON ai_configs(tenant_id);
3.2 提供商类型枚举
from enum import Enum
class AIProvider(str, Enum):
"""支持的 AI 提供商"""
# 中转服务
ONEAPI = "oneapi" # OneAPI 中转
OPENROUTER = "openrouter" # OpenRouter
# 直连厂商 - 国际
ANTHROPIC = "anthropic" # Anthropic Claude
OPENAI = "openai" # OpenAI
# 直连厂商 - 国内
DEEPSEEK = "deepseek" # DeepSeek
QWEN = "qwen" # 阿里云通义千问
DOUBAO = "doubao" # 字节豆包
ZHIPU = "zhipu" # 智谱 GLM
MOONSHOT = "moonshot" # Moonshot (Kimi)
# 提供商默认 Base URL
PROVIDER_DEFAULT_URLS = {
AIProvider.ANTHROPIC: "https://api.anthropic.com/v1",
AIProvider.OPENAI: "https://api.openai.com/v1",
AIProvider.DEEPSEEK: "https://api.deepseek.com/v1",
AIProvider.QWEN: "https://dashscope.aliyuncs.com/compatible-mode/v1",
AIProvider.DOUBAO: "https://ark.cn-beijing.volces.com/api/v3",
AIProvider.ZHIPU: "https://open.bigmodel.cn/api/paas/v4",
AIProvider.MOONSHOT: "https://api.moonshot.cn/v1",
}
3.3 模型能力分类
class ModelCapability(str, Enum):
"""模型能力类型"""
TEXT = "text" # 文字处理
VISION = "vision" # 视觉理解
AUDIO = "audio" # 音频处理
# 已知模型的能力映射(用于分类显示)
MODEL_CAPABILITIES = {
# 文字处理模型
"claude-opus-4-5-20251101": [ModelCapability.TEXT, ModelCapability.VISION],
"claude-sonnet-4-20250514": [ModelCapability.TEXT, ModelCapability.VISION],
"gpt-4o": [ModelCapability.TEXT, ModelCapability.VISION],
"gpt-4o-mini": [ModelCapability.TEXT, ModelCapability.VISION],
"deepseek-chat": [ModelCapability.TEXT],
"deepseek-reasoner": [ModelCapability.TEXT],
"qwen-max": [ModelCapability.TEXT],
"qwen-plus": [ModelCapability.TEXT],
"glm-4": [ModelCapability.TEXT],
"moonshot-v1-128k": [ModelCapability.TEXT],
# 视觉模型
# 注:Logo 检测由系统内置 CV 模型(Grounding DINO)处理,不受此配置影响。
# 品牌方配置的视觉模型仅用于语义场景理解(如环境分析、画面质量判定)。
"qwen-vl-max": [ModelCapability.VISION],
"qwen-vl-plus": [ModelCapability.VISION],
"Doubao-Seed-1.6-thinking": [ModelCapability.VISION],
"doubao-vision-pro": [ModelCapability.VISION],
"glm-4v": [ModelCapability.VISION],
# 音频模型
"whisper-large-v3": [ModelCapability.AUDIO],
"whisper-1": [ModelCapability.AUDIO],
"paraformer-v2": [ModelCapability.AUDIO],
"sensevoice": [ModelCapability.AUDIO],
}
4. API 接口设计
4.1 获取当前配置
GET /api/v1/ai-config
Authorization: Bearer {token}
响应:
{
"provider": "oneapi",
"base_url": "https://oneapi.intelligrow.cn",
"api_key_masked": "sk-****...****",
"models": {
"text": "claude-opus-4-5-20251101",
"vision": "Doubao-Seed-1.6-thinking",
"audio": "whisper-large-v3"
},
"parameters": {
"temperature": 0.7,
"max_tokens": 2000
},
"available_models": {
"text": ["claude-opus-4-5-20251101", "deepseek-chat", "gpt-4o", ...],
"vision": ["Doubao-Seed-1.6-thinking", "qwen-vl-max", "gpt-4o", ...],
"audio": ["whisper-large-v3", "paraformer-v2", ...]
},
"is_configured": true,
"last_test_at": "2026-02-02T10:30:00Z",
"last_test_result": {
"text": {"success": true, "latency_ms": 342},
"vision": {"success": true, "latency_ms": 528},
"audio": {"success": true, "latency_ms": 215}
}
}
模型说明: 同一模型可能同时出现在
text与vision列表中,名称保持一致,仅能力标签不同(如gpt-4o兼具文字与视觉能力)。
未配置说明: 若租户未完成 AI 配置,接口返回明确错误(如 409/404),前端需提示品牌方完成配置;所有 AI 调用在此之前应被阻断。
4.2 更新配置
PUT /api/v1/ai-config
Authorization: Bearer {token}
Content-Type: application/json
{
"provider": "oneapi",
"base_url": "https://oneapi.intelligrow.cn",
"api_key": "sk-xxxxxxxxxxxxxxxxxxxxxxxx",
"models": {
"text": "claude-opus-4-5-20251101",
"vision": "Doubao-Seed-1.6-thinking",
"audio": "whisper-large-v3"
},
"parameters": {
"temperature": 0.7,
"max_tokens": 2000
}
}
4.3 获取可用模型列表
POST /api/v1/ai-config/models
Authorization: Bearer {token}
Content-Type: application/json
{
"provider": "oneapi",
"base_url": "https://oneapi.intelligrow.cn",
"api_key": "sk-xxxxxxxxxxxxxxxxxxxxxxxx"
}
响应:
{
"success": true,
"models": {
"text": [
{"id": "claude-opus-4-5-20251101", "name": "Claude Opus 4.5"},
{"id": "deepseek-chat", "name": "DeepSeek Chat"},
{"id": "gpt-4o", "name": "GPT-4o"},
...
],
"vision": [
{"id": "Doubao-Seed-1.6-thinking", "name": "豆包 Seed 1.6"},
{"id": "qwen-vl-max", "name": "通义千问 VL Max"},
{"id": "gpt-4o", "name": "GPT-4o"},
...
],
"audio": [
{"id": "whisper-large-v3", "name": "Whisper Large V3"},
{"id": "paraformer-v2", "name": "Paraformer V2"},
...
]
}
}
4.4 测试连接
POST /api/v1/ai-config/test
Authorization: Bearer {token}
Content-Type: application/json
{
"provider": "oneapi",
"base_url": "https://oneapi.intelligrow.cn",
"api_key": "sk-xxxxxxxxxxxxxxxxxxxxxxxx",
"models": {
"text": "claude-opus-4-5-20251101",
"vision": "Doubao-Seed-1.6-thinking",
"audio": "whisper-large-v3"
}
}
响应:
{
"success": false,
"results": {
"text": {
"success": true,
"latency_ms": 342,
"model": "claude-opus-4-5-20251101"
},
"vision": {
"success": true,
"latency_ms": 528,
"model": "Doubao-Seed-1.6-thinking"
},
"audio": {
"success": false,
"error": "Model not found or unauthorized",
"model": "whisper-large-v3"
}
},
"message": "1 个模型连接失败,请检查模型名称或 API 权限"
}
5. 核心代码设计
5.1 配置模型 (Pydantic)
# app/models/ai_config.py
from pydantic import BaseModel, Field, SecretStr
from typing import Optional, Dict, List
from decimal import Decimal
from app.models.enums import AIProvider
class AIModelsConfig(BaseModel):
"""三个模型配置"""
text: str = Field(..., description="文字处理模型")
vision: str = Field(..., description="视频分析模型")
audio: str = Field(..., description="音频解析模型")
class AIParametersConfig(BaseModel):
"""参数配置"""
temperature: Decimal = Field(default=Decimal("0.7"), ge=0, le=1)
max_tokens: int = Field(default=2000, ge=100, le=32000)
class AIConfigUpdate(BaseModel):
"""更新 AI 配置请求"""
provider: AIProvider
base_url: str
api_key: SecretStr
models: AIModelsConfig
parameters: AIParametersConfig = AIParametersConfig()
class AIConfigResponse(BaseModel):
"""AI 配置响应"""
provider: AIProvider
base_url: str
api_key_masked: str # 脱敏后的 API Key
models: AIModelsConfig
parameters: AIParametersConfig
available_models: Dict[str, List[dict]]
is_configured: bool
last_test_at: Optional[str]
last_test_result: Optional[dict]
class ModelTestResult(BaseModel):
"""单个模型测试结果"""
success: bool
latency_ms: Optional[int] = None
error: Optional[str] = None
model: str
class TestConnectionResponse(BaseModel):
"""测试连接响应"""
success: bool # 三个都成功才为 True
results: Dict[str, ModelTestResult]
message: str
5.2 AI 服务路由器
# app/services/ai/router.py
from typing import Optional
from uuid import UUID
from app.models.ai_config import AIModelsConfig, AIParametersConfig
from app.repositories.ai_config_repo import AIConfigRepository
from app.services.ai.client_factory import AIClientFactory
class AIServiceRouter:
"""AI 服务路由器 - 根据租户获取配置并调用对应模型"""
def __init__(
self,
config_repo: AIConfigRepository,
client_factory: AIClientFactory,
):
self.config_repo = config_repo
self.client_factory = client_factory
async def get_config(self, tenant_id: UUID) -> dict:
"""获取租户的 AI 配置"""
config = await self.config_repo.get_by_tenant(tenant_id)
if not config or not config.is_configured:
# 未配置时阻断调用并提示品牌方完成配置
raise ValueError(f"AI service not configured for tenant {tenant_id}")
return config
async def chat(
self,
tenant_id: UUID,
messages: list,
model_type: str = "text", # text / vision / audio
**kwargs
) -> dict:
"""统一的对话接口"""
config = await self.get_config(tenant_id)
# 根据类型选择模型
model = getattr(config.models, model_type)
# 获取客户端
client = await self.client_factory.get_client(
base_url=config.base_url,
api_key=config.api_key,
)
# 调用
return await client.chat(
messages=messages,
model=model,
temperature=float(config.parameters.temperature),
max_tokens=config.parameters.max_tokens,
**kwargs
)
async def transcribe(
self,
tenant_id: UUID,
audio_file: bytes,
) -> dict:
"""音频转文字"""
config = await self.get_config(tenant_id)
client = await self.client_factory.get_client(
base_url=config.base_url,
api_key=config.api_key,
)
return await client.transcribe(
audio=audio_file,
model=config.models.audio,
)
5.3 测试连接服务
# app/services/ai/connection_tester.py
import asyncio
from typing import Dict
from openai import AsyncOpenAI
from app.models.ai_config import ModelTestResult, TestConnectionResponse
class AIConnectionTester:
"""AI 连接测试服务"""
async def test_all_models(
self,
base_url: str,
api_key: str,
models: Dict[str, str], # {"text": "...", "vision": "...", "audio": "..."}
) -> TestConnectionResponse:
"""并行测试三个模型"""
# 并行执行测试
tasks = [
self._test_model(base_url, api_key, model_type, model_id)
for model_type, model_id in models.items()
]
results = await asyncio.gather(*tasks, return_exceptions=True)
# 整理结果
result_dict = {}
all_success = True
failed_count = 0
for model_type, result in zip(models.keys(), results):
if isinstance(result, Exception):
result_dict[model_type] = ModelTestResult(
success=False,
error=str(result),
model=models[model_type]
)
all_success = False
failed_count += 1
else:
result_dict[model_type] = result
if not result.success:
all_success = False
failed_count += 1
# 生成消息
if all_success:
message = "所有模型连接成功"
else:
message = f"{failed_count} 个模型连接失败,请检查模型名称或 API 权限"
return TestConnectionResponse(
success=all_success,
results=result_dict,
message=message
)
async def _test_model(
self,
base_url: str,
api_key: str,
model_type: str,
model_id: str,
) -> ModelTestResult:
"""测试单个模型"""
import time
client = AsyncOpenAI(base_url=base_url, api_key=api_key)
start_time = time.time()
try:
if model_type == "audio":
# ASR 模型测试 - 检查模型是否存在
models = await client.models.list()
model_exists = any(m.id == model_id for m in models.data)
if not model_exists:
return ModelTestResult(
success=False,
error="Model not found",
model=model_id
)
else:
# 文字/视觉模型测试 - 发送简单请求
await client.chat.completions.create(
model=model_id,
messages=[{"role": "user", "content": "Hi"}],
max_tokens=5,
)
latency_ms = int((time.time() - start_time) * 1000)
return ModelTestResult(
success=True,
latency_ms=latency_ms,
model=model_id
)
except Exception as e:
return ModelTestResult(
success=False,
error=str(e),
model=model_id
)
6. 安全设计
6.1 API Key 加密存储
- 使用 AES-256-GCM 加密存储 API Key
- 主密钥从环境变量或密钥管理服务 (Vault/KMS) 获取
- API 响应中永不返回完整 API Key,仅返回脱敏版本
def mask_api_key(api_key: str) -> str:
"""API Key 脱敏"""
if len(api_key) <= 8:
return "****"
return f"{api_key[:4]}...{api_key[-4:]}"
6.2 权限控制
| 操作 | 品牌方管理员 | 代理商 | 达人 |
|---|---|---|---|
| 查看 AI 配置 | ✅ (本租户) | ❌ | ❌ |
| 修改 AI 配置 | ✅ (本租户) | ❌ | ❌ |
| 测试连接 | ✅ (本租户) | ❌ | ❌ |
| 查看完整 API Key | ❌ | ❌ | ❌ |
注:系统不设独立的系统管理员角色。AI 配置由各品牌方管理员自行管理本租户的配置。
7. 界面设计
详见 User_Role_Interfaces.md 第 4.6 章「AI 服务配置」
7.1 界面入口
品牌方端 → 系统设置 → AI 服务配置
7.2 界面结构
- 提供商选择 - 下拉选择 AI 提供商
- 连接配置 - Base URL 和 API Key 输入
- 获取模型按钮 - 点击后从 API 获取可用模型列表
- 模型配置 - 三个下拉框分别选择文字/视觉/音频模型
- 参数配置 - Temperature 滑块和 Max Tokens 输入
- 测试连接按钮 - 并行测试三个模型
- 保存配置按钮
8. 相关文档
| 文档 | 说明 |
|---|---|
| UIDesign.md | UI 设计规范(第 10 章 AI 配置界面) |
| User_Role_Interfaces.md | 用户角色与界面规范 |
| tasks.md | 开发任务清单 |
| DevelopmentPlan.md | 开发计划与技术架构 |