fix: all 15 bugs + full agentic RAG pipeline in HF Space

Critical fixes:
- OpenSearch client param names aligned (query_text/query_vector)
- _execute_search preserves _source as nested dict
- health.py: SQLAlchemy text() wrapper + FAISS import path
- llm_config: module-level get_synthesizer() function
- Redis cache: simplified get(key)/set(key,value) API

High priority:
- Infinite rewrite loop prevented via retrieval_attempts counter
- analyze.py: handles both dict and object results

HF Space full pipeline integration:
- retrieve_node: retriever-agnostic (FAISS/OpenSearch/BM25)
- AgenticContext: retriever field added
- grade_documents_node + generate_answer_node: unified doc format
- app.py: full guardrail->retrieve->grade->rewrite->generate pipeline
- asyncio.get_event_loop() -> get_running_loop()

huggingface/app.py

@@ -582,106 +582,149 @@ def format_summary(response: dict, elapsed: float) -> str:
 
 
 # ---------------------------------------------------------------------------
-# Q&A Chat Functions - Streaming Support
+# Q&A Chat Functions — Full Agentic RAG Pipeline
 # ---------------------------------------------------------------------------
 
+_rag_service = None
+_rag_service_error = None
+
+
+def _get_rag_service():
+    """Lazily initialize the full agentic RAG service for Q&A.
+
+    Uses a FAISS-backed retriever wrapped in an AgenticContext so the
+    guardrail → retrieve → grade → rewrite → generate pipeline runs
+    identically to the production API.
+    """
+    global _rag_service, _rag_service_error
+
+    if _rag_service is not None:
+        return _rag_service
+
+    if _rag_service_error is not None:
+        logger.warning("Previous RAG service init failed, retrying...")
+        _rag_service_error = None
+
+    try:
+        from src.services.agents.agentic_rag import AgenticRAGService
+        from src.services.agents.context import AgenticContext
+        from src.services.retrieval.factory import make_retriever
+        from src.llm_config import get_synthesizer
+
+        llm = get_synthesizer()
+        retriever = make_retriever()  # auto-detects FAISS
+
+        # HF Space: skip OpenSearch, Redis, Langfuse
+        # but still get guardrail, grading, rewriting, generation
+        context = AgenticContext(
+            llm=llm,
+            embedding_service=None,
+            opensearch_client=None,
+            cache=None,
+            tracer=None,
+            retriever=retriever,
+        )
+
+        _rag_service = AgenticRAGService(context)
+        logger.info("Agentic RAG service initialized for Q&A")
+        return _rag_service
+
+    except Exception as exc:
+        logger.error(f"Failed to init agentic RAG service: {exc}")
+        _rag_service_error = exc
+        return None
+
+
+def _fallback_qa(question: str, context_text: str = "") -> str:
+    """Direct retriever+LLM fallback when agentic pipeline is unavailable."""
+    from src.services.retrieval.factory import make_retriever
+    from src.llm_config import get_synthesizer
+
+    retriever = make_retriever()
+    search_query = f"{context_text} {question}" if context_text.strip() else question
+    docs = retriever.retrieve(search_query, top_k=5)
+
+    doc_context = ""
+    if docs:
+        doc_texts = [d.content[:500] for d in docs[:5]]
+        doc_context = "\n\n---\n\n".join(doc_texts)
+
+    llm = get_synthesizer()
+    prompt = f"""You are a medical AI assistant. Answer the following medical question based on the provided context.
+Be helpful, accurate, and include relevant medical information. Always recommend consulting a healthcare professional.
+
+Context from medical knowledge base:
+{doc_context if doc_context else "No specific context available."}
+
+Patient Context: {context_text if context_text else "Not provided"}
+
+Question: {question}
+
+Answer:"""
+    response = llm.invoke(prompt)
+    return response.content if hasattr(response, 'content') else str(response)
+
+
 def answer_medical_question(
-    question: str, 
+    question: str,
     context: str = "",
     chat_history: list = None
 ) -> tuple[str, list]:
