Update core/pipeline_2/logic.py

core/pipeline_2/logic.py

@@ -404,9 +404,15 @@ class PipelineRAG:
     def run_stream(self, query: str, ground_truth: str = None):
         """Streaming version of Pipeline 2"""
         start = time.time()
+        
+        yield {"type": "status", "data": "ANALYZING QUERY STRUCTURE..."}
         decomposition = self._decompose_query(query)
+        is_multi_hop = decomposition["is_multi_hop"]
         
-        if decomposition["is_multi_hop"]:
+        yield {"type": "status", "data": f"PLAN: multi_hop={is_multi_hop}"}
+        
+        yield {"type": "status", "data": "EXECUTING HYBRID VECTOR RETRIEVAL..."}
+        if is_multi_hop:
             def _search_sub_query(sub_q: str) -> list[dict]:
                 return self.retriever.search(sub_q, top_k=20)
             all_candidates = []
@@ -414,18 +420,24 @@ class PipelineRAG:
                 futures = [pool.submit(_search_sub_query, sq) for sq in decomposition["sub_queries"]]
                 for fut in as_completed(futures): all_candidates.extend(fut.result())
             unique_candidates = self._deduplicate_papers(all_candidates, top_n=self.retrieval_top_k)
+            
+            yield {"type": "status", "data": f"RETRIEVED {len(unique_candidates)} CANDIDATES. RERANKING..."}
             top_papers = self.retriever.rerank(query, unique_candidates, top_n=self.rerank_top_n)
             top_papers = self._ensure_sub_query_coverage(top_papers, decomposition["sub_queries"], top_n=self.rerank_top_n)
         else:
             candidates = self.retriever.search(query, top_k=self.retrieval_top_k)
+            yield {"type": "status", "data": f"RETRIEVED {len(candidates)} CANDIDATES. RERANKING..."}
             top_papers = self.retriever.rerank(query, candidates, top_n=self.rerank_top_n)
 
         sources = [p.get("title", "") for p in top_papers]
         yield {"type": "sources", "data": sources}
         
-        max_ft = self.max_full_text if decomposition["is_multi_hop"] else min(2, self.max_full_text)
+        yield {"type": "status", "data": "FETCHING FULL-TEXT & BUILDING CONTEXT..."}
+        max_ft = self.max_full_text if is_multi_hop else min(2, self.max_full_text)
         full_context, abstracts, _ = self._build_context(top_papers, query, max_full_text=max_ft)
 
+        yield {"type": "status", "data": "SYNTHESIZING RESPONSE VIA GROQ..."}
+
         system_msg = (
             "You are an expert AI research assistant specializing in scientific literature synthesis. "
             "Answer the query using ONLY the provided paper context. Speak with absolute certainty.\n\n"
@@ -450,7 +462,7 @@ class PipelineRAG:
             
             usage = None
             for chunk in stream:
-                if chunk.choices[0].delta.content:
+                if chunk.choices and chunk.choices[0].delta.content:
                     token = chunk.choices[0].delta.content
                     full_answer += token
                     yield {"type": "token", "data": token}
@@ -460,12 +472,14 @@ class PipelineRAG:
                         candidates_token_count=chunk.x_groq.usage.completion_tokens
                     )
 
+            yield {"type": "status", "data": "GENERATION COMPLETE. CALCULATING BENCHMARK METRICS..."}
             stats = self.metrics.process_metrics(
                 client=None, query=query, answer=full_answer, context=full_context,
                 usage_metadata=usage, start_time=start, abstracts_list=abstracts, ground_truth=ground_truth,
                 model_name=self.model_name
             )
             yield {"type": "metrics", "data": stats}
+            yield {"type": "status", "data": "PIPELINE EXECUTION SUCCESSFUL."}
             
         except Exception as e:
             logger.error(f"Pipeline 2 stream error: {e}")