Directory Structure

src/
  agents/
    planning_agent.py
    supervisor.py
    web_search_retriever.py
  tools/
    exa_tool.py
  utils/
    config.py
.env.example
.gitignore
.python-version
main.py
pyproject.toml
README.md

Files

File: src/agents/planning_agent.py

"""Planning Agent for generating Exa-optimized research subqueries."""

import json
import httpx
from typing import Dict, Any, List
from src.utils.config import Config


class PlanningAgent:
    """
    Agent that decomposes research queries into Exa-optimized subqueries.
    Uses OpenRouter with Gemini 2.5 Flash.
    """

    def __init__(self):
        """Initialize Planning Agent."""
        self.api_key = Config.OPENROUTER_API_KEY
        self.base_url = Config.OPENROUTER_BASE_URL
        self.model = Config.OPENROUTER_MODEL

        self.system_prompt = """Generate 8-12 comprehensive Exa-optimized subqueries for deep research on the topic.

For thorough coverage, create subqueries across multiple dimensions:
- Core concepts and definitions
- Latest developments and breakthroughs (recent news)
- Historical context and evolution
- Technical implementations and applications
- Expert opinions and analysis
- Academic research and papers
- Industry trends and market analysis
- Future implications and predictions
- Challenges and limitations
- Related technologies and comparisons

Consider the following when creating subqueries:
- Neural search formulation: Use natural language questions and descriptive phrases
- Domain filters: Suggest specific domains when relevant (e.g., arxiv.org for papers, news sites)
- Time periods: Specify time relevance (recent, past_week, past_month, past_year, any)
- Content types: Specify type when relevant (news, research paper, pdf, blog, etc.)
- Priority: Assign priority 1-5 (1=highest priority)

Each subquery should focus on a different aspect of the research topic to ensure comprehensive, multi-dimensional coverage.

Output valid JSON in this exact format:
{
  "subqueries": [
    {
      "query": "descriptive natural language query",
      "type": "auto|news|research paper|pdf|etc",
      "time_period": "recent|past_week|past_month|past_year|any",
      "include_domains": ["example.com"],
      "exclude_domains": ["example.org"],
      "priority": 1
    }
  ]
}

Notes:
- include_domains and exclude_domains are optional (can be null or omitted)
- type should be "auto" unless you have specific content type needs
- Ensure queries are diverse and cover different angles of the topic"""

    def plan(self, research_query: str) -> Dict[str, Any]:
        """
        Generate Exa-optimized subqueries for a research topic.

        Args:
            research_query: The main research question or topic

        Returns:
            Dictionary containing subqueries with optimization parameters
        """
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json",
        }

        payload = {
            "model": self.model,
            "messages": [
                {"role": "system", "content": self.system_prompt},
                {
                    "role": "user",
                    "content": f"Research topic: {research_query}\n\nGenerate Exa-optimized subqueries for comprehensive research on this topic.",
                },
            ],
            "temperature": 0.7,
            "max_tokens": 3000,
        }

        try:
            with httpx.Client(timeout=60.0) as client:
                response = client.post(
                    f"{self.base_url}/chat/completions",
                    json=payload,
                    headers=headers,
                )
                response.raise_for_status()
                result = response.json()

                # Extract the content from the response
                content = result["choices"][0]["message"]["content"]

                # Parse the JSON response
                # Handle potential markdown code blocks
                if "```json" in content:
                    content = content.split("```json")[1].split("```")[0].strip()
                elif "```" in content:
                    content = content.split("```")[1].split("```")[0].strip()

                subqueries_data = json.loads(content)

                return {
                    "status": "success",
                    "subqueries": subqueries_data.get("subqueries", []),
                    "research_query": research_query,
                }

        except httpx.HTTPError as e:
            return {
                "status": "error",
                "error": f"HTTP error: {str(e)}",
                "subqueries": [],
            }
        except json.JSONDecodeError as e:
            return {
                "status": "error",
                "error": f"JSON parsing error: {str(e)}",
                "subqueries": [],
            }
        except Exception as e:
            return {
                "status": "error",
                "error": f"Unexpected error: {str(e)}",
                "subqueries": [],
            }

    def execute(self, query: str) -> str:
        """
        Execute planning and return formatted results as a string.
        This method is called by the supervisor via tool execution.

        Args:
            query: Research query to plan for

        Returns:
            JSON string containing the planning results
        """
        result = self.plan(query)
        return json.dumps(result, indent=2)

File: src/agents/supervisor.py

"""Supervisor Agent using Minimax M2 with interleaved thinking."""

import anthropic
from typing import List, Dict, Any
from rich.console import Console
from src.utils.config import Config
from src.agents.planning_agent import PlanningAgent
from src.agents.web_search_retriever import WebSearchRetriever

# Initialize rich console
console = Console()


