# S03 — LLM Extraction Pipeline + Qdrant Integration — Research

**Date:** 2026-03-29

## Summary

S03 is the highest-risk, highest-complexity slice in M001. It builds the entire LLM extraction pipeline (stages 2–5: segmentation → extraction → classification → synthesis), the Qdrant vector embedding pipeline, the Celery worker infrastructure, the prompt template system, and the config extensions for LLM/embedding endpoints. This slice touches requirements R003, R009, R011, R012, and R013.

The codebase currently has **no worker/Celery/pipeline code at all** — it's entirely greenfield for this slice. Docker Compose already declares a `chrysopedia-worker` service that runs `celery -A worker worker`, but the `worker` module doesn't exist yet. The data model is complete — `KeyMoment`, `TechniquePage`, `RelatedTechniqueLink`, and `Tag` tables with their enums (`ProcessingStatus`, `ReviewStatus`, `KeyMomentContentType`, etc.) are all defined in `models.py` and migrated via `001_initial.py`. The S02 ingest endpoint already creates `TranscriptSegment` rows and sets `processing_status = transcribed`, which is the trigger state for the pipeline.

The primary risk is LLM integration: the pipeline must talk to an OpenAI-compatible API (Qwen on DGX Sparks primary, Ollama fallback), parse structured JSON responses, handle failures gracefully, and be resumable per-video per-stage. Secondary risk is Qdrant integration — creating collections with the right schema, generating embeddings via a configurable endpoint, and upserting points with metadata payloads. The Celery+Redis infrastructure is lower risk since both are already in Docker Compose and `celery[redis]` is in requirements.txt.

## Recommendation

Build this slice in a layered approach: (1) config/settings extensions first, (2) Celery app + task skeleton, (3) LLM client with fallback, (4) prompt templates on disk, (5) pipeline stages 2–5 as individual Celery tasks with a chain orchestrator, (6) Qdrant client + embedding pipeline, (7) trigger mechanism (post-ingest hook or API endpoint), (8) integration tests with mocked LLM responses. 

Use the `openai` Python SDK for LLM calls — it works with any OpenAI-compatible API (Open WebUI, Ollama, vLLM). Use `qdrant-client` with `AsyncQdrantClient` for vector operations. Use Pydantic models to define expected LLM output schemas for each stage, enabling `response_format` structured output where the model supports it, with JSON-parse fallback for models that don't. Each pipeline stage should be an individual Celery task chained together, with `processing_status` updated after each stage completes, enabling resumability.

## Implementation Landscape

### Key Files

- `backend/models.py` — Complete. All 7 entities exist with correct enums. `ProcessingStatus` has the progression: `pending → transcribed → extracted → reviewed → published`. `KeyMoment` has `review_status`, `content_type`, `plugins`, `raw_transcript`. `TechniquePage` has `body_sections` (JSONB), `signal_chains` (JSONB), `topic_tags` (ARRAY), `source_quality`. No schema changes needed.
- `backend/database.py` — Provides `engine`, `async_session`, `get_session`. Worker tasks will need their own session management (not FastAPI dependency injection).
- `backend/config.py` — Currently has DB, Redis, CORS, file storage settings. **Must be extended** with: `llm_api_url`, `llm_api_key`, `llm_model`, `llm_fallback_url`, `llm_fallback_model`, `embedding_api_url`, `embedding_model`, `qdrant_url`, `qdrant_collection`, `review_mode` (bool). All referenced in `.env.example` but not yet in `Settings`.
- `backend/requirements.txt` — Has `celery[redis]`, `redis`, `httpx`, `pydantic`. **Must add**: `openai`, `qdrant-client`.
- `backend/routers/ingest.py` — After successful ingest, should trigger the pipeline. Currently sets `processing_status = transcribed` and commits. Add a post-commit Celery task dispatch: `run_pipeline.delay(video_id)`.
- `backend/worker.py` — **New file.** Celery app definition. Docker Compose expects `celery -A worker worker`.
- `backend/pipeline/` — **New directory.** Individual stage modules:
  - `backend/pipeline/__init__.py`
  - `backend/pipeline/llm_client.py` — OpenAI-compatible client with primary/fallback logic
  - `backend/pipeline/embedding_client.py` — Embedding generation via OpenAI-compatible `/v1/embeddings` endpoint
  - `backend/pipeline/qdrant_client.py` — Qdrant collection management and upsert operations
  - `backend/pipeline/stages.py` — Celery tasks for stages 2–5 + orchestrator chain
  - `backend/pipeline/schemas.py` — Pydantic models for LLM input/output per stage
- `prompts/` — **New template files:**
  - `prompts/stage2_segmentation.txt` — Topic boundary detection prompt
  - `prompts/stage3_extraction.txt` — Key moment extraction prompt
  - `prompts/stage4_classification.txt` — Classification/tagging prompt
  - `prompts/stage5_synthesis.txt` — Technique page synthesis prompt
