feat: 可选 TwelveLabs 视频理解 provider（Pegasus 问答 + Marengo 向量）/ opt-in TwelveLabs video understanding

.env.example

@@ -32,6 +32,20 @@ ASR_COMPUTE_TYPE=int8
 ASR_MAX_INPUT_SECONDS=60
 ASR_MAX_INFERENCE_BATCH_SIZE=1
 
+# ============================================
+# TwelveLabs 视频理解（可选，默认关闭）
+# ============================================
+# 设置后可将「视频问答」切换为 Pegasus —— 直接理解视频画面而非转录文本，
+# 并用 Marengo 生成 512 维向量做片段语义检索。
+# 不设置时不影响任何现有功能（问答仍走默认的转录文本 + OpenAI）。
+# 免费 API key: https://twelvelabs.io （有较慷慨的免费额度）
+# 需安装可选依赖: uv sync --extra twelvelabs
+# TWELVELABS_API_KEY=your-twelvelabs-key
+# VIDEO_QA_PROVIDER=twelvelabs          # 设为 twelvelabs 启用 Pegasus 视频问答
+# TWELVELABS_PEGASUS_MODEL=pegasus1.5
+# TWELVELABS_MARENGO_MODEL=marengo3.0
+# TWELVELABS_QA_MAX_TOKENS=2048
+
 # ============================================
 # ANP服务配置（可选）
 # ============================================

README.md

@@ -347,6 +347,31 @@ B站结果：8 个视频
 | `WHISPER_MODEL_SIZE` | Whisper模型大小 | `base`                      | ✅ |
 | `APP_HOST` | 服务监听地址 | `0.0.0.0`                   | ❌ |
 | `APP_PORT` | 服务端口 | `8999`                      | ❌ |
+| `TWELVELABS_API_KEY` | TwelveLabs API 密钥（可选，启用视频问答增强） | -        | ❌ |
+| `VIDEO_QA_PROVIDER` | 视频问答 provider，设为 `twelvelabs` 启用 Pegasus | -    | ❌ |
+
+### 🎬 TwelveLabs 视频理解（可选增强）
+
+默认情况下，视频问答基于**转录文本** + OpenAI。如果希望模型直接理解**视频画面**
+（动作、场景、画面中的物体等转录文本无法表达的信息），可选启用 [TwelveLabs](https://twelvelabs.io)：
+
+- **Pegasus** —— 直接对视频内容进行问答；
+- **Marengo** —— 生成 512 维多模态向量，可对抽取出的视频片段做语义检索。
+
+启用步骤（完全可选，不配置时现有功能不受任何影响）：
+
+```bash
+# 1. 安装可选依赖
+uv sync --extra twelvelabs
+
+# 2. 在 .env 中配置（免费 key: https://twelvelabs.io，有较慷慨的免费额度）
+TWELVELABS_API_KEY=your-twelvelabs-key
+VIDEO_QA_PROVIDER=twelvelabs
+```
+
+> 说明：Pegasus 需要可公开访问的视频 URL（或先上传为 asset，direct 上传上限 200MB），
+> 且被分析的视频时长需 ≥ 4 秒。
+
 ### Whisper 模型选择
 
 | 模型 | 参数量 | GPU 显存需求 (fp16) | CPU 内存需求 (int8) | 相对速度 | 质量 | 推荐场景 |

backend/config/__init__.py

@@ -9,6 +9,7 @@
     get_asr_config,
     get_whisper_config,
     get_openai_config,
+    get_twelvelabs_config,
     get_language_name
 )
 
@@ -20,5 +21,6 @@
     'get_asr_config',
     'get_whisper_config',
     'get_openai_config',
+    'get_twelvelabs_config',
     'get_language_name'
 ]

backend/config/ai_config.py

@@ -161,11 +161,56 @@ def is_configured(self) -> bool:
         return bool(self.api_key)
 
 
