Add TwelveLabs Pegasus video understanding agent (opt-in)

environment/config/config.yml

@@ -13,8 +13,13 @@ llm:
   gpt_base_url: ""  
 
   # MLLM for caption and fine-grained video understanding
-  gemini_api_key: ""  
-  gemini_base_url: ""  
+  gemini_api_key: ""
+  gemini_base_url: ""
+
+  # Optional: TwelveLabs Pegasus for native video understanding/Q&A
+  # (no local transcription needed). Free key at https://twelvelabs.io
+  twelvelabs_api_key: ""
+  twelvelabs_base_url: ""
 
 
 

environment/config/llm.py

@@ -1,6 +1,8 @@
 from openai import OpenAI
+
 from environment.config.config import config
 
+
 def get_client(model_prefix=None):
     """Get OpenAI client with appropriate credentials based on model prefix"""
     # Get model-specific API key and base URL if available, otherwise use defaults
@@ -75,6 +77,46 @@ def gemini(model="gemini-2.5-flash", system=None, user=None, messages=None):
     )
     return response
 
+def twelvelabs_client():
+    """Build a TwelveLabs client from the ``twelvelabs`` credentials in config.yml.
+
+    The ``twelvelabs`` SDK is imported lazily so it is only required when the
+    Pegasus backend is actually used.
+    """
+    from twelvelabs import TwelveLabs
+
+    api_key = config['llm'].get('twelvelabs_api_key')
+    if not api_key:
+        raise ValueError(
+            "TwelveLabs API key not configured. Set llm.twelvelabs_api_key in "
+            "environment/config/config.yml (free key at https://twelvelabs.io)."
+        )
+
+    base_url = config['llm'].get('twelvelabs_base_url') or None
+    if base_url:
+        return TwelveLabs(api_key=api_key, base_url=base_url)
+    return TwelveLabs(api_key=api_key)
+
+def pegasus(video, prompt, model="pegasus1.5", max_tokens=2048, temperature=None):
+    """Analyze a video with TwelveLabs Pegasus and return the generated text.
+
+    ``video`` is either a public URL string or a dict describing a
+    TwelveLabs video context, e.g. ``{"type": "url", "url": "..."}`` or
+    ``{"type": "asset_id", "asset_id": "..."}``. Unlike the chat helpers
+    above, Pegasus understands the video natively (frames + audio), so no
+    transcription step is required.
+    """
+    if isinstance(video, str):
+        video = {"type": "url", "url": video}
+
+    return twelvelabs_client().analyze(
+        model_name=model,
+        video=video,
+        prompt=prompt,
+        max_tokens=max_tokens,
+        temperature=temperature,
+    )
+
 def gpt(model="gpt-4o", system=None, user=None, messages=None):
     # Get client for gpt
     client = get_client("gpt")

environment/config/registry.json

@@ -31,5 +31,6 @@
     "CommentaryContentGenerator": "environment.roles.vid_comm.comm_story_gen",
     "NewsContentGenerator": "environment.roles.vid_news.news_story_gen",
     "VideoContentQA": "environment.roles.vid_qa.content_loader",
+    "PegasusVideoUnderstanding": "environment.roles.vid_qa.pegasus_understanding",
     "VideoSummarizationGenerator": "environment.roles.vid_summ.summ_loader"
 }

environment/roles/vid_qa/pegasus_understanding.py

