Upload 38 files

app/agents/deep_research.py

@@ -0,0 +1,249 @@
+"""Deep Research Orchestrator.
+
+Coordinates the full deep research pipeline:
+1. Planning (query decomposition)
+2. Parallel searching (multiple dimensions)
+3. Report synthesis
+"""
+
+import asyncio
+import json
+import time
+from typing import AsyncIterator, Optional
+
+from app.agents.planner import create_research_plan, ResearchPlan, ResearchDimension
+from app.agents.llm_client import generate_completion_stream
+from app.config import get_settings
+from app.sources.tavily import search_tavily
+from app.sources.duckduckgo import search_duckduckgo
+from app.reranking.pipeline import rerank_results
+from app.temporal.freshness_scorer import calculate_freshness_score
+
+
+class DimensionResult:
+    """Results from researching a single dimension."""
+    
+    def __init__(self, dimension: ResearchDimension):
+        self.dimension = dimension
+        self.results: list[dict] = []
+        self.error: Optional[str] = None
+
+
+async def run_deep_research(
+    query: str,
+    max_dimensions: int = 6,
+    max_sources_per_dim: int = 5,
+    max_total_searches: int = 20,
+) -> AsyncIterator[str]:
+    """
+    Run a deep research pipeline with streaming progress.
+    
+    Yields SSE-formatted events as the research progresses.
+    
+    Args:
+        query: The research query
+        max_dimensions: Maximum dimensions to research
+        max_sources_per_dim: Max results per dimension
+        max_total_searches: Total Tavily API calls allowed
+        
+    Yields:
+        SSE event strings in format: data: {json}\n\n
+    """
+    start_time = time.perf_counter()
+    settings = get_settings()
+    
+    try:
+        # === PHASE 1: PLANNING ===
+        yield _sse_event("status", {"phase": "planning", "message": "Analyzing query..."})
+        
+        plan = await create_research_plan(query, max_dimensions)
+        
+        yield _sse_event("plan_ready", {
+            "refined_query": plan.refined_query,
+            "dimensions": [
+                {"name": d.name, "description": d.description, "priority": d.priority}
+                for d in plan.dimensions
+            ],
+            "estimated_sources": plan.estimated_sources,
+        })
+        
+        # === PHASE 2: PARALLEL SEARCHING ===
+        yield _sse_event("status", {"phase": "searching", "message": "Researching dimensions..."})
+        
+        # Distribute search budget across dimensions
+        num_dimensions = len(plan.dimensions)
+        searches_per_dim = max(1, max_total_searches // num_dimensions)
+        
+        dimension_results: list[DimensionResult] = []
+        
+        # Search dimensions in parallel batches
+        for i, dimension in enumerate(plan.dimensions):
+            yield _sse_event("dimension_start", {
+                "index": i + 1,
+                "total": num_dimensions,
+                "name": dimension.name,
+                "query": dimension.search_query,
+            })
+            
+            # Search this dimension
+            result = await _search_dimension(
+                dimension=dimension,
+                max_results=max_sources_per_dim,
+                max_searches=searches_per_dim,
+            )
+            dimension_results.append(result)
+            
+            yield _sse_event("dimension_complete", {
+                "index": i + 1,
+                "name": dimension.name,
+                "results_count": len(result.results),
+                "error": result.error,
+            })
+            
+            # Small delay to avoid rate limits
+            await asyncio.sleep(0.1)
+        
+        # === PHASE 3: SYNTHESIS ===
+        yield _sse_event("status", {"phase": "synthesizing", "message": "Generating report..."})
+        yield _sse_event("synthesis_start", {})
+        
+        # Stream the report generation
+        async for chunk in _synthesize_report_stream(query, plan, dimension_results):
+            yield _sse_event("report_chunk", {"content": chunk})
+        
+        # === COMPLETE ===
+        total_time = time.perf_counter() - start_time
+        total_sources = sum(len(r.results) for r in dimension_results)
+        
+        yield _sse_event("done", {
+            "total_sources": total_sources,
+            "total_dimensions": num_dimensions,
+            "total_time_seconds": round(total_time, 2),
+        })
+        
+    except Exception as e:
+        yield _sse_event("error", {"message": str(e)})
+
+
+async def _search_dimension(
+    dimension: ResearchDimension,
+    max_results: int = 5,
+    max_searches: int = 2,
+) -> DimensionResult:
+    """Search a single dimension using available sources."""
+    result = DimensionResult(dimension)
+    
+    try:
+        settings = get_settings()
+        all_results = []
+        
+        # Try Tavily first
+        if settings.tavily_api_key and max_searches > 0:
+            tavily_results = await search_tavily(
+                query=dimension.search_query,
+                max_results=max_results,
+                freshness="any",
+            )
+            all_results.extend(tavily_results)
+        
+        # Supplement with DuckDuckGo if needed
+        if len(all_results) < max_results:
+            ddg_results = await search_duckduckgo(
+                query=dimension.search_query,
+                max_results=max_results - len(all_results),
+            )
+            all_results.extend(ddg_results)
+        
+        # Light reranking (without embeddings for speed)
+        if all_results:
+            ranked = await rerank_results(
+                query=dimension.search_query,
+                results=all_results,
+                temporal_urgency=0.5,
+                max_results=max_results,
+                use_embeddings=False,  # Skip for speed
+            )
+            result.results = ranked
+        
+    except Exception as e:
+        result.error = str(e)
+    
+    return result
+
+
+async def _synthesize_report_stream(
+    original_query: str,
+    plan: ResearchPlan,
+    dimension_results: list[DimensionResult],
+) -> AsyncIterator[str]:
+    """Stream the synthesis of the final report."""
+    
+    # Build context from all dimension results
+    context_parts = []
+    all_sources = []
+    source_index = 1
+    
+    for dr in dimension_results:
+        if dr.results:
+            context_parts.append(f"\n## {dr.dimension.name}\n")
+            for r in dr.results:
+                context_parts.append(
+                    f"[{source_index}] {r.get('title', 'Untitled')}\n"
+                    f"   URL: {r.get('url', '')}\n"
+                    f"   Content: {r.get('content', '')[:400]}...\n"
+                )
+                all_sources.append({
+                    "index": source_index,
+                    "title": r.get("title", ""),
+                    "url": r.get("url", ""),
+                })
+                source_index += 1
+    
+    context = "\n".join(context_parts)
+    
+    # Build synthesis prompt
+    prompt = f"""You are a research analyst. Create a comprehensive research report based on the gathered information.
+
+ORIGINAL QUERY: {original_query}
+REFINED QUERY: {plan.refined_query}
+
+RESEARCH DIMENSIONS:
+{', '.join(d.name for d in plan.dimensions)}
+
+GATHERED INFORMATION:
+{context}
+
+INSTRUCTIONS:
+1. Write a comprehensive research report in Markdown format
+2. Start with an Executive Summary (2-3 paragraphs)
+3. Create a section for each research dimension
+4. Use citations [1], [2], etc. to reference sources
+5. Include a Conclusion section
+6. Be thorough but concise
+7. Write in the same language as the query
+8. Use headers (##) to organize sections
+
+Generate the report:"""
+
+    messages = [
+        {"role": "system", "content": "You are a research analyst creating detailed reports."},
+        {"role": "user", "content": prompt},
+    ]
+    
+    try:
+        async for chunk in generate_completion_stream(messages, temperature=0.4):
+            yield chunk
+        
+        # Append sources at the end
+        yield "\n\n---\n\n## Sources\n\n"
+        for src in all_sources:
+            yield f"[{src['index']}] [{src['title']}]({src['url']})\n"
+            
+    except Exception as e:
+        yield f"\n\n**Error generating report:** {e}"
+
+
+def _sse_event(event_type: str, data: dict) -> str:
+    """Format an SSE event."""
+    payload = {"type": event_type, **data}
+    return f"data: {json.dumps(payload)}\n\n"

app/agents/planner.py

@@ -0,0 +1,133 @@
+"""Research Planner Agent.
+
+Decomposes complex queries into multiple research dimensions.
+"""
+
+import json
+from typing import Optional
+
+from pydantic import BaseModel, Field
+
+from app.agents.llm_client import generate_completion
+from app.config import get_settings
+
+
+class ResearchDimension(BaseModel):
+    """A single dimension/aspect to research."""
+    
+    name: str = Field(..., description="Short name for this dimension")
+    description: str = Field(..., description="What this dimension covers")
+    search_query: str = Field(..., description="Optimized search query for this dimension")
+    priority: int = Field(default=1, ge=1, le=3, description="1=high, 2=medium, 3=low")
+
+
+class ResearchPlan(BaseModel):
+    """Complete research plan with all dimensions."""
+    
+    original_query: str
+    refined_query: str = Field(..., description="Clarified version of the query")
+    dimensions: list[ResearchDimension]
+    estimated_sources: int = Field(default=20)
+
+
+PLANNER_PROMPT = """You are a research planning assistant. Your job is to decompose a complex query into multiple research dimensions.
+
+USER QUERY: {query}
+
+INSTRUCTIONS:
+1. Analyze the query and identify 2-6 key dimensions/aspects that need to be researched
+2. Each dimension should be distinct and cover a different angle
+3. Create an optimized search query for each dimension
+4. Assign priority (1=high, 2=medium, 3=low) based on relevance to the main query
+5. Respond ONLY with valid JSON, no other text
+
+OUTPUT FORMAT:
+{{
+    "refined_query": "A clearer version of the user's query",
+    "dimensions": [
+        {{
+            "name": "Short name",
+            "description": "What this covers",
+            "search_query": "Optimized search query",
+            "priority": 1
+        }}
+    ]
+}}
+
+Generate the research plan:"""
+
+
+async def create_research_plan(
+    query: str,
+    max_dimensions: int = 6,
+) -> ResearchPlan:
+    """
+    Create a research plan by decomposing a query into dimensions.
+    
+    Args:
+        query: The user's research query
+        max_dimensions: Maximum number of dimensions to generate
+        
+    Returns:
+        ResearchPlan with dimensions to investigate
+    """
+    settings = get_settings()
+    
+    messages = [
+        {"role": "system", "content": "You are a research planning assistant. Always respond with valid JSON only."},
+        {"role": "user", "content": PLANNER_PROMPT.format(query=query)},
+    ]
+    
+    try:
+        response = await generate_completion(messages, temperature=0.3)
+        
+        # Parse JSON response
+        # Try to extract JSON if there's extra text
+        json_start = response.find("{")
+        json_end = response.rfind("}") + 1
+        if json_start >= 0 and json_end > json_start:
+            response = response[json_start:json_end]
+        
+        data = json.loads(response)
+        
+        # Build dimensions
+        dimensions = []
+        for dim_data in data.get("dimensions", [])[:max_dimensions]:
+            dimensions.append(ResearchDimension(
+                name=dim_data.get("name", "Unknown"),
+                description=dim_data.get("description", ""),
+                search_query=dim_data.get("search_query", query),
+                priority=dim_data.get("priority", 2),
+            ))
+        
+        # Sort by priority
+        dimensions.sort(key=lambda d: d.priority)
+        
+        return ResearchPlan(
+            original_query=query,
+            refined_query=data.get("refined_query", query),
+            dimensions=dimensions,
+            estimated_sources=len(dimensions) * 5,
+        )
+        
+    except (json.JSONDecodeError, KeyError) as e:
+        # Fallback: create a simple 2-dimension plan
+        return ResearchPlan(
+            original_query=query,
+            refined_query=query,
+            dimensions=[
+                ResearchDimension(
+                    name="Main Research",
+                    description=f"Primary research on: {query}",
+                    search_query=query,
+                    priority=1,
+                ),
+                ResearchDimension(
+                    name="Background",
+                    description=f"Background and context for: {query}",
+                    search_query=f"{query} background overview",
+                    priority=2,
+                ),
+            ],
+            estimated_sources=10,
+        )

app/api/routes/search.py

@@ -14,6 +14,7 @@ from app.api.schemas import (
     TemporalContext,
     Citation,
     ErrorResponse,
+    DeepResearchRequest,
 )
 from app.config import get_settings
 from app.temporal.intent_detector import detect_temporal_intent
@@ -252,3 +253,38 @@ async def search_stream(request: SearchRequest):
             "X-Accel-Buffering": "no",
         },
     )