-    """
-    Answer a free-form medical question using retriever + LLM directly.
-    
-    Args:
-        question: The user's medical question
-        context: Optional biomarker/patient context
-        chat_history: Previous conversation history
-        
-    Returns:
-        Tuple of (formatted_answer, updated_chat_history)
+    """Answer a medical question using the full agentic RAG pipeline.
+
+    Pipeline: guardrail → retrieve → grade → rewrite → generate.
+    Falls back to direct retriever+LLM if the pipeline is unavailable.
     """
     if not question.strip():
         return "", chat_history or []
-    
-    # Check API key dynamically
+
     groq_key, google_key = get_api_keys()
     if not groq_key and not google_key:
         error_msg = "❌ Please add your GROQ_API_KEY or GOOGLE_API_KEY in Space Settings → Secrets."
         history = (chat_history or []) + [(question, error_msg)]
         return error_msg, history
-    
-    # Setup provider
+
     provider = setup_llm_provider()
     logger.info(f"Q&A using provider: {provider}")
-    
+
     try:
         start_time = time.time()
-        
-        # Import retriever and LLM
-        from src.services.retrieval import make_retriever
-        from src.llm_config import get_synthesizer
-        
-        # Initialize retriever
-        retriever = make_retriever()
-        
-        # Build search query with context
-        search_query = question
-        if context.strip():
-            search_query = f"{context} {question}"
-        
-        # Retrieve relevant documents
-        docs = retriever.search(search_query, top_k=5)
-        
-        # Format context from retrieved docs
-        doc_context = ""
-        if docs:
-            doc_texts = []
-            for doc in docs[:5]:
-                if hasattr(doc, 'content'):
-                    doc_texts.append(doc.content[:500])
-                elif isinstance(doc, dict) and 'content' in doc:
-                    doc_texts.append(doc['content'][:500])
-            doc_context = "\n\n---\n\n".join(doc_texts)
-        
-        # Get LLM
-        llm = get_synthesizer()
-        
-        # Build prompt
-        prompt = f"""You are a medical AI assistant. Answer the following medical question based on the provided context.
-Be helpful, accurate, and include relevant medical information. Always recommend consulting a healthcare professional for personal medical advice.
-
-Context from medical knowledge base:
-{doc_context if doc_context else "No specific context available - using general medical knowledge."}
-
-Patient Context: {context if context else "Not provided"}
 
-Question: {question}
+        rag_service = _get_rag_service()
+        if rag_service is not None:
+            result = rag_service.ask(query=question, patient_context=context)
+            answer = result.get("final_answer", "")
+            guardrail = result.get("guardrail_score")
+            docs_retrieved = len(result.get("retrieved_documents", []))
+            docs_relevant = len(result.get("relevant_documents", []))
+        else:
+            logger.warning("Using fallback Q&A (agentic pipeline unavailable)")
+            answer = _fallback_qa(question, context)
+            guardrail = None
+            docs_retrieved = 0
+            docs_relevant = 0
 
-Answer:"""
-
-
-        # Generate response
-        response = llm.invoke(prompt)
-        answer = response.content if hasattr(response, 'content') else str(response)
-        
         if not answer:
             answer = "I apologize, but I couldn't generate a response. Please try rephrasing your question."
-        
+
         elapsed = time.time() - start_time
-        
-        # Format response with metadata
+
+        meta_parts = [f"⏱️ {elapsed:.1f}s"]
+        if guardrail is not None:
+            meta_parts.append(f"🛡️ Guardrail: {guardrail:.0f}/100")
+        if docs_retrieved > 0:
+            meta_parts.append(f"📚 {docs_relevant}/{docs_retrieved} relevant docs")
+        meta_parts.append("🤖 Agentic RAG" if rag_service else "🤖 RAG")
+        meta_line = " | ".join(meta_parts)
+
         formatted_answer = f"""{answer}
 
 ---
-*⏱️ Response time: {elapsed:.1f}s | 🤖 Powered by RAG*
+*{meta_line}*
 """
-        
-        # Update chat history
         history = (chat_history or []) + [(question, formatted_answer)]
-        
         return formatted_answer, history
-        
+
     except Exception as exc:
         logger.exception(f"Q&A error: {exc}")
         error_msg = f"❌ Error: {str(exc)}"
@@ -690,83 +733,48 @@ Answer:"""
 
 
 def streaming_answer(question: str, context: str = ""):