@@ -0,0 +1,162 @@
+import logging
+import os
+import time
+from typing import Dict, List
+
+from pydantic import BaseModel, Field
+from tenacity import (
+    retry,
+    retry_if_exception_type,
+    stop_after_attempt,
+    wait_exponential,
+)
+
+from environment.agents.base import BaseTool
+from environment.config.llm import pegasus, twelvelabs_client
+
+# Configure logging
+logging.basicConfig(level=logging.INFO,
+                    format='%(asctime)s - %(levelname)s - %(message)s')
+logger = logging.getLogger(__name__)
+
+
+class PegasusVideoUnderstanding(BaseTool):
+    """
+    Agent that answers questions about videos using TwelveLabs Pegasus.
+    Unlike the transcript-based Q&A agent, Pegasus understands the video
+    natively (visuals + audio), so it works without local transcription and
+    can reason about on-screen actions, scenes, and objects, not just speech.
+    This is an opt-in backend: it only runs when a TwelveLabs API key is set
+    in environment/config/config.yml (free key at https://twelvelabs.io).
+    """
+
+    def __init__(self, max_tokens: int = 2048):
+        super().__init__()
+        self.max_tokens = max_tokens
+        # Video extensions Pegasus can ingest directly.
+        self.video_extensions = {'.mp4', '.avi', '.mov', '.mkv', '.flv', '.wmv', '.webm', '.m4v', '.3gp'}
+
+    class InputSchema(BaseTool.BaseInputSchema):
+        video_path: str = Field(
+            ...,
+            description="Path to a video file, a directory of videos, or a public video URL to analyze"
+        )
+        prompt: str = Field(
+            ...,
+            description="The question or instruction to ask Pegasus about the video(s)"
+        )
+        model: str = Field(
+            "pegasus1.5",
+            description="TwelveLabs Pegasus model name (e.g. pegasus1.5)"
+        )
+
+    class OutputSchema(BaseModel):
+        answers: Dict[str, str] = Field(
+            ...,
+            description="Mapping of each processed video (filename or URL) to Pegasus's answer"
+        )
+        processed_videos: List[str] = Field(
+            ...,
+            description="List of videos that were analyzed"
+        )
+        status: str = Field(
+            ...,
+            description="Overall status of the analysis ('success' or 'error')"
+        )
+
+    def _get_video_files(self, video_path: str) -> List[str]:
+        """Resolve the input into a list of analyzable video sources."""
+        # A URL is passed straight through to Pegasus (analyzed server-side).
+        if video_path.startswith(("http://", "https://")):
+            return [video_path]
+
+        if not os.path.exists(video_path):
+            raise FileNotFoundError(f"Video path not found: {video_path}")
+
+        if os.path.isfile(video_path):
+            return [video_path]
+
+        video_files = []
+        for filename in os.listdir(video_path):
+            file_path = os.path.join(video_path, filename)
+            if os.path.isfile(file_path):
+                _, ext = os.path.splitext(filename.lower())
+                if ext in self.video_extensions:
+                    video_files.append(file_path)
+        video_files.sort()
+        logger.info(f"Found {len(video_files)} video files in directory: {video_path}")
+        return video_files
+
+    def _upload_asset(self, file_path: str):
+        """Upload a local video file as a TwelveLabs asset for analysis."""
+        logger.info(f"Uploading local video as TwelveLabs asset: {file_path}")
+        client = twelvelabs_client()
+        with open(file_path, "rb") as f:
+            return client.assets.create(method="direct", file=f, filename=os.path.basename(file_path))
+
+    @retry(
+        retry=retry_if_exception_type((Exception)),
+        wait=wait_exponential(multiplier=1, min=4, max=60),
+        stop=stop_after_attempt(3),
+        reraise=True
+    )
+    def _analyze_one(self, source: str, prompt: str, model: str) -> str:
+        """Send a single video to Pegasus and return the generated text."""
+        # URLs are passed directly; local files are uploaded as a TwelveLabs
+        # asset first and referenced by id.
+        if source.startswith(("http://", "https://")):
+            video = {"type": "url", "url": source}
+        else:
+            asset = self._upload_asset(source)
+            video = {"type": "asset_id", "asset_id": asset.id}
+
+        logger.info(f"Analyzing with Pegasus ({model}): {source}")
+        start_time = time.time()
+        response = pegasus(video=video, prompt=prompt, model=model, max_tokens=self.max_tokens)
+        logger.info(f"Pegasus analysis completed in {time.time() - start_time:.2f}s")
+        return response.data or ""
+
+    def execute(self, **kwargs):
+        """Analyze the given video(s) with Pegasus and return per-video answers."""
+        params = self.InputSchema(**kwargs)
+
+        print("\n=== PEGASUS VIDEO UNDERSTANDING ===")
+        print(f"Source: {params.video_path}")
+        print(f"Prompt: {params.prompt}")
+
+        try:
+            sources = self._get_video_files(params.video_path)
+            if not sources:
+                raise ValueError(f"No video files found at: {params.video_path}")
+
+            answers: Dict[str, str] = {}
+            processed: List[str] = []
+            for i, source in enumerate(sources, 1):
+                key = source if source.startswith(("http://", "https://")) else os.path.basename(source)
+                print(f"\nAnalyzing {i}/{len(sources)}: {key}")
+                try:
+                    answers[key] = self._analyze_one(source, params.prompt, params.model)
+                    processed.append(source)
+                except Exception as e:
+                    logger.error(f"Failed to analyze {source}: {e}")
+                    answers[key] = f"[Error analyzing {key}: {e}]"
+
+            print("\n=== ANALYSIS COMPLETED ===")
+            return {
+                "answers": answers,
+                "processed_videos": [
+                    s if s.startswith(("http://", "https://")) else os.path.basename(s)
+                    for s in processed
+                ],
+                "status": "success",
+            }
+
+        except Exception as e:
+            error_msg = f"Error in Pegasus video understanding: {e}"
+            logger.error(error_msg)
+            print(f"\nError: {error_msg}")
+            return {
+                "answers": {},
+                "processed_videos": [],
+                "status": "error",
+            }