- `config/canonical_tags.yaml` — Already exists with 6 categories, sub-topics, and genre taxonomy. Pipeline reads this during stage 4 classification.

### Build Order

1. **Config + dependencies first** — Extend `Settings` with LLM/embedding/Qdrant/review_mode vars. Add `openai` and `qdrant-client` to requirements.txt. This unblocks everything.

2. **Celery app + skeleton tasks** — Create `backend/worker.py` with Celery app instance pointing to Redis. Create `backend/pipeline/stages.py` with placeholder tasks for stages 2–5 and a `run_pipeline` orchestrator that chains them. Verify `celery -A worker worker` starts without errors. This proves the Celery infrastructure works.

3. **LLM client with fallback** — Create `backend/pipeline/llm_client.py` using the `openai` SDK's `AsyncOpenAI` client. The key insight: the openai SDK works with any OpenAI-compatible API by setting `base_url`. Build a wrapper that tries the primary endpoint, catches connection/timeout errors, and falls back to the secondary. Define Pydantic schemas for each stage's expected LLM output in `backend/pipeline/schemas.py`.

4. **Prompt templates** — Write the 4 prompt template files. These are the soul of the extraction pipeline. Stage 2 (segmentation) needs the full transcript and asks for topic boundaries. Stage 3 (extraction) takes individual segments and extracts key moments. Stage 4 (classification) takes key moments and assigns tags from the canonical list. Stage 5 (synthesis) takes all moments for a creator+topic and produces technique page content.

5. **Pipeline stages implementation** — Implement the actual logic in each Celery task. Each stage:
   - Reads its input from PostgreSQL (transcript segments, key moments, etc.)
   - Loads the appropriate prompt template from disk
   - Calls the LLM client
   - Parses the structured response
   - Writes results to PostgreSQL (topic_labels on segments, new KeyMoment rows, tags, TechniquePage records)
   - Updates `processing_status` on the SourceVideo
   - Pipeline is resumable: if a video is at `transcribed`, start from stage 2; if at `extracted`, skip to stage 4 (classification), etc.

6. **Qdrant integration** — Create `backend/pipeline/qdrant_client.py` and `backend/pipeline/embedding_client.py`. Create the `chrysopedia` collection (if not exists) with vector size matching the embedding model (768 for nomic-embed-text). After stage 5 synthesis (or after key moment creation), embed summaries and upsert to Qdrant with metadata payloads (creator_id, topic, timestamps, source_video_id).

7. **Ingest trigger** — Modify `backend/routers/ingest.py` to dispatch `run_pipeline.delay(video_id)` after successful commit. Add an API endpoint `POST /api/v1/pipeline/trigger/{video_id}` for manual re-triggering.

