Crea tu equipo de investigación de IA con Agno, Streamlit y uv

curl -LsSf https://astral.sh/uv/install.sh | sh

irm https://astral.sh/uv/install.ps1 | iex

mkdir ai-research-team
cd ai-research-team

# Create and activate a virtual environment
uv venv
source .venv/bin/activate  # On Windows: .venv\Scripts\activate

# Install dependencies at warp speed
uv add agno streamlit python-dotenv duckduckgo-search crawl4ai youtube-transcript-api resend pygithub hackernews

OPENROUTER_API_KEY=your_openrouter_key
[email protected]
[email protected]
GITHUB_ACCESS_TOKEN=your_github_token
RESEND_API_KEY=your_resend_key

search_agent = Agent(
    name="InternetSearcher",
    model=model,
    tools=[DuckDuckGoTools(search=True, news=False)],
    add_history_to_messages=True,
    num_history_responses=3, # Limit history passed to agent
    description="Expert at finding information online.",
    instructions=[
        "Use duckduckgo_search for web queries.",
        "Cite sources with URLs.",
        "Focus on recent, reliable information."
    ],
    add_datetime_to_instructions=True, # Add time context
    markdown=True,
    exponential_backoff=True # Add robustness
)

crawler_agent = Agent(
    name="WebCrawler",
    model=model,
    tools=[Crawl4aiTools(max_length=None)], # No content length limit
    add_history_to_messages=True,
    num_history_responses=3,
    description="Extracts content from specific websites.",
    instructions=[
        "Use web_crawler to extract content from provided URLs.",
        "Summarize key points and include the URL."
    ],
    markdown=True,
    exponential_backoff=True
)

youtube_agent = Agent(
    name="YouTubeAnalyst",
    model=model,
    tools=[YouTubeTools()],
    add_history_to_messages=True,
    num_history_responses=3,
    description="Analyzes YouTube videos.",
    instructions=[
        "Extract captions and metadata for YouTube URLs.",
        "Summarize key points and include the video URL."
    ],
    markdown=True,
    exponential_backoff=True
)

email_agent = Agent(
    name="EmailAssistant",
    model=model,
    tools=[ResendTools(from_email=EMAIL_FROM, api_key=RESEND_API_KEY)],
    add_history_to_messages=True,
    num_history_responses=3,
    description="Sends emails professionally.",
    instructions=[
        "send professional emails based on context or user request.",
        f"Default recipient is {EMAIL_TO}, but use recipient specified in the query if provided.",
        "Include URLs and links clearly.",
        "Ensure the tone is professional and courteous."
    ],
    markdown=True,
    exponential_backoff=True
)

github_agent = Agent(
    name="GitHubResearcher",
    model=model,
    tools=[GithubTools(access_token=GITHUB_ACCESS_TOKEN)],
    add_history_to_messages=True,
    num_history_responses=3,
    description="Explores GitHub repositories.",
    instructions=[
        "Search repositories or list pull requests based on user query.",
        "Include repository URLs and summarize findings concisely."
    ],
    markdown=True,
    exponential_backoff=True,
    add_datetime_to_instructions=True
)

hackernews_agent = Agent(
    name="HackerNewsMonitor",
    model=model,
    tools=[HackerNewsTools()],
    add_history_to_messages=True,
    num_history_responses=3,
    description="Tracks Hacker News trends.",
    instructions=[
        "Fetch top stories using get_top_hackernews_stories.",
        "Summarize discussions and include story URLs."
    ],
    markdown=True,
    exponential_backoff=True,
    add_datetime_to_instructions=True
)

general_agent = Agent(
    name="GeneralAssistant",
    model=model,
    add_history_to_messages=True,
    num_history_responses=5, # More history for context
    description="Handles general queries and synthesizes information from specialists.",
    instructions=[
        "Answer general questions or combine specialist inputs.",
        "If specialists provide information, synthesize it clearly.",
        "If a query doesn't fit other specialists, attempt to answer directly.",
        "Maintain a professional tone."
    ],
    markdown=True,
    exponential_backoff=True
)

