fix(phase-2): resolve console streaming, fallback visibility, and score-based dedup

- Fixed live colored console stream handler in logger.py
- Added visible fallback alert banner in main.py and is_fallback tracking in PlanOutput
- Updated .env.example with MAX_SUB_QUERIES and planner hyperparameters
- Enhanced PlannerAgent JSON extraction for reasoning and prompt robustness
- Fixed limit=1 prompt wording edge case in planner_agent.py
- Implemented highest-score retention for URL deduplication in search_agent.py
- Synchronized README.md project structure and telemetry tables

.env.example

@@ -1,11 +1,17 @@
 # Groq API Configuration (Free tier: https://console.groq.com/)
 GROQ_API_KEY=your_groq_api_key_here
 GROQ_MODEL=openai/gpt-oss-20b
-MAX_TOKENS=1024
 
 # Tavily API Configuration (Free 1,000 searches/mo: https://tavily.com/)
 TAVILY_API_KEY=your_tavily_api_key_here
 
 # Pipeline Settings
 MAX_SEARCH_RESULTS=5
+MAX_SUB_QUERIES=3
+MAX_RESULTS_PER_SUBQUERY=3
+
+# Model Hyperparameters
 TEMPERATURE=0.2
+MAX_TOKENS=1024
+PLANNER_TEMPERATURE=0.1
+PLANNER_MAX_TOKENS=500

README.md

@@ -69,14 +69,15 @@ Traditional LLMs hallucinate or rely on frozen training data. Unlike standard st
 
 ---
 
-## 📊 Benchmarks & Telemetry (Phase 1 Baseline)
+## 📊 Benchmarks & Telemetry
 
-| Metric | Measured Value | Target | Status |
-| :--- | :--- | :--- | :--- |
-| **Search Latency (Tavily)** | ~0.95s – 4.0s | < 10.0s | ✅ Optimal |
-| **Synthesis Latency (Groq)** | ~2.1s – 3.2s | < 10.0s | ✅ Optimal |
-| **Total End-to-End Latency** | **~3.15s – 7.97s** | **< 30.0s** | ✅ **Well within target** |
-| **Cost Per Query** | **$0.00** (Free Developer Tier) | $0.00 | ✅ 100% Free |
+| Metric | Phase 1 (Single-Search) | Phase 2 (Planner + Multi-Search) | Target | Status |
+| :--- | :--- | :--- | :--- | :--- |
+| **Planning Latency (Groq)** | N/A | ~0.8s – 1.2s | < 3.0s | ✅ Sub-second |
+| **Search Latency (Tavily)** | ~0.95s – 4.0s | ~8.0s – 15.0s (3 sub-queries) | < 20.0s | ✅ Optimal |
+| **Synthesis Latency (Groq)** | ~2.1s – 3.2s | ~1.7s – 3.5s | < 10.0s | ✅ Optimal |
+| **Total End-to-End Latency** | **~3.15s – 7.97s** | **~10.5s – 18.0s** | **< 30.0s** | ✅ **Well within target** |
+| **Cost Per Query** | **$0.00** (Free Tier) | **$0.00** (Free Tier) | $0.00 | ✅ 100% Free |
 
 ---
 
@@ -85,22 +86,24 @@ Traditional LLMs hallucinate or rely on frozen training data. Unlike standard st
 ```
 Multi-Agent-Research-Assistant/
 ├── .gitignore               # Protection for secrets, logs, and venvs
-├── .env.example             # Configuration template
+├── .env.example             # Configuration template (includes Phase 2 settings)
 ├── requirements.txt         # Exact-pinned dependencies
-├── main.py                  # CLI entry point
+├── main.py                  # Interactive CLI entry point with fallback alerts
 ├── logs/                    # Runtime logs (gitignored)
 │   └── research_assistant.log
 ├── src/
 │   ├── __init__.py
-│   ├── config.py            # Settings validation & loader
-│   ├── logger.py            # Structured logging setup
-│   ├── pipeline.py          # Two-Agent Pipeline coordinator
+│   ├── config.py            # Settings validation & dynamic configuration
+│   ├── logger.py            # Centralized logging (live console streaming + file logging)
+│   ├── pipeline.py          # Three-Agent Pipeline coordinator (Planner -> Multi-Search -> Writer)
 │   └── agents/
 │       ├── __init__.py
-│       ├── search_agent.py  # Tavily Web Search agent
+│       ├── planner_agent.py # Query decomposition & JSON planning agent
+│       ├── search_agent.py  # Tavily Web Search & score-based URL deduplication
 │       └── writer_agent.py  # Groq Synthesis & Grounding agent
 └── tests/
-    └── test_phase1.py       # Automated unit & mock tests
+    ├── test_phase1.py       # Phase 1 unit & pipeline tests
+    └── test_phase2.py       # Phase 2 planner, deduplication & 3-agent tests
 ```
 
 ---
