Example: Research Agent

import asyncio
from opper_agents import Agent, tool, hook
from pydantic import BaseModel, Field


# Web Researcher tools
@tool
def search_web(query: str) -> str:
    """Search the web for information on a topic."""
    # Simulated search results
    return f"""Search results for '{query}':
1. Study shows 40% increase in AI adoption in 2024
2. Market size expected to reach $500B by 2025
3. Key players include major tech companies
4. Healthcare and finance lead adoption rates"""


@tool
def fetch_page(url: str) -> str:
    """Fetch and extract content from a webpage."""
    # Simulated page content
    return f"""Content from {url}:
The AI industry has seen remarkable growth. Recent studies indicate
that enterprise adoption has doubled year-over-year. Key drivers include
improved accessibility of tools and demonstrated ROI across industries."""


# Data Analyst tools
@tool
def calculate_growth(start: float, end: float) -> dict:
    """Calculate growth metrics between two values."""
    growth = ((end - start) / start) * 100
    return {
        "absolute_change": end - start,
        "percentage_growth": round(growth, 2),
        "growth_factor": round(end / start, 2)
    }


@tool
def analyze_trend(data_points: list[float]) -> dict:
    """Analyze a trend from a series of data points."""
    if len(data_points) < 2:
        return {"error": "Need at least 2 data points"}

    avg = sum(data_points) / len(data_points)
    trend = "increasing" if data_points[-1] > data_points[0] else "decreasing"

    return {
        "average": round(avg, 2),
        "min": min(data_points),
        "max": max(data_points),
        "trend": trend,
        "data_points": len(data_points)
    }


# Synthesis tool for the director
@tool
def synthesize_findings(findings: list[str]) -> str:
    """Combine multiple research findings into a coherent summary."""
    combined = "\n".join(f"• {f}" for f in findings)
    return f"Synthesized findings:\n{combined}"


# Output schema
class ResearchReport(BaseModel):
    title: str = Field(description="Report title")
    summary: str = Field(description="Executive summary (2-3 sentences)")
    key_findings: list[str] = Field(description="Main findings (3-5 points)")
    data_insights: list[str] = Field(description="Quantitative insights")
    recommendations: list[str] = Field(description="Action recommendations")
    confidence: float = Field(description="Confidence score 0-1")


# Create specialist agents
web_researcher = Agent(
    name="WebResearcher",
    description="Searches the web and extracts information from pages",
    instructions="You are a thorough web researcher. Search for relevant information and extract key facts.",
    tools=[search_web, fetch_page]
)

data_analyst = Agent(
    name="DataAnalyst",
    description="Analyzes numerical data and identifies trends",
    instructions="You are a data analyst. Focus on numbers, growth rates, and statistical insights.",
    tools=[calculate_growth, analyze_trend]
)


# Hooks for visibility
@hook("tool_call")
async def log_delegation(context, agent, tool, parameters, **kwargs):
    if agent.name == "ResearchDirector":
        if "researcher" in tool.name.lower() or "analyst" in tool.name.lower():
            print(f"  📋 Delegating to {tool.name}...")


@hook("agent_end")
async def log_completion(context, agent, **kwargs):
    print(f"  ✓ {agent.name} completed ({context.iteration} iterations)")


# Create the director agent
research_director = Agent(
    name="ResearchDirector",
    description="Conducts comprehensive research by coordinating specialists",
    instructions="""You are a research director. When given a topic:

1. First, use the web researcher to gather general information
2. Then, use the data analyst to analyze any numerical data
3. Finally, synthesize all findings into a comprehensive report

Be thorough but efficient. Delegate appropriately to your team.""",
    tools=[
        web_researcher.as_tool(),
        data_analyst.as_tool(),
        synthesize_findings
    ],
    output_schema=ResearchReport,
    hooks=[log_delegation, log_completion],
    max_iterations=15
)


async def main():
    print("=" * 60)
    print("RESEARCH AGENT DEMO")
    print("=" * 60)

    topic = "AI adoption trends in enterprise"
    print(f"\nResearch Topic: {topic}\n")
    print("-" * 60)

    result = await research_director.process(
        f"Research the following topic and provide a comprehensive report: {topic}"
    )

    print("\n" + "=" * 60)
    print("FINAL REPORT")
    print("=" * 60)
    print(f"\nTitle: {result.title}")
    print(f"\nSummary: {result.summary}")
    print(f"\nKey Findings:")
    for i, finding in enumerate(result.key_findings, 1):
        print(f"  {i}. {finding}")
    print(f"\nData Insights:")
    for insight in result.data_insights:
        print(f"  • {insight}")
    print(f"\nRecommendations:")
    for rec in result.recommendations:
        print(f"  → {rec}")
    print(f"\nConfidence: {result.confidence:.0%}")


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

# Add a fact-checker agent
fact_checker = Agent(
    name="FactChecker",
    description="Verifies claims and checks for accuracy",
    tools=[verify_claim, cross_reference]
)

research_director = Agent(
    name="ResearchDirector",
    tools=[
        web_researcher.as_tool(),
        data_analyst.as_tool(),
        fact_checker.as_tool(),  # New specialist
        synthesize_findings
    ]
)

research_director = Agent(
    name="ResearchDirector",
    tools=[...],
    enable_memory=True  # Remember previous research
)

# Research builds on previous findings
await research_director.process("Research AI in healthcare")
await research_director.process("Compare this to AI in finance")