research-agent/
├── agent.py              # Main agent orchestration
├── tools/
│   ├── __init__.py
│   ├── web_search.py     # Web search tool
│   ├── scraper.py        # Web scraping tool
│   ├── calculator.py     # Calculator tool
│   ├── file_writer.py    # File operations tool
│   └── data_analyzer.py  # Data analysis tool
├── prompts.py            # Custom prompts
├── utils.py              # Helper functions
├── requirements.txt      # Dependencies
├── .env                  # API keys
└── outputs/              # Generated reports

langchain==0.1.0
langchain-openai==0.0.2
langchain-community==0.0.13
python-dotenv==1.0.0
requests==2.31.0
beautifulsoup4==4.12.2
duckduckgo-search==4.1.1
pandas==2.1.4
numpy==1.26.3

# Install dependencies
pip install -r requirements.txt

# Create .env file
echo "OPENAI_API_KEY=your_key_here" > .env

# Create directory structure
mkdir tools outputs
touch tools/__init__.py

from langchain.tools import Tool
from duckduckgo_search import DDGS
from typing import Optional

class WebSearchTool:
    """Tool for searching the web using DuckDuckGo"""

    def __init__(self, max_results: int = 5):
        self.max_results = max_results
        self.ddgs = DDGS()

    def search(self, query: str) -> str:
        """
        Search the web for information

        Args:
            query: Search query string

        Returns:
            Formatted search results
        """
        try:
            results = list(self.ddgs.text(query, max_results=self.max_results))

            if not results:
                return f"No results found for: {query}"

            formatted_results = f"Search results for '{query}':\n\n"

            for i, result in enumerate(results, 1):
                title = result.get('title', 'No title')
                snippet = result.get('body', 'No description')
                url = result.get('href', 'No URL')

                formatted_results += f"{i}. {title}\n"
                formatted_results += f"   {snippet}\n"
                formatted_results += f"   URL: {url}\n\n"

            return formatted_results

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

    def get_tool(self) -> Tool:
        """Return LangChain Tool object"""
        return Tool(
            name="WebSearch",
            func=self.search,
            description=(
                "Useful for searching the web for current information, news, facts, "
                "and general knowledge. Input should be a search query string. "
                "Returns top search results with titles, descriptions, and URLs."
            )
        )


# Example usage
if __name__ == "__main__":
    search_tool = WebSearchTool(max_results=3)
    results = search_tool.search("latest developments in AI")
    print(results)

from langchain.tools import Tool
import requests
from bs4 import BeautifulSoup
from typing import Optional

class WebScraperTool:
    """Tool for scraping content from web pages"""

    def __init__(self, max_length: int = 5000):
        self.max_length = max_length

    def scrape(self, url: str) -> str:
        """
        Scrape and extract text content from a URL

        Args:
            url: Web page URL to scrape

        Returns:
            Extracted text content
        """
        try:
            headers = {
                'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
            }

            response = requests.get(url, headers=headers, timeout=10)
            response.raise_for_status()

            soup = BeautifulSoup(response.content, 'html.parser')

            # Remove script and style elements
            for script in soup(["script", "style", "nav", "footer", "aside"]):
                script.decompose()

            # Get text
            text = soup.get_text(separator='\n', strip=True)

            # Clean up whitespace
            lines = [line.strip() for line in text.splitlines() if line.strip()]
            text = '\n'.join(lines)

            # Truncate if too long
            if len(text) > self.max_length:
                text = text[:self.max_length] + "\n\n[Content truncated...]"

            return f"Content from {url}:\n\n{text}"

        except requests.exceptions.RequestException as e:
            return f"Error fetching URL: {str(e)}"
        except Exception as e:
            return f"Error scraping content: {str(e)}"

    def get_tool(self) -> Tool:
        """Return LangChain Tool object"""
        return Tool(
            name="WebScraper",
            func=self.scrape,
            description=(
                "Useful for extracting and reading the full content of a specific web page. "
                "Input should be a complete URL (including http:// or https://). "
                "Returns the text content of the page."
            )
        )


# Example usage
if __name__ == "__main__":
    scraper = WebScraperTool()
    content = scraper.scrape("https://en.wikipedia.org/wiki/Artificial_intelligence")
    print(content[:500])

from langchain.tools import Tool
import re
import math
import numpy as np

