原始内容
Rayflix
This demo showcases Anyscale's platform capabilities through a multi-modal video analysis pipeline that processes YouTube videos using Whisper (audio transcription) and CLIP (visual embeddings), with an intelligent agentic router for queries.
Architecture Overview (Bottom-Up)
┌─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┐
│ 🎯 COMPLETE SYSTEM ARCHITECTURE │
├─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┤
│ │
│ 🎬 PREPROCESSING 📊 BATCH INFERENCE 🏗️ APPLICATION LAYER │
│ (Ray Data) (Ray Serve) │
│ │
│ ┌─────────────────┐ ┌───────────────────────┐ ┌─────────────────────────────┐ │
│ │ │ │ │ │ │ │
│ │ video.mp4 │ │ Audio Processing │ │ User Query │ │
│ │ ↓ │ │ (Whisper + Text │ │ ↓ │ │
│ │ Extract & │ ════> │ Embeddings) │ ════> │ AgentService │ │
│ │ Chunk Data │ │ ↓ │ │ ┌─────────────────────┐ │ │
│ │ ↓ │ │ Transcripts + │ │ │ Planning │ │ │
│ │ audio_chunks/ │ │ Text Embeddings │ │ │ ↓ │ │ │
│ │ frames/ │ │ │ │ │ Tools (actions) │ │ │
│ │ │ │ ───────────────── │ │ │ ↓ │ │ │
│ │ │ │ │ │ │ Generation │ │ │
│ │ │ │ Visual Processing │ │ │ (optional) │ │ │
│ │ │ ════> │ (CLIP) │ ════> │ └─────────────────────┘ │ │
│ │ │ │ ↓ │ │ │ │
│ │ │ │ Frame Embeddings │ │ │ │
│ │ │ │ │ │ │ │
│ └─────────────────┘ └───────────────────────┘ └─────────────────────────────┘ │
│ │
│ Key Benefits: │
│ ✓ Left-to-right pipeline: Preprocessing → Batch Inference → Application │
│ ✓ Independent microservices with intelligent agent routing │
│ ✓ GPU allocation optimized (only where needed) │
│ ✓ Ray Data for scalable batch processing + Ray Serve for real-time inference │
│ ✓ Multi-modal retrieval (visual + audio) with agentic planning │
└─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┘
Preprocessing: We start by preparing and preprocessing the data where we:
- Extract the audio (
extract_audio) and chunk it (chunk_audio) - Extract the frames (
extract_frames) from the video at specified FPS
┌─────────────────────────────────────────────────────────────────────────────────┐
│ 🎬 PREPROCESSING LAYER │
├─────────────────────────────────────────────────────────────────────────────────┤
│ Input: video.mp4 │
│ ↓ │
│ ┌────────────────────────────────────────────────────────────────┐ │
│ │ extract_audio() → audio.wav → chunk_audio() → audio_chunks/ │ │
│ └────────────────────────────────────────────────────────────────┘ │
│ ┌────────────────────────────────────────────────────────────────┐ │
│ │ extract_frames() → frames/ (extracted at 3 FPS) │ │
│ └────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────────┘
Batch Inference: Then we run batch inference Ray jobs to process the data for retrieval:
- Whisper (
WhisperTranscriber): Transcribes audio chunks. Whisper automatically segments into phrases/sentences (0.2-2s each) with timestamps. - CLIP (
CLIPEmbedder): Generates visual embeddings for video frames for semantic visual search (runs in parallel with Whisper). - Sentence-Transformers (
TranscriptEmbedder): Generates text embeddings for each transcript segment for semantic text search (runs after Whisper).
┌─────────────────────────────────────────────────────────────────────────────────┐
│ 📊 BATCH INFERENCE LAYER (Ray Data) │
├─────────────────────────────────────────────────────────────────────────────────┤
│ WhisperTranscriber CLIPEmbedder │
│ ┌──────────────────┐ ┌──────────────────┐ │
│ │ Audio chunks │ │ Video frames │ │
│ │ ↓ │ │ ↓ │ │
│ │ Whisper large-v3 │ │ CLIP ViT-B/32 │ │
│ │ ↓ │ │ ↓ │ │
│ │ Transcripts with │ │ Frame embeddings │ │
│ │ timestamps │ │ (512-dim) │ │
│ │ (0.2-2s segments)│ └────────┬─────────┘ │
│ └────────┬─────────┘ │
│ ↓ │
│ TranscriptEmbedder │
│ ┌────────────────────────┐ │
│ │ Transcript segments │ │
│ │ ↓ │ │
│ │ all-MiniLM-L6-v2 │ │
│ │ ↓ │ │
│ │ Text embeddings │ │
│ │ (384-dim) │ │
│ └────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────────┘
Application: The system is deployed as independent, scalable microservices. Each service can scale independently based on load:
Service 1: LLM Service (/llm/v1/*)
- Model: Qwen2.5-7B-Instruct with vLLM engine
- Resources: 4 GPUs (tensor parallel)
- Autoscaling: 1-4 replicas
- Purpose: Planning and answer generation
Service 2: TranscriptRetrievalService (/transcript/*)
- Resources: CPU-only (2 CPUs per replica)
- Autoscaling: 1-4 replicas
- Search Methods:
- Hybrid search (30% lexical + 70% semantic)
- Lexical search (keyword matching)
- Semantic search (embedding similarity)
- Timestamp-based search
Service 3: FrameRetrievalService (/frame/*)
- Resources: 1 T4 GPU per replica (for CLIP)
- Autoscaling: 1-4 replicas
- Purpose: Semantic visual search using CLIP embeddings
Service 4: AgentService (/agent/*)
- Resources: CPU-only (1 CPU per replica)
- Autoscaling: 1-8 replicas
- Purpose: Lightweight agent that coordinates the three-step agentic reasoning pattern:
- Planning: Calls LLM to decide which retrieval actions to execute
- Retrieval: Calls appropriate services (TranscriptRetrieval, FrameRetrieval, or both)
- Response: Calls LLM to generate final answer or returns data directly
┌──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┐
│ 🏗️ MICROSERVICES ARCHITECTURE │
├──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┤
│ │
│ User Query AgentService Actions/Tools Retrieval Services Generation │
│ /agent/query (optional) │
│ │
│ ┌─────────┐ ┌──────────────┐ ┌─────────────────┐ ┌──────────────────────┐ ┌────────────────┐ │
│ │ │ │ │ │ SEARCH_ │───────────>│ Transcript Service │ │ │ │
│ │ "What │ │ Planning │ │ TRANSCRIPTS │ │ • Hybrid search │ │ Generation │ │
│ │ does │ ════> │ (always) │ ════> │ │ │ • Lexical search │ ════> │ (questions │ ════> LLM │
│ │ the │ │ ↓ │ ├─────────────────┤ │ • Semantic search │ │ & │ Service │
│ │ old │ │ Coordination │ │ READ_ALL_ │───────────>│ • Timestamp search │ │ summaries) │ Qwen2.5-7B │
│ │ man │ │ ↓ │ │ TRANSCRIPTS │ │ │ │ │ │
│ │ say?" │ │ Response │ │ │ └──────────────────────┘ └────────────────┘ │
│ │ │ │ │ ├─────────────────┤ ↕ │
│ └─────────┘ └──────────────┘ │ SEARCH_TRANS- │ ┌──────────────────────┐ │
│ │ CRIPTS_AT_ │───────────>│ Frame Service │ │
│ │ FRAMES ↻ │ │ • CLIP semantic │ │
│ │ (Frame→Trnscpt) │ │ visual search │ │
│ │ │ │ • Returns frames │ │
│ ├─────────────────┤ │ + timestamps │ │
│ │ SEARCH_FRAMES_ │───────────>│ │ │
│ │ SEMANTIC │ └──────────────────────┘ │
│ └─────────────────┘ │
│ │
│ Benefits: │
│ ✓ Independent scaling per service based on load ✓ Fault isolation between services │
│ ✓ GPU allocation only where needed (cost optimization) ✓ Easy A/B testing and gradual rollouts │
│ ✓ Ray Serve handles service mesh, load balancing, autoscaling ✓ Low latency via shared memory when services are co-located │
└──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┘
Project Structure
src/
├── preprocessing/ # Data preparation pipeline
│ ├── prepare_video_data.py # Downloads videos, extracts frames/audio
│ ├── transcribe.py # Whisper batch inference for transcripts
│ ├── generate_frame_embeddings.py # CLIP batch inference for visual embeddings
│ └── generate_transcript_embeddings.py # Sentence-transformers for transcript semantic search
│
├── agent/ # Agent business logic
│ ├── agent.py # Agent class (3-step agentic reasoning: Planning → Retrieval → Response)
│ ├── clients.py # LLM, CLIP, and text encoder clients
│ ├── retrieval.py # Data loading & retrieval operations (lexical + semantic)
│ └── prompts/ # LLM prompt templates
│ ├── planning.txt # Planning phase prompt
│ └── answer.txt # Answer generation prompt
│
├── serving/ # Ray Serve deployment wrappers
│ ├── app.py # Deploys all microservices independently
│ ├── services.py # All service deployments (Agent, Transcript, Frame)
│ └── shutdown.py # Utility to stop all services
│
└── ui/ # User interface
└── streamlit_app.py # Interactive web UI
Query-to-Response Flow (Microservices)
The agent coordinates a three-step agentic pattern across distributed services:
┌───────────────────────────────────────────────────────────────────────────┐
│ 1. PLANNING (AgentService → LLM Service) │
├───────────────────────────────────────────────────────────────────────────┤
│ User: "What does the old man say?" │
│ ↓ │
│ POST /agent/query → AgentService │
│ ↓ │
│ Agent → POST /llm/v1/chat/completions (planning prompt) │
│ ↓ │
│ LLM returns: {"actions": ["SEARCH_TRANSCRIPTS_AT_FRAMES"], │
│ "visual_query": "old man elderly person"} │
└───────────────────────────────────────────────────────────────────────────┘
┌───────────────────────────────────────────────────────────────────────────┐
│ 2. RETRIEVAL (Agent → Frame & Transcript Services) │
├───────────────────────────────────────────────────────────────────────────┤
│ Agent → POST /frame/search │
│ {"query": "old man elderly person", "top_k": 3} │
│ ↓ │
│ FrameService: CLIP encodes query, searches embeddings │
│ Returns: [{frame: "frame_123.jpg", timestamp: "12.5s"}, ...] │
│ ↓ │
│ Agent → POST /transcript/search │
│ {"search_type": "timestamp", "timestamps": [12.5, 45.2, 78.9], │
│ "time_window": 15.0} │
│ ↓ │
│ TranscriptService: Hybrid search at timestamps (±15s) │
│ Returns: [{text: "...", start: 10.2, score: 0.85}, ...] │
└───────────────────────────────────────────────────────────────────────────┘
┌───────────────────────────────────────────────────────────────────────────┐
│ 3. RESPONSE (Agent → LLM Service → User) │
├───────────────────────────────────────────────────────────────────────────┤
│ Decision logic: │
│ • Visual-only → Return frames │
│ • Multi-modal → Return frames + transcripts │
│ • Question/Summary → Generate LLM answer │
│ ↓ │
│ (For questions) Agent → POST /llm/v1/chat/completions │
│ Loads prompts/answer.txt, synthesizes retrieved data │
│ ↓ │
│ Returns: {"answer": "...", "frames": [...], "transcript_matches": [...]} │
└───────────────────────────────────────────────────────────────────────────┘
Action Types:
SEARCH_FRAMES_SEMANTIC: Visual queries (faces, objects, actions) → FrameServiceSEARCH_TRANSCRIPTS: Text queries (dialogue, keywords) → TranscriptServiceSEARCH_TRANSCRIPTS_AT_FRAMES: Multi-modal (visual + speech) → Both servicesREAD_ALL_TRANSCRIPTS: Summarization → TranscriptService + LLM generation
Features
- Batch Inference: Process video frames and audio at scale with Ray Data
- Unity Catalog Integration: Load data from Unity Catalog with
ray.data.read_unity_catalog() - Lineage Tracking: Automatic track datasets and model artifacts across our workloads.
- Data Dashboard: Visualize operator breakdown, queues, and bottlenecks
- Critical events: Critical event detection (OOM, errors) with contextual logs
- Agents: Intelligent routing between audio, visual, and multi-modal agents (w/ LLM serving)
- Developer Workflow: Workspace → Jobs → Services progression
Quick Start
1. Setup Environment
# Install dependencies
pip install -r requirements.txt
# Install system dependencies (Ubuntu/Debian)
sudo apt-get update
sudo apt-get install ffmpeg
# Only is using local model (not cloud storage)
mkdir /mnt/user_storage/keynote_2025
ln -s /mnt/user_storage/keynote_2025 /home/ray/default/keynote_2025
rm /home/ray/default/keynote_2025 # to remvoe the symlink
2. Prepare Data
Process video from S3:
VIDEO_DIR="s3://rayflix-2025/kpop_demon_hunters"
python src/preprocessing/prepare_video_data.py ${VIDEO_DIR}/video.mp4 --video-dir $VIDEO_DIR --fps 3 --chunk-duration 10
3. Run Batch Inference
Process the video with Whisper and CLIP, then generate semantic embeddings:
# Generate transcripts with Whisper
python src/preprocessing/transcribe.py \
--video-dir ${VIDEO_DIR} \
--batch-size 1 \
--concurrency 4
# Generate frame embeddings with CLIP
python src/preprocessing/generate_frame_embeddings.py \
--video-dir ${VIDEO_DIR} \
--batch-size 32 \
--concurrency 4
# Generate transcript embeddings for semantic search
python src/preprocessing/generate_transcript_embeddings.py \
--transcripts-path ${VIDEO_DIR}/transcripts \
--batch-size 64 \
--concurrency 2
4. Deploy Services
Deploy all services independently:
VIDEO_DIR="s3://rayflix-2025/kpop_demon_hunters"
python src/serving/shutdown.py
python src/serving/app.py --video-dir ${VIDEO_DIR}
streamlit run src/ui/streamlit_app.py
Test the services:
# Test main agent endpoint
curl -X POST http://localhost:8000/agent/query \
-H "Content-Type: application/json" \
-d '{"query": "Find dramatic moments where characters are shouting"}'
# Test transcript service directly
curl -X POST http://localhost:8000/transcript/search \
-H "Content-Type: application/json" \
-d '{"query": "shouting", "search_type": "hybrid", "top_k": 5}'
# Test frame service directly
curl -X POST http://localhost:8000/frame/search \
-H "Content-Type: application/json" \
-d '{"query": "person with sword", "top_k": 3}'
# Test LLM service directly
curl -X POST http://localhost:8000/llm/v1/chat/completions \
-H "Content-Type: application/json" \
-d '{"model": "Qwen--Qwen2.5-7B-Instruct", "messages": [{"role": "user", "content": "Capital of France?"}]}'
5. Production
For production, we run the batch inference workloads as Anyscale Jobs and the overall agentic service as an Anyscale Service.
Deploy the preprocessing job:
anyscale job submit -f /home/ray/default/configs/preprocessing_pipeline.yaml
Deploy the microservices:
anyscale service deploy -f /home/ray/default/configs/rayflix_service.yaml --canary-percent 100
Launch Streamlit UI:
# For localhost (default)
streamlit run src/ui/streamlit_app.py
# For production Anyscale Service
export SERVICE_BASE_URL="https://rayflix-multimodal-service-jgz99.cld-kvedzwag2qa8i5bj.s.anyscaleuserdata.com"
export AUTH_TOKEN="mgyjsg3TPLTEpDjp4e_PcJtKHvFy_C6epvc5qTkT5uw"
streamlit run src/ui/streamlit_app.py
6. Reinforcement learning (beta)
Current State: The LLM-based planner in src/agent/agent.py decides which actions to take (SEARCH_FRAMES_SEMANTIC, SEARCH_TRANSCRIPTS, etc.).
RL Approach:
- State: Query embedding + query type features + data availability context
- Actions: Selection of retrieval actions (or action sequences)
- Reward: User feedback on result quality (implicit: click-through, explicit: thumbs up/down)
- Model: Policy network that learns optimal action selection patterns
Why it works: The current LLM planner has fixed logic. RL can learn from user interactions to discover:
- When visual search is more effective than transcript search
- Optimal combinations of actions (e.g., when to do multi-modal vs single-modal)
- Query patterns that benefit from specific retrieval strategies
Architecture: Contextual bandits since the actions are one-shot, we could use DQN if actions affect future states (but A/B testing with a supervised planner might be the best thing to start with).
class RLActionPlanner:
def __init__(self):
self.policy_net = SimpleNN(
input_dim=query_embedding_dim + context_features,
output_dim=num_action_combinations # e.g., 8-10 distinct strategies
)
def select_actions(self, query_embedding, context):
# Epsilon-greedy exploration
if random() < epsilon:
return random_action_combo()
return self.policy_net(query_embedding, context).argmax()
def update(self, query, actions, user_feedback):
# Simple bandit update or policy gradient
reward = self._compute_reward(user_feedback)
loss = -log_prob(actions) * reward
self.policy_net.backward(loss)
Benefits:
- better results for users by identifying the best actions based on their query
- Faster inference (RL policy is 100x faster than LLM planning calls)
- cost reduction (fewer LLM calls)
- Personalization (can segment policies by user type)