Skip to content

Commit 1cfa84a

Browse files
mohit-twelvelabsmohit-twelvelabs
authored
Add TwelveLabs video RAG template (Pegasus parser + Marengo embedder) (#129)
1 parent b8879cc commit 1cfa84a

11 files changed

Lines changed: 883 additions & 1 deletion

File tree

‎README.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,7 @@ The application templates provided in this repo scale up to **millions of pages
2828
| [`Adaptive RAG App`](templates/adaptive_rag/) | A RAG application using Adaptive RAG, a technique developed by Pathway to reduce token cost in RAG up to 4x while maintaining accuracy. |
2929
| [`Private RAG App with Mistral and Ollama`](templates/private_rag/) | A fully private (local) version of the `question_answering_rag` RAG pipeline using Pathway Live Data Framework, Mistral, and Ollama. |
3030
| [`Slides AI Search App`](templates/slides_ai_search/) | An indexing pipeline for retrieving slides. It performs multi-modal of PowerPoint and PDF and maintains live index of your slides."|
31+
| [`Video RAG with TwelveLabs`](templates/video_rag_twelvelabs/) | A RAG pipeline over **video**. It uses [TwelveLabs](https://twelvelabs.io) Pegasus to turn videos into rich text descriptions and Marengo multimodal embeddings to index them, so you can ask questions about your videos on a live connected data source (files, Google Drive,...). |
3132

3233

3334
## How do these AI Pipelines work?

‎pyproject.toml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -57,7 +57,7 @@ profile = "black"
5757

5858
[tool.mypy]
5959
python_version = "3.10"
60-
exclude = ["templates/adaptive_rag", "templates/private_rag", "templates/document_indexing", "templates/multimodal_rag", "templates/question_answering_rag"]
60+
exclude = ["templates/adaptive_rag", "templates/private_rag", "templates/document_indexing", "templates/multimodal_rag", "templates/question_answering_rag", "templates/video_rag_twelvelabs"]
6161
ignore_missing_imports = true
6262
check_untyped_defs = true
6363
warn_redundant_casts = true

‎templates/video_rag_twelvelabs/.env.example‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,6 @@
1+
# TwelveLabs API key, used by the Pegasus video parser and the Marengo embedder.
2+
# Get a free key at https://twelvelabs.io
3+
TWELVELABS_API_KEY=
4+
5+
# OpenAI API key, used by the question-answering LLM.
6+
OPENAI_API_KEY=

‎templates/video_rag_twelvelabs/Dockerfile‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,12 @@
1+
FROM pathwaycom/pathway:latest
2+
3+
WORKDIR /app
4+
5+
COPY requirements.txt .
6+
RUN pip install -U --no-cache-dir -r requirements.txt
7+
8+
COPY . .
9+
10+
EXPOSE 8000
11+
12+
CMD ["python", "app.py"]

‎templates/video_rag_twelvelabs/README.md‎

Lines changed: 122 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,122 @@
1+
<p align="center" class="flex items-center gap-1 justify-center flex-wrap">
2+
<img src="../../assets/gcp-logo.svg?raw=true" alt="GCP Logo" height="20" width="20">
3+
<a href="https://pathway.com/developers/user-guide/deployment/gcp-deploy">Deploy with GCP</a> |
4+
<img src="../../assets/aws-fargate-logo.svg?raw=true" alt="AWS Logo" height="20" width="20">
5+
<a href="https://pathway.com/developers/user-guide/deployment/aws-fargate-deploy">Deploy with AWS</a> |
6+
<img src="../../assets/azure-logo.svg?raw=true" alt="Azure Logo" height="20" width="20">
7+
<a href="https://pathway.com/developers/user-guide/deployment/azure-aci-deploy">Deploy with Azure</a> |
8+
<img src="../../assets/render.png?raw=true" alt="Render Logo" height="20" width="20">
9+
<a href="https://pathway.com/developers/user-guide/deployment/render-deploy"> Deploy with Render </a>
10+
</p>
11+
12+
# Video RAG with Pathway Live Data Framework and TwelveLabs
13+
14+
## Overview
15+
16+
This app template shows how to build a RAG application **over video** using the
17+
Pathway Live Data Framework together with [TwelveLabs](https://twelvelabs.io).
18+
19+
Videos are notoriously hard to put into a RAG pipeline: most stacks first
20+
transcribe the audio and throw away everything that happens on screen. This
21+
template indexes the *whole* video instead, using two TwelveLabs models:
22+
23+
- **[Pegasus](https://docs.twelvelabs.io/docs/concepts/models/pegasus)** — a
24+
video-understanding model that turns each video into a rich text description
25+
(what happens, who and what appears, the setting, on-screen and spoken text,
26+
the overall topic). Pathway indexes that text exactly like it would index a PDF.
27+
- **[Marengo](https://docs.twelvelabs.io/docs/concepts/models/marengo)** — a
28+
multimodal embedding model that produces 512-dimensional vectors in a shared
29+
space for text, image, audio and video. It is used here as the retriever
30+
embedder.
31+
32+
Because the template uses the standard Pathway `DocumentStore` and
33+
`BaseRAGQuestionAnswerer`, everything else works out of the box: live sync with
34+
your data source, the in-memory vector index, caching, and the HTTP API. Drop a
35+
new video into the connected folder and it becomes queryable automatically.
36+
37+
## Architecture
38+
39+
```
40+
video files ─▶ pw.io.fs.read (bytes)
41+
─▶ TwelveLabsVideoParser (Pegasus: video → text)
42+
─▶ TokenCountSplitter (chunking)
43+
─▶ DocumentStore + UsearchKnnFactory
44+
with MarengoEmbedder (text → 512-dim vectors)
45+
─▶ BaseRAGQuestionAnswerer (OpenAI LLM over retrieved context)
46+
─▶ REST API on :8000
47+
```
48+
49+
The TwelveLabs components live in the local
50+
[`pathway_twelvelabs`](pathway_twelvelabs/__init__.py) package:
51+
52+
- `TwelveLabsVideoParser` — a Pathway parser (`pw.UDF`) that uploads the video
53+
bytes as a TwelveLabs asset and asks Pegasus to describe it.
54+
- `MarengoEmbedder` — a Pathway embedder (`BaseEmbedder`) that calls the Marengo
55+
embedding endpoint.
56+
57+
Both are wired in entirely through [`app.yaml`](app.yaml), so you can swap models,
58+
prompts, the data source, or the LLM without touching any Python.
59+
60+
## Customizing the pipeline
61+
62+
- **Change what is extracted from the video.** Set the `prompt` field of
63+
`$parser` in `app.yaml`, e.g. `"Describe this video, focusing on the products
64+
that appear and any prices shown."`
65+
- **Change the data source.** Replace the `!pw.io.fs.read` source with the Google
66+
Drive, SharePoint, or S3 connector (a commented Google Drive example is
67+
included in `app.yaml`).
68+
- **Change the answering LLM.** Edit the `$llm` block — any
69+
[Pathway LLM wrapper](https://pathway.com/developers/api-docs/pathway-xpacks-llm/llms)
70+
works.
71+
72+
## Running the app
73+
74+
### Prerequisites
75+
76+
- A TwelveLabs API key. Get a free one at [twelvelabs.io](https://twelvelabs.io) —
77+
there is a generous free tier.
78+
- An OpenAI API key for the question-answering LLM.
79+
80+
Copy `.env.example` to `.env` and fill in your keys:
81+
82+
```bash
83+
cp .env.example .env
84+
# edit .env and set TWELVELABS_API_KEY and OPENAI_API_KEY
85+
```
86+
87+
Put one or more videos (e.g. `.mp4`, `.mov`) into the `data/` directory.
88+
89+
### With Docker
90+
91+
```bash
92+
docker build -t video-rag-twelvelabs .
93+
docker run -v $(pwd)/data:/app/data --env-file .env -p 8000:8000 video-rag-twelvelabs
94+
```
95+
96+
### Locally
97+
98+
```bash
99+
pip install -r requirements.txt
100+
python app.py
101+
```
102+
103+
### Querying
104+
105+
Once the server is up, ask questions about your videos:
106+
107+
```bash
108+
curl -X POST http://localhost:8000/v1/pw_ai_answer \
109+
-H "Content-Type: application/json" \
110+
-d '{"prompt": "What products are shown in the videos?"}'
111+
```
112+
113+
## Tests
114+
115+
A small test suite lives in [`test_twelvelabs.py`](test_twelvelabs.py). The
116+
no-network tests run without any credentials; the live Marengo embedding test is
117+
skipped unless `TWELVELABS_API_KEY` is set:
118+
119+
```bash
120+
pip install -r requirements.txt pytest
121+
TWELVELABS_API_KEY=... pytest templates/video_rag_twelvelabs/test_twelvelabs.py
122+
```

‎templates/video_rag_twelvelabs/app.py‎

Lines changed: 88 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,88 @@
1+
#!/usr/bin/env python
2+
3+
# Copyright © 2026 Pathway
4+
5+
import logging
6+
from pathlib import Path
7+
from warnings import warn
8+
9+
import pathway as pw
10+
11+
# `pathway_twelvelabs` registers the TwelveLabsVideoParser and MarengoEmbedder
12+
# classes so they can be referenced from `app.yaml` via the `!` YAML tags.
13+
import pathway_twelvelabs # noqa: F401
14+
from dotenv import load_dotenv
15+
from pathway.xpacks.llm.question_answering import BaseRAGQuestionAnswerer
16+
from pathway.xpacks.llm.servers import QASummaryRestServer
17+
from pydantic import BaseModel, ConfigDict, InstanceOf
18+
19+
# To use advanced features with Pathway Live Data Framework Scale, get your free license key from
20+
# https://pathway.com/features and paste it below.
21+
# To use Pathway Live Data Framework Community, comment out the line below.
22+
pw.set_license_key("demo-license-key-with-telemetry")
23+
24+
logging.basicConfig(
25+
level=logging.INFO,
26+
format="%(asctime)s %(name)s %(levelname)s %(message)s",
27+
datefmt="%Y-%m-%d %H:%M:%S",
28+
)
29+
30+
31+
class App(BaseModel):
32+
question_answerer: InstanceOf[BaseRAGQuestionAnswerer]
33+
host: str = "0.0.0.0"
34+
port: int = 8000
35+
36+
with_cache: bool | None = None # deprecated
37+
persistence_backend: pw.persistence.Backend | None = None
38+
persistence_mode: pw.PersistenceMode | None = pw.PersistenceMode.UDF_CACHING
39+
terminate_on_error: bool = False
40+
41+
def run(self) -> None:
42+
server = QASummaryRestServer( # noqa: F841
43+
self.host, self.port, self.question_answerer
44+
)
45+
46+
if self.persistence_mode is None:
47+
if self.with_cache is True:
48+
warn(
49+
"`with_cache` is deprecated. Please use `persistence_mode` instead.",
50+
DeprecationWarning,
51+
)
52+
persistence_mode = pw.PersistenceMode.UDF_CACHING
53+
else:
54+
persistence_mode = None
55+
else:
56+
persistence_mode = self.persistence_mode
57+
58+
if persistence_mode is not None:
59+
if self.persistence_backend is None:
60+
persistence_backend = pw.persistence.Backend.filesystem("./Cache")
61+
else:
62+
persistence_backend = self.persistence_backend
63+
persistence_config = pw.persistence.Config(
64+
persistence_backend,
65+
persistence_mode=persistence_mode,
66+
)
67+
else:
68+
persistence_config = None
69+
70+
pw.run(
71+
persistence_config=persistence_config,
72+
terminate_on_error=self.terminate_on_error,
73+
monitoring_level=pw.MonitoringLevel.NONE,
74+
)
75+
76+
model_config = ConfigDict(extra="forbid", arbitrary_types_allowed=True)
77+
78+
79+
if __name__ == "__main__":
80+
base_dir = Path(__file__).resolve().parent
81+
82+
load_dotenv(base_dir / ".env")
83+
84+
with open(base_dir / "app.yaml") as f:
85+
config = pw.load_yaml(f)
86+
87+
app = App(**config)
88+
app.run()

‎templates/video_rag_twelvelabs/app.yaml‎

Lines changed: 92 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,92 @@
1+
# This YAML configuration file sets up the TwelveLabs video RAG template.
2+
# It indexes videos by turning them into text with the TwelveLabs Pegasus model,
3+
# embedding that text with the TwelveLabs Marengo model, and serving a RAG
4+
# question-answering endpoint on top.
5+
# You can learn more about the YAML syntax here:
6+
# https://pathway.com/developers/templates/configure-yaml
7+
8+
9+
# $sources defines the data sources read and indexed in the RAG.
10+
# Here we read video files from the local `data` directory. The connector emits
11+
# raw bytes, which the TwelveLabs parser sends to Pegasus for analysis.
12+
# You can learn more about configuring data sources here:
13+
# https://pathway.com/developers/templates/yaml-examples/data-sources-examples
14+
$sources:
15+
- !pw.io.fs.read
16+
path: data
17+
format: binary
18+
with_metadata: true
19+
20+
# Uncomment to use the Google Drive connector to index videos from a Drive folder.
21+
# - !pw.io.gdrive.read
22+
# object_id: $DRIVE_ID
23+
# service_user_credentials_file: gdrive_indexer.json
24+
# file_name_pattern:
25+
# - "*.mp4"
26+
# - "*.mov"
27+
# object_size_limit: null
28+
# with_metadata: true
29+
# refresh_interval: 30
30+
31+
32+
# The parser turns each video into text using the TwelveLabs Pegasus model.
33+
# The TWELVELABS_API_KEY environment variable must be set (see .env.example).
34+
# You can customize the prompt to extract exactly what your application needs.
35+
$parser: !pathway_twelvelabs.TwelveLabsVideoParser
36+
model: "pegasus1.5"
37+
max_tokens: 2048
38+
cache_strategy: !pw.udfs.DefaultCache {}
39+
# prompt: "Describe this video, focusing on the products that appear and any prices shown."
40+
41+
# The embedder converts the parsed text into 512-dimensional multimodal
42+
# embeddings using the TwelveLabs Marengo model.
43+
$embedder: !pathway_twelvelabs.MarengoEmbedder
44+
model: "marengo3.0"
45+
cache_strategy: !pw.udfs.DefaultCache {}
46+
retry_strategy: !pw.udfs.ExponentialBackoffRetryStrategy {}
47+
48+
# Splits the parsed video descriptions into chunks for indexing.
49+
$splitter: !pw.xpacks.llm.splitters.TokenCountSplitter
50+
max_tokens: 400
51+
52+
# The LLM used to answer questions over the indexed video descriptions.
53+
# The list of available Pathway LLM wrappers is available here:
54+
# https://pathway.com/developers/api-docs/pathway-xpacks-llm/llms
55+
$llm: !pw.xpacks.llm.llms.OpenAIChat
56+
model: "gpt-4.1-mini"
57+
retry_strategy: !pw.udfs.ExponentialBackoffRetryStrategy
58+
max_retries: 6
59+
cache_strategy: !pw.udfs.DefaultCache {}
60+
temperature: 0
61+
capacity: 8
62+
async_mode: "fully_async"
63+
64+
# Builds the in-memory vector index over the Marengo embeddings.
65+
$retriever_factory: !pw.indexing.UsearchKnnFactory
66+
reserved_space: 1000
67+
embedder: $embedder
68+
metric: !pw.indexing.USearchMetricKind.COS
69+
70+
# Stores and retrieves the indexed video descriptions.
71+
$document_store: !pw.xpacks.llm.document_store.DocumentStore
72+
docs: $sources
73+
parser: $parser
74+
splitter: $splitter
75+
retriever_factory: $retriever_factory
76+
77+
# The RAG question-answering component served over HTTP.
78+
question_answerer: !pw.xpacks.llm.question_answering.BaseRAGQuestionAnswerer
79+
llm: $llm
80+
indexer: $document_store
81+
# You can set the number of documents to be included as the context of the query
82+
# search_topk: 6
83+
# You can use your own prompt for querying.
84+
# prompt_template: "Given these documents: {context}, please answer the question: {query}"
85+
86+
# Change host and port of the webserver by uncommenting these lines
87+
# host: "0.0.0.0"
88+
# port: 8000
89+
90+
# By default, caching is enabled for UDFs with cache_strategy set.
91+
# You can disable it by uncommenting the following line.
92+
# persistence_mode: null

‎templates/video_rag_twelvelabs/data/.gitkeep‎

Whitespace-only changes.

0 commit comments

Comments
 (0)