class CalculatorTool:
    """Advanced calculator tool with support for math operations"""

    def calculate(self, expression: str) -> str:
        """
        Evaluate mathematical expressions safely

        Args:
            expression: Mathematical expression to evaluate

        Returns:
            Calculation result
        """
        try:
            # Remove any potentially dangerous functions
            expression = expression.strip()

            # Create safe namespace with math functions
            safe_dict = {
                'abs': abs,
                'round': round,
                'min': min,
                'max': max,
                'sum': sum,
                'pow': pow,
                'sqrt': math.sqrt,
                'log': math.log,
                'log10': math.log10,
                'exp': math.exp,
                'sin': math.sin,
                'cos': math.cos,
                'tan': math.tan,
                'pi': math.pi,
                'e': math.e,
                'mean': np.mean,
                'median': np.median,
                'std': np.std,
            }

            # Evaluate expression
            result = eval(expression, {"__builtins__": {}}, safe_dict)

            return f"Result: {result}"

        except Exception as e:
            return f"Error in calculation: {str(e)}. Please check your expression."

    def get_tool(self) -> Tool:
        """Return LangChain Tool object"""
        return Tool(
            name="Calculator",
            func=self.calculate,
            description=(
                "Useful for performing mathematical calculations. "
                "Supports basic arithmetic (+, -, *, /), power (**), "
                "and math functions like sqrt, log, sin, cos, mean, median, std. "
                "Input should be a valid mathematical expression. "
                "Example: 'sqrt(16) + log(100)'"
            )
        )


# Example usage
if __name__ == "__main__":
    calc = CalculatorTool()
    print(calc.calculate("sqrt(144) + pow(2, 3)"))
    print(calc.calculate("mean([10, 20, 30, 40])"))

from langchain.tools import Tool
import os
from datetime import datetime

class FileWriterTool:
    """Tool for saving content to files"""

    def __init__(self, output_dir: str = "outputs"):
        self.output_dir = output_dir
        os.makedirs(output_dir, exist_ok=True)

    def write_file(self, content_and_filename: str) -> str:
        """
        Write content to a file

        Args:
            content_and_filename: Format: "filename.txt|Content to write"

        Returns:
            Success message with file path
        """
        try:
            # Parse input
            if '|' not in content_and_filename:
                return "Error: Input must be in format 'filename.txt|Content to write'"

            filename, content = content_and_filename.split('|', 1)
            filename = filename.strip()
            content = content.strip()

            # Sanitize filename
            filename = "".join(c for c in filename if c.isalnum() or c in ('_', '-', '.'))

            if not filename:
                timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
                filename = f"output_{timestamp}.txt"

            filepath = os.path.join(self.output_dir, filename)

            # Write file
            with open(filepath, 'w', encoding='utf-8') as f:
                f.write(content)

            return f"Successfully saved content to: {filepath}\nFile size: {len(content)} characters"

        except Exception as e:
            return f"Error writing file: {str(e)}"

    def get_tool(self) -> Tool:
        """Return LangChain Tool object"""
        return Tool(
            name="FileWriter",
            func=self.write_file,
            description=(
                "Useful for saving text content to a file. "
                "Input format: 'filename.txt|Content to write' "
                "The filename and content should be separated by a pipe character (|). "
                "Example: 'research_report.txt|This is the content of my report.'"
            )
        )


# Example usage
if __name__ == "__main__":
    writer = FileWriterTool()
    result = writer.write_file("test_report.txt|This is a test report.")
    print(result)

from langchain.prompts import PromptTemplate

# Research agent system prompt
RESEARCH_AGENT_PROMPT = """You are an expert research assistant with access to multiple tools.
Your goal is to conduct thorough research and provide comprehensive, well-sourced answers.

When conducting research:
1. Break down complex questions into smaller sub-questions
2. Use web search to find relevant information
3. Scrape web pages for detailed content when needed
4. Perform calculations when analyzing data
5. Save important findings to files for future reference
6. Cite your sources clearly
7. Synthesize information from multiple sources

Always think step-by-step and explain your reasoning process.
If you're unsure about something, acknowledge it and seek more information.

Available tools:
- WebSearch: Search the web for information
- WebScraper: Extract content from specific URLs
- Calculator: Perform mathematical calculations
- FileWriter: Save research findings to files

Format your final answer as a structured report with:
- Executive Summary
- Detailed Findings (with citations)
- Data/Statistics (if applicable)
- Conclusions
- Sources

Begin!
"""

# Few-shot examples for better performance
RESEARCH_EXAMPLES = """
Example 1:
Question: What is the current market size of the AI industry and its growth rate?

Thought: I need to find recent data about the AI industry market size and growth projections.
Action: WebSearch
Action Input: AI industry market size 2024 growth rate statistics
Observation: [Search results showing multiple sources with data]

Thought: I found several sources. Let me get detailed information from a reliable source.
Action: WebScraper
Action Input: https://www.market-research-report.com/ai-industry
Observation: [Detailed content about market size]

Thought: Now I have the market size data. Let me calculate the CAGR.
Action: Calculator
Action Input: ((500 / 200) ** (1/5) - 1) * 100
Observation: Result: 20.11

Thought: Let me save these findings to a file.
Action: FileWriter
Action Input: ai_market_research.txt|AI Industry Market Research...
Observation: Successfully saved content

Final Answer: [Comprehensive report with all findings]

---

Example 2:
Question: Compare the performance of Python vs JavaScript for data processing tasks.

Thought: I need to research performance benchmarks for both languages.
Action: WebSearch
Action Input: Python vs JavaScript data processing performance benchmarks
Observation: [Search results]

[Continue with systematic research process]
"""