@@ -137,12 +140,23 @@ cp .env.example .env
 ```
 Edit `.env`:
 ```env
+# Groq API Configuration (Free tier: https://console.groq.com/)
 GROQ_API_KEY=gsk_your_groq_api_key_here
 GROQ_MODEL=openai/gpt-oss-20b
+
+# Tavily API Configuration (Free 1,000 searches/mo: https://tavily.com/)
 TAVILY_API_KEY=tvly-your_tavily_api_key_here
+
+# Pipeline Settings
 MAX_SEARCH_RESULTS=5
+MAX_SUB_QUERIES=3
+MAX_RESULTS_PER_SUBQUERY=3
+
+# Model Hyperparameters
 TEMPERATURE=0.2
 MAX_TOKENS=1024
+PLANNER_TEMPERATURE=0.1
+PLANNER_MAX_TOKENS=500
 ```
 
 ### 4. Run the Research Assistant

main.py

@@ -34,13 +34,16 @@ def run_pipeline(query: str):
         result = pipeline.run(query=query)
 
         # Display Planner breakdown
-        print(Fore.GREEN + f"✅ Planner generated {len(result.sub_queries)} sub-queries in {result.planning_time_sec}s:")
-        for idx, sq in enumerate(result.sub_queries, 1):
-            print(Fore.CYAN + f"   [{idx}] " + Fore.WHITE + f"{sq}")
-        if result.plan_rationale:
-            print(Fore.LIGHTBLACK_EX + f"   Rationale: {result.plan_rationale}\n")
+        if result.is_fallback:
+            print(Fore.YELLOW + Style.BRIGHT + f"⚠️ [WARNING] Planner fallback was triggered (single query search):\n   Reason: {result.plan_rationale}\n")
         else:
-            print()
+            print(Fore.GREEN + f"✅ Planner generated {len(result.sub_queries)} sub-queries in {result.planning_time_sec}s:")
+            for idx, sq in enumerate(result.sub_queries, 1):
+                print(Fore.CYAN + f"   [{idx}] " + Fore.WHITE + f"{sq}")
+            if result.plan_rationale:
+                print(Fore.LIGHTBLACK_EX + f"   Rationale: {result.plan_rationale}\n")
+            else:
+                print()
 
         print(Fore.BLUE + f"🔍 [Step 2/3] Search Agent executing multi-search with URL deduplication...")
         print(Fore.GREEN + f"✅ Retrieved {len(result.search_results)} unique sources across all sub-queries in {result.search_time_sec}s\n")

src/agents/planner_agent.py

@@ -1,7 +1,6 @@
 import json
 import logging
 import re
-import time
 from typing import List, Dict, Any, Tuple, Optional
 from groq import Groq
 from pydantic import BaseModel, Field
@@ -15,27 +14,40 @@ class PlanOutput(BaseModel):
     """Structured plan output generated by the PlannerAgent."""
     sub_queries: List[str] = Field(
         ...,
-        description="List of 2 to 4 distinct, targeted search queries to thoroughly investigate the topic."
+        description="List of distinct, targeted search queries to investigate the topic."
     )
     rationale: str = Field(
         default="",
         description="Brief explanation of how these sub-queries cover the research problem."
     )
+    is_fallback: bool = Field(
+        default=False,
+        description="True if the plan is an automated fallback due to model or parsing error."
+    )
 
 
 class PlannerAgent:
     """Agent responsible for decomposing a complex research question into focused sub-queries."""
 
-    def __init__(self, api_key: Optional[str] = None, model: Optional[str] = None):
+    def __init__(
+        self,
+        api_key: Optional[str] = None,
+        model: Optional[str] = None,
+        temperature: Optional[float] = None,
+        max_tokens: Optional[int] = None
+    ):
         self.api_key = api_key or settings.groq_api_key
         self.model = model or settings.groq_model
+        self.temperature = temperature if temperature is not None else settings.planner_temperature
+        self.max_tokens = max_tokens if max_tokens is not None else settings.planner_max_tokens
+
         if not self.api_key:
             raise ValueError("Groq API key is missing. Please set GROQ_API_KEY in .env.")
         self.client = Groq(api_key=self.api_key)
 
     def plan(self, query: str, max_sub_queries: Optional[int] = None) -> Tuple[PlanOutput, Dict[str, Any]]:
         """
-        Decomposes user query into 2-4 focused sub-queries.
+        Decomposes user query into focused sub-queries.
 
         Args:
             query: The user's original research question.
@@ -47,21 +59,24 @@ class PlannerAgent:
         limit = max_sub_queries or settings.max_sub_queries
         limit = max(1, min(limit, 4))  # Constrain between 1 and 4
 
+        query_count_phrase = f"{limit} focused search query" if limit == 1 else f"2 to {limit} distinct, highly specific search queries"
+
         system_prompt = (
-            "You are a master Research Strategist and Planner AI. Your goal is to analyze a complex "
-            "research question and decompose it into 2 to {limit} distinct, highly specific search queries.\n\n"
+            f"You are a master Research Strategist and Planner AI. Your goal is to analyze a complex "
+            f"research question and decompose it into {query_count_phrase}.\n\n"
             "GUIDELINES:\n"
             "1. Each sub-query must target a distinct dimension of the question (e.g., technical definitions, comparisons, real-world benchmarks, trade-offs).\n"
             "2. Ensure queries are formatted as effective search engine queries (concise, clear, and keyword-rich).\n"
             "3. Do NOT produce redundant or overlapping queries.\n"
-            "4. Return strictly valid JSON adhering to this structure:\n"
+            "4. Return strictly valid JSON with keys 'sub_queries' (list of strings) and 'rationale' (string).\n"
+            "Example JSON:\n"
             "{\n"
-            '  "sub_queries": ["query 1", "query 2", ...],\n'
-            '  "rationale": "Brief 1-2 sentence reason for choosing these sub-queries"\n'
+            '  "sub_queries": ["query 1", "query 2"],\n'
+            '  "rationale": "Explanation here"\n'
             "}"
-        ).replace("{limit}", str(limit))
+        )
 
-        user_prompt = f"Decompose this research question into {limit} focused search queries:\n\"{query}\""
+        user_prompt = f"Decompose this research question into search queries:\n\"{query}\""
 
         logger.info(f"Invoking PlannerAgent with model '{self.model}' for query: '{query}' (limit={limit})")
 
@@ -72,12 +87,13 @@ class PlannerAgent:
                     {"role": "user", "content": user_prompt}
                 ],
                 model=self.model,
-                temperature=0.1,  # Low temperature for deterministic planning
-                max_tokens=500,
-                response_format={"type": "json_object"}
+                temperature=self.temperature,
+                max_tokens=self.max_tokens,
             )
 
-            raw_content = chat_completion.choices[0].message.content
+            msg = chat_completion.choices[0].message
+            raw_content = msg.content or getattr(msg, "reasoning", "") or ""
+
             usage = {
                 "prompt_tokens": chat_completion.usage.prompt_tokens if chat_completion.usage else 0,
                 "completion_tokens": chat_completion.usage.completion_tokens if chat_completion.usage else 0,
@@ -90,28 +106,45 @@ class PlannerAgent:
             return plan_output, usage
 
         except Exception as e:
-            logger.warning(f"PlannerAgent failed or raised exception: {e}. Falling back to original query.")
+            logger.warning(f"⚠️ PlannerAgent failed or raised exception: {e}. Falling back to original query.")
             fallback_plan = PlanOutput(
                 sub_queries=[query],
-                rationale=f"Fallback to single query due to planner error: {str(e)}"
+                rationale=f"Fallback to single query due to planner error: {str(e)}",
+                is_fallback=True
             )
             return fallback_plan, {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0, "model": self.model}
 
     def _parse_response(self, content: str, fallback_query: str, limit: int) -> PlanOutput:
-        """Extracts and validates PlanOutput from JSON response string."""
+        """Extracts and validates PlanOutput from JSON response string, markdown blocks, or embedded json."""
         try:
-            # Handle potential markdown code block wrappers
             clean_content = content.strip()
+            
+            # Extract content from markdown code fence if present
             if "```json" in clean_content:
                 clean_content = clean_content.split("```json")[1].split("```")[0].strip()
             elif "```" in clean_content:
                 clean_content = clean_content.split("```")[1].split("```")[0].strip()
 
