agent2.py

import os
import requests
import base64
from typing import Dict, Any, List
from langchain.docstore.document import Document
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.retrievers import BM25Retriever
from smolagents import CodeAgent, OpenAIServerModel, tool, Tool
from smolagents.vision_web_browser import initialize_driver, save_screenshot, helium_instructions
from smolagents.agents import ActionStep
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.common.keys import Keys
import helium
from PIL import Image
from io import BytesIO
from time import sleep
from smolagents import PythonInterpreterTool, SpeechToTextTool

# Langfuse observability imports
from opentelemetry.sdk.trace import TracerProvider
from openinference.instrumentation.smolagents import SmolagentsInstrumentor
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
from opentelemetry import trace
from opentelemetry.trace import format_trace_id
from langfuse import Langfuse


class BM25RetrieverTool(Tool):
    """
    BM25 retriever tool for document search when text documents are available
    """
    name = "bm25_retriever"
    description = "Uses BM25 search to retrieve relevant parts of uploaded documents. Use this when the question references an attached file or document."
    inputs = {
        "query": {
            "type": "string",
            "description": "The search query to find relevant document sections.",
        }
    }
    output_type = "string"

    def __init__(self, docs=None, **kwargs):
        super().__init__(**kwargs)
        self.docs = docs or []
        self.retriever = None
        if self.docs:
            self.retriever = BM25Retriever.from_documents(self.docs, k=5)

    def forward(self, query: str) -> str:
        if not self.retriever:
            return "No documents loaded for retrieval."

        assert isinstance(query, str), "Your search query must be a string"

        docs = self.retriever.invoke(query)
        return "\nRetrieved documents:\n" + "".join([
            f"\n\n===== Document {str(i)} =====\n" + doc.page_content
            for i, doc in enumerate(docs)
        ])


@tool
def search_item_ctrl_f(text: str, nth_result: int = 1) -> str:
    """Search for text on the current page via Ctrl + F and jump to the nth occurrence.
    
    Args:
        text: The text string to search for on the webpage
        nth_result: Which occurrence to jump to (default is 1 for first occurrence)
        
    Returns:
        str: Result of the search operation with match count and navigation status
    """
    try:
        driver = helium.get_driver()
        elements = driver.find_elements(By.XPATH, f"//*[contains(text(), '{text}')]")
        if nth_result > len(elements):
            return f"Match n°{nth_result} not found (only {len(elements)} matches found)"
        result = f"Found {len(elements)} matches for '{text}'."
        elem = elements[nth_result - 1]
        driver.execute_script("arguments[0].scrollIntoView(true);", elem)
        result += f"Focused on element {nth_result} of {len(elements)}"
        return result
    except Exception as e:
        return f"Error searching for text: {e}"


@tool
def go_back() -> str:
    """Navigate back to the previous page in browser history.
    
    Returns:
        str: Confirmation message or error description
    """
    try:
        driver = helium.get_driver()
        driver.back()
        return "Navigated back to previous page"
    except Exception as e:
        return f"Error going back: {e}"


@tool
def close_popups() -> str:
    """Close any visible modal or pop-up on the page by sending ESC key.
    
    Returns:
        str: Confirmation message or error description
    """
    try:
        driver = helium.get_driver()
        webdriver.ActionChains(driver).send_keys(Keys.ESCAPE).perform()
        return "Attempted to close popups"
    except Exception as e:
        return f"Error closing popups: {e}"


@tool
def scroll_page(direction: str = "down", amount: int = 3) -> str:
    """Scroll the webpage in the specified direction.
    
    Args:
        direction: Direction to scroll, either 'up' or 'down'
        amount: Number of scroll actions to perform
        
    Returns:
        str: Confirmation message or error description
    """
    try:
        driver = helium.get_driver()
        for _ in range(amount):
            if direction.lower() == "down":
                driver.execute_script("window.scrollBy(0, 300);")
            elif direction.lower() == "up":
                driver.execute_script("window.scrollBy(0, -300);")
            sleep(0.5)
        return f"Scrolled {direction} {amount} times"
    except Exception as e:
        return f"Error scrolling: {e}"


@tool
def get_page_text() -> str:
    """Extract all visible text from the current webpage.
    
    Returns:
        str: The visible text content of the page
    """
    try:
        driver = helium.get_driver()
        text = driver.find_element(By.TAG_NAME, "body").text
        return f"Page text (first 2000 chars): {text[:2000]}"
    except Exception as e:
        return f"Error getting page text: {e}"