class SupervisorAgent:
    """
    Main supervisor agent that coordinates research workflow using Minimax M2.
    Implements interleaved thinking by preserving all content blocks in conversation history.
    """

    def __init__(self):
        """Initialize Supervisor Agent with Minimax M2."""
        self.client = anthropic.Anthropic(
            api_key=Config.MINIMAX_API_KEY,
            base_url=Config.MINIMAX_BASE_URL,
        )
        self.model = Config.MINIMAX_MODEL

        # Initialize sub-agents
        self.planning_agent = PlanningAgent()
        self.web_search_retriever = WebSearchRetriever()

        # Conversation history with interleaved thinking
        self.messages: List[Dict[str, Any]] = []

        self.system_prompt = """You are a deep research coordinator specializing in comprehensive, academic-quality research reports. Your goal is to produce thorough, well-structured, in-depth analysis.

You have access to the following tools:

1. planning_agent - Breaks down research queries into 8-12 Exa-optimized subqueries
   - Input: research_query (string)
   - Returns: JSON with optimized subqueries covering multiple dimensions

2. web_search_retriever - Searches the web using Exa and synthesizes findings
   - Input: research_query (string), subqueries_json (string)
   - Returns: Comprehensive organized findings with sources

Research Workflow:
1. Call planning_agent with the user's research query to generate comprehensive subqueries
2. Call web_search_retriever with the research query and subqueries to gather extensive information
3. Synthesize a COMPREHENSIVE research report (15-30 pages equivalent) with the following structure:

## Required Report Structure:

### Executive Summary (3-5 paragraphs)
   - Overview of research scope
   - Key findings summary
   - Main conclusions and implications

### Introduction (2-3 paragraphs)
   - Context and background
   - Research objectives
   - Methodology overview

### Key Findings (Multiple detailed sections organized by theme)
   - Each major theme gets its own section with subsections
   - Include data, statistics, expert opinions
   - Cite sources inline with URLs
   - Provide examples and case studies

### Detailed Analysis (Deep dive into each area)
   - Technical details and mechanisms
   - Historical context and evolution
   - Current state of the art
   - Comparisons and contrasts
   - Strengths and limitations

### Industry/Application Analysis (if relevant)
   - Real-world applications
   - Market trends and adoption
   - Key players and institutions
   - Success stories and challenges

### Future Implications and Trends
   - Emerging developments
   - Predictions and projections
   - Challenges ahead
   - Opportunities and potential

### Critical Analysis
   - Debates and controversies
   - Limitations and challenges
   - Alternative perspectives
   - Unanswered questions

### Conclusion
   - Summary of main points
   - Broader implications
   - Recommendations (if applicable)

### Sources and Citations
   - Comprehensive list of all sources with URLs
   - Organized by category or theme

## Quality Guidelines:
- Be EXTREMELY thorough and detailed - aim for 5-10x more content than a typical report
- Use specific data, statistics, and concrete examples throughout
- Quote experts and authoritative sources
- Explain technical concepts clearly
- Make connections across different aspects of the topic
- Maintain academic rigor and objectivity
- Use clear section headers and subsections
- Provide context and background for all major points
- Include both breadth (covering many aspects) and depth (detailed analysis)

## CRITICAL: Inline Citations Format
- **ALWAYS include inline citations** immediately after claims, data, or quotes
- Use markdown link format: `[descriptive text](URL)` for all citations
- Place citations right where information is used, not just at the end
- Examples:
  * "The market is projected to reach $47 billion by 2030 [according to Grand View Research](https://www.example.com/report)"
  * "As noted by [Nick Bostrom's research on AI safety](https://example.com/paper), superintelligence poses..."
  * "Studies show a 44% growth rate [Statista Market Analysis](https://example.com/stats)"
- When citing statistics: include the source inline: "Growth rates of 44% [Source](URL)"
- When quoting experts: cite immediately: "According to [Expert Name](URL), '...'"
- Every factual claim, statistic, or data point MUST have an inline citation
- The final Sources section should be a comprehensive list, but inline citations are PRIMARY"""

        # Tool definitions for Anthropic format
        self.tools = [
            {
                "name": "planning_agent",
                "description": "Generates Exa-optimized subqueries for a research topic. Takes a research query and returns JSON with 3-5 subqueries optimized for neural search.",
                "input_schema": {
                    "type": "object",
                    "properties": {
                        "research_query": {
                            "type": "string",
                            "description": "The main research question or topic to plan for",
                        }
                    },
                    "required": ["research_query"],
                },
            },
            {
                "name": "web_search_retriever",
                "description": "Executes web searches using Exa API for provided subqueries and synthesizes findings. Returns organized research findings with sources.",
                "input_schema": {
                    "type": "object",
                    "properties": {
                        "research_query": {
                            "type": "string",
                            "description": "The original research query for context",
                        },
                        "subqueries_json": {
                            "type": "string",
                            "description": "JSON string containing subqueries from planning_agent",
                        },
                    },
                    "required": ["research_query", "subqueries_json"],
                },
            },
        ]

    def execute_tool(self, tool_name: str, tool_input: Dict[str, Any]) -> str:
        """
        Execute a tool and return its result.

        Args:
            tool_name: Name of the tool to execute
            tool_input: Input parameters for the tool

        Returns:
            Tool execution result as string
        """
        if tool_name == "planning_agent":
            research_query = tool_input.get("research_query", "")
            return self.planning_agent.execute(research_query)

        elif tool_name == "web_search_retriever":
            research_query = tool_input.get("research_query", "")
            subqueries_json = tool_input.get("subqueries_json", "")
            return self.web_search_retriever.retrieve(research_query, subqueries_json)

        else:
            return f"Error: Unknown tool '{tool_name}'"

    def research(self, query: str, max_iterations: int = 10) -> str:
        """
        Conduct research on a given query using Minimax M2 with interleaved thinking.

        Args:
            query: Research question or topic
            max_iterations: Maximum number of agent iterations

        Returns:
            Comprehensive research report
        """
        # Initialize conversation with user query
        self.messages = [
            {
                "role": "user",
                "content": query,
            }
        ]

        iteration = 0

        while iteration < max_iterations:
            iteration += 1

            try:
                # Call Minimax M2 with streaming for long requests
                console.print(f"[bold magenta][Iteration {iteration}][/bold magenta] [cyan]Calling Minimax M2...[/cyan]")

                with self.client.messages.stream(
                    model=self.model,
                    max_tokens=32000,
                    system=self.system_prompt,
                    messages=self.messages,
                    tools=self.tools,
                ) as stream:
                    for event in stream:
                        if hasattr(event, 'type') and event.type == 'content_block_start':
                            console.print("[green].[/green]", end="")

                    response = stream.get_final_message()
                    console.print()

                # CRITICAL: Append the COMPLETE response to message history
                # This preserves the interleaved thinking across turns
                assistant_message = {
                    "role": "assistant",
                    "content": response.content,  # Includes thinking, text, and tool_use blocks
                }
                self.messages.append(assistant_message)

                # Check stop reason
                if response.stop_reason == "end_turn":
                    # Model has finished - extract final response
                    final_text = self._extract_text_from_content(response.content)
                    return final_text

                elif response.stop_reason == "tool_use":
                    # Model wants to use tools - execute them
                    num_tools = len([b for b in response.content if hasattr(b, 'type') and b.type == 'tool_use'])
                    console.print(f"[bold blue][Tool execution][/bold blue] M2 requested [yellow]{num_tools}[/yellow] tool(s)")
                    tool_results = []

                    for content_block in response.content:
                        if content_block.type == "tool_use":
                            tool_name = content_block.name
                            tool_input = content_block.input
                            tool_use_id = content_block.id

                            console.print(f"[dim]  → Executing:[/dim] [cyan]{tool_name}[/cyan]")

                            # Execute the tool
                            result = self.execute_tool(tool_name, tool_input)

                            tool_results.append({
                                "type": "tool_result",
                                "tool_use_id": tool_use_id,
                                "content": result,
                            })

                    # Add tool results to conversation
                    self.messages.append({
                        "role": "user",
                        "content": tool_results,
                    })

                else:
                    # Unexpected stop reason
                    console.print(f"[bold red]⚠ Unexpected stop reason:[/bold red] {response.stop_reason}")
                    return f"Research stopped unexpectedly: {response.stop_reason}"

            except Exception as e:
                console.print(f"[bold red]✗ Error during research:[/bold red] {str(e)}")
                return f"Error during research: {str(e)}"

        console.print(f"[bold yellow]⚠ Research reached maximum iterations ({max_iterations}) without completion.[/bold yellow]")
        return "Research reached maximum iterations without completion."

    def _extract_text_from_content(self, content: List[Any]) -> str:
        """
        Extract text content from response content blocks.

        Args:
            content: List of content blocks from API response

        Returns:
            Combined text content
        """
        text_parts = []

        for block in content:
            if hasattr(block, "type") and block.type == "text":
                text_parts.append(block.text)

        return "\n\n".join(text_parts) if text_parts else "No text content in response."

    def get_conversation_history(self) -> List[Dict[str, Any]]:
        """
        Get the complete conversation history including thinking blocks.

        Returns:
            List of message dictionaries
        """
        return self.messages