# --- Team Initialization (in Session State) ---
def initialize_team():
    """Initializes or re-initializes the research team."""
    return Team(
        name="ResearchAssistantTeam",
        mode="coordinate",
        model=model,
        members=[
            search_agent,
            crawler_agent,
            youtube_agent,
            email_agent,
            github_agent,
            hackernews_agent,
            general_agent
        ],
        description="Coordinates specialists to handle research tasks.",
        instructions=[
            "Analyze the query and assign tasks to specialists.",
            "Delegate based on task type:",
            "- Web searches: InternetSearcher",
            "- URL content: WebCrawler",
            "- YouTube videos: YouTubeAnalyst",
            "- Emails: EmailAssistant",
            "- GitHub queries: GitHubResearcher",
            "- Hacker News: HackerNewsMonitor",
            "- General or synthesis: GeneralAssistant",
            "Synthesize responses into a cohesive answer.",
            "Cite sources and maintain clarity.",
            "Always check previous conversations in memory before responding.",
            "When asked about previous information or to recall something mentioned before, refer to your memory of past interactions.",
            "Use all relevant information from memory when answering follow-up questions."
        ],
        success_criteria="The user's query has been thoroughly answered with information from all relevant specialists.",
        enable_agentic_context=True,      # Coordinator maintains context
        share_member_interactions=True, # Members see previous member interactions in context
        show_members_responses=False,     # Don't show raw member responses in final output
        markdown=True,
        show_tool_calls=False,            # Don't show raw tool calls in final output
        enable_team_history=True,         # Pass history between coordinator/members
        num_of_interactions_from_history=5 # Limit history passed
    )

if "team" not in st.session_state:
    st.session_state.team = initialize_team()

success_criteria="The user's query has been thoroughly answered with information from all relevant specialists."

# --- Streamlit UI ---
st.title("🤖 Research Assistant Team")
st.markdown("""
This team coordinates specialists to assist with:
- 🔍 Web searches
- 🌐 Website content extraction
- 📺 YouTube video analysis
- 📧 Email drafting/sending
- 💻 GitHub repository exploration
- 📰 Hacker News trends
- 🧠 General queries and synthesis
""")

# Display chat messages from history
for message in st.session_state.messages:
    with st.chat_message(message["role"]):
        st.markdown(message["content"])

# Handle user input
user_query = st.chat_input("Ask the research team anything...")

if user_query:
    # Add user message to chat history
    st.session_state.messages.append({"role": "user", "content": user_query})

    # Display user message
    with st.chat_message("user"):
        st.markdown(user_query)

    # Display team response (Streaming)
    with st.chat_message("assistant"):
        message_placeholder = st.empty()
        full_response = ""
        try:
            # Use stream=True for the team run
            response_stream: Iterator[RunResponse] = st.session_state.team.run(user_query, stream=True) # Ensure type hint

            for chunk in response_stream:
                # Check if content is present and a string
                if chunk.content and isinstance(chunk.content, str):
                    full_response += chunk.content
                    message_placeholder.markdown(full_response + "▌") # Add cursor effect
            message_placeholder.markdown(full_response) # Final response without cursor

            # Update memory debug information for display
            if hasattr(st.session_state.team, 'memory') and hasattr(st.session_state.team.memory, 'messages'):
                try:
                    # Extract only role and content safely
                    st.session_state.memory_dump = [
                        {"role": m.role if hasattr(m, 'role') else 'unknown',
                         "content": m.content if hasattr(m, 'content') else str(m)}
                        for m in st.session_state.team.memory.messages
                    ]
                except Exception as e:
                    st.session_state.memory_dump = f"Error accessing memory messages: {str(e)}"
            else:
                st.session_state.memory_dump = "Team memory object or messages not found/accessible."

            # Add the final assistant response to Streamlit's chat history
            st.session_state.messages.append({"role": "assistant", "content": full_response})

        except Exception as e:
            st.exception(e) # Show full traceback in Streamlit console for debugging
            error_message = f"An error occurred: {str(e)}\n\nPlease check your API keys and tool configurations. Try rephrasing your query."
            st.error(error_message)
            message_placeholder.markdown(f"⚠️ {error_message}")
            # Add error message to history for context
            st.session_state.messages.append({"role": "assistant", "content": f"Error: {str(e)}"})