+@dataclass
+class TwelveLabsConfig:
+    """TwelveLabs 配置（可选，默认关闭）
+
+    需要可选依赖 ``twelvelabs`` 与环境变量 ``TWELVELABS_API_KEY``。
+    未配置时不影响任何现有功能，问答仍走默认的转录文本 + OpenAI 路径。
+    """
+    api_key: Optional[str] = None
+    # 是否将视频问答切换到 TwelveLabs Pegasus（直接理解视频画面，而非转录文本）
+    use_for_qa: bool = False
+    # Pegasus（视频理解 / 问答）
+    pegasus_model: str = "pegasus1.5"
+    qa_max_tokens: int = 2048
+    # Marengo（视频 / 文本多模态向量，512 维，用于片段语义检索）
+    marengo_model: str = "marengo3.0"
+
+    def __post_init__(self):
+        self.api_key = os.getenv("TWELVELABS_API_KEY")
+
+        env_use = os.getenv("VIDEO_QA_PROVIDER")
+        if env_use and env_use.strip().lower() == "twelvelabs":
+            self.use_for_qa = True
+
+        env_pegasus = os.getenv("TWELVELABS_PEGASUS_MODEL")
+        if env_pegasus:
+            self.pegasus_model = env_pegasus
+
+        env_marengo = os.getenv("TWELVELABS_MARENGO_MODEL")
+        if env_marengo:
+            self.marengo_model = env_marengo
+
+        env_max_tokens = os.getenv("TWELVELABS_QA_MAX_TOKENS")
+        if env_max_tokens:
+            try:
+                self.qa_max_tokens = max(512, int(env_max_tokens))
+            except ValueError:
+                pass
+
+    @property
+    def is_configured(self) -> bool:
+        """API key 已设置时视为可用。"""
+        return bool(self.api_key)
+
+
 @dataclass
 class AIServiceConfig:
     """AI服务统一配置"""
     asr: ASRConfig
     openai: OpenAIConfig
+    twelvelabs: TwelveLabsConfig
 
     # 文本处理配置
     max_chars_per_chunk: int = 4000
@@ -197,7 +242,8 @@ def __post_init__(self):
 # 创建全局配置实例
 ai_config = AIServiceConfig(
     asr=ASRConfig(),
-    openai=OpenAIConfig()
+    openai=OpenAIConfig(),
+    twelvelabs=TwelveLabsConfig()
 )
 
 
@@ -216,6 +262,11 @@ def get_openai_config() -> OpenAIConfig:
     return ai_config.openai
 
 
+def get_twelvelabs_config() -> TwelveLabsConfig:
+    """获取TwelveLabs配置（可选）"""
+    return ai_config.twelvelabs
+
+
 def get_ai_config() -> AIServiceConfig:
     """获取完整AI配置"""
     return ai_config

backend/core/state.py

@@ -54,8 +54,17 @@ def get_video_download_service():
 def get_video_qa_service():
     global _video_qa_service
     if _video_qa_service is None:
-        from backend.services.video_qa_service import VideoQAService
-        _video_qa_service = VideoQAService()
+        # 可选：当 VIDEO_QA_PROVIDER=twelvelabs 且已配置 key 时，
+        # 使用 Pegasus 直接对视频画面问答；否则保持默认的转录文本 + OpenAI 路径。
+        from backend.config.ai_config import get_twelvelabs_config
+        tl_config = get_twelvelabs_config()
+        if tl_config.use_for_qa and tl_config.is_configured:
+            from backend.services.twelvelabs_service import TwelveLabsVideoQAService
+            _video_qa_service = TwelveLabsVideoQAService()
+            logger.info("视频问答使用 TwelveLabs Pegasus provider")
+        else:
+            from backend.services.video_qa_service import VideoQAService
+            _video_qa_service = VideoQAService()
     return _video_qa_service
 
 

backend/routers/qa.py

@@ -211,7 +211,8 @@ async def video_qa_stream(request: Request):
 
         if not question:
             raise HTTPException(status_code=400, detail="问题不能为空")
-        if not transcript:
+        # 默认转录文本路径需要 transcript；TwelveLabs Pegasus 路径基于视频 URL，无需 transcript
+        if not transcript and not video_url:
             raise HTTPException(status_code=400, detail="转录文本不能为空")
         if not get_video_qa_service().is_available():
             raise HTTPException(status_code=503, detail="AI服务暂时不可用，请稍后重试")

backend/services/twelvelabs_service.py

