diff --git a/CHANGELOG.md b/CHANGELOG.md index b4b4a4b..c15e98a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- **Degradation signaling** - `HybridRAGResponse` now includes `source_status` (per-leg `'ok'`/`'failed'`/`'disabled'` for vector, graph, web, volatile, documents) and `degraded` (true when any enabled leg failed). Retrieval legs report errors instead of silently swallowing them; failed legs are logged at WARNING. Both fields are additive and optional, so clients that ignore them are unaffected. + ### Fixed - **Ollama generation model env collision** - Renamed the generation-model setting `ollama_model` to `ollama_llm_model` (env: `OLLAMA_LLM_MODEL`, default `gemma4:e2b`). The container env `OLLAMA_MODEL=nomic-embed-text` (meant for embeddings) was shadowing the generation model, breaking Phase 0 keyword extraction and Phase 4 LLM re-ranking on every request. Startup now logs the resolved generation model. diff --git a/src/models/hybrid_rag.py b/src/models/hybrid_rag.py index 32a2596..80a282f 100644 --- a/src/models/hybrid_rag.py +++ b/src/models/hybrid_rag.py @@ -88,6 +88,11 @@ class HybridRAGResponse(BaseModel): timing: TimingBreakdown = Field(..., description="Performance breakdown") config_used: HybridRAGConfig = Field(..., description="Configuration used") search_id: Optional[str] = Field(None, description="Search ID for Librarian tracking") + source_status: Dict[str, str] = Field( + default={}, + description="Per-leg retrieval status ('ok', 'failed', or 'disabled') keyed by: vector, graph, web, volatile, documents" + ) + degraded: bool = Field(default=False, description="True when any enabled retrieval leg reported 'failed'") class HybridRAGRequest(BaseModel): diff --git a/src/services/hybrid_rag_service.py b/src/services/hybrid_rag_service.py index 074d4fb..7ef48e2 100644 --- a/src/services/hybrid_rag_service.py +++ b/src/services/hybrid_rag_service.py @@ -44,6 +44,15 @@ class HybridRAGService: Service for HybridRAG multi-source search with fusion and re-ranking. """ + # Maps internal retrieval leg names to source_status keys in the response + SOURCE_STATUS_KEYS = { + "vector": "vector", + "graph": "graph", + "web": "web", + "volatile": "volatile", + "document": "documents", + } + def __init__( self, vector_service: VectorService, @@ -185,6 +194,14 @@ class HybridRAGService: ) timing["persistence_ms"] = (time.time() - phase6_start) * 1000 + # Degradation signaling: a failed leg contributes no results, but the + # response says so instead of silently pretending the leg was empty + source_status = raw_results.get("source_status", {}) + degraded = any(status == "failed" for status in source_status.values()) + if degraded: + failed_legs = [leg for leg, status in source_status.items() if status == "failed"] + logger.warning(f"HybridRAG search degraded: failed legs: {failed_legs}") + # Build response return HybridRAGResponse( query=query, @@ -195,7 +212,9 @@ class HybridRAGService: total_results=len(result_models), timing=TimingBreakdown(**timing), config_used=config, - search_id=search_id + search_id=search_id, + source_status=source_status, + degraded=degraded ) async def _extract_keywords_and_synonyms(self, query: str) -> Dict[str, Any]: @@ -293,10 +312,14 @@ JSON:""" user: str, config: HybridRAGConfig, keywords_data: Dict[str, Any] - ) -> Dict[str, List]: + ) -> Dict[str, Any]: """ Phase 1: Retrieve results from all sources in parallel. + Each retrieval leg returns (results, timing_ms, error) so that a + failed leg still contributes no results but is reported in + "source_status" instead of being silently swallowed. + Args: query: Search query user: User identifier @@ -304,7 +327,8 @@ JSON:""" keywords_data: Extracted keywords/synonyms Returns: - Dictionary with results from each source and timing + Dictionary with results from each source, timing, and per-leg + "source_status" ('ok', 'failed', or 'disabled') """ tasks = {} timing = {} @@ -331,10 +355,10 @@ JSON:""" } for r in response.results ] - return results, (time.time() - start) * 1000 + return results, (time.time() - start) * 1000, None except Exception as e: - logger.error(f"Vector search failed: {e}", exc_info=True) - return [], (time.time() - start) * 1000 + logger.warning(f"Vector search failed: {e}", exc_info=True) + return [], (time.time() - start) * 1000, e tasks["vector"] = vector_search() @@ -369,10 +393,10 @@ JSON:""" } for r in results ] - return formatted, (time.time() - start) * 1000 + return formatted, (time.time() - start) * 1000, None except Exception as e: - logger.error(f"Graph search failed: {e}", exc_info=True) - return [], (time.time() - start) * 1000 + logger.warning(f"Graph search failed: {e}", exc_info=True) + return [], (time.time() - start) * 1000, e tasks["graph"] = graph_search() @@ -408,10 +432,10 @@ JSON:""" } for r in results ] - return formatted, (time.time() - start) * 1000 + return formatted, (time.time() - start) * 1000, None except Exception as e: - logger.error(f"Web search failed: {e}", exc_info=True) - return [], (time.time() - start) * 1000 + logger.warning(f"Web search failed: {e}", exc_info=True) + return [], (time.time() - start) * 1000, e tasks["web"] = web_search() @@ -439,10 +463,10 @@ JSON:""" } for r in results ] - return formatted, (time.time() - start) * 1000 + return formatted, (time.time() - start) * 1000, None except Exception as e: - logger.error(f"Volatile search failed: {e}", exc_info=True) - return [], (time.time() - start) * 1000 + logger.warning(f"Volatile search failed: {e}", exc_info=True) + return [], (time.time() - start) * 1000, e tasks["volatile"] = volatile_search() @@ -458,7 +482,7 @@ JSON:""" # Check if collection exists exists = await self.vector.qdrant.collection_exists(collection_name) if not exists: - return [], (time.time() - start) * 1000 + return [], (time.time() - start) * 1000, None # Get query embedding query_embedding = await self.vector.ollama.embed_text(query) @@ -496,22 +520,30 @@ JSON:""" "source": "document" }) - return formatted, (time.time() - start) * 1000 + return formatted, (time.time() - start) * 1000, None except Exception as e: - logger.error(f"Document search failed: {e}", exc_info=True) - return [], (time.time() - start) * 1000 + logger.warning(f"Document search failed: {e}", exc_info=True) + return [], (time.time() - start) * 1000, e tasks["document"] = document_search() # Execute all searches in parallel + # No return_exceptions needed: each leg captures its own exception + # and reports it via the (results, timing, error) tuple. results_dict = await asyncio.gather(*tasks.values()) - # Combine results with timing - output = {"timing": {}} + # Combine results with timing and per-leg status + output = {"timing": {}, "source_status": {}} for i, source in enumerate(tasks.keys()): - results, source_timing = results_dict[i] + results, source_timing, error = results_dict[i] output[source] = results output["timing"][f"{source}_ms"] = source_timing + status_key = self.SOURCE_STATUS_KEYS[source] + output["source_status"][status_key] = "failed" if error is not None else "ok" + + # Legs that were not attempted are reported as disabled + for status_key in self.SOURCE_STATUS_KEYS.values(): + output["source_status"].setdefault(status_key, "disabled") logger.info( f"Parallel retrieval: vector={len(output.get('vector', []))}, "