Deep Research AI Agents: Complete Guide to Autonomous Research Systems 2026

from dataclasses import dataclass
from typing import List, Optional
from enum import Enum

class TaskType(Enum):
    BACKGROUND = "background"
    DEFINITION = "definition"
    CURRENT_STATE = "current_state"
    TECHNICAL = "technical"
    COMPARISON = "comparison"
    FUTURE = "future_outlook"
    CASE_STUDY = "case_study"

@dataclass
class ResearchTask:
    """Represents a single research sub-task."""
    id: str
    description: str
    task_type: TaskType
    search_terms: List[str]
    priority: int  # 1-5, higher = more important
    dependencies: List[str]  # task_ids that must complete first
    estimated_sources: int
    depth: str  # "shallow", "medium", "deep"
    
@dataclass
class ResearchPlan:
    """Complete research plan with task DAG."""
    query: str
    tasks: List[ResearchTask]
    estimated_duration: int  # seconds
    depth_level: str
    
    def get_executable_tasks(self, completed: set[str]) -> List[ResearchTask]:
        """Return tasks whose dependencies are satisfied."""
        return [
            task for task in self.tasks
            if task.id not in completed 
            and all(dep in completed for dep in task.dependencies)
        ]

async def create_research_plan(
    self, 
    query: str, 
    depth: str = "comprehensive"
) -> ResearchPlan:
    """Generate structured research plan from query."""
    
    system_prompt = """You are a research planning expert. Given a query, create a 
    comprehensive research plan with 6-12 sub-tasks covering:
    1. Background/definitions
    2. Current state and key players
    3. Technical details and mechanisms
    4. Comparative analysis (if applicable)
    5. Challenges and limitations
    6. Future outlook and trends
    7. Real-world applications/case studies
    
    Return JSON with tasks array containing: id, description, task_type, 
    search_terms (3-5 per task), priority (1-5), dependencies (task ids), 
    estimated_sources (5-20), depth (shallow/medium/deep).
    
    Task dependencies should form a DAG - no cycles."""
    
    user_prompt = f"""Create research plan for: "{query}"
    Depth: {depth}
    Target: 8-10 tasks covering all major aspects"""
    
    response = await self.llm.complete(
        system=system_prompt,
        user=user_prompt,
        response_format={"type": "json_object"},
        temperature=0.3
    )
    
    plan_dict = json.loads(response)
    tasks = [ResearchTask(**t) for t in plan_dict["tasks"]]
    
    return ResearchPlan(
        query=query,
        tasks=tasks,
        estimated_duration=self._estimate_duration(tasks),
        depth_level=depth
    )
    
def _estimate_duration(self, tasks: List[ResearchTask]) -> int:
    """Estimate total research time in seconds."""
    # Parallel execution model with 3 concurrent tasks
    max_depth = self._calculate_dag_depth(tasks)
    avg_task_time = 40  # seconds per task
    return max_depth * avg_task_time

from abc import ABC, abstractmethod

class SearchBackend(ABC):
    """Abstract base for search providers."""
    
    @abstractmethod
    async def search(
        self, 
        query: str, 
        num_results: int = 10,
        **kwargs
    ) -> List[SearchResult]:
        pass

@dataclass
class SearchResult:
    url: str
    title: str
    snippet: str
    published_date: Optional[str]
    source_domain: str
    score: float

class MultiSearchExecutor:
    """Executes searches across multiple backends with fallback."""
    
    def __init__(self, config: SearchConfig):
        self.backends = {
            "tavily": TavilyBackend(config.tavily_api_key),
            "serper": SerperBackend(config.serper_api_key),
            "brave": BraveBackend(config.brave_api_key),
        }
        self.primary = config.primary_backend
        
    async def search(
        self, 
        task: ResearchTask,
        num_results: int = 15
    ) -> List[SearchResult]:
        """Execute search with fallback on failure."""
        
        results = []
        for query in task.search_terms:
            try:
                backend_results = await self.backends[self.primary].search(
                    query=query,
                    num_results=num_results // len(task.search_terms)
                )
                results.extend(backend_results)
            except Exception as e:
                logger.warning(f"Primary search failed: {e}, trying fallback")
                # Fallback to alternative backend
                for name, backend in self.backends.items():
                    if name != self.primary:
                        try:
                            results.extend(
                                await backend.search(query, num_results)
                            )
                            break
                        except Exception:
                            continue
        
        # Deduplicate by URL
        seen = set()
        unique_results = []
        for r in results:
            if r.url not in seen:
                seen.add(r.url)
                unique_results.append(r)
        
        return unique_results[:num_results]

from bs4 import BeautifulSoup
from trafilatura import extract
import asyncio
import aiohttp