-    """
-    Stream answer tokens for real-time response.
-    Uses retriever + LLM directly (not the guild).
+    """Stream answer using the full agentic RAG pipeline.
+    Falls back to direct retriever+LLM if the pipeline is unavailable.
     """
     if not question.strip():
         yield ""
         return
-    
-    # Check API key
+
     groq_key, google_key = get_api_keys()
     if not groq_key and not google_key:
         yield "❌ Please add your GROQ_API_KEY or GOOGLE_API_KEY in Space Settings → Secrets."
         return
-    
-    # Setup provider
-    setup_llm_provider()
-    
-    try:
-        yield "🔍 Searching medical knowledge base...\n\n"
-        
-        from src.services.retrieval import make_retriever
-        from src.llm_config import get_synthesizer
-        
-        # Initialize retriever
-        retriever = make_retriever()
-        
-        # Build search query
-        search_query = question
-        if context.strip():
-            search_query = f"{context} {question}"
-        
-        yield "🔍 Searching medical knowledge base...\n📚 Retrieving relevant documents...\n\n"
-        
-        # Retrieve docs
-        docs = retriever.search(search_query, top_k=5)
-        
-        # Format context
-        doc_context = ""
-        if docs:
-            doc_texts = []
-            for doc in docs[:5]:
-                if hasattr(doc, 'content'):
-                    doc_texts.append(doc.content[:500])
-                elif isinstance(doc, dict) and 'content' in doc:
-                    doc_texts.append(doc['content'][:500])
-            doc_context = "\n\n---\n\n".join(doc_texts)
-        
-        yield "🔍 Searching medical knowledge base...\n📚 Retrieving relevant documents...\n💭 Generating response...\n\n"
-        
-        # Get LLM
-        llm = get_synthesizer()
-        
-        start_time = time.time()
-        
-        # Build prompt
-        prompt = f"""You are a medical AI assistant. Answer the following medical question based on the provided context.
-Be helpful, accurate, and include relevant medical information. Always recommend consulting a healthcare professional for personal medical advice.
 
-Context from medical knowledge base:
-{doc_context if doc_context else "No specific context available - using general medical knowledge."}
+    setup_llm_provider()
 
-Patient Context: {context if context else "Not provided"}
+    try:
+        yield "🛡️ Checking medical domain relevance...\n\n"
 
-Question: {question}
+        start_time = time.time()
 
-Answer:"""
+        rag_service = _get_rag_service()
+        if rag_service is not None:
+            yield "🛡️ Checking medical domain relevance...\n🔍 Retrieving medical documents...\n\n"
+            result = rag_service.ask(query=question, patient_context=context)
+            answer = result.get("final_answer", "")
+            guardrail = result.get("guardrail_score")
+            docs_relevant = len(result.get("relevant_documents", []))
+            docs_retrieved = len(result.get("retrieved_documents", []))
+        else:
+            yield "🔍 Searching medical knowledge base...\n📚 Retrieving relevant documents...\n\n"
+            answer = _fallback_qa(question, context)
+            guardrail = None
+            docs_relevant = 0
+            docs_retrieved = 0
 
-        # Generate response
-        response = llm.invoke(prompt)
-        answer = response.content if hasattr(response, 'content') else str(response)
-        
         if not answer:
             answer = "I apologize, but I couldn't generate a response. Please try rephrasing your question."
-        
+
+        yield "🛡️ Guardrail ✓\n🔍 Retrieved ✓\n📊 Graded ✓\n💭 Generating response...\n\n"
+
         elapsed = time.time() - start_time
-        
-        # Simulate streaming by revealing text progressively
+
+        # Progressive reveal
         words = answer.split()
         accumulated = ""
         for i, word in enumerate(words):
@@ -774,20 +782,27 @@ Answer:"""
             if i % 5 == 0:
                 yield accumulated
                 time.sleep(0.02)
