---
id: S03
parent: M001
milestone: M001
provides:
  - 6 Celery tasks: stage2-6 + run_pipeline orchestrator
  - LLMClient with primary/fallback for downstream use
  - EmbeddingClient for vector generation
  - QdrantManager for vector store operations
  - POST /api/v1/pipeline/trigger/{video_id} manual re-trigger endpoint
  - 8 Pydantic schemas for pipeline stage I/O
  - 4 editable prompt templates in prompts/
  - 10 integration tests with mock fixtures
requires:
  - slice: S02
    provides: Ingest endpoint, database models (SourceVideo, TranscriptSegment, KeyMoment, TechniquePage, Creator), async SQLAlchemy engine, test infrastructure
affects:
  - S04
  - S05
key_files:
  - backend/config.py
  - backend/worker.py
  - backend/pipeline/__init__.py
  - backend/pipeline/schemas.py
  - backend/pipeline/llm_client.py
  - backend/pipeline/embedding_client.py
  - backend/pipeline/qdrant_client.py
  - backend/pipeline/stages.py
  - backend/routers/pipeline.py
  - backend/routers/ingest.py
  - backend/main.py
  - prompts/stage2_segmentation.txt
  - prompts/stage3_extraction.txt
  - prompts/stage4_classification.txt
  - prompts/stage5_synthesis.txt
  - backend/tests/test_pipeline.py
  - backend/tests/fixtures/mock_llm_responses.py
  - backend/tests/conftest.py
  - backend/requirements.txt
key_decisions:
  - Sync OpenAI/SQLAlchemy/Qdrant throughout Celery tasks — no async in worker context (D004)
  - Embedding/Qdrant stage is non-blocking side-effect — failures don't break pipeline (D005)
  - Stage 4 classification stored in Redis (24h TTL) due to missing KeyMoment columns
  - Pipeline dispatch from ingest is best-effort; manual trigger returns 503 on Celery failure
  - LLMClient retries once with JSON nudge on malformed LLM output before failing
patterns_established:
  - Celery task pattern: @celery_app.task(bind=True, max_retries=3) with sync SQLAlchemy session per task
  - LLM client pattern: primary → fallback → fail, with Pydantic response parsing
  - Non-blocking side-effect pattern: max_retries=0, catch-all exception handler, pipeline continues
  - Prompt template pattern: plain text files in prompts/ dir, XML-style content fencing, loaded at runtime
  - Pipeline test pattern: patch module-level _engine/_SessionLocal globals to redirect stages to test DB
observability_surfaces:
  - INFO log at start/end of each stage with video_id and duration
  - WARNING on LLM fallback trigger
  - ERROR on LLM parse failure with raw response excerpt
  - WARNING on embedding/Qdrant failures with error details
  - source_videos.processing_status tracks pipeline progress per video
  - Celery task registry shows all 6 registered tasks
  - POST /api/v1/pipeline/trigger/{video_id} returns current processing_status
drill_down_paths:
  - .gsd/milestones/M001/slices/S03/tasks/T01-SUMMARY.md
  - .gsd/milestones/M001/slices/S03/tasks/T02-SUMMARY.md
  - .gsd/milestones/M001/slices/S03/tasks/T03-SUMMARY.md
  - .gsd/milestones/M001/slices/S03/tasks/T04-SUMMARY.md
  - .gsd/milestones/M001/slices/S03/tasks/T05-SUMMARY.md
duration: ""
verification_result: passed
completed_at: 2026-03-29T22:59:23.268Z
blocker_discovered: false
---

# S03: LLM Extraction Pipeline + Qdrant Integration

**Built the complete 6-stage LLM extraction pipeline (segmentation → extraction → classification → synthesis → embedding) with Celery workers, sync SQLAlchemy, primary/fallback LLM endpoints, Qdrant vector indexing, configurable prompt templates, auto-dispatch from ingest, manual re-trigger API, and 10 integration tests — all 16 tests pass.**

## What Happened

## What This Slice Delivered