+            # If not pure json, try finding outermost JSON object { ... } or array [ ... ]
+            if not (clean_content.startswith("{") or clean_content.startswith("[")):
+                match = re.search(r"(\{.*\}|\[.*\])", clean_content, re.DOTALL)
+                if match:
+                    clean_content = match.group(1).strip()
+
             data = json.loads(clean_content)
-            sub_queries = data.get("sub_queries", [])
+
+            # Handle case where JSON is a list of strings directly: ["q1", "q2"]
+            if isinstance(data, list):
+                sub_queries = data
+                rationale = "Direct sub-query list generated by planner."
+            elif isinstance(data, dict):
+                sub_queries = data.get("sub_queries") or data.get("queries") or []
+                rationale = data.get("rationale", "")
+            else:
+                raise ValueError("Parsed JSON root is neither an object nor a list.")
 
             # Filter and sanitize sub-queries
-            valid_queries = [q.strip() for q in sub_queries if isinstance(q, str) and q.strip()]
+            valid_queries = [str(q).strip() for q in sub_queries if str(q).strip()]
 
             if not valid_queries:
                 raise ValueError("Parsed JSON contains empty sub_queries list.")
@@ -121,12 +154,14 @@ class PlannerAgent:
 
             return PlanOutput(
                 sub_queries=valid_queries,
-                rationale=data.get("rationale", "")
+                rationale=rationale,
+                is_fallback=False
             )
 
         except Exception as err:
             logger.warning(f"Failed to parse JSON planner output: {err}. Content was: {content[:200]}")
             return PlanOutput(
                 sub_queries=[fallback_query],
-                rationale=f"Fallback used due to parsing exception ({err})"
+                rationale=f"Fallback used due to parsing exception ({err})",
+                is_fallback=True
             )