# --- Sidebar ---
with st.sidebar:
    st.title("Team Settings")

    # Memory debug section
    if st.checkbox("Show Team Memory Contents", value=False):
        st.subheader("Team Memory Contents (Debug)")
        if "memory_dump" in st.session_state:
            try:
                # Use pformat for potentially complex structures
                memory_str = pformat(st.session_state.memory_dump, indent=2, width=80)
                st.code(memory_str, language="python")
            except Exception as format_e:
                st.warning(f"Could not format memory dump: {format_e}")
                st.json(st.session_state.memory_dump) # Fallback to json
        else:
            st.info("No memory contents to display yet. Interact with the team first.")

    st.markdown(f"**Session ID**: `{st.session_state.team_session_id}`")
    st.markdown(f"**Model**: {model_name}")

    # Memory information
    st.subheader("Team Memory")
    st.markdown("This team remembers conversations within this browser session. Clearing the chat resets the memory.")

    # Clear chat button
    if st.button("Clear Chat & Reset Team"):
        st.session_state.messages = []
        st.session_state.team_session_id = f"streamlit-team-session-{int(time.time())}" # New ID for clarity
        st.session_state.team = initialize_team() # Re-initialize the team to reset its state
        if "memory_dump" in st.session_state:
            del st.session_state.memory_dump # Clear the dump
        st.rerun()

    st.title("About")
    st.markdown("""
    **How it works**:
    - The team coordinator analyzes your query.
    - Tasks are delegated to specialists (Searcher, Crawler, YouTube Analyst, Email, GitHub, HackerNews, General).
    - Responses are synthesized into a final answer.
    - Team memory retains context within this session.

    **Example queries**:
    - "What are the latest AI breakthroughs?"
    - "Crawl agno.com and summarize the homepage."
    - "Summarize the YouTube video: https://www.youtube.com/watch?v=dQw4w9WgXcQ"
    - "Draft an email to [email protected] introducing our research services."
    - "Find popular AI repositories on GitHub created in the last month."
    - "What's trending on Hacker News today?"
    - "What was the first question I asked you?" (tests memory)
    """)

# --- Session State Initialization ---
# Initialize team_session_id for this specific browser session
if "team_session_id" not in st.session_state:
    st.session_state.team_session_id = f"streamlit-team-session-{int(time.time())}"
# Initialize chat message history
if "messages" not in st.session_state:
    st.session_state.messages = []

enable_agentic_context=True,      # Coordinator maintains context
share_member_interactions=True,     # Members see previous member interactions
enable_team_history=True,           # Pass history between coordinator/members
num_of_interactions_from_history=5  # Limit history passed

# app.py
import os
import streamlit as st
from dotenv import load_dotenv
import time
from pprint import pformat
from typing import Iterator # Added for type hinting

# Agno Imports
from agno.agent import Agent
from agno.models.openrouter import OpenRouter
from agno.team import Team
from agno.run.response import RunResponse # Added for type hinting
from agno.tools.duckduckgo import DuckDuckGoTools
from agno.tools.crawl4ai import Crawl4aiTools
from agno.tools.youtube import YouTubeTools
from agno.tools.resend import ResendTools
from agno.tools.github import GithubTools
from agno.tools.hackernews import HackerNewsTools

# --- Configuration ---
# Load environment variables from .env file
load_dotenv()

# Check for essential API keys
OPENROUTER_API_KEY = os.getenv("OPENROUTER_API_KEY")
EMAIL_FROM = os.getenv("EMAIL_FROM")
EMAIL_TO = os.getenv("EMAIL_TO") # Default recipient
GITHUB_ACCESS_TOKEN = os.getenv("GITHUB_ACCESS_TOKEN")
RESEND_API_KEY = os.getenv("RESEND_API_KEY") # ResendTools requires this

# Simple validation for required keys
required_keys = {
    "OPENROUTER_API_KEY": OPENROUTER_API_KEY,
    "EMAIL_FROM": EMAIL_FROM,
    "EMAIL_TO": EMAIL_TO,
    "GITHUB_ACCESS_TOKEN": GITHUB_ACCESS_TOKEN,
    "RESEND_API_KEY": RESEND_API_KEY
}

