Add Pipeline visualization

src/agent/router.py

@@ -5,7 +5,7 @@ import math
 
 from langchain_core.runnables import Runnable
 
-from src.models import IntentType, GenerationResponse
+from src.models import IntentType, GenerationResponse, PipelineDetails
 from src.agent.intent_classifier import IntentClassifier
 from src.retrieval.hybrid import HybridRetriever
 from src.retrieval.reranker import Reranker
@@ -104,6 +104,7 @@ class QueryRouter:
 
         # Detect language and translate to Danish for retrieval if needed
         retrieval_query, user_language = self._detect_and_translate_query(query)
+        translated = retrieval_query != query
 
         intent = self._intent_classifier.classify(query)
         logger.info("Classified intent: %s", intent.value)
@@ -112,11 +113,28 @@ class QueryRouter:
         should_retrieve = intent != IntentType.UNKNOWN
         logger.debug("[DEBUG] Retrieval executed: %s (intent=%s)", should_retrieve, intent.value)
 
-        results = self._hybrid_retriever.search(retrieval_query, top_k=top_k) if should_retrieve else []
+        # Use detailed search to capture intermediate results
+        pipeline = PipelineDetails(
+            original_query=query,
+            retrieval_query=retrieval_query,
+            detected_language=user_language,
+            translated=translated,
+        )
+
+        if should_retrieve:
+            hybrid_result = self._hybrid_retriever.search_detailed(retrieval_query, top_k=top_k)
+            pipeline.dense_results = hybrid_result.dense_results
+            pipeline.sparse_results = hybrid_result.sparse_results
+            pipeline.fused_results = hybrid_result.fused_results
+            results = hybrid_result.fused_results
+        else:
+            results = []
+
         logger.info("Retrieved %d results from hybrid search", len(results))
         logger.debug("[DEBUG] Retrieval returned %d results", len(results))
 
         reranked = self._reranker.rerank(retrieval_query, results, top_k=top_k) if results else []
+        pipeline.reranked_results = reranked
         logger.info("Reranked to %d results", len(reranked))
 
         if reranked and intent == IntentType.FACTUAL:
@@ -143,6 +161,7 @@ class QueryRouter:
             sources=reranked,
             intent=intent,
             confidence=confidence,
+            pipeline_details=pipeline,
         )
 
     def _build_prompt(

src/api/routes.py

@@ -62,6 +62,28 @@ class QueryRequest(BaseModel):
     strategy: str = "recursive"
 
 
+class PipelineResultItem(BaseModel):
+    """A single result item in pipeline details."""
+
+    document_id: str
+    chunk_id: str
+    score: float
+    source: str
+
+
+class PipelineDetailsResponse(BaseModel):
+    """Intermediate pipeline data for the query response."""
+
+    original_query: str = ""
+    retrieval_query: str = ""
+    detected_language: str = ""
+    translated: bool = False
+    dense_results: list[PipelineResultItem] = []
+    sparse_results: list[PipelineResultItem] = []
+    fused_results: list[PipelineResultItem] = []
+    reranked_results: list[PipelineResultItem] = []
+
+
 class QueryResponse(BaseModel):
     """Response body for the query endpoint."""
 
@@ -69,6 +91,7 @@ class QueryResponse(BaseModel):
     sources: list[dict[str, str | float]]
     intent: str
     confidence: float
+    pipeline_details: PipelineDetailsResponse = PipelineDetailsResponse()
 
 
 class IngestRequest(BaseModel):
@@ -154,11 +177,35 @@ async def query_documents(request: QueryRequest) -> QueryResponse:
         for result in response.sources
     ]
 
+    def _to_pipeline_items(results: list) -> list[PipelineResultItem]:
+        return [
+            PipelineResultItem(
+                document_id=r.chunk.document_id,
+                chunk_id=r.chunk.chunk_id,
+                score=r.score,
+                source=r.source,
+            )
+            for r in results
+        ]
+
+    pd = response.pipeline_details
+    pipeline_details = PipelineDetailsResponse(
+        original_query=pd.original_query,
+        retrieval_query=pd.retrieval_query,
+        detected_language=pd.detected_language,
+        translated=pd.translated,
+        dense_results=_to_pipeline_items(pd.dense_results),
+        sparse_results=_to_pipeline_items(pd.sparse_results),
+        fused_results=_to_pipeline_items(pd.fused_results),
+        reranked_results=_to_pipeline_items(pd.reranked_results),
+    )
+
     return QueryResponse(
         answer=response.answer,
         sources=sources,
         intent=response.intent.value,
         confidence=response.confidence,
+        pipeline_details=pipeline_details,
     )
 
 