def get_research_prompt() -> PromptTemplate:
    """Get the research agent prompt template"""
    template = RESEARCH_AGENT_PROMPT + "\n\n" + RESEARCH_EXAMPLES + "\n\n{agent_scratchpad}"

    return PromptTemplate(
        input_variables=["agent_scratchpad"],
        template=template
    )

from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_react_agent
from langchain import hub
from tools.web_search import WebSearchTool
from tools.scraper import WebScraperTool
from tools.calculator import CalculatorTool
from tools.file_writer import FileWriterTool
from dotenv import load_dotenv
import sys

# Load environment variables
load_dotenv()

class ResearchAgent:
    """Advanced multi-step research agent"""

    def __init__(self, model: str = "gpt-4", temperature: float = 0, verbose: bool = True):
        self.model = model
        self.temperature = temperature
        self.verbose = verbose

        # Initialize LLM
        self.llm = ChatOpenAI(
            model=model,
            temperature=temperature
        )

        # Initialize tools
        self.tools = self._initialize_tools()

        # Create agent
        self.agent = self._create_agent()

    def _initialize_tools(self):
        """Initialize all available tools"""
        search_tool = WebSearchTool(max_results=5)
        scraper_tool = WebScraperTool(max_length=5000)
        calc_tool = CalculatorTool()
        file_writer_tool = FileWriterTool()

        return [
            search_tool.get_tool(),
            scraper_tool.get_tool(),
            calc_tool.get_tool(),
            file_writer_tool.get_tool()
        ]

    def _create_agent(self):
        """Create the ReAct agent"""
        # Get ReAct prompt from hub
        prompt = hub.pull("hwchase17/react")

        # Create agent
        agent = create_react_agent(
            self.llm,
            self.tools,
            prompt
        )

        # Create agent executor
        agent_executor = AgentExecutor(
            agent=agent,
            tools=self.tools,
            verbose=self.verbose,
            max_iterations=10,
            handle_parsing_errors=True,
            return_intermediate_steps=True
        )

        return agent_executor

    def research(self, query: str) -> dict:
        """
        Conduct research on a query

        Args:
            query: Research question or topic

        Returns:
            Dictionary with answer and intermediate steps
        """
        print(f"\n{'='*70}")
        print(f"RESEARCH QUERY: {query}")
        print(f"{'='*70}\n")

        try:
            result = self.agent.invoke({"input": query})
            return result

        except Exception as e:
            print(f"\nError during research: {str(e)}")
            return {"output": f"Research failed: {str(e)}"}

    def format_report(self, result: dict) -> str:
        """Format the research results as a structured report"""
        output = result.get('output', 'No answer generated')
        intermediate_steps = result.get('intermediate_steps', [])

        report = f"""
{'='*70}
RESEARCH REPORT
{'='*70}

{output}

{'='*70}
RESEARCH PROCESS ({len(intermediate_steps)} steps)
{'='*70}
"""

        for i, (action, observation) in enumerate(intermediate_steps, 1):
            tool_name = action.tool
            tool_input = action.tool_input
            obs_preview = str(observation)[:200]

            report += f"""
Step {i}: {tool_name}
Input: {tool_input}
Output: {obs_preview}{'...' if len(str(observation)) > 200 else ''}
{'-'*70}
"""

        return report


def main():
    """Main execution function"""
    print("""
╔══════════════════════════════════════════════════════════════════╗
║              Multi-Step AI Research Agent                         ║
║              Powered by LangChain and GPT-4                       ║
╚══════════════════════════════════════════════════════════════════╝
    """)

    # Initialize agent
    agent = ResearchAgent(
        model="gpt-4",
        temperature=0,
        verbose=True
    )

    # Interactive mode or single query
    if len(sys.argv) > 1:
        # Single query mode
        query = " ".join(sys.argv[1:])
        result = agent.research(query)
        report = agent.format_report(result)
        print(report)

    else:
        # Interactive mode
        print("\nInteractive Research Mode")
        print("Type your research questions (or 'quit' to exit)\n")

        while True:
            try:
                query = input("\nResearch Query: ").strip()

                if not query:
                    continue

                if query.lower() in ['quit', 'exit', 'q']:
                    print("Goodbye!")
                    break

                # Conduct research
                result = agent.research(query)

                # Display formatted report
                report = agent.format_report(result)
                print(report)

                # Ask if user wants to save
                save = input("\nSave report to file? (y/n): ").strip().lower()
                if save == 'y':
                    filename = input("Filename (without extension): ").strip()
                    if filename:
                        file_writer = FileWriterTool()
                        file_writer.write_file(f"{filename}.txt|{report}")
                        print(f"Report saved to outputs/{filename}.txt")

            except KeyboardInterrupt:
                print("\n\nGoodbye!")
                break
            except Exception as e:
                print(f"\nError: {str(e)}")