missing_keys = [name for name, key in required_keys.items() if not key]

if missing_keys:
    st.error(f"Missing required environment variables: {', '.join(missing_keys)}. Please set them in your .env file or system environment.")
    st.stop() # Stop execution if keys are missing

# Set Streamlit page configuration
st.set_page_config(
    page_title="Research Assistant Team",
    page_icon="🧠",
    layout="wide"
)

# --- Model Initialization ---
# Initialize OpenRouter model only, no fallback
try:
    model = OpenRouter(id="openrouter/optimus-alpha", api_key=OPENROUTER_API_KEY)
    model_name = "OpenRouter (openrouter/optimus-alpha)"
    st.sidebar.info(f"Using model: {model_name}")
except Exception as e:
    st.error(f"Failed to initialize OpenRouter model: {e}")
    st.stop()


# --- Session State Initialization ---
# Initialize team_session_id for this specific browser session
if "team_session_id" not in st.session_state:
    st.session_state.team_session_id = f"streamlit-team-session-{int(time.time())}"
# Initialize chat message history
if "messages" not in st.session_state:
    st.session_state.messages = []

# --- Agent Definitions ---
# Define specialized agents
search_agent = Agent(
    name="InternetSearcher",
    model=model,
    tools=[DuckDuckGoTools(search=True, news=False)],
    add_history_to_messages=True,
    num_history_responses=3, # Limit history passed to agent
    description="Expert at finding information online.",
    instructions=[
        "Use duckduckgo_search for web queries.",
        "Cite sources with URLs.",
        "Focus on recent, reliable information."
    ],
    add_datetime_to_instructions=True, # Add time context
    markdown=True,
    exponential_backoff=True # Add robustness
)

crawler_agent = Agent(
    name="WebCrawler",
    model=model,
    tools=[Crawl4aiTools(max_length=None)], # Consider setting a sensible max_length
    add_history_to_messages=True,
    num_history_responses=3,
    description="Extracts content from specific websites.",
    instructions=[
        "Use web_crawler to extract content from provided URLs.",
        "Summarize key points and include the URL."
    ],
    markdown=True,
    exponential_backoff=True
)

youtube_agent = Agent(
    name="YouTubeAnalyst",
    model=model,
    tools=[YouTubeTools()],
    add_history_to_messages=True,
    num_history_responses=3,
    description="Analyzes YouTube videos.",
    instructions=[
        "Extract captions and metadata for YouTube URLs.",
        "Summarize key points and include the video URL."
    ],
    markdown=True,
    exponential_backoff=True
)

email_agent = Agent(
    name="EmailAssistant",
    model=model,
    tools=[ResendTools(from_email=EMAIL_FROM, api_key=RESEND_API_KEY)], # Pass required args
    add_history_to_messages=True,
    num_history_responses=3,
    description="Sends emails professionally.",
    instructions=[
        "send professional emails based on context or user request.",
        f"Default recipient is {EMAIL_TO}, but use recipient specified in the query if provided.",
        "Include URLs and links clearly.",
        "Ensure the tone is professional and courteous."
    ],
    markdown=True,
    exponential_backoff=True
)

github_agent = Agent(
    name="GitHubResearcher",
    model=model,
    tools=[GithubTools(access_token=GITHUB_ACCESS_TOKEN)], # Pass required args
    add_history_to_messages=True,
    num_history_responses=3,
    description="Explores GitHub repositories.",
    instructions=[
        "Search repositories or list pull requests based on user query.",
        "Include repository URLs and summarize findings concisely."
    ],
    markdown=True,
    exponential_backoff=True,
    add_datetime_to_instructions=True
)

hackernews_agent = Agent(
    name="HackerNewsMonitor",
    model=model,
    tools=[HackerNewsTools()],
    add_history_to_messages=True,
    num_history_responses=3,
    description="Tracks Hacker News trends.",
    instructions=[
        "Fetch top stories using get_top_hackernews_stories.",
        "Summarize discussions and include story URLs."
    ],
    markdown=True,
    exponential_backoff=True,
    add_datetime_to_instructions=True
)

