From 09822dba5891003a57247303806f3c106af7d120 Mon Sep 17 00:00:00 2001 From: Varit Patel Date: Wed, 8 Jul 2026 07:49:52 +0100 Subject: [PATCH] feature: Add recover stale jobs --- .env.example | 29 ++++- .gitignore | 5 +- api/main.py | 50 ++++++- api/routes/ingest.py | 21 ++- api/routes/query.py | 59 +++++++-- config.py | 18 ++- processing/embedder.py | 18 +++ rag/__init__.py | 10 +- rag/pipeline.py | 289 +++++++++++++++++++++++++++++++++++++++++ rag/qa_chain.py | 50 +++++-- rag/retriever.py | 14 +- rag/search.py | 6 +- 12 files changed, 523 insertions(+), 46 deletions(-) create mode 100644 rag/pipeline.py diff --git a/.env.example b/.env.example index 522da81..fbe4fbf 100644 --- a/.env.example +++ b/.env.example @@ -78,6 +78,32 @@ RETRIEVAL_TOP_K=5 CHUNK_DURATION_SECONDS=60 CHUNK_OVERLAP_SECONDS=15 +# HyDE (Hypothetical Document Embeddings) — improves retrieval by generating +# a synthetic transcript paragraph that matches the query before embedding. +# Disabled by default to save time & cost; enable when retrieval quality is critical. +USE_HYDE=false + +# ── Cross-Encoder Reranker ──────────────────────────────────────────────────── +# After vector store retrieval, re-rank results with a cross-encoder model. +# This significantly improves result quality but adds ~50-100ms per query. +USE_RERANKER=True +# Model name for the cross-encoder (must be a sentence-transformers CrossEncoder model) +RERANKER_MODEL=cross-encoder/ms-marco-MiniLM-L-6-v2 +# How many results to fetch from vector store before reranking +RERANKER_TOP_K=10 + +# ── MMR Diversity ───────────────────────────────────────────────────────────── +# Maximal Marginal Relevance — ensures diverse results instead of near-identical chunks. +# 0 = only diversity, 1 = only relevance. 0.5 is a good balance. +USE_MMR=True +MMR_LAMBDA=0.5 + +# ── Multi-Query Expansion ────────────────────────────────────────────────────── +# Generate multiple rephrasings of the question to improve recall. +# Disabled by default: costs 1 extra LLM call + N extra vector searches per query. +USE_MULTI_QUERY=false +MULTI_QUERY_COUNT=3 + # ── Live Stream ─────────────────────────────────────────────────────────────── LIVE_STREAM_SEGMENT_SECONDS=60 @@ -86,6 +112,3 @@ UI_API_BASE_URL=http://localhost:8000 # HTTP request timeout in seconds. Increase this when using local Ollama with long videos. # Default: 120, Recommended for Ollama: 600 (10 minutes) UI_REQUEST_TIMEOUT=120 - -# Query transformation technique -USE_HYDE=True \ No newline at end of file diff --git a/.gitignore b/.gitignore index 1109a4f..57c9932 100644 --- a/.gitignore +++ b/.gitignore @@ -62,4 +62,7 @@ yarn-debug.log* yarn-error.log* .next/ .nuxt/ -dist/ \ No newline at end of file +dist/ + +// Files +Dev-*.md \ No newline at end of file diff --git a/api/main.py b/api/main.py index ddcb96c..9c6fefd 100644 --- a/api/main.py +++ b/api/main.py @@ -19,6 +19,32 @@ from storage.database import init_db +def _recover_stale_jobs() -> None: + """ + Mark jobs that were in progress when the server last stopped as errored. + BackgroundTasks do not survive a restart, so these jobs can never complete. + """ + from storage.database import IngestJob, SessionLocal + + stale_statuses = ("queued", "ingesting", "transcribing", "indexing") + db = SessionLocal() + try: + stale_jobs = ( + db.query(IngestJob).filter(IngestJob.status.in_(stale_statuses)).all() + ) + for job in stale_jobs: + job.status = "error" + job.error_message = "Job interrupted by server restart — please resubmit" + if stale_jobs: + db.commit() + logger.warning(f"Marked {len(stale_jobs)} stale ingestion job(s) as error") + except Exception as e: + db.rollback() + logger.error(f"Failed to recover stale jobs: {e}") + finally: + db.close() + + @asynccontextmanager async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: """Startup and shutdown events.""" @@ -26,6 +52,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: logger.info("Starting Streaming Video-RAG API...") settings.ensure_dirs() init_db() + _recover_stale_jobs() logger.success( f"API ready — LLM={settings.llm_provider.value}/{settings.llm_model}, " f"Whisper={settings.whisper_mode.value}/{settings.whisper_model_size}, " @@ -83,16 +110,35 @@ def health() -> dict[str, str]: if __name__ == "__main__": + from pathlib import Path + import uvicorn + # Only enable SSL if both cert files exist + ssl_kwargs = {} + if settings.api_ssl_certfile and settings.api_ssl_keyfile: + cert_path = Path(settings.api_ssl_certfile) + key_path = Path(settings.api_ssl_keyfile) + if cert_path.exists() and key_path.exists(): + ssl_kwargs["ssl_certfile"] = settings.api_ssl_certfile + ssl_kwargs["ssl_keyfile"] = settings.api_ssl_keyfile + logger.info(f"SSL enabled — cert={cert_path}, key={key_path}") + else: + logger.warning( + f"SSL cert/key files not found at {cert_path}, {key_path}. " + "Running without SSL. Generate certs with: " + "openssl req -x509 -newkey rsa:4096 -days 365 -nodes " + "-keyout certs/localhost-key.pem -out certs/localhost.pem " + "-subj '/CN=localhost' -addext 'subjectAltName=DNS:localhost'" + ) + uvicorn.run( "api.main:app", host=settings.api_host, port=settings.api_port, reload=settings.api_reload, http="h11", - ssl_certfile=settings.api_ssl_certfile, - ssl_keyfile=settings.api_ssl_keyfile, timeout_keep_alive=300, # 5 minute timeout for long running tasks timeout_graceful_shutdown=300, + **ssl_kwargs, ) diff --git a/api/routes/ingest.py b/api/routes/ingest.py index 6b68bdd..8f04a7a 100644 --- a/api/routes/ingest.py +++ b/api/routes/ingest.py @@ -173,6 +173,20 @@ def update_job( ingester = _get_ingester(source_type, platform, api_credentials) asset = ingester.ingest(source) + # Idempotency: skip videos that are already fully indexed to avoid + # re-transcribing and re-embedding (duplicate cost + duplicate vectors). + existing = db.query(Video).filter(Video.id == asset.video_id).first() + if existing is not None and str(existing.status) == "indexed": + update_job( + "done", + f"Video already indexed ({existing.chunk_count} chunks) — skipped", + video_id=asset.video_id, + ) + logger.info( + f"[Ingest] Job {job_id} — video {asset.video_id} already indexed, skipping" + ) + return + update_job( "transcribing", "Transcribing audio with Whisper...", @@ -262,8 +276,11 @@ def update_job( video.status = "error" video.error_message = str(e) db.commit() - except Exception: - pass + except Exception as cleanup_error: + db.rollback() + logger.warning( + f"[Ingest] Job {job_id} — failed to mark video as error: {cleanup_error}" + ) finally: db.close() diff --git a/api/routes/query.py b/api/routes/query.py index ae366a1..1183f02 100644 --- a/api/routes/query.py +++ b/api/routes/query.py @@ -22,7 +22,13 @@ router = APIRouter(tags=["RAG"]) -@router.post("/query", response_model=QueryResponse) +@router.post( + "/query", + response_model=QueryResponse, + responses={ + 503: {"description": "LLM service unavailable (e.g. Ollama not running)"}, + }, +) def query_videos(request: QueryRequest) -> QueryResponse: """ Ask a question and get an answer grounded in indexed video content. @@ -31,11 +37,22 @@ def query_videos(request: QueryRequest) -> QueryResponse: logger.info(f"[API/query] '{request.question[:60]}'") chain = QAChain() - result = chain.ask( - question=request.question, - video_id=request.video_id, - top_k=request.top_k, - ) + try: + result = chain.ask( + question=request.question, + video_id=request.video_id, + top_k=request.top_k, + ) + except Exception as exc: + logger.error(f"[API/query] LLM invocation failed: {exc}") + raise HTTPException( + status_code=503, + detail=( + f"Cannot reach the LLM backend: {exc}. " + "Check that Ollama is running (`ollama serve`) " + "or verify your LLM_PROVIDER and API key configuration." + ), + ) from exc sources = [SourceCitation.model_validate(s) for s in result.source_citations] @@ -49,7 +66,10 @@ def query_videos(request: QueryRequest) -> QueryResponse: @router.post( "/summarize", response_model=SummarizeResponse, - responses={404: {"description": "Video not found"}}, + responses={ + 404: {"description": "Video not found"}, + 503: {"description": "LLM service unavailable (e.g. Ollama not running)"}, + }, ) def summarize_video( request: SummarizeRequest, db: Annotated[Session, Depends(get_db)] @@ -67,11 +87,24 @@ def summarize_video( ) summarizer = Summarizer() - result = summarizer.summarize( - video_id=request.video_id, - title=str(video.title), - include_chapters=request.include_chapters, - ) + try: + result = summarizer.summarize( + video_id=request.video_id, + title=str(video.title), + include_chapters=request.include_chapters, + ) + except Exception as exc: + # Catch connection errors (e.g. Ollama not running), + # timeouts, and any other LLM invocation failures. + logger.error(f"[API/summarize] LLM invocation failed: {exc}") + raise HTTPException( + status_code=503, + detail=( + f"Cannot reach the LLM backend: {exc}. " + "Check that Ollama is running (`ollama serve`) " + "or verify your LLM_PROVIDER and API key configuration." + ), + ) from exc chapter_summaries = None if result.chapter_summaries: @@ -86,4 +119,4 @@ def summarize_video( overall_summary=result.overall_summary, chapter_summaries=chapter_summaries, chunk_count=result.chunk_count, - ) + ) \ No newline at end of file diff --git a/config.py b/config.py index 9c88416..26f6d7b 100644 --- a/config.py +++ b/config.py @@ -43,7 +43,7 @@ class Settings(BaseSettings): # ── Whisper transcription ───────────────────────────────────────────── whisper_mode: WhisperMode = WhisperMode.LOCAL # Local model size: tiny | base | small | medium | large | large-v2 - whisper_model_size: str = "large-v2" + whisper_model_size: str = "base" # ── Embeddings ──────────────────────────────────────────────────────── embedding_mode: EmbeddingMode = EmbeddingMode.LOCAL @@ -76,13 +76,27 @@ class Settings(BaseSettings): retrieval_top_k: int = 5 chunk_duration_seconds: int = 60 chunk_overlap_seconds: int = 15 - use_hyde: bool = False # Enable/disable HyDE query transformation + use_hyde: bool = False # HyDE query transformation (adds 1 LLM call per query) + + # ── Re-ranker ───────────────────────────────────────────────────────── + use_reranker: bool = True + reranker_model: str = "cross-encoder/ms-marco-MiniLM-L-6-v2" + reranker_top_k: int = 10 # Fetch this many from vector store before reranking + + # ── MMR diversity ───────────────────────────────────────────────────── + use_mmr: bool = True + mmr_lambda: float = 0.5 # 0 = only diversity, 1 = only relevance + + # ── Multi-query expansion ───────────────────────────────────────────── + use_multi_query: bool = False # adds 1 LLM call + N extra searches per query + multi_query_count: int = 3 # ── Live stream ─────────────────────────────────────────────────────── live_stream_segment_seconds: int = 60 # capture window per segment # ── UI ──────────────────────────────────────────────────────────────── ui_api_base_url: str = "http://localhost:8000" + ui_request_timeout: int = 120 class Config: env_file = ".env" diff --git a/processing/embedder.py b/processing/embedder.py index 0a645c1..15bc9cb 100644 --- a/processing/embedder.py +++ b/processing/embedder.py @@ -84,6 +84,24 @@ def embed_query(self, query: str) -> list[float]: else: return self._embed_openai([query])[0] + def embed_query_batch(self, texts: list[str]) -> list[list[float]]: + """ + Embed a batch of query strings for retrieval. + Useful for MMR diversity scoring and multi-query expansion. + + Args: + texts: List of text strings to embed + + Returns: + List of float vectors, one per text (same order) + """ + if not texts: + return [] + if self.mode == EmbeddingMode.LOCAL: + return self._embed_local(texts) + else: + return self._embed_openai(texts) + # ── Local (sentence-transformers) ──────────────────────────────────────── def _embed_local(self, texts: list[str]) -> list[list[float]]: diff --git a/rag/__init__.py b/rag/__init__.py index 659f5be..972cf68 100644 --- a/rag/__init__.py +++ b/rag/__init__.py @@ -1,6 +1,14 @@ +from .pipeline import RAGPipeline from .qa_chain import QAChain from .retriever import Retriever, get_retriever from .search import SearchEngine from .summarizer import Summarizer -__all__ = ["QAChain", "Retriever", "SearchEngine", "Summarizer", "get_retriever"] +__all__ = [ + "QAChain", + "RAGPipeline", + "Retriever", + "SearchEngine", + "Summarizer", + "get_retriever", +] diff --git a/rag/pipeline.py b/rag/pipeline.py new file mode 100644 index 0000000..6f630ad --- /dev/null +++ b/rag/pipeline.py @@ -0,0 +1,289 @@ +""" +RAG pipeline orchestrator — chains together multi-query expansion, +cross-encoder reranking, and MMR diversity for high-quality retrieval. + +This sits between the Retriever and QAChain to provide: + 1. Multi-query expansion: generate N rephrasings of the question + 2. Cross-encoder reranking: re-rank vector store results with a more accurate model + 3. MMR (Maximal Marginal Relevance): ensure diversity among top results +""" + +from __future__ import annotations + +from typing import cast + +from langchain_core.messages import HumanMessage, SystemMessage +from loguru import logger + +from config import settings +from llm.factory import get_llm +from processing.embedder import Embedder +from vector_store import SearchResult, get_vector_store + +# numpy is required for MMR — check availability early +try: + import numpy as np +except ImportError: + raise ImportError( + "numpy is required for MMR diversity. Install it with: pip install numpy" + ) + +# ── Multi-query expansion prompts ───────────────────────────────────────────── + +MULTI_QUERY_SYSTEM = """\ +You are a helpful assistant that rephrases user questions to improve \ +semantic search over video transcripts. Generate {count} different versions \ +of the given question. Each version should: +- Use different wording and sentence structure +- Cover different angles or aspects of the original question +- Be self-contained and clearly answerable +- Return ONLY the rephrased questions, one per line, numbered 1-{count}""" + +MULTI_QUERY_USER = """\ +Original question: {question} + +Generate {count} rephrased versions of this question for video transcript search:""" + +# ── Cross-encoder reranker ──────────────────────────────────────────────────── + +_RERANKER_INSTANCE = None + + +def _get_reranker(): + """Lazy-load the cross-encoder reranker model.""" + global _RERANKER_INSTANCE + if _RERANKER_INSTANCE is None: + from sentence_transformers import CrossEncoder + + model_name = settings.reranker_model + logger.info(f"[Reranker] Loading model: {model_name}") + _RERANKER_INSTANCE = CrossEncoder(model_name) + return _RERANKER_INSTANCE + + +# ── MMR (Maximal Marginal Relevance) ────────────────────────────────────────── + + +def _mmr_select( + results: list[SearchResult], + query_embedding: list[float], + top_k: int, + lambda_param: float = 0.5, +) -> list[SearchResult]: + """ + Select diverse results using Maximal Marginal Relevance. + + Args: + results: Candidate results (already scored by reranker or vector store) + query_embedding: The original query embedding vector + top_k: Number of results to return + lambda_param: 0 = only diversity, 1 = only relevance + + Returns: + Top-k results with MMR diversity applied + """ + if len(results) <= top_k: + return results + + query_vec = np.array(query_embedding, dtype=np.float32) + # Normalize query vector + query_norm = np.linalg.norm(query_vec) + if query_norm > 0: + query_vec = query_vec / query_norm + + # Build candidate embeddings from chunk texts (re-embed for diversity scoring) + embedder = Embedder() + candidate_texts = [r.chunk.text for r in results] + candidate_embs = np.array( + embedder.embed_query_batch(candidate_texts), dtype=np.float32 + ) + + # Normalize candidate embeddings + norms = np.linalg.norm(candidate_embs, axis=1, keepdims=True) + norms[norms == 0] = 1.0 + candidate_embs = candidate_embs / norms + + n = len(results) + selected_indices: list[int] = [] + remaining_indices = list(range(n)) + + # Relevance scores from the reranker (or vector store similarity) + relevance = np.array([r.score for r in results], dtype=np.float32) + + for _ in range(min(top_k, n)): + if not remaining_indices: + break + + best_score = -1.0 + best_idx = -1 + + for idx in remaining_indices: + # Relevance component + rel_score = relevance[idx] + + # Diversity component: max similarity to already selected + if selected_indices: + sim_to_selected = max( + float(np.dot(candidate_embs[idx], candidate_embs[s])) + for s in selected_indices + ) + else: + sim_to_selected = 0.0 + + # MMR score + mmr_score = lambda_param * rel_score - (1 - lambda_param) * sim_to_selected + + if mmr_score > best_score: + best_score = mmr_score + best_idx = idx + + if best_idx >= 0: + selected_indices.append(best_idx) + remaining_indices.remove(best_idx) + + return [results[i] for i in selected_indices] + + +# ── Pipeline orchestrator ───────────────────────────────────────────────────── + + +class RAGPipeline: + """ + Orchestrates retrieval with multi-query expansion, cross-encoder reranking, + and MMR diversity. + """ + + def __init__(self) -> None: + self.store = get_vector_store() + self.embedder = Embedder() + + def retrieve( + self, + query: str, + top_k: int | None = None, + video_id: str | None = None, + ) -> list[SearchResult]: + """ + Full retrieval pipeline: multi-query → vector search → rerank → MMR. + + Args: + query: Natural language question + top_k: Number of final results to return + video_id: Optional video filter + + Returns: + List of SearchResult sorted by relevance (highest first) + """ + top_k = top_k or settings.retrieval_top_k + logger.info(f"[RAGPipeline] Query='{query[:80]}', top_k={top_k}") + + # ── Step 1: Generate multiple query variants ────────────────────────── + queries = [query] + if settings.use_multi_query: + try: + extra_queries = self._generate_multi_queries(query) + queries.extend(extra_queries) + logger.debug(f"[RAGPipeline] Multi-query: {len(queries)} variants") + except Exception as e: + logger.warning( + f"[RAGPipeline] Multi-query failed, using original only: {e}" + ) + + # ── Step 2: Retrieve for each query variant ─────────────────────────── + fetch_k = settings.reranker_top_k if settings.use_reranker else top_k + all_results: dict[str, SearchResult] = {} + + for q in queries: + query_vector = self.embedder.embed_query(q) + results = self.store.search( + query_vector, top_k=fetch_k, filter_video_id=video_id + ) + for r in results: + # Deduplicate by chunk_id, keep highest score + if ( + r.chunk.chunk_id not in all_results + or r.score > all_results[r.chunk.chunk_id].score + ): + all_results[r.chunk.chunk_id] = r + + merged = sorted(all_results.values(), key=lambda r: r.score, reverse=True) + logger.debug(f"[RAGPipeline] {len(merged)} unique candidates after merge") + + if not merged: + return [] + + # ── Step 3: Cross-encoder reranking ─────────────────────────────────── + if settings.use_reranker: + try: + merged = self._rerank(query, merged) + logger.debug(f"[RAGPipeline] Reranked {len(merged)} results") + except Exception as e: + logger.warning( + f"[RAGPipeline] Reranker failed, using vector store scores: {e}" + ) + + # ── Step 4: MMR diversity ───────────────────────────────────────────── + if settings.use_mmr: + try: + query_vector = self.embedder.embed_query(query) + merged = _mmr_select( + merged, + query_embedding=query_vector, + top_k=top_k, + lambda_param=settings.mmr_lambda, + ) + logger.debug(f"[RAGPipeline] MMR selected {len(merged)} results") + except Exception as e: + logger.warning(f"[RAGPipeline] MMR failed, using reranked results: {e}") + + # ── Step 5: Trim to final top_k ─────────────────────────────────────── + final = merged[:top_k] + logger.success(f"[RAGPipeline] Returning {len(final)} results") + return final + + # ── Private helpers ──────────────────────────────────────────────────────── + + def _generate_multi_queries(self, query: str) -> list[str]: + """Generate alternative phrasings of the query for better recall.""" + llm = get_llm(temperature=0.3) + count = settings.multi_query_count + prompt = MULTI_QUERY_USER.format(question=query, count=count) + messages = [ + SystemMessage(content=MULTI_QUERY_SYSTEM.format(count=count)), + HumanMessage(content=prompt), + ] + response = llm.invoke(messages) + content = cast(str, response.content) + + # Parse numbered lines: "1. ..." or "1) ..." or "- ..." + queries: list[str] = [] + for line in content.strip().split("\n"): + line = line.strip() + # Remove leading number/prefix + for prefix in [f"{i}." for i in range(1, count + 1)]: + if line.startswith(prefix): + line = line[len(prefix) :].strip() + break + if line and len(line) > 5: + queries.append(line) + + return queries[:count] + + def _rerank(self, query: str, results: list[SearchResult]) -> list[SearchResult]: + """ + Re-rank results using a cross-encoder model. + Cross-encoders are much more accurate than bi-encoder cosine similarity. + """ + reranker = _get_reranker() + pairs = [(query, r.chunk.text) for r in results] + scores = reranker.predict(pairs) + + # Attach new scores and sort + for r, score in zip(results, scores, strict=False): + # Cross-encoder returns logits; convert to 0-1 score via sigmoid + import math + + r.score = 1.0 / (1.0 + math.exp(-float(score))) + + results.sort(key=lambda r: r.score, reverse=True) + return results diff --git a/rag/qa_chain.py b/rag/qa_chain.py index 5808a55..fcfc07b 100644 --- a/rag/qa_chain.py +++ b/rag/qa_chain.py @@ -1,6 +1,8 @@ """ Q&A chain — answers questions grounded in retrieved video transcript chunks. Returns the answer text plus citations (video title + timestamp for each source). + +Uses the RAGPipeline (multi-query + reranker + MMR) for high-quality retrieval. """ from dataclasses import dataclass @@ -12,19 +14,27 @@ from llm.factory import get_llm from vector_store import SearchResult -from .retriever import get_retriever +from .pipeline import RAGPipeline SYSTEM_PROMPT = """\ -You are a helpful assistant that answers questions based on video transcript content. +You are a helpful, precise assistant that answers questions based on video transcript content. You are given relevant excerpts from one or more video transcripts, each labeled with \ a source video title and timestamp range. -Rules: -- Answer only based on the provided transcript excerpts. -- If the answer is not found in the excerpts, say so clearly. -- Be concise and direct. -- Reference source numbers (e.g. [1], [2]) when citing specific content. -- Do not invent information beyond what is in the transcripts. +## Rules: +1. Answer ONLY based on the provided transcript excerpts. Do NOT use outside knowledge. +2. If the answer is not fully found in the excerpts, say so clearly and explain what IS available. +3. Be concise but thorough. Cover all relevant information from the transcripts. +4. Cite sources using bracketed numbers: [1], [2], etc. At the end of your answer, \ +list the source references with their titles and timestamps. +5. If the question asks for a list, use bullet points. If it asks for an explanation, \ +use clear paragraphs. +6. If the question is ambiguous, acknowledge the ambiguity and answer the most likely interpretation. +7. Do NOT invent information beyond what is in the transcripts. + +## Output format: +- Answer: [your answer here with [1], [2] citations] +- Sources: [1] "Video Title" @ 00:01:23-00:02:45 """ QA_PROMPT_TEMPLATE = """\ @@ -35,8 +45,7 @@ Question: {question} -Please answer based on the transcript excerpts above. If citing a source, reference its number (e.g. [1]). -""" +Please answer based on the transcript excerpts above. Cite sources with [1], [2], etc.""" @dataclass @@ -68,11 +77,11 @@ def source_citations(self) -> list[dict[str, object]]: class QAChain: """ Retrieval-augmented Q&A over indexed video content. - Retrieves relevant chunks, builds a prompt, and calls the LLM. + Uses RAGPipeline (multi-query + reranker + MMR) for retrieval and calls the LLM. """ def __init__(self) -> None: - self.retriever = get_retriever() + self.pipeline = RAGPipeline() def ask( self, @@ -94,7 +103,7 @@ def ask( logger.info(f"[QA] Question: '{question[:80]}'") # ── Retrieve relevant chunks ────────────────────────────────────────── - results = self.retriever.retrieve(question, top_k=top_k, video_id=video_id) + results = self.pipeline.retrieve(question, top_k=top_k, video_id=video_id) if not results: return QAResult( @@ -105,7 +114,7 @@ def ask( ) # ── Build context and prompt ────────────────────────────────────────── - context = self.retriever.format_context(results) + context = self._format_context(results) user_message = QA_PROMPT_TEMPLATE.format(context=context, question=question) # ── Call LLM ────────────────────────────────────────────────────────── @@ -126,3 +135,16 @@ def ask( sources=results, video_id=video_id, ) + + @staticmethod + def _format_context(results: list[SearchResult]) -> str: + """ + Format retrieved chunks into a context block for the LLM. + Each chunk is prefixed with its source video title and timestamp. + """ + parts = [] + for i, r in enumerate(results, 1): + parts.append( + f'[{i}] "{r.chunk.title}" @ {r.chunk.start_ts}-{r.chunk.end_ts}\n{r.chunk.text}' + ) + return "\n\n---\n\n".join(parts) \ No newline at end of file diff --git a/rag/retriever.py b/rag/retriever.py index be9ed76..679ab1e 100644 --- a/rag/retriever.py +++ b/rag/retriever.py @@ -13,13 +13,17 @@ _retriever_instance = None -HYDE_SYSTEM_PROMPT = "You are an expert video transcript generator." +HYDE_SYSTEM_PROMPT = """\ +You are an expert at generating realistic video transcript excerpts. \ +Given a question, write a single dense, factual paragraph that would likely appear \ +in a video transcript answering that question. Use first-person ("I", "we") as if \ +spoken by a presenter. Include specific keywords, technical terms, names, and numbers \ +that would help a semantic search engine find this content. Do NOT include meta-talk \ +like "Here is a transcript excerpt". Just provide the transcript content itself.""" HYDE_USER_PROMPT = """\ -Given a question, write a single paragraph that would likely appear in a video transcript answering that question. -The paragraph should be in the first person ("I", "we") as if spoken by a presenter. -Focus on being factual and informative. Do not include any meta-talk like "Here is a transcript excerpt". -Just provide the content of the transcript itself. +Generate a realistic video transcript excerpt that answers the following question. \ +Make it dense with relevant keywords and terminology. Question: {query} diff --git a/rag/search.py b/rag/search.py index 9f8ba88..4940c99 100644 --- a/rag/search.py +++ b/rag/search.py @@ -9,7 +9,7 @@ from vector_store import SearchResult -from .retriever import get_retriever +from .pipeline import RAGPipeline @dataclass @@ -28,7 +28,7 @@ class SearchEngine: """ def __init__(self) -> None: - self.retriever = get_retriever() + self.pipeline = RAGPipeline() def search( self, @@ -51,7 +51,7 @@ def search( """ logger.info(f"[Search] Query='{query[:60]}', top_k={top_k}") - raw_results = self.retriever.retrieve(query, top_k=top_k, video_id=video_id) + raw_results = self.pipeline.retrieve(query, top_k=top_k, video_id=video_id) # Apply score threshold filtered = [r for r in raw_results if r.score >= min_score]