src/models.py

@@ -57,6 +57,31 @@ class QueryResult:
     source: str
 
 
+@dataclass
+class PipelineDetails:
+    """Intermediate pipeline data for debugging and transparency.
+
+    Attributes:
+        original_query: The user's original query text.
+        retrieval_query: The query used for retrieval (may be translated).
+        detected_language: Detected language of the original query.
+        translated: Whether the query was translated for retrieval.
+        dense_results: Results from dense (vector) retrieval.
+        sparse_results: Results from sparse (BM25) retrieval.
+        fused_results: Results after reciprocal rank fusion.
+        reranked_results: Results after cross-encoder reranking.
+    """
+
+    original_query: str = ""
+    retrieval_query: str = ""
+    detected_language: str = ""
+    translated: bool = False
+    dense_results: list[QueryResult] = field(default_factory=list)
+    sparse_results: list[QueryResult] = field(default_factory=list)
+    fused_results: list[QueryResult] = field(default_factory=list)
+    reranked_results: list[QueryResult] = field(default_factory=list)
+
+
 @dataclass
 class GenerationResponse:
     """Structured response from the generation pipeline.
@@ -66,9 +91,11 @@ class GenerationResponse:
         sources: List of source chunks used for generation.
         intent: Classified intent of the original query.
         confidence: Model confidence in the answer (0.0-1.0).
+        pipeline_details: Optional intermediate pipeline data.
     """
 
     answer: str
     sources: list[QueryResult]
     intent: IntentType
-    confidence: float
+    confidence: float
+    pipeline_details: PipelineDetails = field(default_factory=PipelineDetails)

src/retrieval/hybrid.py

@@ -1,6 +1,7 @@
 """Hybrid search combining dense and sparse retrieval with reciprocal rank fusion."""
 
 import logging
+from dataclasses import dataclass
 
 from src.models import QueryResult
 from src.retrieval.bm25_search import BM25Search
@@ -10,6 +11,21 @@ from src.retrieval.vector_store import VectorStore
 logger = logging.getLogger(__name__)
 
 
+@dataclass
+class HybridSearchResult:
+    """Container for hybrid search results including intermediate stages.
+
+    Attributes:
+        dense_results: Results from dense (vector) retrieval.
+        sparse_results: Results from sparse (BM25) retrieval.
+        fused_results: Results after reciprocal rank fusion.
+    """
+
+    dense_results: list[QueryResult]
+    sparse_results: list[QueryResult]
+    fused_results: list[QueryResult]
+
+
 class HybridRetriever:
     """Combines dense (vector) and sparse (BM25) retrieval using rank fusion."""
 