8. **Integration tests** — Test with mocked LLM responses (don't call real LLMs in CI). Verify: pipeline stages update processing_status correctly, key moments are created with correct fields, technique pages are synthesized, Qdrant upsert is called with correct payloads.

### Verification Approach

1. `celery -A worker worker --loglevel=debug` starts cleanly with registered tasks visible in output
2. `python -c "from pipeline.stages import run_pipeline; print('import ok')"` — module imports work
3. `python -c "from pipeline.llm_client import LLMClient; print('client ok')"` — LLM client importable
4. Pipeline integration test: ingest a sample transcript → verify processing_status progresses through `transcribed → extracted → reviewed → published` (with mocked LLM)
5. Key moment creation test: verify stage 3 output produces KeyMoment rows with correct fields
6. Technique page synthesis test: verify stage 5 produces TechniquePage with body_sections, signal_chains, plugins
7. Qdrant test: verify embedding client produces vectors and qdrant_client upserts with correct payloads (mock Qdrant)
8. Prompt template loading: verify all 4 templates load from `prompts/` directory
9. Config test: verify all new Settings fields have correct defaults and load from env vars
10. `docker compose config > /dev/null` still passes with worker service

## Don't Hand-Roll

| Problem | Existing Solution | Why Use It |
|---------|------------------|------------|
| OpenAI-compatible API calls | `openai` Python SDK | Works with Open WebUI, Ollama, vLLM by setting `base_url`. Has async support, structured output parsing, retry logic. No need to build raw httpx calls. |
| Qdrant vector operations | `qdrant-client` | Official Python client with `AsyncQdrantClient`, typed models for `PointStruct`, `VectorParams`, metadata filtering. Already mature. |
| Background job queue | `celery[redis]` | Already in requirements.txt and Docker Compose. Task chaining, retries, result tracking built in. |
| Structured LLM output parsing | Pydantic models + openai SDK `response_format` | The openai SDK supports `response_format={"type": "json_object"}` for structured output. Combined with Pydantic `model_validate_json()` for robust parsing. |
| YAML config loading | `PyYAML` (already installed) | For loading `canonical_tags.yaml`. Already a dependency. |

## Constraints

- **Celery is sync, DB is async.** The Celery worker runs in a sync event loop. Pipeline tasks using SQLAlchemy async require either: (a) running `asyncio.run()` inside each task, (b) using `asgiref.sync_to_async`, or (c) creating a sync SQLAlchemy engine for the worker. Option (a) is simplest — each Celery task wraps an async function with `asyncio.run()`. The openai SDK has both sync and async clients; use the sync client in Celery tasks to avoid complexity.
- **Qdrant is at 10.0.0.10 from the hypervisor, not localhost.** The `QDRANT_URL` env var must be configured per-environment. In Docker Compose on the hypervisor, it should point to `http://10.0.0.10:6333`. Tests should mock Qdrant entirely.
- **LLM structured output support varies by model.** Qwen via Open WebUI may or may not support `response_format: json_object`. The pipeline must handle both: try structured output first, fall back to prompting for JSON and parsing the raw text response. Always validate with Pydantic regardless.
- **Prompts are bind-mounted read-only.** Docker Compose mounts `./prompts:/prompts:ro` for the worker. Prompt templates should be loaded from `/prompts/` in production and `./prompts/` in development. Config should have a `prompts_path` setting.
- **`processing_status` enum progression is: `pending → transcribed → extracted → reviewed → published`.** The pipeline must respect this: stage 2+3 set `extracted`, stage 4 doesn't change status (classification is metadata), stage 5 sets `reviewed` (or `published` if `review_mode=false`).
- **Naive datetimes required.** Per KNOWLEDGE.md, all datetime values must use `_now()` (naive UTC) for asyncpg compatibility with TIMESTAMP WITHOUT TIME ZONE columns.

## Common Pitfalls

- **Celery `asyncio.run()` nesting** — Celery tasks run in a sync context. If you try to call async functions without `asyncio.run()`, you get "no running event loop" errors. If a Celery worker happens to use an async-aware event loop (e.g., with gevent), `asyncio.run()` may fail with "cannot run nested event loop". Safest approach: use sync clients (sync `openai.OpenAI`, sync `QdrantClient`) in Celery tasks, or use a dedicated sync SQLAlchemy engine for the worker.
- **LLM JSON parse failures** — LLMs don't always produce valid JSON even when asked. Every LLM response must be wrapped in try/except for `json.JSONDecodeError` and `pydantic.ValidationError`. Log the raw response on failure for debugging. Consider retry with a "please output valid JSON" follow-up.
- **Qdrant collection recreation** — `create_collection` fails if collection exists. Always use `collection_exists()` check first, or use `recreate_collection()` only during initial setup.
- **Token limits on long transcripts** — A 4-hour livestream transcript could be 50k+ tokens. Stage 2 (segmentation) must handle this via chunking — split the transcript into overlapping windows and segment each independently, then merge boundary segments. This is the hardest part of stage 2.
- **Prompt template injection** — Transcript text is user-generated content inserted into prompts. While this is a self-hosted system with trusted content, prompts should clearly delineate the transcript from instructions using XML-style tags or markdown fences.

## Open Risks

- **LLM endpoint availability during testing.** Neither DGX Sparks nor local Ollama are available in the CI/test environment. All LLM calls in tests must use mocked responses. This means test coverage proves the pipeline plumbing but not extraction quality — quality validation requires real LLM calls in a staging environment.
- **Token window sizing for long transcripts.** The spec mentions 4-hour livestreams. Qwen 2.5-72B has a 128K context window, but sending a full 4-hour transcript (~50K tokens) in one call may degrade quality. The chunking strategy for stage 2 needs careful design — overlapping windows with merge logic at boundaries.
- **Embedding model dimension mismatch.** `nomic-embed-text` produces 768-dimensional vectors. If the embedding model is changed later, the Qdrant collection schema must be recreated. The collection creation should read the dimension from config, not hardcode it.
- **Open WebUI API compatibility.** Open WebUI wraps Ollama/vLLM with its own API layer. The exact endpoint path and authentication method (API key format) may differ from standard OpenAI API conventions. The LLM client should be tested against the actual endpoint before the pipeline processes real data.

## Skills Discovered

| Technology | Skill | Status |
|------------|-------|--------|
| Celery | `bobmatnyc/claude-mpm-skills@celery` (143 installs) | available |
| Qdrant | `davila7/claude-code-templates@qdrant-vector-search` (470 installs) | available |
| Qdrant | `giuseppe-trisciuoglio/developer-kit@qdrant` (339 installs) | available |

Install commands (user decision):
- `npx skills add bobmatnyc/claude-mpm-skills@celery`
- `npx skills add davila7/claude-code-templates@qdrant-vector-search`
