multimedia-backend-integrator

多媒体后端集成指南

为 MassGen 统一的 generate_media 工具添加新媒体生成后端的参考指南。

架构概览

_base.py          -- Registration: API keys, default models, priority lists
_selector.py      -- Auto-selection logic: picks best backend by key + priority
_image.py         -- Image backends: OpenAI, Google (Gemini/Imagen), Grok, OpenRouter
_video.py         -- Video backends: Grok, Google Veo, OpenAI Sora
_audio.py         -- Audio backends: ElevenLabs, OpenAI TTS
generate_media.py -- Entry point: routing, validation, batch mode, image-to-image

添加新后端的完整检查清单

1. 注册（_base.py）

• 在 BACKEND_API_KEYS 中添加映射，将后端名称映射到环境变量
• 在 DEFAULT_MODELS 中添加映射，将后端名称映射到每种受支持类型对应的 {MediaType: model_name}
• 在 BACKEND_PRIORITY 中，根据媒体类型将其插入正确位置

2. 实现（_image.py / _video.py / _audio.py）

• 在模块顶部添加 SDK 的 import
• 实现 _generate_{media}_{backend}(config) -> GenerationResult
• 首先检查 API 密钥；如果缺失，则返回错误结果
• 使用 API 密钥创建 SDK 客户端
• 将 config.* 字段映射到 SDK 参数
• 处理续接（如适用）——请参阅续接存储模式
• 将输出字节写入 config.output_path
• 返回包含元数据的 GenerationResult
• 使用 try/except 包装并记录错误

3. 更新分发器

• 在对应媒体类型的 generate_{media}() 函数中添加 elif backend == "new_backend":

4. 支持图像到图像（generate_media.py）

• 在 _generate_single_with_input_images 中，将后端名称添加到 selected_backend not in (...) 检查中
• 添加回退逻辑：在自动选择链中添加 elif has_api_key("new_backend"):
• 更新错误消息，注明新后端和环境变量

5. 文档

• TOOL.md：在 frontmatter 中添加环境变量，在表格中添加后端，并添加关键词
• generate_media.py 文档字符串：更新 backend_type 列表和 Supported Backends

6. 测试

• 后端注册测试（API 密钥、默认模型、优先级顺序）
• 自动选择测试（仅使用此后端的密钥，以及使用多个密钥）
• SDK 调用验证（传入的参数正确）
• 正确写入输出文件
• 续接流程（如适用）
• 错误处理（缺少密钥、API 错误）
• 参数映射（aspect_ratio、size、duration）
• 更新断言优先级列表长度或内容的现有测试

续接存储模式

支持迭代编辑的每个后端都需要一种续接机制：

后端	存储类型	键格式	存储内容	续接方式
OpenAI	无状态（服务端）	response.id	本地不存储任何内容	在下一次调用中传入 previous_response_id
Gemini	_GeminiChatStore（内存中）	gemini_chat_{uuid12}	(client, chat) 元组	重用 chat 对象调用 send_message()；让 client 保持活动状态，以防 HTTP 连接被垃圾回收
Grok	_GrokImageStore（内存中）	grok_img_{uuid12}	Base64 字符串	将存储的 Base64 作为 image_url 数据 URI 传入

存储模式模板

class _NewBackendStore:
    def __init__(self, max_items: int = 50):
        self._store: OrderedDict[str, Any] = OrderedDict()
        self._max = max_items

    def save(self, data: Any) -> str:
        store_id = f"prefix_{uuid.uuid4().hex[:12]}"
        if len(self._store) >= self._max:
            self._store.popitem(last=False)  # LRU eviction
        self._store[store_id] = data
        return store_id

    def get(self, store_id: str) -> Any | None:
        return self._store.get(store_id)

_store = _NewBackendStore()

常见问题

1. 未加入优先级列表——显式指定时后端可以正常工作，但永远不会被自动选中
2. 同步与异步——某些 SDK 仅支持同步；如有需要，请使用 asyncio.to_thread() 包装
3. 临时 URL——某些 API 会返回临时 URL；始终优先使用 Base64，或立即下载
4. 无效的 duration——duration or default 会将 0 视为假值；请使用 if duration is not None
5. 现有测试受影响——添加到优先级列表会改变自动选择行为；请更新会清除环境变量的现有测试
6. 图像到图像的限制——_generate_single_with_input_images 函数有后端允许列表

参考文件

文件	用途
massgen/tool/_multimodal_tools/generation/_base.py	API 密钥、默认模型、优先级
massgen/tool/_multimodal_tools/generation/_selector.py	后端自动选择逻辑
massgen/tool/_multimodal_tools/generation/_image.py	图像生成后端
massgen/tool/_multimodal_tools/generation/_video.py	视频生成后端
massgen/tool/_multimodal_tools/generation/_audio.py	音频生成后端
massgen/tool/_multimodal_tools/generation/generate_media.py	入口点和路由
massgen/tool/_multimodal_tools/TOOL.md	面向用户的文档
massgen/tests/test_grok_multimedia_generation.py	参考：Grok 后端测试
massgen/tests/test_grok_multimedia_backend_selection.py	参考：Grok 选择测试
massgen/tests/test_multimodal_image_backend_selection.py	参考：图像选择测试