pyproject.toml

@@ -113,7 +113,8 @@ dependencies = [
   "pydub==0.25.1",
   "FreeSimpleGUI==5.1.1",
   "sounddevice==0.5.0",
-  "python-dotenv"
+  "python-dotenv",
+  "twelvelabs>=1.2.8"
 ]
 license = {text = "Apache"}
 

readme.md

@@ -381,8 +381,26 @@ llm:
   # MLLM for caption and fine-grained video understanding
   gemini_api_key: ""  
   gemini_base_url: ""  
+
+  # Optional: TwelveLabs Pegasus for native video understanding/Q&A
+  twelvelabs_api_key: ""
+  twelvelabs_base_url: ""
 ```
 
+#### 🎥 Optional: TwelveLabs Pegasus (native video understanding)
+
+The `PegasusVideoUnderstanding` agent answers questions about a video using
+[TwelveLabs](https://twelvelabs.io) Pegasus. Unlike the transcript-based Q&A
+agent, Pegasus understands the video natively (visuals **and** audio), so it
+works without local transcription and can reason about on-screen actions,
+scenes, and objects — not just speech. It accepts a local video file, a
+directory of videos, or a public video URL.
+
+This backend is **opt-in and non-breaking**: it only activates when you set
+`twelvelabs_api_key` above, and it leaves the default transcript-based flow
+untouched. You can grab a free API key at https://twelvelabs.io — there's a
+generous free tier.
+
 ### 🎯 **Usage**
 
 ```bash
@@ -460,6 +478,7 @@ We express our deepest gratitude to the numerous individuals and organizations t
 - [ImageBind](https://github.com/facebookresearch/ImageBind)
 - [Whisper](https://github.com/openai/whisper)
 - [Librosa](https://github.com/librosa/librosa)
+- [TwelveLabs](https://twelvelabs.io)
 
 
 ### 🎨 **Content Creators and Inspiration**

tests/test_pegasus_understanding.py

@@ -0,0 +1,67 @@
+"""Tests for the TwelveLabs Pegasus video-understanding backend.
+
+The no-network tests run anywhere. The live test is skipped unless
+TWELVELABS_API_KEY is set (free key at https://twelvelabs.io).
+"""
+import os
+import sys
+from unittest import mock
+
+import pytest
+
+# Make the project root importable when running pytest from the repo root.
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
+
+from environment.roles.vid_qa.pegasus_understanding import PegasusVideoUnderstanding
+
+
+def test_url_is_passed_through_without_filesystem_access():
+    """A public URL should be analyzed directly, with no disk lookup."""
+    tool = PegasusVideoUnderstanding()
+    sources = tool._get_video_files("https://example.com/clip.mp4")
+    assert sources == ["https://example.com/clip.mp4"]
+
+
+def test_execute_routes_url_to_pegasus():
+    """execute() should call the pegasus() helper and surface its text."""
+    tool = PegasusVideoUnderstanding()
+    fake_response = mock.Mock(data="A rabbit wakes up in a meadow.")
+
+    with mock.patch(
+        "environment.roles.vid_qa.pegasus_understanding.pegasus",
+        return_value=fake_response,
+    ) as mock_pegasus:
+        result = tool.execute(
+            video_path="https://example.com/clip.mp4",
+            prompt="Describe this video.",
+        )
+
+    assert result["status"] == "success"
+    assert result["answers"]["https://example.com/clip.mp4"] == "A rabbit wakes up in a meadow."
+    # The URL is forwarded to Pegasus untouched.
+    _, kwargs = mock_pegasus.call_args
+    assert kwargs["video"] == {"type": "url", "url": "https://example.com/clip.mp4"}
+    assert kwargs["prompt"] == "Describe this video."
+
+
+def test_missing_path_reports_error():
+    tool = PegasusVideoUnderstanding()
+    result = tool.execute(video_path="/no/such/video.mp4", prompt="What is this?")
+    assert result["status"] == "error"
+
+
+@pytest.mark.skipif(
+    not os.environ.get("TWELVELABS_API_KEY"),
+    reason="requires TWELVELABS_API_KEY",
+)
+def test_twelvelabs_credentials_and_sdk_live():
+    """Live check: the SDK + key reach the TwelveLabs backend.
+
+    Uses a fast Marengo text embedding (512-dim) to confirm wiring without
+    waiting on a full Pegasus video analysis.
+    """
+    from twelvelabs import TwelveLabs
+
+    client = TwelveLabs(api_key=os.environ["TWELVELABS_API_KEY"])
+    resp = client.embed.create(model_name="marengo3.0", text="a person walking a dog")
+    assert len(resp.text_embedding.segments[0].float_) == 512