def save_screenshot_callback(memory_step: ActionStep, agent: CodeAgent) -> None:
    """Save screenshots for web browser automation"""
    try:
        sleep(1.0)
        driver = helium.get_driver()
        if driver is not None:
            # Clean up old screenshots
            for previous_memory_step in agent.memory.steps:
                if isinstance(previous_memory_step, ActionStep) and previous_memory_step.step_number <= memory_step.step_number - 2:
                    previous_memory_step.observations_images = None

            png_bytes = driver.get_screenshot_as_png()
            image = Image.open(BytesIO(png_bytes))
            memory_step.observations_images = [image.copy()]

            # Update observations with current URL
            url_info = f"Current url: {driver.current_url}"
            memory_step.observations = (
                url_info if memory_step.observations is None 
                else memory_step.observations + "\n" + url_info
            )
    except Exception as e:
        print(f"Error in screenshot callback: {e}")


class GAIAAgent:
    """
    GAIA agent using smolagents with Gemini 2.0 Flash and Langfuse observability
    """

    def __init__(self, user_id: str = None, session_id: str = None):
        """Initialize the agent with Gemini 2.0 Flash, tools, and Langfuse observability"""

        # Get API keys
        gemini_api_key = os.environ.get("GOOGLE_API_KEY")
        if not gemini_api_key:
            raise ValueError("GOOGLE_API_KEY environment variable not found")

        # Initialize Langfuse observability
        self._setup_langfuse_observability()

        # Initialize Gemini 2.0 Flash model
        self.model = OpenAIServerModel(
            model_id="gemini-2.0-flash",
            api_base="https://generativelanguage.googleapis.com/v1beta/openai/",
            api_key=gemini_api_key,
        )

        # Store user and session IDs for tracking
        self.user_id = user_id or "gaia-user"
        self.session_id = session_id or "gaia-session"

        # GAIA system prompt from the leaderboard
        self.system_prompt = """You are a general AI assistant. I will ask you a question. Report your thoughts and reasoning process clearly. You should use the available tools to gather information and solve problems step by step.

When using web browser automation:
- Use helium commands like go_to(), click(), scroll_down()
- Take screenshots to see what's happening
- Handle popups and forms appropriately
- Be patient with page loading

For document retrieval:
- Use the BM25 retriever when there are text documents attached
- Search with relevant keywords from the question

Your final answer should be as few words as possible, a number, or a comma-separated list. Don't use articles, abbreviations, or units unless specified."""

        # Initialize retriever tool (will be updated when documents are loaded)
        self.retriever_tool = BM25RetrieverTool()

        # Initialize web driver for browser automation
        self.driver = None

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

        # Initialize Langfuse client
        self.langfuse = Langfuse()

    def _setup_langfuse_observability(self):
        """Set up Langfuse observability with OpenTelemetry"""
        # Get Langfuse keys from environment variables
        langfuse_public_key = os.environ.get("LANGFUSE_PUBLIC_KEY")
        langfuse_secret_key = os.environ.get("LANGFUSE_SECRET_KEY")
        
        if not langfuse_public_key or not langfuse_secret_key:
            print("Warning: LANGFUSE_PUBLIC_KEY or LANGFUSE_SECRET_KEY not found. Observability will be limited.")
            return

        # Set up Langfuse environment variables
        os.environ["LANGFUSE_HOST"] = os.environ.get("LANGFUSE_HOST", "https://cloud.langfuse.com")
        
        langfuse_auth = base64.b64encode(
            f"{langfuse_public_key}:{langfuse_secret_key}".encode()
        ).decode()
        
        os.environ["OTEL_EXPORTER_OTLP_ENDPOINT"] = os.environ.get("LANGFUSE_HOST") + "/api/public/otel"
        os.environ["OTEL_EXPORTER_OTLP_HEADERS"] = f"Authorization=Basic {langfuse_auth}"

        # Create a TracerProvider for OpenTelemetry
        trace_provider = TracerProvider()
        
        # Add a SimpleSpanProcessor with the OTLPSpanExporter to send traces
        trace_provider.add_span_processor(SimpleSpanProcessor(OTLPSpanExporter()))
        
        # Set the global default tracer provider
        trace.set_tracer_provider(trace_provider)
        self.tracer = trace.get_tracer(__name__)
        
        # Instrument smolagents with the configured provider
        SmolagentsInstrumentor().instrument(tracer_provider=trace_provider)

    def _create_agent(self):
        """Create the CodeAgent with tools"""
        base_tools = [
            self.retriever_tool, 
            search_item_ctrl_f, 
            go_back, 
            close_popups,
            scroll_page,
            get_page_text
        ]

        self.agent = CodeAgent(
            tools=base_tools + [PythonInterpreterTool(), SpeechToTextTool()],
            model=self.model,
            add_base_tools=False,
            planning_interval=2,
            additional_authorized_imports=["helium", "requests", "BeautifulSoup", "json"],
            step_callbacks=[save_screenshot_callback] if self.driver else [],
            max_steps=10,
            description=self.system_prompt,
            verbosity_level=2,
        )

    def initialize_browser(self):
        """Initialize browser for web automation tasks"""
        try:
            chrome_options = webdriver.ChromeOptions()
            chrome_options.add_argument("--force-device-scale-factor=1")
            chrome_options.add_argument("--window-size=1000,1350")
            chrome_options.add_argument("--disable-pdf-viewer")
            chrome_options.add_argument("--window-position=0,0")
            chrome_options.add_argument("--no-sandbox")
            chrome_options.add_argument("--disable-dev-shm-usage")

            self.driver = helium.start_chrome(headless=False, options=chrome_options)

            # Recreate agent with browser tools
            self._create_agent()

            # Import helium for the agent
            self.agent.python_executor("from helium import *")

            return True
        except Exception as e:
            print(f"Failed to initialize browser: {e}")
            return False

    def load_documents_from_file(self, file_path: str):
        """Load and process documents from a file for BM25 retrieval"""
        try:
            # Read file content
            with open(file_path, 'r', encoding='utf-8') as f:
                content = f.read()

            # Split into chunks
            text_splitter = RecursiveCharacterTextSplitter(
                chunk_size=1000,
                chunk_overlap=200,
                separators=["\n\n", "\n", ".", " ", ""]
            )

            # Create documents
            chunks = text_splitter.split_text(content)
            docs = [Document(page_content=chunk, metadata={"source": file_path}) 
                    for chunk in chunks]

            # Update retriever tool
            self.retriever_tool = BM25RetrieverTool(docs)

            # Recreate agent with updated retriever
            self._create_agent()

            print(f"Loaded {len(docs)} document chunks from {file_path}")
            return True

        except Exception as e:
            print(f"Error loading documents from {file_path}: {e}")
            return False

    def download_gaia_file(self, task_id: str, api_url: str = "https://agents-course-unit4-scoring.hf.space") -> str:
        """Download file associated with GAIA task_id"""
        try:
            response = requests.get(f"{api_url}/files/{task_id}", timeout=30)
            response.raise_for_status()

            filename = f"task_{task_id}_file.txt"
            with open(filename, 'wb') as f:
                f.write(response.content)

            return filename
        except Exception as e:
            print(f"Failed to download file for task {task_id}: {e}")
            return None

    def solve_gaia_question(self, question_data: Dict[str, Any], tags: List[str] = None) -> str:
        """
        Solve a GAIA question with full Langfuse observability
        """
        question = question_data.get("Question", "")
        task_id = question_data.get("task_id", "")
        
        # Prepare tags for observability
        trace_tags = ["gaia-agent", "question-solving"]
        if tags:
            trace_tags.extend(tags)
        if task_id:
            trace_tags.append(f"task-{task_id}")

        # Start Langfuse trace with OpenTelemetry
        with self.tracer.start_as_current_span("GAIA-Question-Solving") as span:
            try:
                # Set span attributes for tracking
                span.set_attribute("langfuse.user.id", self.user_id)
                span.set_attribute("langfuse.session.id", self.session_id)
                span.set_attribute("langfuse.tags", trace_tags)
                span.set_attribute("gaia.task_id", task_id)
                span.set_attribute("gaia.question_length", len(question))

                # Get trace ID for Langfuse linking
                current_span = trace.get_current_span()
                span_context = current_span.get_span_context()
                trace_id = span_context.trace_id
                formatted_trace_id = format_trace_id(trace_id)

                # Create Langfuse trace
                langfuse_trace = self.langfuse.trace(
                    id=formatted_trace_id,
                    name="GAIA Question Solving",
                    input={"question": question, "task_id": task_id},
                    user_id=self.user_id,
                    session_id=self.session_id,
                    tags=trace_tags,
                    metadata={
                        "model": self.model.model_id,
                        "question_length": len(question),
                        "has_file": bool(task_id)
                    }
                )

                # Download and load file if task_id provided
                file_loaded = False
                if task_id:
                    file_path = self.download_gaia_file(task_id)
                    if file_path:
                        file_loaded = self.load_documents_from_file(file_path)
                        span.set_attribute("gaia.file_loaded", file_loaded)
                        print(f"Loaded file for task {task_id}")

                # Check if this requires web browsing
                web_indicators = ["navigate", "browser", "website", "webpage", "url", "click", "search on"]
                needs_browser = any(indicator in question.lower() for indicator in web_indicators)
                span.set_attribute("gaia.needs_browser", needs_browser)

                if needs_browser and not self.driver:
                    print("Initializing browser for web automation...")
                    browser_initialized = self.initialize_browser()
                    span.set_attribute("gaia.browser_initialized", browser_initialized)

                # Prepare the prompt
                prompt = f"""
Question: {question}
{f'Task ID: {task_id}' if task_id else ''}
{f'File loaded: Yes' if file_loaded else 'File loaded: No'}

Solve this step by step. Use the available tools to gather information and provide a precise answer.
                """

                if needs_browser:
                    prompt += "\n" + helium_instructions

                print("=== AGENT REASONING ===")
                result = self.agent.run(prompt)
                print("=== END REASONING ===")

                # Update Langfuse trace with result
                langfuse_trace.update(
                    output={"answer": str(result)},
                    end_time=None  # Will be set automatically
                )

                # Add success attributes
                span.set_attribute("gaia.success", True)
                span.set_attribute("gaia.answer_length", len(str(result)))

                # Flush Langfuse data
                self.langfuse.flush()

                return str(result)

            except Exception as e:
                error_msg = f"Error processing question: {str(e)}"
                print(error_msg)
                
                # Log error to span and Langfuse
                span.set_attribute("gaia.success", False)
                span.set_attribute("gaia.error", str(e))
                
                if 'langfuse_trace' in locals():
                    langfuse_trace.update(
                        output={"error": error_msg},
                        level="ERROR"
                    )
                
                self.langfuse.flush()
                return error_msg
                
            finally:
                # Clean up browser if initialized
                if self.driver:
                    try:
                        helium.kill_browser()
                    except:
                        pass

    def evaluate_answer(self, question: str, answer: str, expected_answer: str = None) -> Dict[str, Any]:
        """
        Evaluate the agent's answer using LLM-as-a-Judge and optionally compare with expected answer
        """
        evaluation_prompt = f"""
Please evaluate the following answer to a question on a scale of 1-5:

Question: {question}
Answer: {answer}
{f'Expected Answer: {expected_answer}' if expected_answer else ''}

Rate the answer on:
1. Accuracy (1-5)
2. Completeness (1-5) 
3. Clarity (1-5)

Provide your rating as JSON: {{"accuracy": X, "completeness": Y, "clarity": Z, "overall": W, "reasoning": "explanation"}}
        """

        try:
            # Use the same model to evaluate
            evaluation_result = self.agent.run(evaluation_prompt)
            
            # Try to parse JSON response
            import json
            try:
                scores = json.loads(evaluation_result)
                return scores
            except:
                # Fallback if JSON parsing fails
                return {
                    "accuracy": 3,
                    "completeness": 3,
                    "clarity": 3,
                    "overall": 3,
                    "reasoning": "Could not parse evaluation response",
                    "raw_evaluation": evaluation_result
                }
                
        except Exception as e:
            return {
                "accuracy": 1,
                "completeness": 1,
                "clarity": 1,
                "overall": 1,
                "reasoning": f"Evaluation failed: {str(e)}"
            }

    def add_user_feedback(self, trace_id: str, feedback_score: int, comment: str = None):
        """
        Add user feedback to a specific trace
        
        Args:
            trace_id: The trace ID to add feedback to
            feedback_score: Score from 0-5 (0=very bad, 5=excellent)
            comment: Optional comment from user
        """
        try:
            self.langfuse.score(
                trace_id=trace_id,
                name="user-feedback",
                value=feedback_score,
                comment=comment
            )
            self.langfuse.flush()
            print(f"User feedback added: {feedback_score}/5")
        except Exception as e:
            print(f"Error adding user feedback: {e}")


# Example usage with observability
if __name__ == "__main__":
    # Set up environment variables (you need to set these)
    # os.environ["GOOGLE_API_KEY"] = "your-gemini-api-key"
    # os.environ["LANGFUSE_PUBLIC_KEY"] = "pk-lf-..."
    # os.environ["LANGFUSE_SECRET_KEY"] = "sk-lf-..."
    
    # Test the agent with observability
    agent = GAIAAgent(
        user_id="test-user-123",
        session_id="test-session-456"
    )

    # Example question
    question_data = {
        "Question": "How many studio albums Mercedes Sosa has published between 2000-2009?",
        "task_id": ""
    }

    # Solve with full observability
    answer = agent.solve_gaia_question(
        question_data, 
        tags=["music-question", "discography"]
    )
    print(f"Answer: {answer}")

    # Evaluate the answer
    evaluation = agent.evaluate_answer(
        question_data["Question"], 
        answer
    )
    print(f"Evaluation: {evaluation}")