多 Provider 图片生成服务 — 架构设计文档

多 Provider 图片生成服务 — 架构设计文档文档版本 v2.0脱敏声明 文档中所有具体域名、API Key、SDK 包名、模型名称均已替换为占位符描述正式部署时请替换为实际值。适用场景 文档生成、图片创作、AI 图像生成等需要调用外部图像 API 的服务。背景与目标1.1 为什么需要多 Provider在 AI 图像生成应用中依赖单一模型供应商存在以下风险风险类型描述可用性风险单点故障导致服务不可用影响用户体验成本风险高峰时段 API 定价波动影响成本可控性能力覆盖风险单一模型在特定类型图片如文字排版、风格一致性上表现不稳定网络依赖风险某些 Provider 在特定地区无法访问或响应缓慢1.2 本服务的设计目标目标说明高可用主备 Provider 自动切换单一节点故障不影响整体服务成本可控优先使用低单价 Provider必要时切换高单价兜底能力互补不同模型各有优势按场景类型选择最适合的生成器灵活扩展新增 Provider 只需实现统一接口无需修改核心调度逻辑网络兼容自动适配不同 Provider 的网络访问要求如代理/直连整体架构2.1 架构总览┌─────────────────────────────────────────────────────────┐│ 调用入口层 ││ 文档生成服务 / 任务池 / SSE 流式响应 │└──────────────────┬──────────────────────────────────────┘│▼┌─────────────────────────────────────────────────────────┐│ ImageGenerationService ││ ┌─────────────────────────────────────────────────┐ ││ │ 调度策略Primary / Fallback │ ││ │ · 有本地参考图 → 优先 Primary支持图片理解 │ ││ │ · Primary 重试耗尽 → 自动切换 Fallback │ ││ │ · 无本地参考图 有远端资产ID → 直接用 Fallback │ ││ └─────────────────────────────────────────────────┘ │└──────────┬──────────────────────────┬──────────────────┘│ primary_provider │ fallback_provider▼ ▼┌──────────────────────┐ ┌────────────────────────────────┐│ PrimaryGenerator │ │ FallbackGenerator ││ ┌────────────────┐ │ │ ┌────────────────────────┐ ││ │ Provider A │ │ │ │ Provider B资产引用 │ ││ │ · 本地参考图 │ │ │ │ · 资产 ID 列表 │ ││ │ · base64 上传 │ │ │ │ · 无需本地文件 │ ││ │ · 流式 SSE │ │ │ · 异步轮询结果 │ ││ └────────────────┘ │ │ └────────────────────────┘ │└──────────────────────┘ └────────────────────────────────┘│ │▼ ▼┌─────────────┐ ┌─────────────────┐│ 供应商 A │ │ 供应商 B ││ (海外 API) │ │ (国内直连 API) ││ VPN 路由 │ │ 直连 │└─────────────┘ └─────────────────┘2.2 核心设计思想注册表模式所有 Provider 在模块加载时自动注册新增 Provider 零改动主备降级Primary 失败自动切换 Fallback保证服务可用性滑动窗口并发池控制并发数、失败重试、自动入队调度网络兼容不同 Provider 的网络要求代理/直连在 Provider 层各自处理核心组件详解3.1 Provider 注册中心文件src/core/providers/provider_registry.py采用注册表模式Registry Pattern所有 Provider 在模块加载时自动注册到全局字典from typing import Dict, TypePROVIDER_REGISTRY: Dict[str, Type[“BaseImageGenerator”]] {}def register_provider(name: str, generator_class: Type[“BaseImageGenerator”]) - None:“”“注册 Provider自动校验接口兼容性”“”if not issubclass(generator_class, BaseImageGenerator):raise TypeError(f{generator_class.name} 必须继承 BaseImageGenerator)PROVIDER_REGISTRY[name] generator_classdef create_generator(provider: str, **config) - “BaseImageGenerator”:“”“根据名称创建 Provider 实例”“”if provider not in PROVIDER_REGISTRY:available , “.join(PROVIDER_REGISTRY.keys())raise ValueError(fUnknown provider: ‘{provider}’. Available: {available}”)return PROVIDER_REGISTRYprovider扩展新 Provider 的标准流程实现 BaseImageGenerator 抽象类在对应模块末尾调用 register_provider(“your_provider”, YourProviderGenerator)在配置文件中添加 Provider 名称即可启用3.2 统一接口契约文件src/core/providers/base_generator.py所有 Provider 必须继承 BaseImageGenerator 并实现以下方法from abc import ABC, abstractmethodfrom dataclasses import dataclass, fieldfrom typing import Optional, Dict, Any, Listdataclassclass ImageGenerationResult:“”“统一返回格式”“”success: boolerror: Optional[str] Noneerror_code: Optional[str] None # 结构化错误码便于 Fallback 判断image_path: Optional[str] None # 本地保存路径image_data: Optional[bytes] None # 原始图片数据image_url: Optional[str] None # 远程图片 URLmetadata: Dict[str, Any] field(default_factorydict) # Provider 原始响应class BaseImageGenerator(ABC):“”“图片生成器抽象基类”“”property abstractmethod def provider_name(self) - str: Provider 标识名称 pass property abstractmethod def supports_reference_images(self) - bool: 是否支持参考图输入影响调度策略 pass property abstractmethod def requires_async_mode(self) - bool: 是否需要异步轮询模式 pass abstractmethod def generate( self, prompt: str, reference_image_paths: Optional[List[str]] None, reference_asset_ids: Optional[List[str]] None, **kwargs ) - ImageGenerationResult: 同步生成 pass abstractmethod async def generate_async( self, prompt: str, reference_image_paths: Optional[List[str]] None, reference_asset_ids: Optional[List[str]] None, **kwargs ) - ImageGenerationResult: 异步生成流式/SSE 场景 pass abstractmethod def download_and_save_image(self, url: str, save_path: str) - bool: 下载并保存图片 pass3.3 Primary / Fallback 调度器文件src/core/services/image_generation_service.pyimport asynciofrom typing import Optional, List, Dict, Anyclass ImageGenerationService:“”统一调度器自动选择 Primary / Fallback Provider“”def __init__( self, primary_provider: str, fallback_provider: str, providers_config: Dict[str, Dict[str, Any]], ): self.primary create_provider(primary_provider, providers_config.get(primary_provider, {})) self.fallback create_provider(fallback_provider, providers_config.get(fallback_provider, {})) async def generate( self, prompt: str, reference_image_paths: Optional[List[str]] None, reference_asset_ids: Optional[List[str]] None, **kwargs ) - ImageGenerationResult: # ── 调度策略 ── # 1. 有本地参考图 → 优先 Primary if reference_image_paths and self.primary.supports_reference_images: result await self._try_primary(prompt, reference_image_paths, **kwargs) if result.success: return result # 2. Primary 重试耗尽 → 切换 Fallback if self._is_retry_exhausted(result): if self.fallback.requires_async_mode: return await self._try_fallback(prompt, reference_asset_ids, **kwargs) else: return await self._try_fallback(prompt, reference_image_paths, **kwargs) return result # 3. 无本地参考图 有资产ID → 直接走 Fallback if reference_asset_ids and self.fallback.requires_async_mode: return await self._try_fallback(prompt, reference_asset_ids, **kwargs) # 4. 默认走 Primary return await self._try_primary(prompt, reference_image_paths, **kwargs) async def _try_primary(self, prompt: str, assets, **kwargs) - ImageGenerationResult: # 实际调用 重试逻辑指数退避 pass async def _try_fallback(self, prompt: str, assets, **kwargs) - ImageGenerationResult: # Fallback 调用 pass def _is_retry_exhausted(self, result: ImageGenerationResult) - bool: 判断是否应该触发 Fallback # 429 限流、内容审核拒绝、网络超时 → 应切换 RETRYABLE_CODES {RATE_LIMIT, CONTENT_FILTER, TIMEOUT, NETWORK_ERROR} return result.error_code in RETRYABLE_CODES3.4 并发任务池文件src/core/services/image_task_pool.py采用滑动窗口模式保证容器中始终有 N 个并发任务运行import asynciofrom typing import List, Callable, Any, Coroutineclass ImageGenerationTaskPool:“”并发任务池滑动窗口- 控制最大并发数避免 Provider 限流- 失败自动重试自动入队“”def __init__(self, max_workers: int 5, max_retries: int 2): self.max_workers max_workers self.max_retries max_retries self.running: int 0 self.pending_queue: asyncio.Queue asyncio.Queue() self.retry_counts: Dict[int, int] {} async def submit_batch( self, tasks: List[Coroutine], on_progress: Callable[[int, int, Any], None] None, ) - List[Any]: 提交一批任务自动控制并发 Args: tasks: 任务列表 on_progress: 进度回调 (idx, total, result) results [None] * len(tasks) for idx, task in enumerate(tasks): await self.pending_queue.put((idx, task)) async def worker(): while True: try: idx, task_coro self.pending_queue.get_nowait() except asyncio.QueueEmpty: break self.running 1 try: result await task_coro results[idx] result if not self._is_success(result) and self.retry_counts.get(idx, 0) self.max_retries: self.retry_counts[idx] self.retry_counts.get(idx, 0) 1 await self.pending_queue.put((idx, task_coro)) if on_progress: await on_progress(idx, len(tasks), result) finally: self.running - 1 self.pending_queue.task_done() workers [asyncio.create_task(worker()) for _ in range(self.max_workers)] await asyncio.gather(*workers) return results def _is_success(self, result: Any) - bool: return getattr(result, success, False)3.5 提示词管理器按场景类型文件src/core/prompts/prompt_manager.py每个 Provider 维护独立的 YAML 提示词目录按场景类型精准匹配yaml_prompts/provider_a/ ← Provider A 专用提示词封面.yaml背景页.yaml内容页.yaml数据图表.yaml…provider_b/ ← Provider B 专用提示词封面.yaml背景页.yaml内容页.yaml…使用方式class PromptManager:definit(self, prompts_dir: str “yaml_prompts”):self.prompts_dir prompts_dirself._cache: Dict[str, Dict[str, str]] {}def get_prompt(self, provider: str, page_type: str) - str: cache_key f{provider}:{page_type} if cache_key not in self._cache: path Path(self.prompts_dir) / provider / f{page_type}.yaml self._cache[cache_key] yaml.safe_load(path.read_text()) return self._cache[cache_key][prompt]调度决策树用户请求│├── 有本地参考图路径│ ││ ├─ 是 ──→ 检查 Primary 是否支持参考图│ │ ││ │ ├─ 支持 ──→ 调用 Primary│ │ │ ├─ 成功 ──→ 返回结果│ │ │ ├─ 限流/超时/审核拒绝 ──→ 切换 Fallback│ │ │ └─ 其他错误 ──→ 返回错误│ │ ││ │ └─ 不支持 ──→ 直接调用 Fallback│ ││ └─ 否│ ││ ├── 有远端资产 ID 列表│ │ ││ │ ├─ 是 ──→ 调用 Fallback支持资产引用│ │ └─ 否 ──→ 调用 Primary无参考图模式│ │└── 任务池层处理并发、重试、进度推送Provider 配置说明5.1 Provider APrimary适用于需要图片理解/风格迁移、支持本地参考图输入的场景。配置项说明API 端点https://api.provider-a.example/v1/chat/completions鉴权方式Bearer Token通过环境变量注入参考图输入本地文件路径 → base64 压缩后内嵌请求体输出格式流式 SSE内含 base64 编码图片数据重试策略429 限流 / 内容审核拒绝指数退避最多 3 次5.2 Provider BFallback适用于国内直连、成本较低、支持资产 ID 引用的场景。配置项说明API 端点国内直连域名无需代理鉴权方式云服务商 SecretId / SecretKey通过配置文件注入参考图输入预上传后获得的资产 ID 列表0~3 个输出格式异步任务 → 轮询获取结果 → 下载图片 URL轮询策略每 3 秒轮询一次最多 120 次约 6 分钟超时如何接入新的 Provider以接入一个新 Provider 为例完整步骤如下6.1 实现 Provider 类文件src/core/providers/generator_openai.pyfrom .base_generator import BaseImageGenerator, ImageGenerationResult, register_providerfrom .network_helper import download_image_bytesimport requests, base64, osclass OpenAIImageGenerator(BaseImageGenerator):def __init__(self, api_key: str, base_url: str https://api.openai.com, **kwargs): self.api_key api_key self.base_url base_url self.max_retries kwargs.get(max_retries, 3) property def provider_name(self) - str: return openai property def supports_reference_images(self) - bool: return True # 支持图片参考 property def requires_async_mode(self) - bool: return False # 同步返回 def _prepare_reference_images(self, paths: List[str]) - List[Dict]: 将本地图片转为 base64 格式 images [] for path in paths: with open(path, rb) as f: data base64.b64encode(f.read()).decode() images.append({type: image_url, image_url: {url: fdata:image/png;base64,{data}}}) return images async def generate_async( self, prompt: str, reference_image_paths: List[str] None, **kwargs ) - ImageGenerationResult: try: headers {Authorization: fBearer {self.api_key}, Content-Type: application/json} body {prompt: prompt, n: 1, size: 1024x1024} if reference_image_paths: body[images] self._prepare_reference_images(reference_image_paths) resp requests.post( f{self.base_url}/v1/images/generations, headersheaders, jsonbody, timeout60 ) if resp.status_code 200: image_url resp.json()[data][0][url] image_data download_image_bytes(image_url) return ImageGenerationResult(successTrue, image_dataimage_data, metadataresp.json()) else: return ImageGenerationResult( successFalse, errorresp.text, error_codeself._classify_error(resp.status_code) ) except Exception as e: return ImageGenerationResult(successFalse, errorstr(e), error_codeNETWORK_ERROR) def generate(self, prompt: str, **kwargs) - ImageGenerationResult: return asyncio.run(self.generate_async(prompt, **kwargs)) def download_and_save_image(self, url: str, save_path: str) - bool: data download_image_bytes(url) if data: os.makedirs(os.path.dirname(save_path), exist_okTrue) with open(save_path, wb) as f: f.write(data) return True return False def _classify_error(self, status_code: int) - str: mapping {429: RATE_LIMIT, 400: BAD_REQUEST, 401: AUTH_ERROR, 500: SERVER_ERROR} return mapping.get(status_code, UNKNOWN_ERROR)注册到注册中心register_provider(“openai”, OpenAIImageGenerator)6.2 添加配置文件config/image_providers.yamlproviders:openai:enabled: trueapi_key: ${OPENAI_API_KEY}base_url: “https://api.openai.com”max_retries: 3timeout: 60siliconflow: # Fallback Providerenabled: trueapi_key: ${SILICONFLOW_API_KEY}base_url: “https://api.siliconflow.cn/v1”max_retries: 3timeout: 1206.3 启用 Provider在应用初始化时from src.core.providers import create_generatorprimary create_generator(“openai”, config{“api_key”: os.getenv(“OPENAI_API_KEY”)})fallback create_generator(“siliconflow”, config{“api_key”: os.getenv(“SILICONFLOW_API_KEY”)})service ImageGenerationService(primaryprimary, fallbackfallback)网络路由策略流量类型路由方式说明Provider A海外 APIVPN / 代理海外 endpoint需配置出口代理Provider B国内 API直连国内云服务直接访问结果图片下载直连CDN / 云存储均为国内域名推荐实现 在系统代理层如 Clash Verge TUN 模式按域名规则分流业务代码无感知。详见多网络环境隔离方案目录结构src/├── core/│ ├── providers/│ │ ├── provider_registry.py ← Provider 注册中心│ │ ├── base_generator.py ← 统一接口契约抽象基类│ │ ├── generator_provider_a.py ← Provider A 实现示例│ │ ├── generator_provider_b.py ← Provider B 实现示例│ │ ├── generator_openai.py ← OpenAI Provider 示例│ │ └── network_helper.py ← 网络请求辅助工具│ └── services/│ ├── image_generation_service.py ← 调度器Primary/Fallback│ └── image_task_pool.py ← 并发任务池├── prompts/│ └── page_type_manager.py ← 提示词管理器└── config/└── image_providers.yaml ← Provider 配置文件已知限制与改进方向9.1 当前版本已知限制编号限制描述改进建议状态L1全局单例无法重置多租户/测试场景串台会话级实例工厂 TTL 清理✅ 已修复L2Fallback 触发条件依赖字符串关键字随时失效结构化 ErrorCode 枚举✅ 已修复L3提示词与 Provider 强耦合多级兜底链 自动降级日志✅ 已修复L4部分 Provider 接受语义尺寸如2K不支持精确像素映射在调度层统一归一化为像素尺寸待处理9.2 可选增强方向增强项说明指标采集接入 Prometheus采集每个 Provider 的成功率、平均耗时、成本熔断器当某个 Provider 连续失败 N 次后自动熔断降级不再尝试灰度切换支持按用户、按场景类型选择 Provider逐步放量成本优化增加用量统计高峰期自动切换到更经济的 Provider多级兜底从 Primary → Fallback1 → Fallback2支持更长的降级链安全注意事项密钥不硬编码所有 API Key / Secret 通过环境变量注入不提交到代码仓库文件路径校验reference_image_paths 必须为已存在的本地文件防止路径穿越攻击资产 ID 隔离资产 ID 仅来自可信数据源数据库/Redis不接受用户直接传入错误信息脱敏向上层暴露的错误信息不含内部路径、堆栈、Key 信息部署检查清单[ ] 所有 Provider 的 API Key / Secret 已正确注入环境变量[ ] 海外 Provider 的 VPN 代理已启动并验证连通性[ ] 国内 Provider 的云端配额充足未达上限[ ] 参考图预上传流程已验证可用[ ] max_workers 根据服务器 CPU/内存调整建议 CPU 核数 / 2[ ] 任务池并发数与 Provider 侧 Rate Limit 匹配[ ] 图片输出目录已创建且有写权限[ ] 域名分流规则已配置详见网络隔离方案