Loop #1 – RAG Pipeline

Project Overview

Building a YouTube video RAG pipeline for an AI coach agent. The pipeline pulls transcripts from a specific YouTube channel, chunks them, embeds them, and stores them in a vector database for agent retrieval.

---

Tech Stack

Component	Choice
Language	Python
Transcript source	Supadata API
Channel listing	YouTube Data API v3
Database	Supabase (PostgreSQL + pgvector)
Embedding model	OpenAI text-embedding-3-small (1536 dims)
Chunking	Custom character-based sliding window

---

Architecture & Pipeline Flow

YouTube Data API v3 (publishedAfter filter — last 7 days)
  → filter already-processed video IDs against Supabase
  → Supadata /v1/transcript (only for new videos)
  → character-based chunker with sentence boundary detection
  → OpenAI text-embedding-3-small
  → Supabase upsert (youtube_videos + youtube_chunks tables)

Trigger: simple polling script run manually or via cron.

---

Decisions

Fetching Strategy

• Use YouTube Data API v3 to list videos from the configured channel with a publishedAfter filter
• Only call Supadata for transcripts on videos that pass the date filter and dedup check
• Rationale: precise date filtering before any Supadata credits are spent; no wasted API calls
• Channel ID specified via environment variable (YOUTUBE_CHANNEL_ID)

Video Filtering

• Only process videos published within the last 7 days
• Check youtube_videos table before fetching transcript — skip if video ID already exists

Transcript Fetching

• Use Supadata GET /v1/transcript with timestamp segments
• If transcript fetch fails: retry once, then log error and skip the video (do not crash pipeline)
• Handle "no transcript available" as a graceful skip, not an error

Chunking

• Strategy: Docling HybridChunker (token-aware + semantic boundaries)
• Tokenizer: sentence-transformers/all-MiniLM-L6-v2
• Max tokens per chunk: 512 (configurable via CHUNK_MAX_TOKENS)
• merge_peers=True — merges small adjacent chunks for better coherence
• Chunk timestamp: locate each Docling chunk's start position in the concatenated transcript string and binary-search the position map to assign offset_ms of the first matching segment
• Reference: PRPs/examples/docling_hybrid_chunking.py

Embeddings

• Model: text-embedding-3-small
• Dimensions: 1536
• Provider: OpenAI

Database Tables

youtube_videos — dedup tracking and video metadata

id            TEXT PRIMARY KEY,   -- YouTube video ID
channel_id    TEXT NOT NULL,
title         TEXT,
url           TEXT NOT NULL,
published_at  TIMESTAMPTZ,
processed_at  TIMESTAMPTZ,
status        TEXT DEFAULT 'pending',  -- pending | processed | failed
created_at    TIMESTAMPTZ DEFAULT NOW()

youtube_chunks — RAG chunks with embeddings

id             BIGSERIAL PRIMARY KEY,
video_id       TEXT REFERENCES youtube_videos(id),
channel_id     TEXT NOT NULL,
content        TEXT NOT NULL,
chunk_index    INT NOT NULL,
start_time_ms  BIGINT,       -- timestamp of first segment in this chunk
embedding      vector(1536),
created_at     TIMESTAMPTZ DEFAULT NOW()

Extensibility

• Pipeline structured so other data sources can be added alongside the YouTube source in future
• Channel ID driven by env var to support switching channels without code changes

---

Environment Variables

Variable	Description
YOUTUBE_CHANNEL_ID	Target YouTube channel ID
YOUTUBE_API_KEY	YouTube Data API v3 key
SUPADATA_API_KEY	Supadata API key
OPENAI_API_KEY	OpenAI API key
SUPABASE_URL	Supabase project URL
SUPABASE_KEY	Supabase service role key
CHUNK_MAX_TOKENS	Max tokens per chunk for HybridChunker (512)

---

Known Considerations & Gotchas

• Supadata createdAt field in /v1/metadata had a known bug after Jan 2026 — use YouTube Data API for publish date filtering instead
• Getting video title + channel name via Supadata costs 2 credits per video (metadata + transcript) — avoided by using YouTube Data API for metadata
• Transcript language: some videos may have no transcript available — handle as graceful skip
• Chunk timestamp mapping: chunk boundaries won't align with transcript segment boundaries — assign timestamp of first segment within the chunk

---

Loop #2 – RAG Agent (YouTube AI Coach)

Project Overview

A Pydantic AI agent with a FastAPI streaming API. The agent acts as a YouTube AI coach, searching the ingested transcript knowledge base via RAG tools and answering questions about video content.

---

Tech Stack

Component	Choice
Language	Python
Agent framework	Pydantic AI
API framework	FastAPI (streaming SSE)
LLM	Configurable via env (LLM_CHOICE, any OpenAI-compat provider)
Embeddings	Configurable via env (EMBEDDING_MODEL_CHOICE)
Database	Supabase (existing project noyddkiaggbhqhsliybl)
Auth	Supabase JWT verification

---

Architecture