-        
-        # Final complete response
+
+        # Final response with metadata
+        meta_parts = [f"⏱️ {elapsed:.1f}s"]
+        if guardrail is not None:
+            meta_parts.append(f"🛡️ Guardrail: {guardrail:.0f}/100")
+        if docs_retrieved > 0:
+            meta_parts.append(f"📚 {docs_relevant}/{docs_retrieved} relevant docs")
+        meta_parts.append("🤖 Agentic RAG" if rag_service else "🤖 RAG")
+        meta_line = " | ".join(meta_parts)
+
         yield f"""{answer}
 
 ---
-*⏱️ Response time: {elapsed:.1f}s | 🤖 Powered by RAG*
+*{meta_line}*
 """
-        
+
     except Exception as exc:
         logger.exception(f"Streaming Q&A error: {exc}")
         yield f"❌ Error: {str(exc)}"
 
 
-
 # ---------------------------------------------------------------------------
 # Gradio Interface
 # ---------------------------------------------------------------------------

src/llm_config.py

@@ -387,6 +387,11 @@ class LLMConfig:
 llm_config = LLMConfig()
 
 
+def get_synthesizer(model_name: Optional[str] = None):
+    """Module-level convenience: get a synthesizer LLM instance."""
+    return llm_config.get_synthesizer(model_name)
+
+
 def check_api_connection():
     """Verify API connection and keys are configured"""
     provider = DEFAULT_LLM_PROVIDER

src/routers/analyze.py