File: src/agents/web_search_retriever.py

"""Web Search Retriever Agent using Exa API."""

import json
import httpx
from typing import Dict, Any, List
from src.tools.exa_tool import ExaTool
from src.utils.config import Config


class WebSearchRetriever:
    """
    Agent that executes Exa searches for provided subqueries and synthesizes findings.
    Uses OpenRouter with Gemini 2.5 Flash for synthesis.
    """

    def __init__(self):
        """Initialize Web Search Retriever."""
        self.exa = ExaTool()
        self.api_key = Config.OPENROUTER_API_KEY
        self.base_url = Config.OPENROUTER_BASE_URL
        self.model = Config.OPENROUTER_MODEL

        self.system_prompt = """You are a web search retrieval specialist.

Your job is to:
1. Execute Exa searches for each provided subquery
2. Use find_similar() on the best results to discover related content
3. Organize findings by relevance and topic
4. Extract key insights from the search results

Return structured results with:
- URLs and titles
- Summaries of key findings
- Relevant quotes/highlights
- How each source contributes to answering the research query

Be comprehensive but focused. Prioritize high-quality, authoritative sources."""

    def search_with_subqueries(self, subqueries: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
        """
        Execute Exa searches for all subqueries.

        Args:
            subqueries: List of subquery dictionaries from planning agent

        Returns:
            List of search results for each subquery
        """
        all_results = []

        for subquery in subqueries:
            query_text = subquery.get("query", "")
            content_type = subquery.get("type", "auto")
            time_period = subquery.get("time_period", "any")
            include_domains = subquery.get("include_domains")
            exclude_domains = subquery.get("exclude_domains")
            priority = subquery.get("priority", 3)

            # Map time period to date filters (approximate)
            start_date = None
            if time_period == "recent" or time_period == "past_week":
                # Exa uses ISO format dates
                start_date = "2025-11-17T00:00:00.000Z"  # Approximate for recent
            elif time_period == "past_month":
                start_date = "2025-10-24T00:00:00.000Z"
            elif time_period == "past_year":
                start_date = "2024-11-24T00:00:00.000Z"

            # Execute search with more results for deeper research
            search_results = self.exa.search(
                query=query_text,
                num_results=20 if priority <= 2 else 15,  # Increased for deeper research
                start_published_date=start_date,
                include_domains=include_domains,
                exclude_domains=exclude_domains,
                type=content_type,
            )

            # Format results
            formatted_results = self.exa.format_results(search_results)

            # Find similar content for more queries (priority <= 3 instead of <= 2)
            similar_results = []
            if formatted_results and priority <= 3:
                top_url = formatted_results[0].get("url")
                if top_url:
                    similar_response = self.exa.find_similar(
                        url=top_url,
                        num_results=5,  # Increased from 3 to 5
                    )
                    similar_results = self.exa.format_results(similar_response)

            all_results.append({
                "subquery": query_text,
                "priority": priority,
                "results": formatted_results,
                "similar_results": similar_results,
            })

        return all_results

    def synthesize_findings(
        self,
        research_query: str,
        search_results: List[Dict[str, Any]]
    ) -> str:
        """
        Use LLM to synthesize search results into organized findings.

        Args:
            research_query: Original research query
            search_results: Results from Exa searches

        Returns:
            Synthesized findings as a string
        """
        # Prepare context from search results
        context_parts = []
        for result_set in search_results:
            subquery = result_set.get("subquery", "")
            results = result_set.get("results", [])

            context_parts.append(f"\n## Subquery: {subquery}")

            for i, result in enumerate(results[:10], 1):  # Top 10 per subquery for deeper research
                title = result.get("title", "No title")
                url = result.get("url", "")
                highlights = result.get("highlights", [])
                text_excerpt = result.get("text", "")[:1000]  # Increased to 1000 chars for more context

                context_parts.append(f"\n### Result {i}: {title}")
                context_parts.append(f"URL: {url}")
                if highlights:
                    context_parts.append(f"Highlights: {', '.join(highlights[:5])}")  # More highlights
                if text_excerpt:
                    context_parts.append(f"Excerpt: {text_excerpt}...")

        context = "\n".join(context_parts)

        # Create synthesis request
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json",
        }

        payload = {
            "model": self.model,
            "messages": [
                {"role": "system", "content": self.system_prompt},
                {
                    "role": "user",
                    "content": f"""Research Query: {research_query}

Search Results:
{context}

Organize these findings into a comprehensive, detailed summary. Include:
1. Key findings organized by topic/theme with extensive details
2. Important sources with URLs and brief descriptions
3. Relevant quotes and highlights with context
4. How the sources address the research query
5. Connections and patterns across different sources
6. Notable experts, institutions, or authoritative voices
7. Data, statistics, or concrete examples when available

Be thorough and detailed - this will feed into a comprehensive research report.""",
                },
            ],
            "temperature": 0.5,
            "max_tokens": 6000,  # Increased for more comprehensive synthesis
        }

        try:
            with httpx.Client(timeout=60.0) as client:
                response = client.post(
                    f"{self.base_url}/chat/completions",
                    json=payload,
                    headers=headers,
                )
                response.raise_for_status()
                result = response.json()
                return result["choices"][0]["message"]["content"]

        except Exception as e:
            return f"Error synthesizing findings: {str(e)}"

    def retrieve(self, research_query: str, subqueries_json: str) -> str:
        """
        Execute web search retrieval for given subqueries.
        This method is called by the supervisor via tool execution.

        Args:
            research_query: Original research query
            subqueries_json: JSON string containing subqueries from planning agent

        Returns:
            Synthesized findings as a string
        """
        try:
            # Parse subqueries
            subqueries_data = json.loads(subqueries_json)
            subqueries = subqueries_data.get("subqueries", [])

            if not subqueries:
                return "Error: No subqueries provided"

            # Execute searches
            search_results = self.search_with_subqueries(subqueries)

            # Synthesize findings
            findings = self.synthesize_findings(research_query, search_results)

            return findings

        except json.JSONDecodeError as e:
            return f"Error parsing subqueries: {str(e)}"
        except Exception as e:
            return f"Error in retrieval: {str(e)}"