# Generalist Agent (No KB in this version)
general_agent = Agent(
    name="GeneralAssistant",
    model=model,
    add_history_to_messages=True,
    num_history_responses=5, # Can access slightly more history
    description="Handles general queries and synthesizes information from specialists.",
    instructions=[
        "Answer general questions or combine specialist inputs.",
        "If specialists provide information, synthesize it clearly.",
        "If a query doesn't fit other specialists, attempt to answer directly.",
        "Maintain a professional tone."
    ],
    markdown=True,
    exponential_backoff=True
)

# --- Team Initialization (in Session State) ---
def initialize_team():
    """Initializes or re-initializes the research team."""
    return Team(
        name="ResearchAssistantTeam",
        mode="coordinate",
        model=model,
        members=[
            search_agent,
            crawler_agent,
            youtube_agent,
            email_agent,
            github_agent,
            hackernews_agent,
            general_agent
        ],
        description="Coordinates specialists to handle research tasks.",
        instructions=[
            "Analyze the query and assign tasks to specialists.",
            "Delegate based on task type:",
            "- Web searches: InternetSearcher",
            "- URL content: WebCrawler",
            "- YouTube videos: YouTubeAnalyst",
            "- Emails: EmailAssistant",
            "- GitHub queries: GitHubResearcher",
            "- Hacker News: HackerNewsMonitor",
            "- General or synthesis: GeneralAssistant",
            "Synthesize responses into a cohesive answer.",
            "Cite sources and maintain clarity.",
            "Always check previous conversations in memory before responding.",
            "When asked about previous information or to recall something mentioned before, refer to your memory of past interactions.",
            "Use all relevant information from memory when answering follow-up questions."
        ],
        success_criteria="The user's query has been thoroughly answered with information from all relevant specialists.",
        enable_agentic_context=True,      # Coordinator maintains context
        share_member_interactions=True, # Members see previous member interactions in context
        show_members_responses=False,     # Don't show raw member responses in final output
        markdown=True,
        show_tool_calls=False,            # Don't show raw tool calls in final output
        enable_team_history=True,         # Pass history between coordinator/members
        num_of_interactions_from_history=5 # Limit history passed
    )

if "team" not in st.session_state:
    st.session_state.team = initialize_team()


# --- Streamlit UI ---
st.title("🤖 Research Assistant Team")
st.markdown("""
This team coordinates specialists to assist with:
- 🔍 Web searches
- 🌐 Website content extraction
- 📺 YouTube video analysis
- 📧 Email drafting/sending
- 💻 GitHub repository exploration
- 📰 Hacker News trends
- 🧠 General queries and synthesis
""")

# Display chat messages from history
for message in st.session_state.messages:
    with st.chat_message(message["role"]):
        st.markdown(message["content"])

# Handle user input
user_query = st.chat_input("Ask the research team anything...")

if user_query:
    # Add user message to chat history
    st.session_state.messages.append({"role": "user", "content": user_query})

    # Display user message
    with st.chat_message("user"):
        st.markdown(user_query)

    # Display team response (Streaming)
    with st.chat_message("assistant"):
        message_placeholder = st.empty()
        full_response = ""
        try:
            # Use stream=True for the team run
            response_stream: Iterator[RunResponse] = st.session_state.team.run(user_query, stream=True) # Ensure type hint

            for chunk in response_stream:
                # Check if content is present and a string
                if chunk.content and isinstance(chunk.content, str):
                    full_response += chunk.content
                    message_placeholder.markdown(full_response + "▌") # Add cursor effect
            message_placeholder.markdown(full_response) # Final response without cursor

            # Update memory debug information for display
            if hasattr(st.session_state.team, 'memory') and hasattr(st.session_state.team.memory, 'messages'):
                try:
                    # Extract only role and content safely
                    st.session_state.memory_dump = [
                        {"role": m.role if hasattr(m, 'role') else 'unknown',
                         "content": m.content if hasattr(m, 'content') else str(m)}
                        for m in st.session_state.team.memory.messages
                    ]
                except Exception as e:
                    st.session_state.memory_dump = f"Error accessing memory messages: {str(e)}"
            else:
                st.session_state.memory_dump = "Team memory object or messages not found/accessible."

            # Add the final assistant response to Streamlit's chat history
            st.session_state.messages.append({"role": "assistant", "content": full_response})

        except Exception as e:
            st.exception(e) # Show full traceback in Streamlit console for debugging
            error_message = f"An error occurred: {str(e)}\n\nPlease check your API keys and tool configurations. Try rephrasing your query."
            st.error(error_message)
            message_placeholder.markdown(f"⚠️ {error_message}")
            # Add error message to history for context
            st.session_state.messages.append({"role": "assistant", "content": f"Error: {str(e)}"})