class ContentExtractor:
    """Extract clean content from web pages."""
    
    def __init__(self):
        self.timeout = aiohttp.ClientTimeout(total=15)
        
    async def extract_content(
        self, 
        url: str
    ) -> Optional[ExtractedContent]:
        """Fetch and extract main content from URL."""
        
        try:
            async with aiohttp.ClientSession(timeout=self.timeout) as session:
                async with session.get(
                    url, 
                    headers={"User-Agent": "ResearchBot/1.0"}
                ) as response:
                    if response.status != 200:
                        return None
                    
                    html = await response.text()
                    
            # Use trafilatura for main content extraction
            text = extract(
                html,
                include_comments=False,
                include_tables=True,
                include_images=False
            )
            
            if not text or len(text) < 200:
                return None
            
            # Extract metadata
            soup = BeautifulSoup(html, 'html.parser')
            title = soup.find('title')
            meta_desc = soup.find('meta', attrs={'name': 'description'})
            
            return ExtractedContent(
                url=url,
                title=title.text if title else "",
                text=text,
                description=meta_desc.get('content') if meta_desc else "",
                word_count=len(text.split()),
                extracted_at=datetime.now()
            )
            
        except asyncio.TimeoutError:
            logger.warning(f"Timeout extracting {url}")
            return None
        except Exception as e:
            logger.error(f"Error extracting {url}: {e}")
            return None
    
    async def batch_extract(
        self, 
        urls: List[str], 
        max_concurrent: int = 5
    ) -> List[ExtractedContent]:
        """Extract content from multiple URLs concurrently."""
        
        semaphore = asyncio.Semaphore(max_concurrent)
        
        async def extract_with_limit(url):
            async with semaphore:
                return await self.extract_content(url)
        
        results = await asyncio.gather(
            *[extract_with_limit(url) for url in urls],
            return_exceptions=True
        )
        
        return [r for r in results if isinstance(r, ExtractedContent)]

from typing import List, Dict, Optional
from dataclasses import dataclass, field
from datetime import datetime
import asyncio

@dataclass
class Finding:
    """A single research finding with source."""
    content: str
    source_url: str
    source_title: str
    relevance_score: float
    extracted_at: datetime
    task_id: str

@dataclass
class ResearchReport:
    """Final research output."""
    query: str
    summary: str
    sections: List[Dict[str, str]]
    findings: List[Finding]
    sources: List[Source]
    metadata: Dict
    generated_at: datetime

