feat(gpt): 接入 TwelveLabs Pegasus 视频理解后端（可选）

README.md

@@ -71,6 +71,7 @@ BiliNote 是一个开源的 AI 视频笔记助手，支持通过哔哩哔哩、Y
 - 支持多模态视频理解
 - 支持多版本记录保留
 - 支持自行配置 GPT 大模型（OpenAI、DeepSeek、Qwen 等）
+- 可选 TwelveLabs Pegasus 视频理解：直接「看」视频从画面 + 语音生成笔记（详见 [doc/twelvelabs.md](./doc/twelvelabs.md)）
 - 本地模型音频转写（支持 Fast-Whisper、MLX-Whisper、Groq、BCut）
 - GPT 大模型总结视频内容
 - 自动生成结构化 Markdown 笔记

backend/app/gpt/gpt_factory.py

@@ -9,5 +9,11 @@
 class GPTFactory:
     @staticmethod
     def from_config(config: ModelConfig) -> GPT:
+        # TwelveLabs Pegasus 视频理解后端：直接「看」视频生成笔记（可选）。
+        # 仅当供应商 type == 'twelvelabs' 时路由到这里；其余供应商行为不变。
+        if (config.provider or "").lower() == "twelvelabs":
+            from app.gpt.twelvelabs_gpt import TwelveLabsGPT
+            return TwelveLabsGPT(api_key=config.api_key, model=config.model_name or "pegasus1.5")
+
         client = OpenAICompatibleProvider(api_key=config.api_key, base_url=config.base_url).get_client
         return UniversalGPT(client=client, model=config.model_name)

backend/app/gpt/twelvelabs_gpt.py

@@ -0,0 +1,98 @@
+"""TwelveLabs Pegasus 视频理解后端（可选）。
+
+与其它 GPT 后端不同，本后端不依赖音频转写文本，而是把视频 URL 直接交给
+TwelveLabs Pegasus 模型「看」视频本身，从画面 + 语音两路信息生成笔记。
+对于演示、操作录屏、图表讲解类视频，画面里往往承载了转写文本拿不到的信息。
+
+为什么是可选：
+  - 仅当供应商 type == 'twelvelabs' 时，GPTFactory 才会路由到这里；
+    未配置 TwelveLabs 供应商时整条链路行为不变。
+  - 没有视频 URL（如本地文件、纯音频）时优雅退回，提示用户改用常规模型。
+
+契约（针对官方 SDK twelvelabs>=1.2.8，已对线上 API 实测）：
+  - TwelveLabs(api_key=...).analyze(model_name='pegasus1.5', video=VideoContext_Url(url=...),
+    prompt=..., max_tokens=...).data 返回 Markdown 文本。
+  - Pegasus 1.5 不接受裸 video_id，必须传公网 URL 或已上传 asset。
+  - max_tokens 取值区间为 [512, 98304]，低于 512 会报 parameter_invalid。
+  - 被分析视频时长需 >= 4s。
+"""
+from typing import List, Optional
+
+from app.gpt.base import GPT
+from app.gpt.prompt_builder import generate_base_prompt
+from app.models.gpt_model import GPTSource
+from app.utils.logger import get_logger
+
+logger = get_logger(__name__)
+
+# Pegasus 输出 token 下限；低于该值 API 直接报 parameter_invalid。
+_MIN_MAX_TOKENS = 512
+_DEFAULT_MAX_TOKENS = 2048
+
+
+class TwelveLabsGPT(GPT):
+    """用 TwelveLabs Pegasus 直接理解视频内容生成笔记。"""
+
+    def __init__(self, api_key: str, model: str = "pegasus1.5", temperature: float = 0.7):
+        # 延迟 import：未安装 twelvelabs 的用户不应在加载 gpt 包时就报错。
+        try:
+            from twelvelabs import TwelveLabs
+        except ImportError as exc:  # pragma: no cover - 取决于运行环境是否装了依赖
+            raise ImportError(
+                "使用 TwelveLabs 视频理解需要先安装 twelvelabs SDK："
+                "pip install 'twelvelabs>=1.2.8'"
+            ) from exc
+
+        if not api_key or not str(api_key).strip():
+            raise ValueError("TwelveLabs 的 API Key 未配置，请先在「设置」里填写后再使用")
+
+        self.client = TwelveLabs(api_key=str(api_key).strip())
+        self.model = model or "pegasus1.5"
+        self.temperature = temperature
+
+    def _build_prompt(self, source: GPTSource) -> str:
+        """复用现有笔记 prompt 生成器；Pegasus 直接看视频，故不喂转写文本。"""
+        prompt = generate_base_prompt(
+            title=source.title,
+            segment_text="（本次由视频理解模型直接观看视频，无需转写文本）",
+            tags=source.tags,
+            _format=source._format,
+            style=source.style,
+            extras=source.extras,
+        )
+        return (
+            prompt
+            + "\n\n你正在直接观看这段视频，请结合**画面与语音**两路信息生成笔记，"
+            "充分利用画面中的演示、图表、UI、文字等转写文本无法体现的内容。"
+        )
+
+    def summarize(self, source: GPTSource) -> str:
+        video_url = getattr(source, "video_url", None)
+        if not video_url:
+            raise ValueError(
+                "TwelveLabs Pegasus 视频理解需要可公开访问的视频 URL，"
+                "当前任务没有可用 URL（如本地文件 / 纯音频）。请改用常规文本模型。"
+            )
+
+        from twelvelabs.types.video_context import VideoContext_Url
+
+        max_tokens = max(_MIN_MAX_TOKENS, _DEFAULT_MAX_TOKENS)
+        logger.info(f"使用 TwelveLabs Pegasus 直接理解视频：{video_url}")
+        response = self.client.analyze(
+            model_name=self.model,
+            video=VideoContext_Url(url=video_url),
+            prompt=self._build_prompt(source),
+            temperature=self.temperature,
+            max_tokens=max_tokens,
+        )
+        text = (response.data or "").strip()
+        if not text:
+            raise RuntimeError("TwelveLabs Pegasus 返回空结果")
+        return text
+
+    # 与 base.GPT 接口保持一致；视频理解后端不走分段消息拼装。
+    def create_messages(self, segments: List, **kwargs) -> list:
+        return []
+
+    def list_models(self) -> Optional[list]:
+        return [self.model]

