--- slug: "keynote-demo-2025" source_type: "readme" source_url: "https://cdn.jsdelivr.net/gh/anyscale/keynote-demo-2025@main/README.md" repo: "https://github.com/anyscale/keynote-demo-2025" source_file: "README.md" branch: "main" --- # 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: 1. **Planning**: Calls LLM to decide which retrieval actions to execute 2. **Retrieval**: Calls appropriate services (TranscriptRetrieval, FrameRetrieval, or both) 3. **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) → FrameService - `SEARCH_TRANSCRIPTS`: Text queries (dialogue, keywords) → TranscriptService - `SEARCH_TRANSCRIPTS_AT_FRAMES`: Multi-modal (visual + speech) → Both services - `READ_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 ```bash # Install dependencies pip install -r requirements.txt # Install system dependencies (Ubuntu/Debian) sudo apt-get update sudo apt-get install ffmpeg ``` ```bash # 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: ```bash 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: ```bash # 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: ```bash 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: ```bash # 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:** ```bash anyscale job submit -f /home/ray/default/configs/preprocessing_pipeline.yaml ``` **Deploy the microservices:** ```bash anyscale service deploy -f /home/ray/default/configs/rayflix_service.yaml --canary-percent 100 ``` **Launch Streamlit UI:** ```bash # 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). ```python 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)