File: src/tools/exa_tool.py

"""Exa API wrapper for neural web search."""

import httpx
from typing import List, Dict, Any, Optional
from src.utils.config import Config


class ExaTool:
    """Wrapper for Exa API endpoints."""

    def __init__(self):
        """Initialize Exa API client."""
        self.api_key = Config.EXA_API_KEY
        self.base_url = Config.EXA_BASE_URL
        self.headers = {
            "x-api-key": self.api_key,
            "Content-Type": "application/json",
        }

    def search(
        self,
        query: str,
        num_results: int = 10,
        start_published_date: Optional[str] = None,
        end_published_date: Optional[str] = None,
        include_domains: Optional[List[str]] = None,
        exclude_domains: Optional[List[str]] = None,
        type: str = "auto",
        use_autoprompt: bool = True,
        text: bool = True,
        highlights: bool = True,
    ) -> Dict[str, Any]:
        """
        Perform neural search using Exa API.

        Args:
            query: Search query
            num_results: Number of results to return
            start_published_date: Filter results published after this date (ISO format)
            end_published_date: Filter results published before this date (ISO format)
            include_domains: List of domains to include
            exclude_domains: List of domains to exclude
            type: Content type filter (auto, news, research paper, pdf, etc.)
            use_autoprompt: Let Exa optimize the query
            text: Include full text content
            highlights: Include relevant highlights

        Returns:
            Dictionary containing search results
        """
        url = f"{self.base_url}/search"

        payload = {
            "query": query,
            "numResults": num_results,
            "useAutoprompt": use_autoprompt,
            "type": type,
            "contents": {
                "text": text,
                "highlights": highlights,
            },
        }

        if start_published_date:
            payload["startPublishedDate"] = start_published_date
        if end_published_date:
            payload["endPublishedDate"] = end_published_date
        if include_domains:
            payload["includeDomains"] = include_domains
        if exclude_domains:
            payload["excludeDomains"] = exclude_domains

        try:
            with httpx.Client(timeout=30.0) as client:
                response = client.post(url, json=payload, headers=self.headers)
                response.raise_for_status()
                return response.json()
        except httpx.HTTPError as e:
            return {
                "error": str(e),
                "status": "failed",
                "results": [],
            }

    def find_similar(
        self,
        url: str,
        num_results: int = 5,
        exclude_source_domain: bool = True,
        text: bool = True,
        highlights: bool = True,
    ) -> Dict[str, Any]:
        """
        Find similar content to a given URL using Exa's neural similarity search.

        Args:
            url: URL to find similar content for
            num_results: Number of similar results to return
            exclude_source_domain: Exclude results from the same domain
            text: Include full text content
            highlights: Include relevant highlights

        Returns:
            Dictionary containing similar search results
        """
        api_url = f"{self.base_url}/findSimilar"

        payload = {
            "url": url,
            "numResults": num_results,
            "excludeSourceDomain": exclude_source_domain,
            "contents": {
                "text": text,
                "highlights": highlights,
            },
        }

        try:
            with httpx.Client(timeout=30.0) as client:
                response = client.post(api_url, json=payload, headers=self.headers)
                response.raise_for_status()
                return response.json()
        except httpx.HTTPError as e:
            return {
                "error": str(e),
                "status": "failed",
                "results": [],
            }

    def get_contents(
        self,
        ids: List[str],
        text: bool = True,
        highlights: bool = True,
    ) -> Dict[str, Any]:
        """
        Get full content for specific result IDs.

        Args:
            ids: List of Exa result IDs
            text: Include full text content
            highlights: Include relevant highlights

        Returns:
            Dictionary containing content for the specified IDs
        """
        url = f"{self.base_url}/contents"

        payload = {
            "ids": ids,
            "contents": {
                "text": text,
                "highlights": highlights,
            },
        }

        try:
            with httpx.Client(timeout=30.0) as client:
                response = client.post(url, json=payload, headers=self.headers)
                response.raise_for_status()
                return response.json()
        except httpx.HTTPError as e:
            return {
                "error": str(e),
                "status": "failed",
                "contents": [],
            }

    def format_results(self, results: Dict[str, Any]) -> List[Dict[str, Any]]:
        """
        Format Exa search results into a standardized structure.

        Args:
            results: Raw results from Exa API

        Returns:
            List of formatted result dictionaries
        """
        if "error" in results or "results" not in results:
            return []

        formatted = []
        for result in results.get("results", []):
            formatted_result = {
                "title": result.get("title", "No title"),
                "url": result.get("url", ""),
                "author": result.get("author"),
                "published_date": result.get("publishedDate"),
                "score": result.get("score", 0),
                "text": result.get("text", ""),
                "highlights": result.get("highlights", []),
                "summary": result.get("summary", ""),
            }
            formatted.append(formatted_result)

        return formatted

