diff --git a/README.md b/README.md index 7052016e..b00a734e 100644 --- a/README.md +++ b/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 笔记 diff --git a/backend/app/gpt/gpt_factory.py b/backend/app/gpt/gpt_factory.py index 8e8b60eb..800e4d8b 100644 --- a/backend/app/gpt/gpt_factory.py +++ b/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) \ No newline at end of file diff --git a/backend/app/gpt/twelvelabs_gpt.py b/backend/app/gpt/twelvelabs_gpt.py new file mode 100644 index 00000000..e5a294fc --- /dev/null +++ b/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] diff --git a/backend/app/models/gpt_model.py b/backend/app/models/gpt_model.py index a5fc936e..e984d0dd 100644 --- a/backend/app/models/gpt_model.py +++ b/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 diff --git a/backend/app/services/note.py b/backend/app/services/note.py index ebbe83a6..f25b1f31 100644 --- a/backend/app/services/note.py +++ b/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: diff --git a/backend/requirements.txt b/backend/requirements.txt index c0726462..852b65ce 100644 --- a/backend/requirements.txt +++ b/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 diff --git a/backend/tests/test_twelvelabs_gpt.py b/backend/tests/test_twelvelabs_gpt.py new file mode 100644 index 00000000..baf2d876 --- /dev/null +++ b/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() diff --git a/doc/twelvelabs.md b/doc/twelvelabs.md new file mode 100644 index 00000000..833c9daf --- /dev/null +++ b/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 +```