class DeepResearchAgent:
    """Complete deep research agent implementation."""

    def __init__(self, config: ResearchConfig):
        self.config = config
        self.llm = create_llm(config.model)
        self.search_executor = MultiSearchExecutor(config.search_config)
        self.content_extractor = ContentExtractor()
        self.source_evaluator = SourceEvaluator()
        self.vector_store = VectorStore(config.vector_db)
        
        self.findings: List[Finding] = []
        self.sources: List[Source] = []
        self.completed_tasks: set[str] = set()

    async def research(
        self, 
        query: str, 
        depth: str = "comprehensive",
        max_time: int = 600  # 10 minutes
    ) -> ResearchReport:
        """Execute full research pipeline."""
        
        start_time = datetime.now()
        
        # Phase 1: Planning
        logger.info(f"Creating research plan for: {query}")
        self.research_plan = await self.create_research_plan(query, depth)
        logger.info(
            f"Generated {len(self.research_plan.tasks)} tasks, "
            f"estimated {self.research_plan.estimated_duration}s"
        )
        
        # Phase 2: Iterative execution
        iteration = 0
        max_iterations = 20
        
        while len(self.completed_tasks) < len(self.research_plan.tasks):
            if iteration >= max_iterations:
                logger.warning("Max iterations reached")
                break
            
            if (datetime.now() - start_time).seconds > max_time:
                logger.warning("Max time reached")
                break
            
            iteration += 1
            
            # Get executable tasks (dependencies satisfied)
            executable = self.research_plan.get_executable_tasks(
                self.completed_tasks
            )
            
            if not executable:
                logger.error("No executable tasks but plan incomplete - circular dependency?")
                break
            
            # Execute up to 3 tasks in parallel
            batch = executable[:3]
            logger.info(
                f"Iteration {iteration}: Executing {len(batch)} tasks in parallel"
            )
            
            results = await asyncio.gather(
                *[self.execute_task(task) for task in batch],
                return_exceptions=True
            )
            
            for task, result in zip(batch, results):
                if isinstance(result, Exception):
                    logger.error(f"Task {task.id} failed: {result}")
                else:
                    self.completed_tasks.add(task.id)
                    logger.info(f"Completed task: {task.id}")
        
        # Phase 3: Synthesis
        logger.info("Starting synthesis phase")
        synthesis = await self.synthesize_findings()
        
        # Phase 4: Report generation
        report = await self.generate_report(query, synthesis)
        
        logger.info(
            f"Research complete. Found {len(self.findings)} findings "
            f"from {len(self.sources)} sources"
        )
        
        return report
    
    async def execute_task(self, task: ResearchTask) -> None:
        """Execute a single research task."""
        
        # Step 1: Search
        search_results = await self.search_executor.search(
            task, 
            num_results=task.estimated_sources
        )
        logger.info(f"Task {task.id}: Found {len(search_results)} results")
        
        # Step 2: Content extraction
        urls = [r.url for r in search_results]
        extracted = await self.content_extractor.batch_extract(urls)
        logger.info(f"Task {task.id}: Extracted {len(extracted)} pages")
        
        # Step 3: Source evaluation
        validated_sources = []
        for content in extracted:
            eval_result = await self.source_evaluator.evaluate(
                url=content.url,
                title=content.title,
                text=content.text[:1000],  # First 1000 chars for eval
                query=task.description
            )
            
            if eval_result.final_score >= self.config.min_source_score:
                validated_sources.append(
                    Source(
                        url=content.url,
                        title=content.title,
                        content=content.text,
                        score=eval_result.final_score,
                        evaluation=eval_result
                    )
                )
        
        logger.info(
            f"Task {task.id}: Validated {len(validated_sources)} sources"
        )
        self.sources.extend(validated_sources)
        
        # Step 4: Extract findings using LLM
        for source in validated_sources:
            findings = await self.extract_findings(source, task)
            self.findings.extend(findings)
        
        # Step 5: Store in vector DB for synthesis
        await self.vector_store.add_documents([
            {"text": f.content, "metadata": {"task_id": task.id, "url": f.source_url}}
            for f in self.findings
            if f.task_id == task.id
        ])
    
    async def extract_findings(
        self, 
        source: Source, 
        task: ResearchTask
    ) -> List[Finding]:
        """Extract relevant findings from a source."""
        
        prompt = f"""Extract 1-3 key findings from this source that are relevant 
        to the research task: "{task.description}"

        Source: {source.title}
        Content: {source.content[:3000]}

        Return JSON array with objects containing:
        - content: The finding (1-2 sentences)
        - relevance: Score 0-1 indicating relevance to task

        Focus on factual claims, data points, expert opinions, and insights."""
        
        response = await self.llm.complete(
            prompt, 
            response_format={"type": "json_object"}
        )
        
        findings_data = json.loads(response).get("findings", [])
        
        return [
            Finding(
                content=f["content"],
                source_url=source.url,
                source_title=source.title,
                relevance_score=f["relevance"],
                extracted_at=datetime.now(),
                task_id=task.id
            )
            for f in findings_data
            if f["relevance"] >= 0.6
        ]
    
    async def synthesize_findings(self) -> Dict:
        """Synthesize all findings into coherent narrative."""
        
        # Group findings by task
        findings_by_task = {}
        for task in self.research_plan.tasks:
            findings_by_task[task.id] = [
                f for f in self.findings if f.task_id == task.id
            ]
        
        # Generate section for each task
        sections = []
        for task in self.research_plan.tasks:
            task_findings = findings_by_task.get(task.id, [])
            
            if not task_findings:
                continue
            
            findings_text = "\n".join([
                f"- {f.content} (Source: {f.source_title})"
                for f in task_findings[:10]  # Top 10 findings per task
            ])
            
            prompt = f"""Synthesize these findings into a coherent section about: 
            {task.description}

            Findings:
            {findings_text}

            Write 2-3 paragraphs that:
            1. Introduce the topic
            2. Present key findings with inline citations [1], [2], etc.
            3. Highlight any contradictions or gaps
            
            Use an informative, objective tone."""
            
            section_text = await self.llm.complete(prompt)
            
            sections.append({
                "task_id": task.id,
                "title": task.description,
                "content": section_text,
                "findings": task_findings
            })
        
        return {
            "sections": sections,
            "total_findings": len(self.findings),
            "total_sources": len(self.sources)
        }
    
    async def generate_report(
        self, 
        query: str, 
        synthesis: Dict
    ) -> ResearchReport:
        """Generate final research report."""
        
        # Create executive summary
        all_section_content = "\n\n".join([
            s["content"] for s in synthesis["sections"]
        ])
        
        summary_prompt = f"""Create an executive summary (150-200 words) for this 
        research on: "{query}"

        Full report content:
        {all_section_content[:5000]}

        Summary should highlight the most important findings and conclusions."""
        
        summary = await self.llm.complete(summary_prompt)
        
        return ResearchReport(
            query=query,
            summary=summary,
            sections=synthesis["sections"],
            findings=self.findings,
            sources=self.sources,
            metadata={
                "total_sources": len(self.sources),
                "total_findings": len(self.findings),
                "tasks_completed": len(self.completed_tasks),
                "tasks_planned": len(self.research_plan.tasks),
                "depth": self.research_plan.depth_level
            },
            generated_at=datetime.now()
        )

