Research Report Generator

from pathlib import Path
from shutil import rmtree
from textwrap import dedent
from typing import Iterator

from agno.agent import Agent, RunResponse
from agno.storage.sqlite import SqliteStorage
from agno.tools.duckduckgo import DuckDuckGoTools
from agno.utils.log import logger
from agno.utils.pprint import pprint_run_response
from agno.workflow import Workflow

reports_dir = Path(__file__).parent.joinpath("reports", "research")
if reports_dir.is_dir():
    rmtree(path=reports_dir, ignore_errors=True)
reports_dir.mkdir(parents=True, exist_ok=True)
research_analyst_report = str(reports_dir.joinpath("research_analyst_report.md"))
research_analyst_report = str(reports_dir.joinpath("research_analyst_report.md"))
research_report = str(reports_dir.joinpath("research_report.md"))


class ResearchReportGenerator(Workflow):
    """Advanced workflow for generating professional research analysis with strategic recommendations."""

    description: str = dedent("""\
    An intelligent research analysis system that produces comprehensive market intelligence and
    strategic insights. This workflow orchestrates multiple AI agents to analyze trends,
    evaluate developments, and create detailed research recommendations.
    The system excels at combining quantitative analysis with qualitative insights to deliver
    actionable research insights.
    """)

    research_analyst: Agent = Agent(
        name="Research Analyst",
        tools=[
            DuckDuckGoTools(
                search=True, news=True
            )
        ],
        description=dedent("""\
        You are ResearchMaster-X, an elite Senior Research Analyst at a top consulting firm with expertise in:

        - Comprehensive information analysis
        - Market trend evaluation
        - Industry development identification
        - News impact assessment
        - Risk factor analysis
        - Growth potential evaluation\
        """),
        instructions=dedent("""\
        1. Topic Research 📊
           - Analyze topic fundamentals and key developments
           - Review recent market trends and innovations
           - Evaluate competitive landscape
           - Assess industry dynamics and growth patterns
        2. Information Analysis 💹
           - Examine key metrics and indicators
           - Review expert opinions and insights
           - Analyze recent news and developments
           - Identify growth catalysts and opportunities
        3. Risk Assessment 🎯
           - Evaluate market risks and challenges
           - Assess topic-specific obstacles
           - Consider macroeconomic factors
           - Identify potential disruptions
        Note: This analysis is for educational and research purposes only.\
        """),
        expected_output="Comprehensive research analysis report in markdown format",
        save_response_to_file=research_analyst_report,
    )

    research_analyst: Agent = Agent(
        name="Research Analyst",
        description=dedent("""\
        You are InsightPro-X, an elite Senior Research Strategist at a top consulting firm specializing in:

        - Research opportunity evaluation
        - Comparative trend analysis
        - Impact assessment
        - Growth potential ranking
        - Strategic recommendations\
        """),
        instructions=dedent("""\
        1. Research Analysis 🔍
           - Evaluate each topic's potential and impact
           - Compare relative importance and trends
           - Assess competitive advantages
           - Consider market positioning and opportunities
        2. Impact Evaluation 📈
           - Analyze growth factors and drivers
           - Consider market conditions and adoption
           - Evaluate sustainability and scalability
           - Assess innovation potential
        3. Topic Ranking 🏆
           - Rank based on research potential and impact
           - Provide detailed rationale and insights
           - Consider long-term trends
           - Explain competitive advantages and opportunities\
        """),
        expected_output="Detailed research analysis and ranking report in markdown format",
        save_response_to_file=research_analyst_report,
    )

    research_lead: Agent = Agent(
        name="Research Lead",
        description=dedent("""\
        You are StrategySage-X, a distinguished Senior Research Lead at a top consulting firm expert in:

        - Research strategy development
        - Strategic insights optimization
        - Trend analysis
        - Research rationale articulation
        - Strategic recommendation delivery\
        """),
        instructions=dedent("""\
        1. Research Strategy 💼
           - Develop strategic research approach
           - Optimize insight delivery
           - Consider trend diversification
           - Set research timeframes and priorities
        2. Strategic Rationale 📝
           - Explain research decisions and focus areas
           - Support with comprehensive analysis
           - Address potential concerns and limitations
           - Highlight key trends and catalysts
        3. Recommendation Delivery 📊
           - Present clear strategic insights
           - Explain research thesis and implications
           - Provide actionable recommendations
           - Include trend considerations and opportunities\
        """),
        save_response_to_file=research_report,
    )

    def run(self, topics: str) -> Iterator[RunResponse]:
        logger.info(f"Getting research reports for topics: {topics}")
        initial_report: RunResponse = self.research_analyst.run(topics)
        if initial_report is None or not initial_report.content:
            yield RunResponse(
                run_id=self.run_id,
                content="Sorry, could not get the research analyst report.",
            )
            return

        logger.info("Ranking topics based on research potential.")
        ranked_topics: RunResponse = self.research_analyst.run(
            initial_report.content
        )
        if ranked_topics is None or not ranked_topics.content:
            yield RunResponse(
                run_id=self.run_id, content="Sorry, could not get the ranked topics."
            )
            return

        logger.info(
            "Reviewing the research report and producing strategic recommendations."
        )
        yield from self.research_lead.run(ranked_topics.content, stream=True)


# Run the workflow if the script is executed directly
if __name__ == "__main__":
    import random

    from rich.prompt import Prompt

    # Example research scenarios to showcase the analyzer's capabilities
    example_scenarios = [
        "AI, Machine Learning, Automation",  # Technology Trends
        "Renewable Energy, Solar, Wind",  # Energy Innovation
        "Electric Vehicles, Autonomous Driving, Mobility",  # Transportation
        "Cloud Computing, Cybersecurity, Data Analytics",  # Digital Infrastructure
        "E-commerce, Digital Payments, Fintech",  # Digital Economy
        "Biotechnology, Pharmaceuticals, Health Tech",  # Healthcare Innovation
        "Sustainable Technology, Green Energy, Climate Solutions",  # Sustainability
    ]

    # Get topics from user with example suggestion
    topics = Prompt.ask(
        "[bold]Enter research topics (comma-separated)[/bold] "
        "(or press Enter for a suggested research focus)\n✨",
        default=random.choice(example_scenarios),
    )

    # Convert topics to URL-safe string for session_id
    url_safe_topics = topics.lower().replace(" ", "-").replace(",", "")

    # Initialize the research analyst workflow
    research_report_generator = ResearchReportGenerator(
        session_id=f"research-report-{url_safe_topics}",
        storage=SqliteStorage(
            table_name="research_report_workflows",
            db_file="tmp/agno_workflows.db",
        ),
    )

    # Execute the workflow
    report: Iterator[RunResponse] = research_report_generator.run(topics=topics)

    # Print the report
    pprint_run_response(report, markdown=True)