File: src/utils/config.py

"""Configuration management for Deep Research Agent."""

import os
from dotenv import load_dotenv

# Load environment variables from .env file
load_dotenv()


class Config:
    """Configuration class for managing API keys and settings."""

    # Minimax API Configuration
    MINIMAX_API_KEY = os.getenv("MINIMAX_API_KEY")
    MINIMAX_BASE_URL = "https://api.minimax.io/anthropic"
    MINIMAX_MODEL = "MiniMax-M2"

    # OpenRouter API Configuration
    OPENROUTER_API_KEY = os.getenv("OPENROUTER_API_KEY")
    OPENROUTER_BASE_URL = "https://openrouter.ai/api/v1"
    OPENROUTER_MODEL = "google/gemini-2.5-flash"

    # Exa API Configuration
    EXA_API_KEY = os.getenv("EXA_API_KEY")
    EXA_BASE_URL = "https://api.exa.ai"

    @classmethod
    def validate(cls):
        """Validate that all required API keys are set."""
        missing = []

        if not cls.MINIMAX_API_KEY:
            missing.append("MINIMAX_API_KEY")
        if not cls.OPENROUTER_API_KEY:
            missing.append("OPENROUTER_API_KEY")
        if not cls.EXA_API_KEY:
            missing.append("EXA_API_KEY")

        if missing:
            raise ValueError(
                f"Missing required API keys: {', '.join(missing)}. "
                f"Please set them in your .env file."
            )

        return True


# Validate configuration on import
try:
    Config.validate()
except ValueError as e:
    print(f"Configuration Error: {e}")
    print("Please copy .env.example to .env and fill in your API keys.")

File: .env.example

# Minimax API Configuration (for Supervisor Agent with M2 model)
MINIMAX_API_KEY=your_minimax_api_key_here

# OpenRouter API Configuration (for Planning Agent and Web Search Retriever)
OPENROUTER_API_KEY=your_openrouter_api_key_here

# Exa API Configuration (for web search)
EXA_API_KEY=your_exa_api_key_here

File: .gitignore

# Python-generated files
__pycache__/
*.py[oc]
build/
dist/
wheels/
*.egg-info

# Virtual environments
.venv

# reports
reports/

# env
.env

File: .python-version

3.12

File: main.py

