From 5c2284922da8f4ec078cdfdb82e2bd39716445a4 Mon Sep 17 00:00:00 2001 From: Jeroen Schweitzer Date: Tue, 14 Jul 2026 10:01:04 +0200 Subject: [PATCH] feat: add source_status and degraded fields to HybridRAG response Each retrieval leg (vector, graph, web, volatile, documents) now returns (results, timing, error) instead of swallowing exceptions to an empty list. The response reports per-leg status ('ok'/'failed'/'disabled') in source_status and sets degraded=true when any enabled leg failed. Failed legs still contribute no results (behavior unchanged) and are logged at WARNING. Both fields are additive with defaults, so deploy order relative to consumers does not matter. Co-Authored-By: Claude Fable 5 --- CHANGELOG.md | 4 ++ src/models/hybrid_rag.py | 5 ++ src/services/hybrid_rag_service.py | 76 +++++++++++++++++++++--------- 3 files changed, 63 insertions(+), 22 deletions(-) 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', []))}, "