if __name__ == "__main__":
    main()

# Example 1: Market Research
query1 = """
Research the current state of the electric vehicle market.
Include market size, growth rate, major players, and future projections.
Save your findings to a file.
"""

# Example 2: Technical Comparison
query2 = """
Compare the performance and use cases of PostgreSQL vs MongoDB.
Include benchmarks if available and provide recommendations for when to use each.
"""

# Example 3: Data Analysis
query3 = """
Find the GDP growth rates of the top 5 economies for 2023.
Calculate the average growth rate and identify the fastest-growing economy.
"""

# Example 4: Trend Analysis
query4 = """
What are the latest trends in generative AI? Find recent articles,
summarize the key developments, and predict the impact on software development.
"""

# Interactive mode
python agent.py

# Single query mode
python agent.py "What are the latest developments in quantum computing?"

# With specific research task
python agent.py "Research the top 5 programming languages in 2024, their growth trends, and calculate which has the highest year-over-year growth rate"

import time
from typing import Callable, Any

def retry_with_exponential_backoff(
    func: Callable,
    max_retries: int = 3,
    initial_delay: float = 1.0
) -> Any:
    """Retry a function with exponential backoff"""
    for attempt in range(max_retries):
        try:
            return func()
        except Exception as e:
            if attempt == max_retries - 1:
                raise e

            delay = initial_delay * (2 ** attempt)
            print(f"Attempt {attempt + 1} failed. Retrying in {delay}s...")
            time.sleep(delay)

class ProgressTracker:
    """Track agent progress and display status"""

    def __init__(self):
        self.steps = []
        self.current_step = None

    def start_step(self, tool_name: str, description: str):
        """Start a new step"""
        self.current_step = {
            "tool": tool_name,
            "description": description,
            "start_time": time.time(),
            "status": "in_progress"
        }
        print(f"\n→ {tool_name}: {description}")

    def complete_step(self, result: str):
        """Complete the current step"""
        if self.current_step:
            self.current_step["status"] = "completed"
            self.current_step["end_time"] = time.time()
            self.current_step["duration"] = (
                self.current_step["end_time"] - self.current_step["start_time"]
            )
            self.current_step["result"] = result
            self.steps.append(self.current_step)

            print(f"✓ Completed in {self.current_step['duration']:.2f}s")

    def get_summary(self) -> str:
        """Get execution summary"""
        total_time = sum(step["duration"] for step in self.steps)
        summary = f"\nExecution Summary:\n"
        summary += f"Total steps: {len(self.steps)}\n"
        summary += f"Total time: {total_time:.2f}s\n"
        return summary

from functools import lru_cache
import hashlib
import json

class CachedResearchAgent(ResearchAgent):
    """Research agent with caching"""

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.cache_file = "cache.json"
        self.cache = self._load_cache()

    def _load_cache(self) -> dict:
        """Load cache from file"""
        try:
            with open(self.cache_file, 'r') as f:
                return json.load(f)
        except:
            return {}

    def _save_cache(self):
        """Save cache to file"""
        with open(self.cache_file, 'w') as f:
            json.dump(self.cache, f)

    def _get_cache_key(self, query: str) -> str:
        """Generate cache key for query"""
        return hashlib.md5(query.encode()).hexdigest()

    def research(self, query: str, use_cache: bool = True) -> dict:
        """Research with caching"""
        cache_key = self._get_cache_key(query)

        if use_cache and cache_key in self.cache:
            print("\n[Using cached result]")
            return self.cache[cache_key]

        result = super().research(query)

        if use_cache:
            self.cache[cache_key] = result
            self._save_cache()

        return result

class SafeResearchAgent(ResearchAgent):
    """Research agent with enhanced error handling"""

    def validate_query(self, query: str) -> bool:
        """Validate research query"""
        if not query or len(query.strip()) < 10:
            print("Error: Query too short. Please provide more details.")
            return False

        if len(query) > 1000:
            print("Error: Query too long. Please be more concise.")
            return False

        return True

    def research(self, query: str) -> dict:
        """Research with validation"""
        if not self.validate_query(query):
            return {"output": "Invalid query"}

        try:
            return super().research(query)
        except Exception as e:
            error_msg = f"Research failed: {str(e)}"
            print(f"\n{error_msg}")
            return {"output": error_msg}