S03 implemented the core intelligence of Chrysopedia: the background worker pipeline that transforms raw transcripts into structured knowledge (technique pages, key moments, topic tags, embeddings).

### T01 — Infrastructure Foundation
Extended Settings with 12 config fields (LLM primary/fallback endpoints, embedding config, Qdrant connection, prompt path, review mode). Created the Celery app in `worker.py` using Redis as broker. Built `LLMClient` with sync `openai.OpenAI` and primary→fallback logic that catches `APIConnectionError` and `APITimeoutError`. Defined 8 Pydantic schemas (`TopicSegment`, `SegmentationResult`, `ExtractedMoment`, `ExtractionResult`, `ClassifiedMoment`, `ClassificationResult`, `SynthesizedPage`, `SynthesisResult`) matching the pipeline stage inputs/outputs.

### T02 — Pipeline Stages + Prompt Templates
Created 4 prompt template files in `prompts/` with XML-style content fencing. Implemented 5 Celery tasks in `pipeline/stages.py`:
- **stage2_segmentation**: Groups transcript segments into topic boundaries, updates `topic_label` on TranscriptSegment rows
- **stage3_extraction**: Extracts key moments from topic groups, creates KeyMoment rows, sets `processing_status=extracted`
- **stage4_classification**: Classifies moments against `canonical_tags.yaml`, stores results in Redis (24h TTL) since KeyMoment lacks tag columns
- **stage5_synthesis**: Synthesizes TechniquePage rows from grouped moments, links KeyMoments, sets `processing_status=reviewed` (or `published` if `review_mode=False`)
- **run_pipeline**: Orchestrator that checks `processing_status` and chains only the remaining stages for resumability

All tasks use sync SQLAlchemy sessions (psycopg2) with `bind=True, max_retries=3`. `_safe_parse_llm_response` retries once with a JSON nudge on malformed output.

### T03 — Embedding & Qdrant Integration
Created `EmbeddingClient` (sync `openai.OpenAI` for `/v1/embeddings`) that returns empty list on errors. Created `QdrantManager` with idempotent `ensure_collection()` (cosine distance, config-driven dimensions) and `upsert_technique_pages()`/`upsert_key_moments()` with full metadata payloads. Added `stage6_embed_and_index` as a non-blocking side-effect (`max_retries=0`, catches all exceptions) appended to the pipeline chain.