src/agents/search_agent.py

@@ -70,8 +70,9 @@ class SearchAgent:
         limit_per_query = max_results_per_query or settings.max_results_per_subquery
         logger.info(f"Executing multi-search across {len(queries)} sub-queries (limit={limit_per_query} each)")
 
+        # Map url -> index in all_results to retain highest-scoring result
+        url_to_index = {}
         all_results = []
-        seen_urls = set()
 
         for idx, sub_query in enumerate(queries, 1):
             try:
@@ -83,15 +84,21 @@ class SearchAgent:
 
                 for item in sub_results:
                     url = item.get("url", "").strip()
-                    # Deduplicate by URL
-                    if url and url in seen_urls:
-                        logger.debug(f"Skipping duplicate URL already retrieved: {url}")
-                        continue
-                    if url:
-                        seen_urls.add(url)
-                    
+                    score = item.get("score", 0.0)
                     item["matched_sub_query"] = sub_query
-                    all_results.append(item)
+
+                    if url and url in url_to_index:
+                        existing_idx = url_to_index[url]
+                        existing_score = all_results[existing_idx].get("score", 0.0)
+                        if score > existing_score:
+                            logger.debug(f"Updating duplicate URL '{url}' with higher score: {score} > {existing_score}")
+                            all_results[existing_idx] = item
+                        else:
+                            logger.debug(f"Skipping duplicate URL '{url}' with lower/equal score: {score} <= {existing_score}")
+                    else:
+                        if url:
+                            url_to_index[url] = len(all_results)
+                        all_results.append(item)
 
             except Exception as e:
                 logger.warning(f"Sub-query search failed for '{sub_query}': {e}. Continuing with remaining queries.")

src/config.py

@@ -23,6 +23,8 @@ class Settings(BaseModel):
     max_results_per_subquery: int = Field(default_factory=lambda: int(os.getenv("MAX_RESULTS_PER_SUBQUERY", "3")))
     temperature: float = Field(default_factory=lambda: float(os.getenv("TEMPERATURE", "0.2")))
     max_tokens: int = Field(default_factory=lambda: int(os.getenv("MAX_TOKENS", "1024")))
+    planner_temperature: float = Field(default_factory=lambda: float(os.getenv("PLANNER_TEMPERATURE", "0.1")))
+    planner_max_tokens: int = Field(default_factory=lambda: int(os.getenv("PLANNER_MAX_TOKENS", "500")))
 
     def validate_keys(self) -> None:
         """Ensure required API keys are populated."""