Frontend  →  POST /api/pydantic-agent  →  FastAPI (main.py)
                                            │
                                     verify JWT (Supabase /auth/v1/user)
                                            │
                                     fetch conversation history
                                            │
                                     agent.iter() — Pydantic AI
                                            │
                              ┌─────────────┼──────────────┐
                              ▼             ▼              ▼
                     search_chunks   get_full_transcript  get_video_url_with_timestamp
                     (vector search) (full video text)    (URL + timestamp)
                              └─────────────┼──────────────┘
                                            │
                                     stream tokens to client
                                            │
                                     store messages in DB

---

File Structure

src/
├── agent/
│   ├── __init__.py
│   ├── agent.py        # Pydantic AI agent, AgentDeps dataclass, get_model()
│   ├── tools.py        # Tool implementation functions (search, transcript, URL)
│   ├── prompt.py       # AGENT_SYSTEM_PROMPT
│   └── db_utils.py     # Conversation/message DB helpers (copied from example)
├── main.py             # FastAPI app — streaming endpoint + JWT auth
├── rag_pipeline/       # Unchanged from Loop 1
└── utils/
    └── config.py       # Extended with agent env vars
supabase/migrations/
└── 004_agent_tables.sql

---

Decisions

Agent & Model

• get_model(config) reads LLM_CHOICE, LLM_BASE_URL, LLM_API_KEY from PipelineConfig
• api_key falls back to config.openai_api_key if llm_api_key is empty
• Returns OpenAIModel with OpenAIProvider — supports OpenAI, OpenRouter, Ollama
• Same pattern for embedding client: EMBEDDING_BASE_URL, EMBEDDING_API_KEY

AgentDeps

@dataclass
class AgentDeps:
    supabase: Client
    embedding_client: AsyncOpenAI
    http_client: AsyncClient
    transcript_max_chars: int

• http_client included for consistency with example pattern (future web search tool)
• No memories field — Mem0 deliberately excluded to keep it simple
• transcript_max_chars passed from config so tool respects env setting

Three RAG Tools

Tool	When to use	Implementation
search_youtube_knowledge_base	Semantic similarity search across all chunks	Embeds query → match_youtube_chunks RPC → formatted results
get_full_video_transcript	Need full context of a specific video	Fetches all chunks by video_id, ordered by chunk_index, joined
get_video_url_with_timestamp	Cite a specific moment in a video	Returns https://www.youtube.com/watch?v={id}&t={seconds}s

SQL Function: match_youtube_chunks

• Joins youtube_chunks with youtube_videos to return video_title and video_url inline
• Filters WHERE yc.embedding IS NOT NULL
• Accepts match_count parameter (default 5)
• Returns: id, video_id, content, chunk_index, start_time_ms, video_title, video_url, similarity

API Endpoint (/api/pydantic-agent)

• Mirrors example agent_api.py exactly, with deliberate omissions
• Kept: JWT verification, rate limiting, conversation history, title generation, streaming via agent.iter(), error handling
• Removed: Mem0 (memory), Langfuse (tracing), file attachments (binary upload)
• Streaming: PartStartEvent / PartDeltaEvent → chunked JSON lines
• Final chunk includes session_id, conversation_title, complete: true

Database Migration (004_agent_tables.sql)

Adds agent tables to existing Supabase project (does NOT touch youtube_videos/youtube_chunks):

• user_profiles — auto-created on Supabase auth signup via trigger
• requests — per-request tracking for rate limiting (5 req/min default)
• conversations — session records with auto-generated titles
• messages — stores both human and AI turns; message_data TEXT stores Pydantic AI serialized message format for history replay
• computed_session_user_id generated column on messages extracts UUID from {uuid}~{random} session_id format
• All tables: RLS enabled, no DELETE policy

Config Extensions (PipelineConfig)

New optional fields added (all have defaults):

• environment: str = "development"
• llm_base_url: str = "https://api.openai.com/v1"
• llm_api_key: str = ""
• llm_choice: str = "gpt-4o-mini"
• embedding_base_url: str = "https://api.openai.com/v1"
• embedding_api_key: str = ""
• embedding_model_choice: str = "text-embedding-3-small"
• transcript_max_chars: int = 20000

Rate Limiting

• Default 5 requests/minute per user
• Tracked via requests table (checked before agent run)
• Returns streaming error response if exceeded

---

Environment Variables (Loop 2 additions)

Variable	Description	Default
ENVIRONMENT	development or production	development
LLM_BASE_URL	LLM provider base URL	https://api.openai.com/v1
LLM_API_KEY	LLM API key (falls back to OPENAI_API_KEY)	—
LLM_CHOICE	Model name (e.g. gpt-4o-mini)	gpt-4o-mini
EMBEDDING_BASE_URL	Embedding provider base URL	https://api.openai.com/v1
EMBEDDING_API_KEY	Embedding API key (falls back to OPENAI_API_KEY)	—
EMBEDDING_MODEL_CHOICE	Embedding model name	text-embedding-3-small
TRANSCRIPT_MAX_CHARS	Max chars returned by full transcript tool	20000
SUPABASE_SERVICE_KEY	Alias for SUPABASE_KEY (already supported)	—