# --- Sidebar ---
with st.sidebar:
    st.title("Team Settings")

    # Memory debug section
    if st.checkbox("Show Team Memory Contents", value=False):
        st.subheader("Team Memory Contents (Debug)")
        if "memory_dump" in st.session_state:
            try:
                # Use pformat for potentially complex structures
                memory_str = pformat(st.session_state.memory_dump, indent=2, width=80)
                st.code(memory_str, language="python")
            except Exception as format_e:
                st.warning(f"Could not format memory dump: {format_e}")
                st.json(st.session_state.memory_dump) # Fallback to json
        else:
            st.info("No memory contents to display yet. Interact with the team first.")

    st.markdown(f"**Session ID**: `{st.session_state.team_session_id}`")
    st.markdown(f"**Model**: {model_name}")

    # Memory information
    st.subheader("Team Memory")
    st.markdown("This team remembers conversations within this browser session. Clearing the chat resets the memory.")

    # Clear chat button
    if st.button("Clear Chat & Reset Team"):
        st.session_state.messages = []
        st.session_state.team_session_id = f"streamlit-team-session-{int(time.time())}" # New ID for clarity
        st.session_state.team = initialize_team() # Re-initialize the team to reset its state
        if "memory_dump" in st.session_state:
            del st.session_state.memory_dump # Clear the dump
        st.rerun()

    st.title("About")
    st.markdown("""
    **How it works**:
    - The team coordinator analyzes your query.
    - Tasks are delegated to specialists (Searcher, Crawler, YouTube Analyst, Email, GitHub, HackerNews, General).
    - Responses are synthesized into a final answer.
    - Team memory retains context within this session.

    **Example queries**:
    - "What are the latest AI breakthroughs?"
    - "Crawl agno.com and summarize the homepage."
    - "Summarize the YouTube video: https://www.youtube.com/watch?v=dQw4w9WgXcQ"
    - "Draft an email to [email protected] introducing our research services."
    - "Find popular AI repositories on GitHub created in the last month."
    - "What's trending on Hacker News today?"
    - "What was the first question I asked you?" (tests memory)
    """)

uv run streamlit run main.py

# Simple validation for required keys
required_keys = {
    "OPENROUTER_API_KEY": OPENROUTER_API_KEY,
    "EMAIL_FROM": EMAIL_FROM,
    "EMAIL_TO": EMAIL_TO,
    "GITHUB_ACCESS_TOKEN": GITHUB_ACCESS_TOKEN,
    "RESEND_API_KEY": RESEND_API_KEY
}

missing_keys = [name for name, key in required_keys.items() if not key]

if missing_keys:
    st.error(f"Missing required environment variables: {', '.join(missing_keys)}. Please set them in your .env file or system environment.")
    st.stop() # Stop execution if keys are missing

exponential_backoff=True  # Add robustness

# Alternative implementation (not in current code)
model = OpenRouter(
    id="openrouter/optimus-alpha",
    api_key=OPENROUTER_API_KEY,
    fallback_models=[
        "openai/gpt-4-turbo",
        "anthropic/claude-3-opus"
    ]
)

# Memory debug section
if st.checkbox("Show Team Memory Contents", value=False):
    st.subheader("Team Memory Contents (Debug)")
    if "memory_dump" in st.session_state:
        try:
            # Use pformat for potentially complex structures
            memory_str = pformat(st.session_state.memory_dump, indent=2, width=80)
            st.code(memory_str, language="python")
        except Exception as format_e:
            st.warning(f"Could not format memory dump: {format_e}")
            st.json(st.session_state.memory_dump) # Fallback to json
    else:
        st.info("No memory contents to display yet. Interact with the team first.")