diff --git a/environment/config/config.yml b/environment/config/config.yml index 231aec2..c5f4d5b 100644 --- a/environment/config/config.yml +++ b/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: "" diff --git a/environment/config/llm.py b/environment/config/llm.py index 36876b1..28379ee 100644 --- a/environment/config/llm.py +++ b/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") diff --git a/environment/config/registry.json b/environment/config/registry.json index 0f0c820..4400249 100644 --- a/environment/config/registry.json +++ b/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" } \ No newline at end of file diff --git a/environment/roles/vid_qa/pegasus_understanding.py b/environment/roles/vid_qa/pegasus_understanding.py new file mode 100644 index 0000000..22f621f --- /dev/null +++ b/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", + } diff --git a/pyproject.toml b/pyproject.toml index f2bfe77..85723a6 100644 --- a/pyproject.toml +++ b/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"} diff --git a/readme.md b/readme.md index 5f4805e..c6360f2 100644 --- a/readme.md +++ b/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** diff --git a/tests/test_pegasus_understanding.py b/tests/test_pegasus_understanding.py new file mode 100644 index 0000000..ccf1477 --- /dev/null +++ b/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