Add TwelveLabs Pegasus as a hosted captioning backend

README.md

@@ -58,6 +58,16 @@ Edit `backend/config.py` to adjust:
 | `MAX_TOKENS` | 512 | Max caption length |
 | `TEMPERATURE` | 0.3 | Generation creativity |
 
+### Hosted captioning with TwelveLabs Pegasus (optional)
+
+Besides the local VLMs, you can use [TwelveLabs](https://twelvelabs.io) Pegasus
+as a hosted captioning backend — no GPU required, the video is analysed
+server-side. Set `TWELVELABS_API_KEY` in your environment and pick the
+**TwelveLabs Pegasus 1.5 (hosted)** preset in Settings. Get a free API key at
+https://twelvelabs.io (generous free tier). See
+[`documentation/CONFIGURATION.md`](documentation/CONFIGURATION.md#twelvelabs-pegasus-hosted)
+for details.
+
 ## Multi-GPU Processing
 
 On systems with multiple CUDA GPUs, the suite automatically detects available devices and enables parallel processing:

backend/config.py

@@ -4,6 +4,7 @@
 """
 
 import json
+import os
 from pathlib import Path
 from typing import Optional
 
@@ -144,6 +145,10 @@ def set_include_images(include: bool) -> None:
 DEFAULT_PRESET_ID = "qwen3-vl-8b"
 MODEL_ID = "Qwen/Qwen3-VL-8B-Instruct"
 
+# TwelveLabs Pegasus (hosted) preset. The API key is read from the environment
+# and never persisted to disk. Get a free key at https://twelvelabs.io.
+TWELVELABS_API_KEY = os.environ.get("TWELVELABS_API_KEY")
+
 # Device: "cuda", "cpu", or "auto"
 DEVICE = "cuda"
 

backend/model_loader.py

@@ -451,6 +451,25 @@ def load_model(
 
     total_start = time.time()
 
+    # Hosted presets (e.g. Pegasus) have no local weights and no torch device —
+    # build their client and return before any download/torch setup.
+    if preset["loader"] == "pegasus":
+        from backend.twelvelabs_provider import load_pegasus
+
+        strategy_info = load_pegasus(preset)
+        model_info = {
+            **strategy_info,
+            "model_id": effective_model_id,
+            "model_path": None,
+            "device": None,
+            "dtype": None,
+            "preset": preset,
+            "preset_id": _find_preset_id(preset),
+        }
+        _MODEL_CACHE[cache_key] = model_info
+        print(f"[Model Loader] Hosted model ready: {preset['label']}")
+        return model_info
+
     sage_enabled = enable_sage_attention() if use_sage_attention else False
 
     model_path = download_model(effective_model_id, config.MODELS_DIR)
@@ -510,8 +529,14 @@ def generate_caption(
     max_tokens: int = None,
     temperature: float = None,
     video_fps: float = None,
+    video_path=None,
 ) -> Tuple[str, Dict[str, Any]]:
-    """Generate a caption via the strategy selected at load time."""
+    """Generate a caption via the strategy selected at load time.
+
+    `video_path` is the source media file; local strategies ignore it (they
+    consume pre-extracted `images`), but hosted strategies like Pegasus analyse
+    the file directly and require it.
+    """
     max_tokens = max_tokens or config.MAX_TOKENS
     temperature = temperature if temperature is not None else config.TEMPERATURE
     preset = model_info["preset"]
@@ -520,6 +545,10 @@ def generate_caption(
         return _generate_image_text_to_text(model_info, images, prompt, max_tokens, temperature)
     elif preset["loader"] == "gemma4":
         return _generate_gemma4(model_info, images, prompt, max_tokens, temperature, preset, video_fps=video_fps)
+    elif preset["loader"] == "pegasus":
+        from backend.twelvelabs_provider import generate_pegasus
+
+        return generate_pegasus(model_info, images, prompt, max_tokens, temperature, video_path=video_path)
     else:
         raise ValueError(f"Unknown loader strategy: {preset['loader']}")
 

backend/model_presets.py

@@ -18,9 +18,10 @@
 #   model_id                  HuggingFace repo id
 #   label                     Human-readable label for the UI dropdown
 #   description               One-line help text
-#   loader                    "image_text_to_text" | "gemma4"
+#   loader                    "image_text_to_text" | "gemma4" | "pegasus"
 #   content_type              "image_list" (one image block per frame)
 #                             | "video_block" (single video block w/ frame list)
+#                             | "hosted_video" (source file analysed server-side)
 #   supports_sage_attention   Whether SageAttention is safe to use
 #   supports_torch_compile    Whether torch.compile is safe to use
 #   supports_multi_gpu_shard  If True, loads with device_map="auto" on one
@@ -87,6 +88,26 @@
         "enable_thinking": False,
         "gen_defaults": {"temperature": 1.0, "top_p": 0.95, "top_k": 64},
     },
+    "pegasus-1.5": {
+        "model_id": "twelvelabs/pegasus1.5",
+        "label": "TwelveLabs Pegasus 1.5 (hosted)",
+        "description": "Hosted video-understanding model. No GPU/VRAM needed; "
+        "analyses the source video server-side. Requires TWELVELABS_API_KEY.",
+        "loader": "pegasus",
+        "content_type": "hosted_video",
+        "supports_sage_attention": False,
+        "supports_torch_compile": False,
+        "supports_multi_gpu_shard": False,
+        "quantization": None,
+        "default_max_frames": 16,
+        "default_frame_size": 336,
+        "approx_vram_gb": 0,
+        "frame_size_divisor": None,
+        # Hosted-model marker: no local weights, requires an API key.
+        "is_hosted": True,
+        "requires_api_key": True,
+        "pegasus_model_name": "pegasus1.5",
+    },
 }
 
 DEFAULT_PRESET = "qwen3-vl-8b"
@@ -144,7 +165,9 @@ def list_presets_public() -> list:
             "quantization": p["quantization"],
             "supports_sage_attention": p["supports_sage_attention"],
             "supports_torch_compile": p["supports_torch_compile"],
-            "is_video_native": p["content_type"] == "video_block",
+            "is_video_native": p["content_type"] in ("video_block", "hosted_video"),
+            "is_hosted": p.get("is_hosted", False),
+            "requires_api_key": p.get("requires_api_key", False),
         })
     return out
 

backend/processing.py

@@ -396,6 +396,7 @@ async def process_single_video(worker_id: int, video_path: Path):
                             max_tokens=settings.max_tokens,
                             temperature=settings.temperature,
                             video_fps=video_meta.get("fps"),
+                            video_path=video_path,
                         )
                     )
 
@@ -597,6 +598,7 @@ async def _process_videos_sequential(
                             max_tokens=settings.max_tokens,
                             temperature=settings.temperature,
                             video_fps=video_meta.get("fps"),
+                            video_path=video_path,
                         )
                     )
 

backend/schemas.py

@@ -187,6 +187,8 @@ class ModelPresetInfo(BaseModel):
     supports_sage_attention: bool
     supports_torch_compile: bool
     is_video_native: bool
+    is_hosted: bool = False
+    requires_api_key: bool = False
 
 
 class ModelPresetListResponse(BaseModel):

backend/tests/test_twelvelabs_provider.py

@@ -0,0 +1,102 @@
+"""
+Tests for the TwelveLabs Pegasus captioning backend.
+
+The unit tests run with no network access. The live test is skipped unless
+TWELVELABS_API_KEY is set (a free key is available at https://twelvelabs.io).
+"""
+
+import os
+import pytest
+
+import sys
+from pathlib import Path
+sys.path.insert(0, str(Path(__file__).parent.parent.parent))
+
+from backend import twelvelabs_provider as tl
+from backend.model_presets import MODEL_PRESETS, get_preset, list_presets_public
+
+
+class TestPegasusPreset:
+    """The Pegasus preset is registered and wired to the pegasus loader."""
+
+    def test_preset_registered(self):
+        preset = get_preset("pegasus-1.5")
+        assert preset is not None
+        assert preset["loader"] == "pegasus"
+        assert preset["is_hosted"] is True
+        assert preset["requires_api_key"] is True
+        assert preset["approx_vram_gb"] == 0
+
+    def test_preset_in_public_list(self):
+        ids = {p["id"] for p in list_presets_public()}
+        assert "pegasus-1.5" in ids
+        entry = next(p for p in list_presets_public() if p["id"] == "pegasus-1.5")
+        assert entry["is_hosted"] is True
+        assert entry["requires_api_key"] is True
+
+    def test_default_preset_unchanged(self):
+        # Opt-in only: Pegasus must never become the default.
+        from backend.model_presets import DEFAULT_PRESET
+        assert DEFAULT_PRESET == "qwen3-vl-8b"
+
+
+class TestBuildGenerateMeta:
+    """build_generate_meta is pure and matches the local-strategy meta shape."""
+
+    def test_meta_shape(self):
+        meta = tl.build_generate_meta("a caption", elapsed=2.0, num_frames=16)
+        for key in (
+            "input_tokens", "output_tokens", "encode_time",
+            "generate_time", "total_time", "tokens_per_sec", "num_frames",
+        ):
+            assert key in meta
+        assert meta["num_frames"] == 16
+        assert meta["generate_time"] == 2.0
+        assert meta["total_time"] == 2.0
+
+    def test_uses_usage_tokens_when_provided(self):
+        meta = tl.build_generate_meta(
+            "hello world", elapsed=1.0, num_frames=4, usage_output_tokens=42
+        )
+        assert meta["output_tokens"] == 42
+        assert meta["tokens_per_sec"] == 42.0
+
+    def test_estimates_tokens_without_usage(self):
+        meta = tl.build_generate_meta("x" * 40, elapsed=2.0, num_frames=1)
+        assert meta["output_tokens"] == 10  # ~4 chars/token
+
+    def test_zero_elapsed_no_divide_by_zero(self):
+        meta = tl.build_generate_meta("text", elapsed=0.0, num_frames=1)
+        assert meta["tokens_per_sec"] == 0
+
+
+class TestPegasusGuards:
+    """Pegasus surfaces clear errors instead of silently misbehaving."""
+
+    def test_generate_requires_video_path(self):
+        model_info = {"client": object(), "pegasus_model_name": "pegasus1.5"}
+        with pytest.raises(ValueError, match="source video path"):
+            tl.generate_pegasus(model_info, images=[], prompt="x",
+                                max_tokens=512, temperature=0.3, video_path=None)
+
+    def test_load_requires_api_key(self, monkeypatch):
+        monkeypatch.delenv("TWELVELABS_API_KEY", raising=False)
+        with pytest.raises(RuntimeError, match="TWELVELABS_API_KEY"):
+            tl.load_pegasus(get_preset("pegasus-1.5"))
+
+
+@pytest.mark.skipif(
+    not os.environ.get("TWELVELABS_API_KEY"),
+    reason="requires TWELVELABS_API_KEY (free key at https://twelvelabs.io)",
+)
+class TestPegasusLive:
+    """Live wiring check against the real API (skipped without a key)."""
+
+    def test_load_builds_client(self):
+        info = tl.load_pegasus(get_preset("pegasus-1.5"))
+        assert info["client"] is not None
+        assert info["pegasus_model_name"] == "pegasus1.5"
+
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-v"])