backend/app/models/gpt_model.py

@@ -16,4 +16,6 @@ class GPTSource:
     _format: Optional[list] = None
     video_img_urls:  Optional[list] = None
     checkpoint_key: Optional[str] = None
+    # 视频原始 URL；仅 TwelveLabs Pegasus 视频理解后端需要（直接「看」视频）。
+    video_url: Optional[str] = None
 

backend/app/services/note.py

@@ -211,6 +211,7 @@ def generate(
                 style=style,
                 extras=extras,
                 video_img_urls=self.video_img_urls,
+                video_url=str(video_url),
             )
 
             # 4. 截图 & 链接替换
@@ -577,6 +578,7 @@ def _summarize_text(
         style: Optional[str],
         extras: Optional[str],
             video_img_urls: List[str],
+        video_url: Optional[str] = None,
     ) -> str | None:
         """
         调用 GPT 对转写结果进行总结，生成 Markdown 文本并缓存。
@@ -606,6 +608,7 @@ def _summarize_text(
             style=style,
             extras=extras,
             checkpoint_key=task_id,
+            video_url=video_url,
         )
 
         try:

backend/requirements.txt

@@ -111,6 +111,7 @@ tinyhtml5==2.0.0
 tokenizers==0.21.1
 tornado==6.4.2
 tqdm==4.67.1
+twelvelabs>=1.2.8
 typer==0.15.2
 typing-inspection==0.4.0
 tzdata==2025.2

backend/tests/test_twelvelabs_gpt.py

@@ -0,0 +1,161 @@
+"""TwelveLabs Pegasus 视频理解后端测试。
+
+无网络单测（默认运行）：用 stub 替掉 twelvelabs SDK，验证
+  - GPTSource 携带 video_url 时调用 Pegasus URL analyze，返回其文本
+  - 没有 video_url 时优雅抛错（本地文件 / 纯音频场景）
+
+可选联网测试（仅设置 TWELVELABS_API_KEY 时运行）：对线上 Pegasus 实测一次。
+"""
+import importlib.util
+import os
+import pathlib
+import sys
+import types
+import unittest
+
+ROOT = pathlib.Path(__file__).resolve().parents[1]
+
+
+def _install_stubs(captured):
+    """装一套最小 app.* + twelvelabs stub，让 twelvelabs_gpt 可在隔离环境里加载。"""
+    app_mod = types.ModuleType("app")
+    gpt_pkg = types.ModuleType("app.gpt")
+    models_pkg = types.ModuleType("app.models")
+    utils_pkg = types.ModuleType("app.utils")
+
+    base_mod = types.ModuleType("app.gpt.base")
+
+    class _GPT:
+        pass
+
+    base_mod.GPT = _GPT
+
+    prompt_builder_mod = types.ModuleType("app.gpt.prompt_builder")
+
+    def _generate_base_prompt(**kwargs):
+        captured["prompt_kwargs"] = kwargs
+        return "PROMPT_BODY"
+
+    prompt_builder_mod.generate_base_prompt = _generate_base_prompt
+
+    gpt_model_mod = types.ModuleType("app.models.gpt_model")
+
+    class _GPTSource:
+        def __init__(self, **kwargs):
+            for k, v in kwargs.items():
+                setattr(self, k, v)
+
+    gpt_model_mod.GPTSource = _GPTSource
+
+    logger_mod = types.ModuleType("app.utils.logger")
+    logger_mod.get_logger = lambda *_a, **_k: types.SimpleNamespace(
+        info=lambda *a, **k: None, warning=lambda *a, **k: None
+    )
+
+    # twelvelabs SDK stub
+    tl_mod = types.ModuleType("twelvelabs")
+    tl_types_mod = types.ModuleType("twelvelabs.types")
+    tl_vc_mod = types.ModuleType("twelvelabs.types.video_context")
+
+    class _VideoContextUrl:
+        def __init__(self, url):
+            self.url = url
+
+    tl_vc_mod.VideoContext_Url = _VideoContextUrl
+
+    class _FakeTwelveLabs:
+        def __init__(self, api_key):
+            captured["api_key"] = api_key
+
+        def analyze(self, *, model_name, video, prompt, temperature, max_tokens):
+            captured["analyze"] = {
+                "model_name": model_name,
+                "url": video.url,
+                "prompt": prompt,
+                "max_tokens": max_tokens,
+            }
+            return types.SimpleNamespace(data="# 视频笔记\n来自画面与语音")
+
+    tl_mod.TwelveLabs = _FakeTwelveLabs
+
+    mods = {
+        "app": app_mod,
+        "app.gpt": gpt_pkg,
+        "app.gpt.base": base_mod,
+        "app.gpt.prompt_builder": prompt_builder_mod,
+        "app.models": models_pkg,
+        "app.models.gpt_model": gpt_model_mod,
+        "app.utils": utils_pkg,
+        "app.utils.logger": logger_mod,
+        "twelvelabs": tl_mod,
+        "twelvelabs.types": tl_types_mod,
+        "twelvelabs.types.video_context": tl_vc_mod,
+    }
+    for name, mod in mods.items():
+        sys.modules[name] = mod
+    return _GPTSource
+
+
+def _load_module():
+    path = ROOT / "app" / "gpt" / "twelvelabs_gpt.py"
+    spec = importlib.util.spec_from_file_location("twelvelabs_gpt", path)
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+    return mod
+
+
+class TestTwelveLabsGPTUnit(unittest.TestCase):
+    def setUp(self):
+        self._saved = dict(sys.modules)
+        self.captured = {}
+        self.GPTSource = _install_stubs(self.captured)
+        self.mod = _load_module()
+
+    def tearDown(self):
+        sys.modules.clear()
+        sys.modules.update(self._saved)
+
+    def test_summarize_calls_pegasus_with_url(self):
+        gpt = self.mod.TwelveLabsGPT(api_key="sk-test")
+        src = self.GPTSource(
+            title="演示视频", tags="tag", _format=None, style=None, extras=None,
+            video_url="https://example.com/v.mp4",
+        )
+        out = gpt.summarize(src)
+        self.assertIn("视频笔记", out)
+        self.assertEqual(self.captured["analyze"]["url"], "https://example.com/v.mp4")
+        self.assertEqual(self.captured["analyze"]["model_name"], "pegasus1.5")
+        # Pegasus 要求 max_tokens >= 512
+        self.assertGreaterEqual(self.captured["analyze"]["max_tokens"], 512)
+
+    def test_missing_url_raises(self):
+        gpt = self.mod.TwelveLabsGPT(api_key="sk-test")
+        src = self.GPTSource(title="本地", tags="", _format=None, style=None, extras=None, video_url=None)
+        with self.assertRaises(ValueError):
+            gpt.summarize(src)
+
+    def test_empty_api_key_raises(self):
+        with self.assertRaises(ValueError):
+            self.mod.TwelveLabsGPT(api_key="")
+
+
+@unittest.skipUnless(os.getenv("TWELVELABS_API_KEY"), "需要 TWELVELABS_API_KEY 才跑联网测试")
+class TestTwelveLabsGPTLive(unittest.TestCase):
+    def test_live_pegasus_url_analyze(self):
+        # 真实 SDK，真实 API：Pegasus 直接看一段公开短视频返回文本笔记。
+        from twelvelabs import TwelveLabs
+        from twelvelabs.types.video_context import VideoContext_Url
+
+        client = TwelveLabs(api_key=os.environ["TWELVELABS_API_KEY"])
+        url = "https://test-videos.co.uk/vids/bigbuckbunny/mp4/h264/720/Big_Buck_Bunny_720_10s_1MB.mp4"
+        resp = client.analyze(
+            model_name="pegasus1.5",
+            video=VideoContext_Url(url=url),
+            prompt="In one sentence, what happens in this video?",
+            max_tokens=512,
+        )
+        self.assertTrue((resp.data or "").strip())
+
+
+if __name__ == "__main__":
+    unittest.main()

doc/twelvelabs.md

@@ -0,0 +1,60 @@
+# TwelveLabs Pegasus 视频理解（可选）
+
+BiliNote 默认链路是「下载 → 转写 → LLM 总结转写文本 → 笔记」。本可选后端用
+[TwelveLabs](https://twelvelabs.io) 的 **Pegasus** 视频理解模型，让模型直接「看」视频，
+从**画面 + 语音**两路信息生成笔记 —— 对演示、操作录屏、图表讲解类视频，画面里往往
+承载了转写文本拿不到的信息。
+
+> 这是一个**可选、不改默认行为**的供应商。不配置 TwelveLabs 供应商时，整条链路与原来完全一致。
+
+## 它做了什么
+
+- 新增 GPT 后端 `app/gpt/twelvelabs_gpt.py`（`TwelveLabsGPT`），实现既有 `GPT.summarize` 接口。
+- `GPTFactory.from_config` 在供应商 `type == "twelvelabs"` 时路由到该后端；其余供应商不受影响。
+- 复用现有笔记 prompt（`generate_base_prompt`），把风格 / 格式 / 标签等选项一并带给 Pegasus，
+  但**不喂转写文本**——Pegasus 直接观看视频本身。
+
+## 使用前提
+
+1. 安装依赖（已加入 `backend/requirements.txt`）：
+
+   ```bash
+   pip install 'twelvelabs>=1.2.8'
+   ```
+
+2. 在 [twelvelabs.io](https://twelvelabs.io) 注册并获取 API Key（有较慷慨的免费额度）。
+
+## 配置
+
+在「设置 → 模型供应商」里新增一个供应商：
+
+| 字段 | 值 |
+|---|---|
+| 类型(type) | `twelvelabs` |
+| API Key | 你的 TwelveLabs API Key |
+| 模型(model_name) | `pegasus1.5`（默认）|
+| base_url | 留空即可（SDK 自带）|
+
+之后在生成笔记时选择该供应商即可。API Key 完全从供应商配置读取，**不会硬编码、不写入仓库**。
+
+## 已知限制
+
+- **需要可公开访问的视频 URL**：Pegasus 1.5 直接拉取视频 URL 分析。本地文件 / 纯音频任务
+  没有可用 URL，此时该后端会优雅报错并提示改用常规文本模型。
+- **输出 token 下限**：Pegasus 1.5 要求 `max_tokens >= 512`（低于会被 API 拒绝）；本实现已固定满足。
+- **被分析视频时长需 ≥ 4s**。
+- 大文件上传（asset 直传）上限 200MB；公网 URL 上限 4GB。本集成走 URL 路径。
+
+## 测试
+
+无网络单测（默认随 `pytest` 运行）：
+
+```bash
+cd backend && pytest tests/test_twelvelabs_gpt.py
+```
+
+可选联网测试（仅在设置 `TWELVELABS_API_KEY` 时运行，会真实调用一次 Pegasus）：
+
+```bash
+TWELVELABS_API_KEY=tlk_xxx pytest tests/test_twelvelabs_gpt.py -k Live
+```