src/logger.py

@@ -6,11 +6,27 @@ BASE_DIR = Path(__file__).resolve().parent.parent
 LOGS_DIR = BASE_DIR / "logs"
 
 
-def setup_logging(level: int = logging.INFO, log_to_file: bool = True, log_to_console: bool = False) -> None:
-    """Configures application-wide logging with file and optional console handlers."""
+class ColoredConsoleFormatter(logging.Formatter):
+    """Format console output with subtle colors for debug and warnings."""
+    
+    COLORS = {
+        logging.DEBUG: "\033[90m[DEBUG]\033[0m",
+        logging.INFO: "\033[36m[INFO]\033[0m",
+        logging.WARNING: "\033[33m[WARNING]\033[0m",
+        logging.ERROR: "\033[31m[ERROR]\033[0m",
+        logging.CRITICAL: "\033[41m[CRITICAL]\033[0m"
+    }
+
+    def format(self, record: logging.LogRecord) -> str:
+        level_tag = self.COLORS.get(record.levelno, f"[{record.levelname}]")
+        return f"{level_tag} [{record.name}] {record.getMessage()}"
+
+
+def setup_logging(level: int = logging.INFO, log_to_file: bool = True, log_to_console: bool = True) -> None:
+    """Configures application-wide logging with file and live console handlers."""
     handlers = []
 
-    # File Handler
+    # 1. File Handler (captures all events with timestamp and lineno)
     if log_to_file:
         LOGS_DIR.mkdir(parents=True, exist_ok=True)
         file_handler = logging.FileHandler(LOGS_DIR / "research_assistant.log", encoding="utf-8")
@@ -22,13 +38,10 @@ def setup_logging(level: int = logging.INFO, log_to_file: bool = True, log_to_co
         file_handler.setLevel(level)
         handlers.append(file_handler)
 
-    # Console Handler (enabled if log_to_console=True or in DEBUG mode)
-    if log_to_console or level <= logging.DEBUG:
+    # 2. Live Console Stream Handler
+    if log_to_console:
         console_handler = logging.StreamHandler(sys.stdout)
-        console_formatter = logging.Formatter(
-            "\033[90m[DEBUG] [%(name)s] %(message)s\033[0m"
-        )
-        console_handler.setFormatter(console_formatter)
+        console_handler.setFormatter(ColoredConsoleFormatter())
         console_handler.setLevel(level)
         handlers.append(console_handler)
 

src/pipeline.py

@@ -22,6 +22,7 @@ class ResearchResult(BaseModel):
     search_time_sec: float = 0.0
     synthesis_time_sec: float = 0.0
     total_time_sec: float = 0.0
+    is_fallback: bool = False
     usage: Dict[str, Any] = Field(default_factory=dict)
 
 
@@ -108,6 +109,7 @@ class ResearchPipeline:
             search_time_sec=round(search_time, 2),
             synthesis_time_sec=round(synthesis_time, 2),
             total_time_sec=round(total_time, 2),
+            is_fallback=plan_output.is_fallback,
             usage=total_usage
         )
 

tests/test_phase2.py

@@ -47,11 +47,14 @@ class TestPhase2PlannerAndMultiSearch(unittest.TestCase):
 
         results = search_agent.search_multi(["query 1", "query 2"], max_results_per_query=2)
 
-        # Should deduplicate https://example.com/shared and return 3 total unique items
+        # Should deduplicate https://example.com/shared and return 3 total unique items, retaining highest-scored version
         self.assertEqual(len(results), 3)
         unique_urls = [r["url"] for r in results]
         self.assertEqual(unique_urls, ["https://example.com/shared", "https://example.com/doc-b", "https://example.com/doc-c"])
-        self.assertEqual(results[0]["matched_sub_query"], "query 1")
+        # Verify higher-scored Doc A (0.95 from query 2) replaced the lower-scored one (0.9 from query 1)
+        self.assertEqual(results[0]["score"], 0.95)
+        self.assertEqual(results[0]["title"], "Doc A (Duplicate)")
+        self.assertEqual(results[0]["matched_sub_query"], "query 2")
         self.assertEqual(results[2]["matched_sub_query"], "query 2")
 
     @patch("src.pipeline.settings")