From 5ee8fb4bf7918ae0dd257f64419d7e5751bc8d91 Mon Sep 17 00:00:00 2001 From: Mohit Varikuti Date: Sat, 27 Jun 2026 00:45:24 -0700 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=96=B0=E5=A2=9E=E5=8F=AF=E9=80=89=20?= =?UTF-8?q?TwelveLabs=20=E8=A7=86=E9=A2=91=E7=90=86=E8=A7=A3=20provider?= =?UTF-8?q?=EF=BC=88Pegasus=20=E9=97=AE=E7=AD=94=20+=20Marengo=20=E5=90=91?= =?UTF-8?q?=E9=87=8F=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 TwelveLabsVideoQAService：通过 Pegasus 直接理解视频画面进行问答， 接口与现有 VideoQAService 一致，可作为视频问答的可选 provider - 新增 Marengo 512 维多模态向量接口，用于视频片段语义检索 - 通过 VIDEO_QA_PROVIDER=twelvelabs + TWELVELABS_API_KEY 显式启用， 默认关闭，不配置时现有转录文本 + OpenAI 路径完全不受影响 - twelvelabs 作为可选依赖（uv sync --extra twelvelabs） - 补充 .env.example、README 文档与单元/实时契约测试 --- .env.example | 14 ++ README.md | 25 ++++ backend/config/__init__.py | 2 + backend/config/ai_config.py | 53 +++++++- backend/core/state.py | 13 +- backend/routers/qa.py | 3 +- backend/services/twelvelabs_service.py | 156 +++++++++++++++++++++++ backend/tests/__init__.py | 0 backend/tests/test_twelvelabs_service.py | 77 +++++++++++ pyproject.toml | 3 + 10 files changed, 342 insertions(+), 4 deletions(-) create mode 100644 backend/services/twelvelabs_service.py create mode 100644 backend/tests/__init__.py create mode 100644 backend/tests/test_twelvelabs_service.py diff --git a/.env.example b/.env.example index 31a5ad3..c145050 100644 --- a/.env.example +++ b/.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服务配置（可选） # ============================================ diff --git a/README.md b/README.md index 2014f6f..130dd04 100644 --- a/README.md +++ b/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) | 相对速度 | 质量 | 推荐场景 | diff --git a/backend/config/__init__.py b/backend/config/__init__.py index d649c83..3bee820 100644 --- a/backend/config/__init__.py +++ b/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' ] diff --git a/backend/config/ai_config.py b/backend/config/ai_config.py index 177b035..16f7670 100644 --- a/backend/config/ai_config.py +++ b/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 diff --git a/backend/core/state.py b/backend/core/state.py index 6a6219f..3078680 100644 --- a/backend/core/state.py +++ b/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 diff --git a/backend/routers/qa.py b/backend/routers/qa.py index 3044fbb..e144e64 100644 --- a/backend/routers/qa.py +++ b/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服务暂时不可用，请稍后重试") diff --git a/backend/services/twelvelabs_service.py b/backend/services/twelvelabs_service.py new file mode 100644 index 0000000..97c8c46 --- /dev/null +++ b/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] diff --git a/backend/tests/__init__.py b/backend/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/backend/tests/test_twelvelabs_service.py b/backend/tests/test_twelvelabs_service.py new file mode 100644 index 0000000..29e647d --- /dev/null +++ b/backend/tests/test_twelvelabs_service.py @@ -0,0 +1,77 @@ +"""TwelveLabs 服务测试。 + +- 无网络单元测试：验证配置开关、provider 选择、文本切分等纯逻辑。 +- 实时契约测试：仅在设置了 TWELVELABS_API_KEY 时运行，验证 Marengo 返回 512 维向量。 + +运行: uv run --extra dev --extra twelvelabs pytest backend/tests/test_twelvelabs_service.py +""" +import asyncio +import os + +import pytest + + +# ────────────────────────────────────────────────────────── +# 无网络单元测试 +# ────────────────────────────────────────────────────────── +def test_chunk_text_splits_on_sentence_boundaries(): + from backend.services.twelvelabs_service import _chunk_text + + out = _chunk_text("第一句。第二句！第三句？") + assert "".join(out) == "第一句。第二句！第三句？" + assert len(out) == 3 + assert _chunk_text("") == [] + + +def test_config_disabled_by_default(monkeypatch): + monkeypatch.delenv("TWELVELABS_API_KEY", raising=False) + monkeypatch.delenv("VIDEO_QA_PROVIDER", raising=False) + from backend.config.ai_config import TwelveLabsConfig + + cfg = TwelveLabsConfig() + assert cfg.is_configured is False + assert cfg.use_for_qa is False + + +def test_config_enabled_via_env(monkeypatch): + monkeypatch.setenv("TWELVELABS_API_KEY", "test-key") + monkeypatch.setenv("VIDEO_QA_PROVIDER", "twelvelabs") + monkeypatch.setenv("TWELVELABS_QA_MAX_TOKENS", "100") # < 512 应被钳制 + from backend.config.ai_config import TwelveLabsConfig + + cfg = TwelveLabsConfig() + assert cfg.is_configured is True + assert cfg.use_for_qa is True + assert cfg.qa_max_tokens == 512 # 钳制到模型下限 + + +def test_qa_requires_video_url(monkeypatch): + """未配置 key 时 is_available 为 False；空 video_url 抛 ValueError。""" + monkeypatch.setenv("TWELVELABS_API_KEY", "test-key") + from backend.services.twelvelabs_service import TwelveLabsVideoQAService + + svc = TwelveLabsVideoQAService() + + async def run(): + gen = svc.answer_question_stream(question="hi", video_url="") + with pytest.raises(ValueError): + await gen.__anext__() + + asyncio.run(run()) + + +# ────────────────────────────────────────────────────────── +# 实时契约测试（需 TWELVELABS_API_KEY） +# ────────────────────────────────────────────────────────── +@pytest.mark.skipif( + not os.getenv("TWELVELABS_API_KEY"), + reason="需要 TWELVELABS_API_KEY 才能运行实时测试", +) +def test_marengo_embedding_is_512_dim(): + from backend.services.twelvelabs_service import TwelveLabsVideoQAService + + svc = TwelveLabsVideoQAService() + assert svc.is_available() + vec = svc.embed_text("a person riding a bicycle") + assert len(vec) == 512 + assert all(isinstance(x, float) for x in vec[:8]) diff --git a/pyproject.toml b/pyproject.toml index 8d77698..c23861e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -41,6 +41,9 @@ dependencies = [ [project.optional-dependencies] dev = ["pytest>=8.0.0", "pytest-asyncio>=0.23.0", "httpx>=0.27.0"] +# 可选：TwelveLabs 视频理解（Pegasus 视频问答 + Marengo 片段语义检索） +# 安装: uv sync --extra twelvelabs +twelvelabs = ["twelvelabs>=1.2.8"] [build-system] requires = ["hatchling"]