#!/usr/bin/env python3
"""
Deep Research Agent CLI

A command-line interface for conducting deep research using:
- Minimax M2 (supervisor with interleaved thinking)
- Exa API (neural web search)
- OpenRouter (subagent LLMs)
"""

import sys
import os
import argparse
from datetime import datetime
from rich.console import Console
from rich.panel import Panel
from rich.text import Text
from rich.markdown import Markdown
from rich import print as rprint
from src.agents.supervisor import SupervisorAgent
from src.utils.config import Config

# Initialize rich console
console = Console()


def print_banner():
    """Print welcome banner with colors."""
    banner_text = Text()
    banner_text.append("DEEP RESEARCH AGENT", style="bold cyan")
    banner_text.append("\nPowered by ", style="white")
    banner_text.append("Minimax M2", style="bold magenta")
    banner_text.append(" with ", style="white")
    banner_text.append("Interleaved Thinking", style="bold green")

    console.print(Panel(
        banner_text,
        border_style="cyan",
        padding=(1, 2)
    ))


def print_section(title: str):
    """Print a formatted section header with colors."""
    console.print(f"\n[bold cyan]{'=' * 80}[/bold cyan]")
    console.print(f"[bold yellow] {title}[/bold yellow]")
    console.print(f"[bold cyan]{'=' * 80}[/bold cyan]")


def save_report(content: str, query: str):
    """Save research report to a file in the reports/ folder."""
    # Create reports directory if it doesn't exist
    reports_dir = "reports"
    os.makedirs(reports_dir, exist_ok=True)

    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    # Create a safe filename from query
    safe_query = "".join(c if c.isalnum() or c in (' ', '-', '_') else '' for c in query)
    safe_query = safe_query.strip()[:50]  # Limit length
    filename = f"research_report_{safe_query}_{timestamp}.md"
    filepath = os.path.join(reports_dir, filename)

    try:
        with open(filepath, 'w') as f:
            f.write(f"# Research Report\n\n")
            f.write(f"**Query:** {query}\n\n")
            f.write(f"**Generated:** {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n\n")
            f.write("---\n\n")
            f.write(content)

        console.print(f"\n[bold green]✓ Report saved to:[/bold green] [cyan]{filepath}[/cyan]")
        return filepath
    except Exception as e:
        console.print(f"\n[bold red]✗ Error saving report:[/bold red] {e}")
        return None


def run_research(query: str, verbose: bool = False, save: bool = False):
    """
    Run research for a given query.

    Args:
        query: Research question or topic
        verbose: Show detailed progress
        save: Save report to file
    """
    try:
        # Initialize supervisor
        if verbose:
            console.print("\n[dim]Initializing Minimax M2 Supervisor Agent...[/dim]")

        supervisor = SupervisorAgent()

        # Conduct research
        print_section("CONDUCTING RESEARCH")
        console.print(f"[bold white]Query:[/bold white] [italic cyan]{query}[/italic cyan]\n")

        if verbose:
            console.print("[dim]→ Planning research strategy...[/dim]")
            console.print("[dim]→ Searching web with Exa...[/dim]")
            console.print("[dim]→ Synthesizing findings...[/dim]\n")

        result = supervisor.research(query)

        # Display results
        print_section("RESEARCH REPORT")

        # Display markdown-formatted result with colors
        console.print(Markdown(result))
        console.print(f"\n[bold cyan]{'=' * 80}[/bold cyan]")

        # Save if requested
        if save:
            save_report(result, query)

        # Show thinking in verbose mode
        if verbose:
            print_section("CONVERSATION HISTORY (VERBOSE)")
            history = supervisor.get_conversation_history()
            for i, msg in enumerate(history, 1):
                console.print(f"\n[bold magenta]--- Message {i} ({msg['role']}) ---[/bold magenta]")
                if isinstance(msg['content'], list):
                    for block in msg['content']:
                        if hasattr(block, 'type'):
                            console.print(f"[cyan]  Block type: {block.type}[/cyan]")
                            if block.type == "thinking":
                                console.print(f"[dim]  Thinking: {block.thinking[:200]}...[/dim]")
                            elif block.type == "text":
                                console.print(f"[white]  Text: {block.text[:200]}...[/white]")
                else:
                    console.print(f"[white]  Content: {str(msg['content'])[:200]}...[/white]")

    except Exception as e:
        console.print(f"\n[bold red]✗ Error:[/bold red] {e}")
        sys.exit(1)


def interactive_mode():
    """Run in interactive mode with multiple queries."""
    print_banner()
    console.print("\n[bold cyan]Interactive mode[/bold cyan] - Enter your research queries (type [yellow]'exit'[/yellow] to quit)\n")

    while True:
        try:
            query = input("\n🔍 Research Query: ").strip()

            if not query:
                continue

            if query.lower() in ['exit', 'quit', 'q']:
                console.print("\n[bold green]👋 Goodbye![/bold green]")
                break

            # Check for special commands
            save = False
            verbose = False

            if query.startswith('/'):
                if query.startswith('/save '):
                    save = True
                    query = query[6:].strip()
                elif query.startswith('/verbose '):
                    verbose = True
                    query = query[9:].strip()
                elif query == '/help':
                    console.print("\n[bold yellow]Commands:[/bold yellow]")
                    console.print("  [cyan]/save <query>[/cyan]     - Save report to file")
                    console.print("  [cyan]/verbose <query>[/cyan]  - Show detailed progress")
                    console.print("  [cyan]/help[/cyan]             - Show this help")
                    console.print("  [cyan]exit/quit/q[/cyan]       - Exit")
                    continue

            run_research(query, verbose=verbose, save=save)

        except KeyboardInterrupt:
            console.print("\n\n[bold green]👋 Goodbye![/bold green]")
            break
        except Exception as e:
            console.print(f"\n[bold red]✗ Error:[/bold red] {e}")