@@ -122,7 +122,7 @@ async def _run_guild_analysis(
 
     try:
         # Run sync function in thread pool
-        loop = asyncio.get_event_loop()
+        loop = asyncio.get_running_loop()
         result = await loop.run_in_executor(
             _executor,
             lambda: ragbot.analyze(
@@ -142,6 +142,16 @@ async def _run_guild_analysis(
     elapsed = (time.time() - t0) * 1000
 
     # Build response from result
+    # Guild workflow returns a dict; ragbot.analyze() may return dict or object
+    if isinstance(result, dict):
+        prediction = result.get('prediction')
+        analysis = result.get('analysis')
+        conversational_summary = result.get('conversational_summary')
+    else:
+        prediction = getattr(result, 'prediction', None)
+        analysis = getattr(result, 'analysis', None)
+        conversational_summary = getattr(result, 'conversational_summary', None)
+
     return AnalysisResponse(
         status="success",
         request_id=request_id,
@@ -150,9 +160,9 @@ async def _run_guild_analysis(
         input_biomarkers=biomarkers,
         patient_context=patient_ctx,
         processing_time_ms=round(elapsed, 1),
-        prediction=result.prediction if hasattr(result, 'prediction') else None,
-        analysis=result.analysis if hasattr(result, 'analysis') else None,
-        conversational_summary=result.conversational_summary if hasattr(result, 'conversational_summary') else None,
+        prediction=prediction,
+        analysis=analysis,
+        conversational_summary=conversational_summary,
     )
 
 

src/routers/ask.py

@@ -88,7 +88,7 @@ async def _stream_rag_response(
         await asyncio.sleep(0)  # Allow event loop to flush
         
         # Run the RAG pipeline (synchronous, but we yield progress)
-        loop = asyncio.get_event_loop()
+        loop = asyncio.get_running_loop()
         result = await loop.run_in_executor(
             None,
             lambda: rag_service.ask(

src/routers/health.py

@@ -40,11 +40,12 @@ async def readiness_check(request: Request) -> HealthResponse:
     # --- PostgreSQL ---
     try:
         from src.database import get_engine
+        from sqlalchemy import text
         engine = get_engine()
         if engine is not None:
             t0 = time.time()
             with engine.connect() as conn:
-                conn.execute("SELECT 1")
+                conn.execute(text("SELECT 1"))
             latency = (time.time() - t0) * 1000
             services.append(ServiceHealth(name="postgresql", status="ok", latency_ms=round(latency, 1)))
         else:
@@ -106,8 +107,8 @@ async def readiness_check(request: Request) -> HealthResponse:
 
     # --- FAISS (local retriever) ---
     try:
-        from src.services.retrieval import make_retriever
-        retriever = make_retriever("faiss")
+        from src.services.retrieval.factory import make_retriever
+        retriever = make_retriever(backend="faiss")
         if retriever is not None:
             doc_count = retriever.doc_count()
             services.append(ServiceHealth(name="faiss", status="ok", detail=f"{doc_count} docs indexed"))

src/services/agents/context.py

@@ -21,3 +21,4 @@ class AgenticContext:
     cache: Any                       # RedisCache
     tracer: Any                      # LangfuseTracer
     guild: Optional[Any] = None      # ClinicalInsightGuild (original workflow)
+    retriever: Optional[Any] = None  # BaseRetriever (FAISS or OpenSearch)

src/services/agents/nodes/generate_answer_node.py

@@ -24,9 +24,10 @@ def generate_answer_node(state: dict, *, context: Any) -> dict:
     # Build evidence block
     evidence_parts: list[str] = []
     for i, doc in enumerate(documents, 1):
-        title = doc.get("title", "Unknown")
-        section = doc.get("section", "")
-        text = doc.get("text", "")[:2000]
+        meta = doc.get("metadata", {})
+        title = meta.get("title", doc.get("title", "Unknown"))
+        section = meta.get("section_title", doc.get("section", ""))
+        text = (doc.get("content") or doc.get("text", ""))[:2000]
         header = f"[{i}] {title}"
         if section:
             header += f" — {section}"

src/services/agents/nodes/grade_documents_node.py

@@ -31,7 +31,7 @@ def grade_documents_node(state: dict, *, context: Any) -> dict:
     grading_results: list[dict] = []
 
     for doc in documents:
-        text = doc.get("text", "")
+        text = doc.get("content") or doc.get("text", "")
         user_msg = f"Query: {query}\n\nDocument:\n{text[:2000]}"
         try:
             response = context.llm.invoke(
@@ -51,11 +51,13 @@ def grade_documents_node(state: dict, *, context: Any) -> dict:
             logger.warning("Grading LLM failed for doc %s: %s — marking relevant", doc.get("id"), exc)
             is_relevant = True  # benefit of the doubt
 
-        grading_results.append({"doc_id": doc.get("id"), "relevant": is_relevant})
+        grading_results.append({"doc_id": doc.get("id", doc.get("_id")), "relevant": is_relevant})
         if is_relevant:
             relevant.append(doc)
 
-    needs_rewrite = len(relevant) < 2 and not state.get("rewritten_query")
+    attempts = state.get("retrieval_attempts", 1)
+    max_attempts = state.get("max_retrieval_attempts", 2)
+    needs_rewrite = len(relevant) < 2 and attempts < max_attempts
 
     return {
         "grading_results": grading_results,

src/services/agents/nodes/retrieve_node.py

@@ -1,7 +1,10 @@
 """
 MediGuard AI — Retrieve Node
 
-Performs hybrid search (BM25 + vector KNN) and merges results.
+Performs document retrieval using the best available backend:
+  1. Generic retriever (FAISS, OpenSearch wrapper, etc.)
+  2. OpenSearch hybrid search (BM25 + KNN)
+  3. BM25 keyword fallback
 """
 
 from __future__ import annotations
@@ -13,56 +16,90 @@ logger = logging.getLogger(__name__)
 
 
 def retrieve_node(state: dict, *, context: Any) -> dict:
-    """Retrieve documents from OpenSearch via hybrid search."""
-    query = state.get("rewritten_query") or state.get("query", "")
+    """Retrieve documents using the best available backend.
 
-    # 1. Try cache first
+    Priority:
+      1. context.retriever (generic BaseRetriever — works with FAISS & OpenSearch)
+      2. context.opensearch_client + context.embedding_service (hybrid search)
+      3. BM25 keyword fallback
+      4. Empty list
+    """
+    query = state.get("rewritten_query") or state.get("query", "")
     cache_key = f"retrieve:{query}"
+
+    # 1. Try cache
     if context.cache:
         cached = context.cache.get(cache_key)
         if cached is not None:
-            logger.debug("Cache hit for retrieve query")
-            return {"retrieved_documents": cached}
+            logger.info("Cache HIT for query: %s…", query[:50])
+            attempts = state.get("retrieval_attempts", 0) + 1
+            return {"retrieved_documents": cached, "retrieval_attempts": attempts}
 
-    # 2. Embed the query
-    try:
-        query_embedding = context.embedding_service.embed_query(query)
-    except Exception as exc:
-        logger.error("Embedding failed: %s", exc)
-        return {"retrieved_documents": [], "errors": [str(exc)]}
+    documents: list = []
 
-    # 3. Hybrid search
-    try:
-        results = context.opensearch_client.search_hybrid(
-            query_text=query,
-            query_vector=query_embedding,
-            top_k=10,
-        )
-    except Exception as exc:
-        logger.error("OpenSearch hybrid search failed: %s — falling back to BM25", exc)
+    # 2. Generic retriever (FAISS, OpenSearch wrapper, etc.)
+    if getattr(context, "retriever", None) is not None:
         try:
-            results = context.opensearch_client.search_bm25(
+            results = context.retriever.retrieve(query, top_k=8)
+            documents = [
+                {
+                    "content": getattr(r, "content", ""),
+                    "metadata": getattr(r, "metadata", {}),
+                    "score": getattr(r, "score", 0.0),
+                }
+                for r in results
+            ]
+            backend = getattr(context.retriever, "backend_name", "unknown")
+            logger.info("Retrieved %d docs via %s", len(documents), backend)
+        except Exception as exc:
+            logger.warning("Retriever failed (%s), trying OpenSearch fallback…", exc)
+
+    # 3. OpenSearch hybrid fallback
+    if not documents and context.opensearch_client and context.embedding_service:
+        try:
+            embedding = context.embedding_service.embed_query(query)
+            raw_hits = context.opensearch_client.search_hybrid(
                 query_text=query,
-                top_k=10,
+                query_vector=embedding,
+                top_k=8,
             )
-        except Exception as exc2:
-            logger.error("BM25 fallback also failed: %s", exc2)
-            return {"retrieved_documents": [], "errors": [str(exc), str(exc2)]}
+            documents = [
+                {
+                    "content": h.get("_source", {}).get("chunk_text", ""),
+                    "metadata": {
+                        k: v for k, v in h.get("_source", {}).items()
+                        if k != "chunk_text"
+                    },
+                    "score": h.get("_score", 0.0),
+                }
+                for h in raw_hits
+            ]
+            logger.info("Retrieved %d docs via OpenSearch hybrid", len(documents))
+        except Exception as exc:
+            logger.error("OpenSearch retrieval failed: %s", exc)
 
-    documents = [
-        {
-            "id": hit.get("_id", ""),
-            "score": hit.get("_score", 0.0),
-            "text": hit.get("_source", {}).get("chunk_text", ""),
-            "title": hit.get("_source", {}).get("title", ""),
-            "section": hit.get("_source", {}).get("section_title", ""),
-            "metadata": hit.get("_source", {}),
-        }
-        for hit in results
-    ]
+    # 4. Optional BM25 fallback if still nothing
+    if not documents and context.opensearch_client:
+        try:
+            raw_hits = context.opensearch_client.search_bm25(query_text=query, top_k=8)
+            documents = [
+                {
+                    "content": h.get("_source", {}).get("chunk_text", ""),
+                    "metadata": {
+                        k: v for k, v in h.get("_source", {}).items()
+                        if k != "chunk_text"
+                    },
+                    "score": h.get("_score", 0.0),
+                }
+                for h in raw_hits
+            ]
+            logger.info("Retrieved %d docs via BM25 fallback", len(documents))
+        except Exception as exc:
+            logger.error("BM25 fallback also failed: %s", exc)
 
-    # 4. Store in cache (5 min TTL)
-    if context.cache:
+    # 5. Store in cache (5 min TTL)
+    if context.cache and documents:
         context.cache.set(cache_key, documents, ttl=300)
 
-    return {"retrieved_documents": documents}
+    attempts = state.get("retrieval_attempts", 0) + 1
+    return {"retrieved_documents": documents, "retrieval_attempts": attempts}

src/services/cache/redis_cache.py

@@ -11,7 +11,7 @@ import hashlib
 import json
 import logging
 from functools import lru_cache
-from typing import Any, Dict, Optional
+from typing import Any, Optional
 
 from src.settings import get_settings
 
@@ -48,12 +48,13 @@ class RedisCache:
         raw = "|".join(parts)
         return f"mediguard:{hashlib.sha256(raw.encode()).hexdigest()}"
 
-    def get(self, *key_parts: str) -> Optional[Dict[str, Any]]:
+    def get(self, key: str) -> Optional[Any]:
+        """Get a cached value by key."""
         if not self._enabled:
             return None
-        key = self._make_key(*key_parts)
+        cache_key = self._make_key(key)
         try:
-            value = self._client.get(key)
+            value = self._client.get(cache_key)
             if value is None:
                 return None
             return json.loads(value)
@@ -61,23 +62,25 @@ class RedisCache:
             logger.warning("Cache GET failed: %s", exc)
             return None
 
-    def set(self, value: Dict[str, Any], *key_parts: str, ttl: Optional[int] = None) -> bool:
+    def set(self, key: str, value: Any, *, ttl: Optional[int] = None) -> bool:
+        """Set a cached value with optional TTL."""
         if not self._enabled:
             return False
-        key = self._make_key(*key_parts)
+        cache_key = self._make_key(key)
         try:
-            self._client.setex(key, ttl or self._default_ttl, json.dumps(value, default=str))
+            self._client.setex(cache_key, ttl or self._default_ttl, json.dumps(value, default=str))
             return True
         except Exception as exc:
             logger.warning("Cache SET failed: %s", exc)
             return False
 
-    def delete(self, *key_parts: str) -> bool:
+    def delete(self, key: str) -> bool:
+        """Delete a cached value by key."""
         if not self._enabled:
             return False
-        key = self._make_key(*key_parts)
+        cache_key = self._make_key(key)
         try:
-            self._client.delete(key)
+            self._client.delete(cache_key)
             return True
         except Exception as exc:
             logger.warning("Cache DELETE failed: %s", exc)

src/services/opensearch/client.py

@@ -85,7 +85,7 @@ class OpenSearchClient:
 
     def search_bm25(
         self,
-        query: str,
+        query_text: str,
         *,
         top_k: int = 10,
         filters: Optional[Dict[str, Any]] = None,
@@ -97,7 +97,7 @@ class OpenSearchClient:
                     "must": [
                         {
                             "multi_match": {
-                                "query": query,
+                                "query": query_text,
                                 "fields": [
                                     "chunk_text^3",
                                     "title^2",
@@ -119,7 +119,7 @@ class OpenSearchClient:
 
     def search_vector(
         self,
-        embedding: List[float],
+        query_vector: List[float],
         *,
         top_k: int = 10,
         filters: Optional[Dict[str, Any]] = None,
@@ -129,7 +129,7 @@ class OpenSearchClient:
             "query": {
                 "knn": {
                     "embedding": {
-                        "vector": embedding,
+                        "vector": query_vector,
                         "k": top_k,
                     }
                 }
@@ -141,8 +141,8 @@ class OpenSearchClient:
 
     def search_hybrid(
         self,
-        query: str,
-        embedding: List[float],
+        query_text: str,
+        query_vector: List[float],
         *,
         top_k: int = 10,
         filters: Optional[Dict[str, Any]] = None,
@@ -150,8 +150,8 @@ class OpenSearchClient:
         vector_weight: float = 0.6,
     ) -> List[Dict[str, Any]]:
         """Reciprocal Rank Fusion of BM25 + KNN results."""
-        bm25_results = self.search_bm25(query, top_k=top_k, filters=filters)
-        vector_results = self.search_vector(embedding, top_k=top_k, filters=filters)
+        bm25_results = self.search_bm25(query_text, top_k=top_k, filters=filters)
+        vector_results = self.search_vector(query_vector, top_k=top_k, filters=filters)
         return self._rrf_fuse(bm25_results, vector_results, top_k=top_k)
 
     # ── Internal helpers ─────────────────────────────────────────────────
@@ -166,7 +166,7 @@ class OpenSearchClient:
             {
                 "_id": h["_id"],
                 "_score": h.get("_score", 0.0),
-                **h.get("_source", {}),
+                "_source": h.get("_source", {}),
             }
             for h in hits
         ]