### T04 — API Wiring
Wired `run_pipeline.delay()` dispatch after ingest commit (best-effort — failures don't break ingest response). Added `POST /api/v1/pipeline/trigger/{video_id}` for manual re-processing (returns 404 for missing video, 503 on Celery failure). Mounted pipeline router in `main.py`.

### T05 — Integration Tests
Created 10 integration tests with mocked LLM/Qdrant and real PostgreSQL: stages 2-6 produce correct DB records, pipeline resumes from `extracted` status, trigger endpoint returns 200/404, ingest dispatches pipeline, LLM falls back on primary failure. All 16 tests (6 ingest + 10 pipeline) pass.

### Key Deviation
Stage 4 stores classification data in Redis rather than DB columns because `KeyMoment` model lacks `topic_tags`/`topic_category` columns. This is an intentional simplification — stage 5 reads from Redis during synthesis.

## Verification

All slice verification checks pass:

**T01 verification (5/5):** Settings prints correct defaults, all 8 schema classes import, LLMClient imports clean, celery_app.main prints 'chrysopedia', openai/qdrant-client in requirements.txt.

**T02 verification (3/3):** 4 prompt files exist, all 5 stage functions import, worker shows 6 registered tasks.

**T03 verification (3/3):** EmbeddingClient, QdrantManager, stage6_embed_and_index all import successfully.

**T04 verification (3/3):** Pipeline router has /trigger/{video_id} route, pipeline in main.py, run_pipeline in ingest.py.

**T05 verification (2/2):** `cd backend && python -m pytest tests/test_pipeline.py -v` — 10/10 pass. `cd backend && python -m pytest tests/ -v` — 16/16 pass (122s).

All 16 registered tests pass. 6 Celery tasks registered in worker.

## Requirements Advanced

- R003 — Full pipeline stages 2-6 implemented and tested: segmentation, extraction, classification, synthesis, embedding. 10 integration tests verify all stages with mocked LLM and real PostgreSQL.
- R009 — Write path implemented: EmbeddingClient generates vectors, QdrantManager upserts with metadata payloads. Read path (search query) deferred to S05.
- R011 — Stage 4 loads canonical_tags.yaml for classification. Tag taxonomy is config-driven.
- R012 — run_pipeline orchestrator resumes from last completed stage. Auto-dispatch from ingest handles new videos. Manual trigger supports re-processing.
- R013 — 4 prompt template files loaded from configurable prompts_path. Manual trigger enables re-processing after prompt edits.

## Requirements Validated

- R003 — 10 integration tests prove full pipeline: stage2 updates topic_labels, stage3 creates KeyMoments, stage4 classifies tags, stage5 creates TechniquePages, stage6 embeds to Qdrant. Resumability and LLM fallback tested.
- R013 — 4 prompt files in prompts/, loaded from configurable path, POST /api/v1/pipeline/trigger/{video_id} enables re-processing.

## New Requirements Surfaced

None.

## Requirements Invalidated or Re-scoped

None.

## Deviations

Stage 4 classification data stored in Redis (not DB columns) because KeyMoment model lacks topic_tags/topic_category columns. Added psycopg2-binary to requirements.txt for sync SQLAlchemy. Created pipeline/stages.py stub in T01 so worker.py import chain succeeds ahead of T02. Pipeline router uses lazy import of run_pipeline inside handler to avoid circular imports.

## Known Limitations

Stage 4 classification stored in Redis with 24h TTL — if Redis is flushed between stage 4 and stage 5, classification data is lost. QdrantManager uses random UUIDs for point IDs — re-indexing creates duplicates rather than updating existing points. KeyMoment model needs topic_tags/topic_category columns for a permanent solution.

## Follow-ups

Add topic_tags and topic_category columns to KeyMoment model to eliminate Redis dependency for classification. Add deterministic point IDs to QdrantManager based on content hash for idempotent re-indexing. Consider adding a /api/v1/pipeline/status/{video_id} endpoint for monitoring pipeline progress.

## Files Created/Modified

- `backend/config.py` — Extended Settings with 12 LLM/embedding/Qdrant/prompt config fields
- `backend/requirements.txt` — Added openai, qdrant-client, pyyaml, psycopg2-binary
- `backend/worker.py` — Created Celery app with Redis broker, imports pipeline.stages
- `backend/pipeline/__init__.py` — Created empty package init
- `backend/pipeline/schemas.py` — 8 Pydantic models for pipeline stage I/O
- `backend/pipeline/llm_client.py` — Sync LLMClient with primary/fallback logic
- `backend/pipeline/embedding_client.py` — Sync EmbeddingClient for /v1/embeddings
- `backend/pipeline/qdrant_client.py` — QdrantManager with idempotent collection mgmt and metadata upserts
- `backend/pipeline/stages.py` — 6 Celery tasks: stages 2-6 + run_pipeline orchestrator
- `backend/routers/pipeline.py` — POST /trigger/{video_id} manual re-trigger endpoint
- `backend/routers/ingest.py` — Added run_pipeline.delay() dispatch after ingest commit
- `backend/main.py` — Mounted pipeline router under /api/v1
- `prompts/stage2_segmentation.txt` — LLM prompt for topic boundary detection
- `prompts/stage3_extraction.txt` — LLM prompt for key moment extraction
- `prompts/stage4_classification.txt` — LLM prompt for canonical tag classification
- `prompts/stage5_synthesis.txt` — LLM prompt for technique page synthesis
- `backend/tests/test_pipeline.py` — 10 integration tests covering all pipeline stages
- `backend/tests/fixtures/mock_llm_responses.py` — Mock LLM response fixtures for all stages
- `backend/tests/conftest.py` — Added sync engine/session fixtures and pre_ingested_video fixture