def main():
    """Main entry point."""
    parser = argparse.ArgumentParser(
        description="Deep Research Agent - Conduct thorough research using Minimax M2 and Exa",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  # Interactive mode
  python main.py

  # Single query
  python main.py -q "What are the latest developments in quantum computing?"

  # Save report to file
  python main.py -q "AI trends in 2025" --save

  # Verbose mode with thinking details
  python main.py -q "Climate change solutions" --verbose
        """
    )

    parser.add_argument(
        '-q', '--query',
        type=str,
        help='Research query (if not provided, runs in interactive mode)'
    )

    parser.add_argument(
        '-v', '--verbose',
        action='store_true',
        help='Show detailed progress and thinking blocks'
    )

    parser.add_argument(
        '-s', '--save',
        action='store_true',
        help='Save research report to file'
    )

    args = parser.parse_args()

    # Validate configuration
    try:
        Config.validate()
    except ValueError as e:
        console.print(f"[bold red]✗ Configuration Error:[/bold red] {e}")
        console.print("\n[yellow]Please ensure you have:[/yellow]")
        console.print("[dim]1. Copied .env.example to .env[/dim]")
        console.print("[dim]2. Added your API keys to .env[/dim]")
        sys.exit(1)

    # Run mode based on arguments
    if args.query:
        # Single query mode
        print_banner()
        run_research(args.query, verbose=args.verbose, save=args.save)
    else:
        # Interactive mode
        interactive_mode()


if __name__ == "__main__":
    main()

File: pyproject.toml

[project]
name = "deep-research-agent"
version = "0.1.0"
description = "Add your description here"
readme = "README.md"
requires-python = ">=3.12"
dependencies = [
    "anthropic>=0.74.1",
    "httpx>=0.28.1",
    "python-dotenv>=1.2.1",
    "rich>=14.2.0",
]

File: README.md

# MiniMax-M2 Deep Research Agent

A sophisticated research tool powered by **Minimax M2** with interleaved thinking, **Exa** neural search, and multi-agent orchestration.

## Features

- **Minimax M2 Supervisor**: Uses interleaved thinking to maintain reasoning state across multi-step research
- **Intelligent Planning**: Automatically decomposes research queries into optimized subqueries
- **Neural Web Search**: Leverages Exa API for high-quality, AI-powered web search
- **Comprehensive Reports**: Generates detailed research reports with citations and analysis
- **CLI Interface**: Simple command-line interface with interactive and single-query modes

## Architecture

```
+-----------------------------------------------+
|            Supervisor Agent                   |
|    (Minimax M2 + Interleaved Thinking)        |
+-----------------------------------------------+
                      |
       +--------------+--------------+
       |              |              |
       v              v              v
+------------+ +-------------+ +-----------+
|  Planning  | | Web Search  | | Synthesis |
|   Agent    | |  Retriever  | |   (M2)    |
|  (Gemini)  | |             | |           |
+------------+ +-------------+ +-----------+
                      |
                      v
               +------------+
               |  Exa API   |
               +------------+
```

### Agent Descriptions

1. **Supervisor Agent**:
   - Uses Minimax M2 with Anthropic SDK
   - Implements interleaved thinking (preserves reasoning across turns)
   - Coordinates planning, search, and synthesis
   - Generates final comprehensive research report

2. **Planning Agent**:
   - Uses Gemini 2.5 Flash via OpenRouter
   - Generates 3-5 Exa-optimized subqueries
   - Considers time periods, domains, content types, and priorities

3. **Web Search Retriever**:
   - Uses Gemini 2.5 Flash for synthesis
   - Executes Exa searches with neural search capabilities
   - Finds similar content using Exa's similarity search
   - Organizes findings with sources and highlights

## Installation

### Prerequisites

- Python 3.9+
- [uv](https://github.com/astral-sh/uv) package manager
- API keys for:
  - Minimax (M2 model)
  - OpenRouter (for Gemini)
  - Exa (web search)

### Setup

1. **Install dependencies**:
```bash
cd deep-research-agent
uv sync
```

2. **Configure environment variables**:
```bash
# Copy the example file
cp .env.example .env

# Edit .env and add your API keys
MINIMAX_API_KEY=your_minimax_api_key_here
OPENROUTER_API_KEY=your_openrouter_api_key_here
EXA_API_KEY=your_exa_api_key_here
```

3. **Activate virtual environment** (if needed):
```bash
source .venv/bin/activate
```

## Usage

### Interactive Mode

Run without arguments to enter interactive mode:

```bash
python main.py
```

Then enter your research queries at the prompt:

```
Research Query: What are the latest developments in quantum computing?
```

### Single Query Mode

Run a single research query:

```bash
python main.py -q "What are the latest developments in quantum computing?"
```

### Save Report to File

Save the research report automatically:

```bash
python main.py -q "AI trends in 2025" --save
```

### Verbose Mode

Show detailed progress and thinking blocks:

```bash
python main.py -q "Climate change solutions" --verbose
```

### Interactive Mode Commands

While in interactive mode, you can use these commands:

- `/save <query>` - Save the report to a file
- `/verbose <query>` - Show detailed progress
- `/help` - Show help message
- `exit`, `quit`, or `q` - Exit the program

## How It Works

### 1. Query Planning

When you submit a research query, the **Planning Agent** decomposes it into 3-5 optimized subqueries:

```json
{
  "subqueries": [
    {
      "query": "quantum computing breakthroughs 2025",
      "type": "news",
      "time_period": "recent",
      "priority": 1
    },
    {
      "query": "quantum computing applications cryptography",
      "type": "auto",
      "time_period": "any",
      "priority": 2
    }
  ]
}
```

### 2. Web Search

The **Web Search Retriever** executes each subquery using Exa:

- Performs neural search for each subquery
- Finds similar content for high-priority results
- Extracts highlights and key information
- Organizes findings by relevance

### 3. Synthesis

The **Supervisor Agent** (using Minimax M2 with interleaved thinking):

- Receives all search findings
- Maintains reasoning state across the research process
- Synthesizes a comprehensive report with:
  - Executive summary
  - Key findings organized by theme
  - Detailed analysis
  - Cited sources with URLs

### 4. Interleaved Thinking

The key innovation is **interleaved thinking**:

- The supervisor preserves ALL content blocks (thinking + text + tool_use) in conversation history
- This maintains the reasoning chain across multiple turns
- Results in more coherent, contextualized research reports
- Prevents "state drift" in multi-step workflows

## Project Structure

```
deep-research-agent/
├── README.md                  # This file
├── .env.example               # Environment variables template
├── .env                       # Your API keys (create this)
├── pyproject.toml             # Project dependencies
├── main.py                    # CLI entry point
└── src/
    ├── agents/
    │   ├── supervisor.py           # Minimax M2 supervisor
    │   ├── planning_agent.py       # Query planning
    │   └── web_search_retriever.py # Exa search integration
    ├── tools/
    │   └── exa_tool.py             # Exa API wrapper
    └── utils/
        └── config.py               # Configuration management
```

## API Keys

### Getting API Keys

1. **Minimax M2**: Sign up at [platform.minimax.io](https://platform.minimax.io)
2. **OpenRouter**: Get key at [openrouter.ai](https://openrouter.ai)
3. **Exa**: Register at [exa.ai](https://exa.ai)

### Why These Services?

- **Minimax M2**: Advanced reasoning model with native interleaved thinking support
- **OpenRouter**: Unified API for accessing Gemini and other LLMs
- **Exa**: Neural search engine optimized for research and discovery

## Examples

### Example 1: Technology Research

```bash
python main.py -q "What are the latest breakthroughs in artificial general intelligence?"
```

**Output**: Comprehensive report covering recent AGI developments, key research papers, major announcements, and expert opinions with citations.

### Example 2: Business Intelligence

```bash
python main.py -q "What are the emerging trends in electric vehicle adoption?" --save
```

**Output**: Market analysis with statistics, industry trends, regional adoption rates, and future projections. Report saved to file.

### Example 3: Scientific Research

```bash
python main.py -q "What are the most promising approaches to carbon capture technology?" --verbose
```

**Output**: Technical analysis of carbon capture methods with detailed thinking process visible.

## Advanced Usage

### Customizing Prompts

Edit the system prompts in:
- `src/agents/supervisor.py` - Main coordinator logic
- `src/agents/planning_agent.py` - Query decomposition strategy
- `src/agents/web_search_retriever.py` - Search and synthesis approach

### Adjusting Search Parameters

Modify Exa search parameters in `src/agents/web_search_retriever.py`:
- `num_results`: Number of results per query (default: 5-10)
- `time_period`: Date filtering (recent, past_week, past_month, past_year, any)
- `content_type`: Filter by type (news, research paper, pdf, blog, etc.)

## Troubleshooting

### Configuration Errors

If you see "Missing required API keys":
1. Ensure `.env` file exists (copy from `.env.example`)
2. Verify all API keys are set correctly
3. Check that API keys don't have extra spaces or quotes

### API Errors

If you encounter API errors:
1. Verify your API keys are valid and active
2. Check your API rate limits and quotas
3. Ensure you have sufficient credits/balance

### Import Errors

If you see module import errors:
1. Activate the virtual environment: `source .venv/bin/activate`
2. Install dependencies: `uv sync`
3. Ensure you're running from the project root directory

## Performance

- Average research query: 30-60 seconds
- Depends on:
  - Number of subqueries (3-5)
  - Complexity of search results
  - LLM response times

## Future Enhancements

Potential improvements:
- [ ] Support for additional search engines (Tavily, Perplexity)
- [ ] PDF and document upload for context
- [ ] Multi-turn conversations with follow-up questions
- [ ] Export to different formats (PDF, Markdown, JSON)
- [ ] Web UI interface
- [ ] Caching and result persistence
- [ ] Custom research templates

## License

MIT License - feel free to use and modify for your own projects.

## Acknowledgments

Built with:
- [Minimax M2](https://www.minimax.io/) - Advanced reasoning model
- [Exa](https://exa.ai/) - Neural web search
- [Anthropic SDK](https://github.com/anthropics/anthropic-sdk-python) - API client
- [OpenRouter](https://openrouter.ai/) - LLM routing

davo/m2-deep-research.md

Select an option

No results found

Select an option

No results found

Directory Structure

Files

File: src/agents/planning_agent.py

File: src/agents/supervisor.py

File: src/agents/web_search_retriever.py

File: src/tools/exa_tool.py

File: src/utils/config.py

File: .env.example

File: .gitignore

File: .python-version

File: main.py

File: pyproject.toml

File: README.md