@@ -0,0 +1,156 @@
+"""
+TwelveLabs 服务（可选）
+
+提供两项基于视频本身（而非转录文本）的能力，作为现有功能的**可选增强**：
+
+1. ``TwelveLabsVideoQAService`` —— 使用 Pegasus 直接对视频画面进行问答。
+   与 ``VideoQAService`` 接口一致（``answer_question_stream`` / ``is_available``），
+   因此可作为视频问答的可选 provider 无缝替换；未配置时不影响默认的
+   "转录文本 + OpenAI" 路径。
+
+2. ``embed_text`` / ``embed_video`` —— 使用 Marengo 生成 512 维多模态向量，
+   可用于对抽取出的视频片段做语义检索。
+
+依赖可选包 ``twelvelabs``（``pip install "twelvelabs>=1.2.8"`` 或
+``uv sync --extra twelvelabs``）以及环境变量 ``TWELVELABS_API_KEY``。
+免费额度可在 https://twelvelabs.io 获取。
+
+设计说明：
+- Pegasus 1.5 不接受裸 ``video_id``，需要可公开访问的 URL 或已上传的
+  ``asset_id``。本服务对带有公开 URL 的视频直接走 URL 路径；
+  对本地文件可由调用方先上传为 asset（``upload_asset``，direct 上传上限 200MB）。
+- Pegasus 要求被分析的视频时长 >= 4 秒。
+- Marengo REST 接口的原始向量字段名为 ``float``，Python SDK 暴露为 ``float_``。
+"""
+import asyncio
+import logging
+from typing import List, Optional
+
+from backend.config.ai_config import get_twelvelabs_config
+
+logger = logging.getLogger(__name__)
+
+
+def _get_client():
+    """惰性创建 TwelveLabs 客户端；缺少依赖或 key 时返回 None。"""
+    config = get_twelvelabs_config()
+    if not config.is_configured:
+        return None
+    try:
+        from twelvelabs import TwelveLabs
+    except ImportError:
+        logger.warning(
+            "未安装 twelvelabs 依赖，TwelveLabs 功能不可用。"
+            "请执行: pip install \"twelvelabs>=1.2.8\""
+        )
+        return None
+    try:
+        return TwelveLabs(api_key=config.api_key)
+    except Exception as e:  # pragma: no cover - 依赖外部 SDK
+        logger.error(f"TwelveLabs 客户端初始化失败: {e}")
+        return None
+
+
+class TwelveLabsVideoQAService:
+    """基于 Pegasus 的视频问答服务（直接理解视频画面）。
+
+    与 ``VideoQAService`` 接口保持一致，可作为视频问答的可选 provider。
+    """
+
+    def __init__(self):
+        self.config = get_twelvelabs_config()
+        self._client = None
+
+    @property
+    def client(self):
+        if self._client is None:
+            self._client = _get_client()
+        return self._client
+
+    def is_available(self) -> bool:
+        """key 已配置且依赖已安装时返回 True。"""
+        return self.client is not None
+
+    async def answer_question_stream(
+        self,
+        question: str,
+        transcript: str = "",
+        video_url: str = "",
+    ):
+        """基于视频画面回答问题（流式输出）。
+
+        Args:
+            question: 用户问题
+            transcript: 转录文本（可选，作为补充上下文）
+            video_url: 视频的公开 URL（Pegasus 必需）
+
+        Yields:
+            回答文本片段
+        """
+        if not self.client:
+            raise Exception("TwelveLabs API 不可用")
+        if not question.strip():
+            raise ValueError("问题不能为空")
+        if not video_url.strip():
+            raise ValueError("TwelveLabs 视频问答需要可公开访问的视频 URL")
+
+        from twelvelabs.types.video_context import VideoContext_Url
+
+        prompt = (
+            "你是一个专业的视频内容分析助手。请基于视频画面与音频，"
+            "准确、详细且有帮助地回答用户的问题；若视频中没有相关信息，请诚实说明。\n\n"
+            f"用户问题：{question}"
+        )
+        if transcript.strip():
+            prompt += f"\n\n（参考转录文本）：\n{transcript[:4000]}"
+
+        logger.info(f"Pegasus 视频问答: {question[:50]}...")
+
+        # SDK 为同步调用，放到线程池避免阻塞事件循环
+        def _analyze():
+            resp = self.client.analyze(
+                model_name=self.config.pegasus_model,
+                video=VideoContext_Url(url=video_url),
+                prompt=prompt,
+                max_tokens=self.config.qa_max_tokens,
+            )
+            return resp.data or ""
+
+        try:
+            answer = await asyncio.to_thread(_analyze)
+        except Exception as e:
+            logger.error(f"Pegasus 问答异常: {e}")
+            raise Exception(f"问答失败: {str(e)}")
+
+        # Pegasus 返回完整文本，按句子切分以模拟流式体验
+        for chunk in _chunk_text(answer):
+            yield chunk
+
+    def upload_asset(self, file_path: str) -> str:
+        """将本地视频上传为 TwelveLabs asset，返回 asset_id。
+
+        direct 上传上限约 200MB；更大的文件请改用公开 URL。
+        """
+        if not self.client:
+            raise Exception("TwelveLabs API 不可用")
+        with open(file_path, "rb") as f:
+            asset = self.client.assets.create(
+                method="direct", file=f, filename=file_path.split("/")[-1]
+            )
+        return asset.id
+
+    def embed_text(self, text: str) -> List[float]:
+        """用 Marengo 生成查询文本的 512 维向量（用于片段语义检索）。"""
+        if not self.client:
+            raise Exception("TwelveLabs API 不可用")
+        resp = self.client.embed.create(model_name=self.config.marengo_model, text=text)
+        return resp.text_embedding.segments[0].float_
+
+
+def _chunk_text(text: str) -> List[str]:
+    """将整段文本切成小片段，用于 SSE 流式输出。"""
+    if not text:
+        return []
+    import re
+    parts = re.split(r"(?<=[。！？.!?\n])", text)
+    return [p for p in parts if p]