+
+
+# === Deep Research Endpoints ===
+
+@router.post(
+    "/research/deep",
+    summary="Deep research with multi-dimensional analysis",
+    description="Decompose a query into dimensions, search each in parallel, and generate a comprehensive report.",
+)
+async def deep_research(request: DeepResearchRequest):
+    """
+    Run deep research with streaming progress updates.
+    
+    Returns SSE events:
+    - plan_ready: Research plan with dimensions
+    - dimension_start/complete: Progress per dimension
+    - report_chunk: Streaming report content
+    - done: Final summary
+    """
+    from app.agents.deep_research import run_deep_research
+    
+    return StreamingResponse(
+        run_deep_research(
+            query=request.query,
+            max_dimensions=request.max_dimensions,
+            max_sources_per_dim=request.max_sources_per_dim,
+            max_total_searches=request.max_total_searches,
+        ),
+        media_type="text/event-stream",
+        headers={
+            "Cache-Control": "no-cache",
+            "Connection": "keep-alive",
+            "X-Accel-Buffering": "no",
+        },
+    )

app/api/schemas.py

@@ -110,3 +110,29 @@ class ErrorResponse(BaseModel):
     
     error: str = Field(..., description="Error message")
     detail: str | None = Field(default=None, description="Detailed error information")
+
+
+# === Deep Research Models ===
+
+class DeepResearchRequest(BaseModel):
+    """Deep research request payload."""
+    
+    query: str = Field(..., min_length=1, max_length=2000, description="Research query")
+    max_dimensions: int = Field(
+        default=5,
+        ge=2,
+        le=8,
+        description="Maximum research dimensions to explore"
+    )
+    max_sources_per_dim: int = Field(
+        default=5,
+        ge=1,
+        le=10,
+        description="Maximum sources per dimension"
+    )
+    max_total_searches: int = Field(
+        default=20,
+        ge=5,
+        le=30,
+        description="Maximum total API searches"
+    )

app/config.py

@@ -38,6 +38,11 @@ class Settings(BaseSettings):
     max_search_results: int = 20
     max_final_results: int = 10
     
+    # Deep Research Settings
+    max_research_dimensions: int = 6
+    max_tavily_calls_per_research: int = 20
+    deep_research_model: str | None = None  # Use main model if None
+    
     @property
     def llm_api_key(self) -> str:
         """Get the appropriate API key based on provider."""