diff --git a/docs/concepts/tool_system/twelvelabs_video.md b/docs/concepts/tool_system/twelvelabs_video.md new file mode 100644 index 00000000..c48b708e --- /dev/null +++ b/docs/concepts/tool_system/twelvelabs_video.md @@ -0,0 +1,95 @@ +# Introduce to TwelveLabs Video tool + +# Introduction + +[TwelveLabs](https://twelvelabs.io) provides state-of-the-art video understanding models. This tool wraps two of them: + +- **Pegasus** (`task: analyze`) — video understanding. Answer a natural language prompt about a video, e.g. summarize it, list the objects shown, or describe what happens. +- **Marengo** (`task: embed`) — multimodal embeddings. Produce a 512-dim embedding vector from text or an image. Text, image and video share one embedding space, so a text/image vector can be used to retrieve relevant videos. + +This lets OmAgent's multimodal agents reason about videos by URL without running a local video model. + +# How to use + +## Use the TwelveLabs Video tool + +YAML file in the configs folder defines available tools. The TwelveLabs Video tool can be enabled by adding the following config. + +```yaml +llm: ${sub|text_res} +tools: + - ...other tools... + - name: TwelveLabsVideo # enable the TwelveLabs video tool + api_key: ${env|twelvelabs_api_key, null} # set the TwelveLabs API key via environment variable +``` + +## Get a TwelveLabs API key + +1. Open [https://twelvelabs.io](https://twelvelabs.io) and sign up (there is a generous free tier). + +2. Create an API key from the dashboard. + +3. Copy it and set the environment variable, e.g. `export twelvelabs_api_key=tlk-xxx` in your terminal, or `os.environ['twelvelabs_api_key'] = "tlk-xxx"` in `run_cli/app/webpage.py`. + +## Input Parameters + +1. task + 1. type: string + 2. enum: ["analyze", "embed"] + 3. description: Which capability to use. `analyze` runs Pegasus to answer a prompt about a video. `embed` runs Marengo to produce an embedding vector. + 4. required: True +2. prompt + 1. type: string + 2. description: For `analyze`: the question or instruction about the video. +3. text + 1. type: string + 2. description: For `embed`: the text to embed into a multimodal vector. +4. image_url + 1. type: string + 2. description: For `embed`: a public URL to an image to embed. +5. video_url + 1. type: string + 2. description: For `analyze`: a public URL to the video file. TwelveLabs fetches it server-side. +6. max_tokens + 1. type: integer + 2. description: For `analyze`: maximum number of tokens to generate. Default is `2048`. + +## Output Data + +For `task: analyze`: + +1. text + 1. type: string + 2. description: The model's answer about the video. +2. finish_reason + 1. type: string + 2. description: Why generation stopped. + +For `task: embed`: + +1. embedding + 1. type: List[float] + 2. description: The 512-dim embedding vector. +2. dimension + 1. type: integer + 2. description: The vector dimension (512). + +## Quick Experience + +```python +from omagent_core.tool_system.tools.twelvelabs_video.twelvelabs_video import TwelveLabsVideo + +tool = TwelveLabsVideo(api_key="tlk-xxx") + +# Marengo embedding +res = tool.run({"task": "embed", "text": "a cat playing piano"}) +print(res["dimension"], res["embedding"][:3]) + +# Pegasus video understanding +res = tool.run({ + "task": "analyze", + "video_url": "https://example.com/your-video.mp4", + "prompt": "Describe what happens in this video.", +}) +print(res["text"]) +``` diff --git a/omagent-core/pyproject.toml b/omagent-core/pyproject.toml index 8f390aea..56f30420 100644 --- a/omagent-core/pyproject.toml +++ b/omagent-core/pyproject.toml @@ -46,6 +46,7 @@ pydub = "^0.25.1" qwen_vl_utils = "0.0.8" face-recognition = "^1.3.0" tavily-python = "^0.5.0" +twelvelabs = "^1.2.8" httpx = {extras = ["http2"], version = "^0.28.1"} func-timeout = "^4.3.5" gradio = "^5.7.1" diff --git a/omagent-core/src/omagent_core/tool_system/tools/twelvelabs_video/__init__.py b/omagent-core/src/omagent_core/tool_system/tools/twelvelabs_video/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/omagent-core/src/omagent_core/tool_system/tools/twelvelabs_video/twelvelabs_video.py b/omagent-core/src/omagent_core/tool_system/tools/twelvelabs_video/twelvelabs_video.py new file mode 100644 index 00000000..d2903936 --- /dev/null +++ b/omagent-core/src/omagent_core/tool_system/tools/twelvelabs_video/twelvelabs_video.py @@ -0,0 +1,156 @@ +from typing import Any, Dict + +from pydantic import field_validator + +from ....utils.logger import logging +from ....utils.registry import registry +from ...base import ArgSchema, BaseTool + +ARGSCHEMA = { + "task": { + "type": "string", + "enum": ["analyze", "embed"], + "description": "Which TwelveLabs capability to use. `analyze` runs the Pegasus model to " + "understand a video and answer a prompt (requires `video_url`). `embed` runs the Marengo " + "model to produce a multimodal embedding vector for a piece of text or an image " + "(requires `text` or `image_url`).", + "required": True, + }, + "prompt": { + "type": "string", + "description": "For `analyze`: the question or instruction about the video, " + "e.g. 'Describe what happens in this video' or 'List every product shown'.", + "required": False, + }, + "text": { + "type": "string", + "description": "For `embed`: the text to embed into a multimodal vector " + "(shares an embedding space with video, so it can be used for video retrieval).", + "required": False, + }, + "image_url": { + "type": "string", + "description": "For `embed`: a public URL to an image to embed into a multimodal vector.", + "required": False, + }, + "video_url": { + "type": "string", + "description": "For `analyze`: a public URL to the video file. TwelveLabs fetches it server-side.", + "required": False, + }, + "max_tokens": { + "type": "integer", + "description": "For `analyze`: maximum number of tokens to generate. Default is 2048.", + "required": False, + }, +} + + +@registry.register_tool() +class TwelveLabsVideo(BaseTool): + """Video understanding and multimodal embedding tool backed by the TwelveLabs API. + + Wraps two TwelveLabs models: + - Pegasus (``analyze``): video understanding -- answer a natural language prompt about a video. + - Marengo (``embed``): produce a 512-dim multimodal embedding vector from text or an image. + Text/image/video share one embedding space, so a text or image vector can retrieve videos. + + Get a free API key at https://twelvelabs.io and provide it via ``api_key`` + (typically ``${env|twelvelabs_api_key}`` in a tool config). + """ + + class Config: + """Configuration for this pydantic object.""" + + extra = "allow" + arbitrary_types_allowed = True + + args_schema: ArgSchema = ArgSchema(**ARGSCHEMA) + description: str = ( + "Understand videos and generate multimodal embeddings with TwelveLabs. " + "Use task='analyze' to answer questions about a video (Pegasus), or " + "task='embed' to get an embedding vector for text or an image (Marengo)." + ) + api_key: str + analyze_model_name: str = "pegasus1.5" + embed_model_name: str = "marengo3.0" + + @field_validator("api_key") + @classmethod + def api_key_validator(cls, api_key: str) -> str: + if not api_key: + raise ValueError( + "TwelveLabs API key is not provided. Get a free key at https://twelvelabs.io." + ) + return api_key + + def __init__(self, **data: Any) -> None: + super().__init__(**data) + # Imported here so the dependency is only required when the tool is used. + from twelvelabs import TwelveLabs + + self.client = TwelveLabs(api_key=self.api_key) + + def _run( + self, + task: str, + prompt: str = None, + text: str = None, + image_url: str = None, + video_url: str = None, + max_tokens: int = 2048, + ) -> Dict[str, Any]: + if task == "analyze": + return self._analyze( + prompt=prompt, video_url=video_url, max_tokens=max_tokens + ) + elif task == "embed": + return self._embed(text=text, image_url=image_url) + else: + raise ValueError( + "Unknown task {!r}. Must be one of 'analyze' or 'embed'.".format(task) + ) + + def _analyze(self, prompt: str, video_url: str, max_tokens: int) -> Dict[str, Any]: + if not video_url: + raise ValueError("`video_url` is required for task='analyze'.") + if not prompt: + raise ValueError("`prompt` is required for task='analyze'.") + from twelvelabs.types.video_context import VideoContext_Url + + try: + res = self.client.analyze( + model_name=self.analyze_model_name, + video=VideoContext_Url(url=video_url), + prompt=prompt, + max_tokens=max_tokens, + ) + return {"text": res.data, "finish_reason": res.finish_reason} + except Exception as e: + logging.error(f"TwelveLabs analyze failed: {e}") + return {"text": "", "error": str(e)} + + def _embed(self, text: str, image_url: str) -> Dict[str, Any]: + if not text and not image_url: + raise ValueError( + "Either `text` or `image_url` is required for task='embed'." + ) + if text and image_url: + raise ValueError( + "Provide exactly one of `text` or `image_url` for task='embed', not both." + ) + try: + if text: + res = self.client.embed.create( + model_name=self.embed_model_name, text=text + ) + vector = res.text_embedding.segments[0].float_ + else: + res = self.client.embed.create( + model_name=self.embed_model_name, image_url=image_url + ) + vector = res.image_embedding.segments[0].float_ + return {"embedding": vector, "dimension": len(vector)} + except Exception as e: + logging.error(f"TwelveLabs embed failed: {e}") + return {"embedding": [], "error": str(e)}