@@ -46,6 +62,19 @@ class HybridRetriever:
         Returns:
             List of QueryResult objects sorted by fused score.
         """
+        result = self.search_detailed(query, top_k)
+        return result.fused_results
+
+    def search_detailed(self, query: str, top_k: int) -> HybridSearchResult:
+        """Execute hybrid search and return all intermediate results.
+
+        Args:
+            query: The search query string.
+            top_k: Number of top results to return after fusion.
+
+        Returns:
+            HybridSearchResult containing dense, sparse, and fused results.
+        """
         query_embedding = self._embedder.embed_text(query)
         dense_results = self._vector_store.search(query_embedding, top_k)
         sparse_results = self._bm25_search.search(query, top_k)
@@ -57,7 +86,11 @@ class HybridRetriever:
         )
 
         fused = self.reciprocal_rank_fusion(dense_results, sparse_results, k=60)
-        return fused[:top_k]
+        return HybridSearchResult(
+            dense_results=dense_results,
+            sparse_results=sparse_results,
+            fused_results=fused[:top_k],
+        )
 
     def reciprocal_rank_fusion(
         self,

src/ui/app.py

@@ -70,6 +70,21 @@ TEXTS: Dict[str, Dict[str, str]] = {
         "model_llm": "LLM",
         "model_embedding": "Embedding",
         "model_unavailable": "Kunne ikke hente modelinfo.",
+        "pipeline_heading": "Pipeline-detaljer",
+        "pipeline_translation": "Oversaettelse",
+        "pipeline_original": "Original foresporgsel",
+        "pipeline_translated": "Oversat til dansk",
+        "pipeline_lang": "Sprog registreret",
+        "pipeline_no_translation": "Ingen oversaettelse (foresporgsel allerede paa dansk)",
+        "pipeline_bm25": "BM25-resultater (leksikalsk soegning)",
+        "pipeline_dense": "Vektorsoegning (semantisk)",
+        "pipeline_fused": "RRF-fusioneret raekkefoelge",
+        "pipeline_reranked": "Reranking (endelig raekkefoelge)",
+        "pipeline_doc": "Dokument",
+        "pipeline_score": "Score",
+        "pipeline_rank": "#",
+        "pipeline_no_results": "Ingen resultater",
+        "pipeline_score_change": "Score-aendring",
     },
     "en": {
         "page_title": "Document Assistant",
@@ -125,6 +140,21 @@ TEXTS: Dict[str, Dict[str, str]] = {
         "model_llm": "LLM",
         "model_embedding": "Embedding",
         "model_unavailable": "Could not fetch model info.",
+        "pipeline_heading": "Pipeline Details",
+        "pipeline_translation": "Query Translation",
+        "pipeline_original": "Original query",
+        "pipeline_translated": "Translated to Danish",
+        "pipeline_lang": "Detected language",
+        "pipeline_no_translation": "No translation (query already in Danish)",
+        "pipeline_bm25": "BM25 Results (lexical search)",
+        "pipeline_dense": "Vector Search (semantic)",
+        "pipeline_fused": "RRF Fused Ranking",
+        "pipeline_reranked": "Reranked (final ranking)",
+        "pipeline_doc": "Document",
+        "pipeline_score": "Score",
+        "pipeline_rank": "#",
+        "pipeline_no_results": "No results",
+        "pipeline_score_change": "Score change",
     },
 }
 
@@ -449,5 +479,80 @@ if search_clicked and question.strip():
     else:
         st.info(t["no_sources"])
 
+    # -- Pipeline Details --
+    pd = data.get("pipeline_details", {})
+    if pd:
+        with st.expander(t["pipeline_heading"], expanded=False):
+            # 1) Query translation
+            st.markdown(f'**{t["pipeline_translation"]}**')
+            if pd.get("translated"):
+                st.markdown(
+                    f'- {t["pipeline_lang"]}: **{pd.get("detected_language", "")}**\n'
+                    f'- {t["pipeline_original"]}: {pd.get("original_query", "")}\n'
+                    f'- {t["pipeline_translated"]}: {pd.get("retrieval_query", "")}'
+                )
+            else:
+                st.markdown(f'_{t["pipeline_no_translation"]}_')
+
+            st.markdown("---")
+
+            def _render_result_table(results: list[dict], label: str) -> None:
+                """Render a ranked results table."""
+                st.markdown(f"**{label}**")
+                if not results:
+                    st.caption(t["pipeline_no_results"])
+                    return
+                header = f'| {t["pipeline_rank"]} | {t["pipeline_doc"]} | {t["pipeline_score"]} |\n|---|---|---|'
+                rows = "\n".join(
+                    f'| {i + 1} | {r.get("document_id", "")} | {r.get("score", 0):.4f} |'
+                    for i, r in enumerate(results)
+                )
+                st.markdown(f"{header}\n{rows}")
+
+            # 2) BM25 results
+            _render_result_table(pd.get("sparse_results", []), t["pipeline_bm25"])
+
+            st.markdown("---")
+
+            # 3) Vector search results
+            _render_result_table(pd.get("dense_results", []), t["pipeline_dense"])
+
+            st.markdown("---")
+
+            # 4) RRF fused ranking
+            _render_result_table(pd.get("fused_results", []), t["pipeline_fused"])
+
+            st.markdown("---")
+
+            # 5) Reranked results with score change
+            reranked = pd.get("reranked_results", [])
+            st.markdown(f'**{t["pipeline_reranked"]}**')
+            if reranked:
+                # Build a map from chunk_id -> fused score for comparison
+                fused_scores: dict[str, float] = {
+                    r.get("chunk_id", ""): r.get("score", 0.0)
+                    for r in pd.get("fused_results", [])
+                }
+                header = (
+                    f'| {t["pipeline_rank"]} | {t["pipeline_doc"]} | '
+                    f'{t["pipeline_score"]} | {t["pipeline_score_change"]} |\n'
+                    f"|---|---|---|---|"
+                )
+                rows_list = []
+                for i, r in enumerate(reranked):
+                    cid = r.get("chunk_id", "")
+                    new_score = r.get("score", 0.0)
+                    old_score = fused_scores.get(cid)
+                    if old_score is not None:
+                        change = f"RRF {old_score:.4f} -> {new_score:.4f}"
+                    else:
+                        change = "-"
+                    rows_list.append(
+                        f'| {i + 1} | {r.get("document_id", "")} | {new_score:.4f} | {change} |'
+                    )
+                st.markdown(f"{header}\n" + "\n".join(rows_list))
+            else:
+                st.caption(t["pipeline_no_results"])
+
 elif search_clicked:
     st.warning(t["empty_warning"])