### Advanced Source Evaluation System

Source quality directly impacts research output quality. A sophisticated evaluator considers multiple dimensions:

```mermaid
flowchart LR
    URL[Source URL] --> D[Domain Analysis]
    URL --> F[Freshness Check]
    URL --> R[Relevance Scoring]
    URL --> RF[Red Flag Detection]
    URL --> B[Bias Detection]
    
    D --> Score{Weighted\nScoring}
    F --> Score
    R --> Score
    RF --> Score
    B --> Score
    
    Score -->|>= 70| Accept[Use Source]
    Score -->|< 70| Review{Manual\nReview?}
    Review -->|High Priority| Manual[Flag for Review]
    Review -->|Low Priority| Reject[Discard]
    
    style Accept fill:#50C878,stroke:#2D7A4A,stroke-width:2px,color:#fff
    style Reject fill:#E74C3C,stroke:#922B21,stroke-width:2px,color:#fff
    style Score fill:#F39C12,stroke:#CA6F1E,stroke-width:2px,color:#fff

from urllib.parse import urlparse
from datetime import datetime, timedelta
from typing import List, Dict
import re

@dataclass
class SourceEvaluation:
    url: str
    domain_score: float
    freshness_score: float
    relevance_score: float
    authority_score: float
    bias_score: float  # 0 = heavily biased, 100 = neutral
    red_flags: List[str]
    final_score: float
    recommendation: str  # "accept", "review", "reject"
    reasoning: str

class SourceEvaluator:
    """Multi-dimensional source quality evaluation."""
    
    # Domain reputation database (expandable)

    TRUSTED_DOMAINS = {
        # Academic and research
        "arxiv.org": 98, "pubmed.ncbi.nlm.nih.gov": 98, "scholar.google.com": 95,
        "nature.com": 97, "science.org": 97, "cell.com": 96,
        "ieee.org": 95, "acm.org": 95, "springer.com": 92,
        
        # News and media (tier 1)
        "reuters.com": 92, "apnews.com": 92, "bbc.com": 90,
        "bloomberg.com": 90, "wsj.com": 89, "ft.com": 89,
        
        # Government
        ".gov": 95, ".edu": 88,
        
        # Technical documentation
        "github.com": 85, "gitlab.com": 85,
        "stackoverflow.com": 82, "docs.python.org": 90,
        
        # Industry analysis
        "gartner.com": 88, "forrester.com": 87, "mckinsey.com": 86,
        
        # News and media (tier 2)
        "theguardian.com": 85, "nytimes.com": 85, "economist.com": 87,
        "techcrunch.com": 75, "wired.com": 78, "arstechnica.com": 80,
        
        # Wikipedia (useful but requires verification)
        "wikipedia.org": 70, "wikimedia.org": 70,
    }
    
    # Suspicious TLDs that require extra scrutiny
    SUSPICIOUS_TLDS = {
        ".xyz", ".top", ".click", ".loan", ".work", ".gq", ".cf", ".ml"
    }
    
    # Content quality indicators
    QUALITY_INDICATORS = [
        "peer-reviewed", "published in", "according to", "study found",
        "research shows", "data indicates", "analysis reveals", "experts say"
    ]
    
    # Bias indicators
    BIAS_INDICATORS = {
        "strong_left": ["socialist", "progressive", "liberal", "leftist"],
        "strong_right": ["conservative", "right-wing", "patriot"],
        "sensational": ["BREAKING", "SHOCKING", "UNBELIEVABLE", "You won't believe"],
        "opinion": ["I think", "in my opinion", "I believe", "personally"]
    }

    async def evaluate(
        self, 
        url: str, 
        title: str, 
        text: str, 
        query: str
    ) -> SourceEvaluation:
        """Comprehensive source evaluation."""
        
        domain = urlparse(url).netloc
        
        # Component scores
        domain_score = self.get_domain_credibility(url)
        freshness_score = await self.check_freshness(url, text)
        relevance_score = self.calculate_relevance(text, query)
        authority_score = self.check_authority_signals(text)
        bias_score = self.detect_bias(text, title)
        red_flags = self.check_red_flags(url, title, text)
        
        # Weighted final score
        weights = {
            "domain": 0.30,
            "freshness": 0.15,
            "relevance": 0.25,
            "authority": 0.20,
            "bias": 0.10
        }
        
        final_score = (
            domain_score * weights["domain"] +
            freshness_score * weights["freshness"] +
            relevance_score * weights["relevance"] +
            authority_score * weights["authority"] +
            bias_score * weights["bias"]
        )
        
        # Red flags penalty
        final_score -= len(red_flags) * 10
        final_score = max(0, min(100, final_score))
        
        # Recommendation logic
        if final_score >= 75 and not red_flags:
            recommendation = "accept"
        elif final_score >= 60:
            recommendation = "review"
        else:
            recommendation = "reject"
        
        reasoning = self._generate_reasoning(
            domain_score, freshness_score, relevance_score, 
            authority_score, bias_score, red_flags
        )
        
        return SourceEvaluation(
            url=url,
            domain_score=domain_score,
            freshness_score=freshness_score,
            relevance_score=relevance_score,
            authority_score=authority_score,
            bias_score=bias_score,
            red_flags=red_flags,
            final_score=final_score,
            recommendation=recommendation,
            reasoning=reasoning
        )

    def get_domain_credibility(self, url: str) -> float:
        """Score domain based on reputation database."""
        domain = urlparse(url).netloc.lower()
        
        # Exact match
        if domain in self.TRUSTED_DOMAINS:
            return self.TRUSTED_DOMAINS[domain]
        
        # TLD match (e.g., .gov, .edu)
        for trusted, score in self.TRUSTED_DOMAINS.items():
            if trusted.startswith('.') and domain.endswith(trusted):
                return score
        
        # Subdomain match (e.g., docs.python.org)
        for trusted, score in self.TRUSTED_DOMAINS.items():
            if domain.endswith(trusted):
                return score * 0.9  # Slight penalty for subdomain
        
        # Unknown domain - neutral score
        return 50.0
    
    async def check_freshness(self, url: str, text: str) -> float:
        """Score based on content recency."""
        
        # Try to extract date from content
        date_patterns = [
            r'(\d{4})-(\d{2})-(\d{2})',  # YYYY-MM-DD
            r'(\w+)\s+(\d{1,2}),\s+(\d{4})',  # Month DD, YYYY
            r'(\d{1,2})/(\d{1,2})/(\d{4})',  # MM/DD/YYYY
        ]
        
        dates_found = []
        for pattern in date_patterns:
            matches = re.findall(pattern, text[:1000])  # Check first 1000 chars
            for match in matches:
                try:
                    if len(match) == 3 and match[0].isdigit():
                        # Try to parse as date
                        year = int(match[0]) if len(match[0]) == 4 else int(match[2])
                        if 2020 <= year <= 2026:
                            dates_found.append(year)
                except:
                    pass
        
        if not dates_found:
            return 50.0  # No date found - neutral score
        
        most_recent_year = max(dates_found)
        current_year = datetime.now().year
        
        age = current_year - most_recent_year
        
        if age == 0:
            return 100.0  # Current year
        elif age == 1:
            return 90.0
        elif age == 2:
            return 75.0
        elif age <= 5:
            return 60.0
        else:
            return 30.0  # Very old content
    
    def calculate_relevance(self, text: str, query: str) -> float:
        """Calculate semantic relevance to query."""
        
        text_lower = text.lower()[:3000]  # First 3000 chars
        query_lower = query.lower()
        
        # Extract key terms from query (simple tokenization)
        query_terms = set(re.findall(r'\b\w{4,}\b', query_lower))
        
        # Count term occurrences
        matches = sum(1 for term in query_terms if term in text_lower)
        
        if not query_terms:
            return 50.0
        
        # Calculate match percentage
        match_ratio = matches / len(query_terms)
        
        # Bonus for query appearing verbatim
        if query_lower in text_lower:
            match_ratio += 0.2
        
        return min(100.0, match_ratio * 100 + 30)
    
    def check_authority_signals(self, text: str) -> float:
        """Check for authority and quality indicators."""
        
        text_lower = text.lower()[:2000]
        
        indicator_count = sum(
            1 for indicator in self.QUALITY_INDICATORS 
            if indicator in text_lower
        )
        
        # Check for citations/references
        has_references = any(
            marker in text_lower 
            for marker in ["references", "bibliography", "cited", "doi:"]
        )
        
        # Check for author credentials
        has_author = "author:" in text_lower or "by " in text_lower[:500]
        
        score = 50.0
        score += indicator_count * 10
        score += 15 if has_references else 0
        score += 10 if has_author else 0
        
        return min(100.0, score)
    
    def detect_bias(self, text: str, title: str) -> float:
        """Detect potential bias in content (100 = neutral, 0 = heavily biased)."""
        
        text_sample = (title + " " + text[:1000]).lower()
        
        bias_count = 0
        
        for category, indicators in self.BIAS_INDICATORS.items():
            for indicator in indicators:
                if indicator.lower() in text_sample:
                    bias_count += 1
        
        # All caps title (sensationalism)
        if title.isupper() and len(title) > 15:
            bias_count += 2
        
        # Excessive punctuation
        exclamation_count = text_sample.count('!')
        if exclamation_count > 3:
            bias_count += 1
        
        # Score: 100 = neutral, decreases with bias indicators
        return max(0, 100 - (bias_count * 15))
    
    def check_red_flags(
        self, 
        url: str, 
        title: str, 
        text: str
    ) -> List[str]:
        """Identify content that should be flagged."""
        
        flags = []
        domain = urlparse(url).netloc
        
        # Suspicious TLD
        if any(domain.endswith(tld) for tld in self.SUSPICIOUS_TLDS):
            flags.append("suspicious_domain")
        
        # Clickbait title
        if title.isupper() and len(title) > 20:
            flags.append("clickbait_title")
        
        # Too short
        if len(text) < 200:
            flags.append("insufficient_content")
        
        # Excessive ads/promotion
        promo_keywords = ["buy now", "limited time", "special offer", "click here"]
        if sum(1 for kw in promo_keywords if kw in text.lower()) >= 3:
            flags.append("promotional_content")
        
        # Paywalled (common indicators)
        if any(phrase in text.lower() for phrase in ["subscribe to continue", "members only", "sign up to read"]):
            flags.append("paywall_detected")
        
        return flags
    
    def _generate_reasoning(
        self, 
        domain: float, 
        freshness: float,
        relevance: float, 
        authority: float, 
        bias: float,
        flags: List[str]
    ) -> str:
        """Generate human-readable explanation."""
        
        parts = []
        
        if domain >= 90:
            parts.append("Highly trusted domain")
        elif domain >= 70:
            parts.append("Reputable domain")
        elif domain < 50:
            parts.append("Unknown or low-reputation domain")
        
        if freshness >= 90:
            parts.append("very recent content")
        elif freshness < 50:
            parts.append("outdated content")
        
        if relevance >= 80:
            parts.append("highly relevant to query")
        elif relevance < 60:
            parts.append("marginal relevance")
        
        if authority >= 80:
            parts.append("strong authority signals")
        
        if bias < 70:
            parts.append("potential bias detected")
        
        if flags:
            parts.append(f"red flags: {', '.join(flags)}")
        
        return "; ".join(parts) + "."

#!/usr/bin/env python3
"""Minimal Deep Research Agent with production patterns."""

import asyncio
import json
import logging
from dataclasses import dataclass, field
from typing import List, Dict, Optional
from openai import AsyncOpenAI
from tavily import TavilyClient

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@dataclass
class ResearchConfig:
    openai_model: str = "gpt-4o"
    tavily_api_key: str = ""
    max_sources: int = 30
    search_depth: str = "advanced"  # "basic" or "advanced"
    timeout: int = 600  # 10 minutes

@dataclass
class SimpleResearchResult:
    query: str
    summary: str
    detailed_sections: List[Dict[str, str]]
    sources: List[Dict[str, str]]
    metadata: Dict

class ProductionResearchAgent:
    """Production-ready minimal research agent."""
    
    def __init__(self, config: ResearchConfig):
        self.config = config
        self.openai_client = AsyncOpenAI()
        self.tavily_client = TavilyClient(api_key=config.tavily_api_key)
        
    async def research(self, query: str) -> SimpleResearchResult:
        """Execute full research pipeline with error handling."""
        
        try:
            # Phase 1: Create plan
            logger.info(f"Planning research for: {query}")
            plan = await self._create_plan(query)
            logger.info(f"Plan created with {len(plan['subtopics'])} subtopics")
            
            # Phase 2: Execute searches in parallel
            logger.info("Executing searches")
            search_results = await self._parallel_search(
                query, 
                plan['subtopics']
            )
            logger.info(f"Found {len(search_results)} sources")
            
            # Phase 3: Synthesize findings
            logger.info("Synthesizing findings")
            synthesis = await self._synthesize(query, search_results, plan)
            
            return SimpleResearchResult(
                query=query,
                summary=synthesis['summary'],
                detailed_sections=synthesis['sections'],
                sources=search_results,
                metadata={
                    'subtopics': len(plan['subtopics']),
                    'sources_found': len(search_results)
                }
            )
            
        except Exception as e:
            logger.error(f"Research failed: {e}", exc_info=True)
            raise
    
    async def _create_plan(self, query: str) -> Dict:
        """Generate structured research plan."""
        
        prompt = f"""Create a research plan for: "{query}"

Generate 5-8 subtopics that provide comprehensive coverage.
Each subtopic should be specific and answerable.

Return JSON:
{{
  "subtopics": ["subtopic 1", "subtopic 2", ...],
  "focus": "brief description of research focus"
}}"""
        
        response = await self.openai_client.chat.completions.create(
            model=self.config.openai_model,
            messages=[
                {"role": "system", "content": "You are a research planning expert."},
                {"role": "user", "content": prompt}
            ],
            response_format={"type": "json_object"},
            temperature=0.3
        )
        
        return json.loads(response.choices[0].message.content)
    
    async def _parallel_search(
        self, 
        main_query: str, 
        subtopics: List[str]
    ) -> List[Dict]:
        """Execute searches in parallel and aggregate results."""
        
        async def search_subtopic(subtopic: str):
            try:
                # Tavily search (synchronous, but run in executor)
                loop = asyncio.get_event_loop()
                response = await loop.run_in_executor(
                    None,
                    lambda: self.tavily_client.search(
                        query=f"{main_query} {subtopic}",
                        max_results=5,
                        search_depth=self.config.search_depth,
                        include_answer=True,
                        include_raw_content=True
                    )
                )
                
                return [{
                    "url": r["url"],
                    "title": r["title"],
                    "content": r.get("content", r.get("raw_content", "")),
                    "score": r.get("score", 0),
                    "subtopic": subtopic
                } for r in response.get("results", [])]
                
            except Exception as e:
                logger.warning(f"Search failed for '{subtopic}': {e}")
                return []
        
        # Execute all searches in parallel
        results = await asyncio.gather(
            *[search_subtopic(st) for st in subtopics],
            return_exceptions=True
        )
        
        # Flatten and deduplicate
        all_sources = []
        seen_urls = set()
        
        for result_list in results:
            if isinstance(result_list, Exception):
                continue
            for source in result_list:
                if source["url"] not in seen_urls:
                    seen_urls.add(source["url"])
                    all_sources.append(source)
        
        return all_sources[:self.config.max_sources]
    
    async def _synthesize(
        self, 
        query: str, 
        sources: List[Dict],
        plan: Dict
    ) -> Dict:
        """Synthesize findings into structured report."""
        
        # Group sources by subtopic
        sources_by_topic = {}
        for source in sources:
            topic = source.get("subtopic", "general")
            if topic not in sources_by_topic:
                sources_by_topic[topic] = []
            sources_by_topic[topic].append(source)
        
        # Generate section for each subtopic
        sections = []
        for subtopic, topic_sources in sources_by_topic.items():
            section = await self._generate_section(
                query, subtopic, topic_sources
            )
            sections.append(section)
        
        # Generate executive summary
        all_content = "\n\n".join([s["content"] for s in sections])
        summary = await self._generate_summary(query, all_content[:8000])
        
        return {
            "summary": summary,
            "sections": sections
        }
    
    async def _generate_section(
        self, 
        main_query: str,
        subtopic: str, 
        sources: List[Dict]
    ) -> Dict:
        """Generate narrative section from sources."""
        
        sources_text = "\n\n".join([
            f"Source: {s['title']}\nURL: {s['url']}\n{s['content'][:1000]}"
            for s in sources[:5]  # Top 5 sources per subtopic
        ])
        
        prompt = f"""Write a comprehensive section about: {subtopic}

Context: This is part of research on "{main_query}"

Available sources:
{sources_text}

Write 3-4 paragraphs that:
1. Introduce the subtopic
2. Present key findings with citations [1], [2], etc.
3. Synthesize information from multiple sources
4. Note any contradictions or gaps

Use an objective, informative tone. Cite sources inline."""
        
        response = await self.openai_client.chat.completions.create(
            model=self.config.openai_model,
            messages=[
                {"role": "system", "content": "You are a research synthesis expert."},
                {"role": "user", "content": prompt}
            ],
            temperature=0.4
        )
        
        return {
            "subtopic": subtopic,
            "content": response.choices[0].message.content,
            "source_count": len(sources)
        }
    
    async def _generate_summary(self, query: str, full_content: str) -> str:
        """Generate executive summary."""
        
        prompt = f"""Create a concise executive summary (200-250 words) for research on:
"{query}"

Full content:
{full_content}

Summary should:
- Highlight key findings
- Note important trends or patterns
- Remain objective and factual"""
        
        response = await self.openai_client.chat.completions.create(
            model=self.config.openai_model,
            messages=[
                {"role": "system", "content": "You are a research summarization expert."},
                {"role": "user", "content": prompt}
            ],
            temperature=0.3
        )
        
        return response.choices[0].message.content

# Usage example
async def main():
    config = ResearchConfig(
        tavily_api_key="your_api_key_here",
        max_sources=30
    )
    
    agent = ProductionResearchAgent(config)
    result = await agent.research("Latest developments in quantum computing")
    
    print("=" * 80)
    print(f"RESEARCH REPORT: {result.query}")
    print("=" * 80)
    print(f"\n{result.summary}\n")
    
    for section in result.detailed_sections:
        print(f"\n## {section['subtopic']}")
        print(f"{section['content']}\n")
    
    print(f"\nTotal sources: {len(result.sources)}")

if __name__ == "__main__":
    asyncio.run(main())

from tenacity import retry, stop_after_attempt, wait_exponential

class RateLimitedSearchExecutor:
    """Search executor with rate limiting."""
    
    def __init__(self, max_concurrent: int = 3):
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.rate_limiter = AsyncLimiter(10, 60)  # 10 requests per 60 seconds
    
    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=4, max=10)
    )
    async def search_with_retry(self, query: str) -> List[Dict]:
        async with self.semaphore:
            async with self.rate_limiter:
                return await self._do_search(query)

class CostTracker:
    """Track API costs across providers."""
    
    def __init__(self):
        self.costs = {"llm": 0.0, "search": 0.0}
        
        # Pricing per 1M tokens
        self.llm_pricing = {
            "gpt-4o": {"input": 5.0, "output": 15.0},
            "claude-3.5-sonnet": {"input": 3.0, "output": 15.0}
        }
        
        # Pricing per search
        self.search_pricing = {
            "tavily": 0.005,
            "serper": 0.002
        }
    
    def track_llm_call(
        self, 
        model: str, 
        input_tokens: int, 
        output_tokens: int
    ):
        pricing = self.llm_pricing[model]
        cost = (
            (input_tokens / 1_000_000) * pricing["input"] +
            (output_tokens / 1_000_000) * pricing["output"]
        )
        self.costs["llm"] += cost
        return cost
    
    def track_search(self, provider: str, count: int = 1):
        cost = self.search_pricing[provider] * count
        self.costs["search"] += cost
        return cost
    
    def get_total(self) -> float:
        return sum(self.costs.values())

import hashlib
import redis.asyncio as redis
import pickle

class ResearchCache:
    """Cache search results and LLM outputs."""
    
    def __init__(self, redis_url: str):
        self.redis = redis.from_url(redis_url)
    
    async def get_search(self, query: str) -> Optional[List[Dict]]:
        key = f"search:{self._hash(query)}"
        data = await self.redis.get(key)
        return pickle.loads(data) if data else None
    
    async def set_search(
        self, 
        query: str, 
        results: List[Dict],
        ttl: int = 3600  # 1 hour
    ):
        key = f"search:{self._hash(query)}"
        await self.redis.setex(key, ttl, pickle.dumps(results))
    
    def _hash(self, text: str) -> str:
        return hashlib.sha256(text.encode()).hexdigest()[:16]

from collections import Counter
from urllib.parse import urlparse

def ensure_source_diversity(
    sources: List[Source], 
    min_domains: int = 5
) -> tuple[bool, str]:
    """Ensure diversity across domains."""
    domains = {urlparse(s.url).netloc for s in sources}
    
    if len(domains) < min_domains:
        return False, f"Only {len(domains)} unique domains (minimum: {min_domains})"
    
    # Check for domain dominance
    domain_counts = Counter(urlparse(s.url).netloc for s in sources)
    max_count = max(domain_counts.values())
    if max_count / len(sources) > 0.4:  # No single domain should exceed 40%
        dominant_domain = domain_counts.most_common(1)[0][0]
        return False, f"Domain {dominant_domain} dominates with {max_count} sources"
    
    return True, "Source diversity acceptable"

async def verify_claim(
    claim: str, 
    sources: List[Source],
    llm_client
) -> VerificationResult:
    """Verify claim across multiple sources."""
    supporting = []
    opposing = []
    
    for source in sources:
        # Use LLM to check if source supports/opposes claim
        prompt = f"""Does this source support or oppose the following claim?

Claim: {claim}
Source: {source.content[:800]}

Return JSON: {{"stance": "support"|"oppose"|"neutral", "confidence": 0-1}}"""
        
        response = await llm_client.complete(prompt, response_format={"type": "json_object"})
        result = json.loads(response)
        
        if result["stance"] == "support" and result["confidence"] > 0.7:
            supporting.append(source)
        elif result["stance"] == "oppose" and result["confidence"] > 0.7:
            opposing.append(source)
    
    # Consensus rules
    if len(supporting) >= 3 and len(opposing) == 0:
        status = "verified"
    elif len(supporting) >= 2:
        status = "likely"
    elif len(supporting) > 0 and len(opposing) > 0:
        status = "conflicting"
    else:
        status = "unverified"
    
    return VerificationResult(
        claim=claim,
        status=status,
        supporting=supporting,
        opposing=opposing
    )

def qa_checklist(report: ResearchReport) -> Dict:
    """Run quality checks on generated report."""
    return {
        "min_sources": len(report.sources) >= 15,
        "min_word_count": len(report.summary.split()) >= 150,
        "has_citations": all(f.source_url for f in report.findings),
        "domain_diversity": len(set(urlparse(s.url).netloc for s in report.sources)) >= 5,
        "no_contradictions": check_internal_consistency(report),
        "avg_source_quality": sum(s.score for s in report.sources) / len(report.sources) >= 70
    }