agent.py

import os
import ast
import re
import json
import operator as op
from pathlib import Path
from typing import List, TypedDict, Annotated, Optional
import requests
from urllib.parse import urlparse
import shutil
import io
from typing import Dict, Any

from langchain.tools import tool, StructuredTool
from langchain_community.document_loaders import (
    CSVLoader, PyPDFLoader, YoutubeLoader
)
from langchain_community.document_loaders import AssemblyAIAudioTranscriptLoader
from langchain.chat_models import init_chat_model
from langchain.agents import initialize_agent, AgentType
from langchain_community.retrievers import BM25Retriever
from langchain.schema import BaseMessage, SystemMessage, HumanMessage, AIMessage
from langgraph.graph.message import add_messages
from langgraph.graph import START, END, StateGraph
from langgraph.prebuilt import ToolNode, tools_condition
from langchain_core.documents import Document

from youtube_transcript_api import YouTubeTranscriptApi
from PIL import Image
import pytesseract
import fitz  # PyMuPDF
from dotenv import load_dotenv
from contextlib import redirect_stdout
from langchain_community.tools import TavilySearchResults
from tavily import TavilyClient
from langchain.text_splitter import RecursiveCharacterTextSplitter

# Load environment variables from .env file
# in HF Spaces, the .env file is saved in Variables and secrets in settings
load_dotenv()

# Initialize Tavily client (after loading environment variables)
tavily_client = TavilyClient(api_key=os.environ.get("TAVILY_API_KEY"))
print(tavily_client)

# === System Prompt ===
SYSTEM_PROMPT = """
You are a general AI assistant. I will ask you a question. Report your thoughts, and finish your answer with the following template:
FINAL ANSWER: [YOUR FINAL ANSWER].
YOUR FINAL ANSWER should be a number OR as few words as possible OR a comma separated list of numbers and/or strings. If you are asked for a number, don't use comma to write your number nor use units such as $ or percent sign unless specified otherwise. If you are asked for a string, don't use articles, neither abbreviations (e.g. for cities), and write the digits in plain text unless specified otherwise. If you are asked for a comma separated list, apply the above rules depending on whether the element to be put in the list is a number or a string.
""".strip()

@tool
def calculate(expr: str) -> str:
    """Evaluate a math expression. Supports operations, numpy and math functions."""
    try:
        import math
        import numpy as np
        
        # Comprehensive math namespace
        safe_dict = {
            **{k: v for k, v in math.__dict__.items() if not k.startswith('_')},
            'np': np,
            'array': np.array,
            'mean': np.mean,
            'median': np.median,
            'std': np.std,
            'sum': np.sum,
            'abs': abs,
            'round': round,
            'max': max,
            'min': min
        }
        
        result = eval(expr, {"__builtins__": {}}, safe_dict)
        # Format result appropriately
        if isinstance(result, (np.ndarray, list)):
            return str(result)
        if isinstance(result, (int, float)):
            return str(float(result))
        return str(result)
    except Exception as e:
        return f"Error calculating expression: {e}"

@tool
def web_search(query: str) -> str:
    """Search the web using DuckDuckGo and SerpAPI for comprehensive results."""
    try:
        from langchain.utilities import DuckDuckGoSearchRun
        from langchain_community.utilities import SerpAPIWrapper
        
        # Try multiple search engines
        results = []
        
        # DuckDuckGo search
        ddg = DuckDuckGoSearchRun()
        ddg_results = ddg.run(query)
        results.append(ddg_results)
        
        # SerpAPI search if API key available
        if os.getenv("SERPAPI_API_KEY"):
            serpapi = SerpAPIWrapper()
            serp_results = serpapi.run(query)
            results.append(serp_results)
            
        # Combine and summarize results
        combined_results = "\n\n".join(results)
        return combined_results[:1000]  # Limit length for better handling
    except Exception as e:
        return f"Error performing web search: {e}"

@tool
def wikipedia_search(query: str) -> str:
    """Search Wikipedia for a general-topic query."""
    try:
        from langchain.utilities import WikipediaAPIWrapper
        return WikipediaAPIWrapper().run(query)
    except Exception as e:
        return f"Error searching Wikipedia: {e}"
    
    
@tool
def tavily_search(query: str) -> str:
    """Search the web using Tavily for comprehensive results."""
    try:
        results = tavily_client.search(query)
        # You can format the results as needed; here we just return the summary or first result
        if isinstance(results, dict) and "results" in results:
            return results["results"][0].get("content", "No content found.")
        elif isinstance(results, list) and results:
            return results[0].get("content", "No content found.")
        return str(results)
    except Exception as e:
        return f"Error performing Tavily search: {e}"
        
@tool
def advanced_search(query: str, max_results: int = 5) -> str:
    """Advanced web search with multiple strategies and better result parsing."""
    try:
        # Try multiple search approaches
        search_results = []
        
        # Primary search
        results = tavily_client.search(
            query, 
            search_depth="advanced",
            max_results=max_results,
            include_answer=True,
            include_raw_content=True,
            include_domains=["arxiv.org", "usgs.gov", "nih.gov", "pubmed.ncbi.nlm.nih.gov"]
        )
        
        if isinstance(results, dict):
            # Include direct answer if available
            if results.get("answer"):
                search_results.append(f"DIRECT ANSWER: {results['answer']}")
            
            # Process search results
            if results.get("results"):
                for i, result in enumerate(results["results"], 1):
                    title = result.get("title", "")
                    content = result.get("content", "")
                    url = result.get("url", "")
                    
                    # Extract more content for academic sources
                    if any(domain in url for domain in ["arxiv.org", "usgs.gov", "nih.gov"]):
                        content = content[:1000]  # More content for academic sources
                    else:
                        content = content[:500]
                    
                    search_results.append(
                        f"RESULT {i}:\nTitle: {title}\nURL: {url}\nContent: {content}\n"
                    )
        
        return "\n".join(search_results)
    
    except Exception as e:
        return f"Search error: {e}"

@tool
def arxiv_search(query: str, date_filter: str = "") -> str:
    """Specialized search for arXiv papers with date filtering."""
    try:
        # Construct arXiv-specific search
        arxiv_query = f"site:arxiv.org {query}"
        if date_filter:
            arxiv_query += f" {date_filter}"
        
        results = tavily_client.search(
            arxiv_query,
            search_depth="advanced", 
            max_results=8,
            include_raw_content=True
        )
        
        if isinstance(results, dict) and results.get("results"):
            arxiv_results = []
            for result in results["results"]:
                if "arxiv.org" in result.get("url", ""):
                    title = result.get("title", "")
                    content = result.get("content", "")
                    url = result.get("url", "")
                    
                    arxiv_results.append(f"ArXiv Paper:\nTitle: {title}\nURL: {url}\nContent: {content[:800]}\n")
            
            return "\n".join(arxiv_results) if arxiv_results else "No arXiv papers found"
        
        return "No results found"
    
    except Exception as e:
        return f"ArXiv search error: {e}"

@tool
def targeted_search(base_query: str, additional_terms: List[str]) -> str:
    """Perform multiple targeted searches with different term combinations."""
    try:
        all_results = []
        
        for terms in additional_terms:
            query = f"{base_query} {terms}"
            results = tavily_client.search(query, max_results=3)
            
            if isinstance(results, dict) and results.get("results"):
                all_results.append(f"=== Search: {query} ===")
                for result in results["results"]:
                    all_results.append(f"Title: {result.get('title', '')}")
                    all_results.append(f"URL: {result.get('url', '')}")
                    all_results.append(f"Content: {result.get('content', '')[:400]}\n")
        
        return "\n".join(all_results)
    
    except Exception as e:
        return f"Targeted search error: {e}"

@tool
def extract_zip_codes(text: str) -> str:
    """Extract 5-digit zip codes from text."""
    try:
        # Look for 5-digit zip codes
        zip_pattern = r'\b\d{5}\b'
        zip_codes = re.findall(zip_pattern, text)
        
        # Remove duplicates and sort
        unique_zips = sorted(list(set(zip_codes)))
        
        if unique_zips:
            return f"Found zip codes: {', '.join(unique_zips)}"
        else:
            return "No 5-digit zip codes found in text"
    
    except Exception as e:
        return f"Zip code extraction error: {e}"

@tool
def academic_citation_search(paper_info: str) -> str:
    """Search for academic papers that cite or are cited by the given paper."""
    try:
        # Search for papers that reference the given paper
        citation_queries = [
            f'"{paper_info}" citations references',
            f'{paper_info} "cited by"',
            f'{paper_info} bibliography references',
            f'site:scholar.google.com {paper_info}'
        ]
        
        results = []
        for query in citation_queries:
            search_result = tavily_client.search(query, max_results=3)
            if isinstance(search_result, dict) and search_result.get("results"):
                results.extend(search_result["results"])
        
        formatted_results = []
        for result in results[:5]:  # Top 5 citation results
            formatted_results.append(
                f"Citation Source: {result.get('title', '')}\n"
                f"URL: {result.get('url', '')}\n"
                f"Content: {result.get('content', '')[:500]}\n"
            )
        
        return "\n".join(formatted_results)
    
    except Exception as e:
        return f"Citation search error: {e}"

@tool
def image_recognition(image_path: str) -> str:
    """Analyze and extract text from an image using Tesseract OCR."""
    try:
        img = Image.open(image_path)
        return pytesseract.image_to_string(img)
    except Exception as e:
        return f"Error processing image: {e}"

@tool
def read_pdf(pdf_path: str) -> str:
    """Read and extract text from a PDF document."""
    try:
        doc = fitz.open(pdf_path)
        return "".join(page.get_text() for page in doc)
    except Exception as e:
        return f"Error reading PDF: {e}"

@tool
def read_csv(csv_path: str) -> str:
    """Read and extract text from a CSV file, row by row."""
    try:
        loader = CSVLoader(csv_path, encoding='utf-8')
        docs = loader.load()
        return "\n".join(doc.page_content for doc in docs)
    except Exception as e:
        return f"Error reading CSV: {e}"

@tool
def read_spreadsheet(spreadsheet_path: str) -> str:
    """Read a spreadsheet into a DataFrame and return CSV text."""
    try:
        import pandas as pd
        df = pd.read_excel(spreadsheet_path)
        return df.to_csv(index=False)
    except Exception as e:
        return f"Error reading spreadsheet: {e}"

@tool
def youtube_transcript_tool(video_url: str) -> str:
    """Download the transcript of a YouTube video using LangChain YoutubeLoader."""
    try:
        loader = YoutubeLoader.from_youtube_url(video_url)
        docs = loader.load()
        return "\n".join(doc.page_content for doc in docs)
    except Exception as e:
        return f"Error fetching YouTube transcript: {e}"

@tool
def youtube_transcript_api(video_url_or_id: str) -> str:
    """Download transcript from YouTube using youtube-transcript-api."""
    try:
        match = re.search(r"(?:v=|youtu\.be/)([A-Za-z0-9_-]{11})", video_url_or_id)
        vid = match.group(1) if match else video_url_or_id
        entries = YouTubeTranscriptApi.get_transcript(vid)
        return " ".join(segment["text"] for segment in entries)
    except Exception as e:
        return f"Error fetching transcript via API: {e}"

@tool
def transcribe_audio(audio_path: str) -> str:
    """Transcribe audio file (e.g., MP3) using AssemblyAI."""
    try:

        loader = AssemblyAIAudioTranscriptLoader(file_path=audio_path)
        docs = loader.load()
        return "\n".join(doc.page_content for doc in docs)
    except Exception as e:
        return f"Error transcribing audio: {e}"


@tool
def read_jsonl(jsonl_path: str) -> str:
    """Read and extract data from a JSONL (JSON Lines) file."""
    try:
        data = []
        with open(jsonl_path, 'r', encoding='utf-8') as file:
            for line in file:
                data.append(json.loads(line))
        return json.dumps(data, indent=2)
    except Exception as e:
        return f"Error reading JSONL file: {e}"

@tool
def python_interpreter(code: str) -> str:
    """Execute Python code and return the output. Supports data analysis, plotting, and general Python operations."""
    try:

        
        # Set up a safe globals environment
        safe_globals = {
            'pd': __import__('pandas'),
            'np': __import__('numpy'),
            'plt': __import__('matplotlib.pyplot'),
            'json': __import__('json'),
            're': __import__('re'),
            'math': __import__('math'),
        }
        
        # Capture output
        buffer = io.StringIO()
        with redirect_stdout(buffer):
            # Execute the code in a safe environment
            exec(code, safe_globals)
            
        return buffer.getvalue() or "Code executed successfully (no output)"
    except Exception as e:
        return f"Error executing Python code: {e}"

@tool
def download_file(url_or_path: str, save_dir: str = "./downloads") -> str:
    """Download a file from URL or copy from local path to the downloads directory."""
    try:
        # Create downloads directory if it doesn't exist
        save_dir = Path(save_dir)
        save_dir.mkdir(parents=True, exist_ok=True)
        
        # Check if input is URL or local path
        if url_or_path.startswith(('http://', 'https://')):
            # Handle URL download
            response = requests.get(url_or_path, stream=True)
            response.raise_for_status()
            
            # Get filename from URL or Content-Disposition header
            filename = response.headers.get('Content-Disposition')
            if filename and 'filename=' in filename:
                filename = filename.split('filename=')[1].strip('"')
            else:
                filename = Path(urlparse(url_or_path).path).name
                
            save_path = save_dir / filename
            
            # Download file
            with open(save_path, 'wb') as f:
                shutil.copyfileobj(response.raw, f)
                
        else:
            # Handle local file copy
            src_path = Path(url_or_path)
            if not src_path.exists():
                return f"Error: Source file {url_or_path} not found"
                
            save_path = save_dir / src_path.name
            shutil.copy2(src_path, save_path)
            
        return f"File successfully saved to {save_path}"
        
    except Exception as e:
        return f"Error downloading/copying file: {e}"


@tool
def extract_table(file_path: str, query: str = "") -> str:
    """Extract relevant rows from a CSV or Excel file based on a query."""
    import pandas as pd
    ext = Path(file_path).suffix.lower()
    if ext in [".csv"]:
        df = pd.read_csv(file_path)
    elif ext in [".xlsx", ".xls"]:
        df = pd.read_excel(file_path)
        text_content = df.to_string()
        loaded_docs = [Document(page_content=text_content)]
    else:
        return "Unsupported file type."
    # Simple filter: return all if no query, else filter columns containing query
    if query:
        mask = df.apply(lambda row: row.astype(str).str.contains(query, case=False).any(), axis=1)
        df = df[mask]
    return df.head(10).to_csv(index=False)

@tool
def summarize(text: str, llm=None) -> str:
    """Summarize a long text chunk."""
    if llm is None:
        return "No LLM provided for summarization."
    return llm.invoke([
        SystemMessage(content="Summarize the following:"),
        HumanMessage(content=text)
    ]).content

# Update tools list
tools: List[StructuredTool] = [
    calculate, tavily_search, advanced_search, arxiv_search, targeted_search,
    academic_citation_search, extract_zip_codes, wikipedia_search, image_recognition,
    read_pdf, read_csv, read_spreadsheet, transcribe_audio,
    youtube_transcript_tool, youtube_transcript_api, read_jsonl,
    python_interpreter, download_file, extract_table,
    # Wrap summarize to inject self.llm at runtime
]

class AgentState(TypedDict):
    # The document provided
    input_file: Optional[List[str]]  # Contains file path (PDF/PNG)
    messages: Annotated[List[BaseMessage], add_messages]


# === Agent Class ===
class MyAgent:
    def __init__(
        self,
        model_name: str = "anthropic:claude-3-5-sonnet-latest",  # <-- Use a valid model name
        temperature: float = 0.0
    ):
        try:
            self.llm = init_chat_model(
                model_name,
                temperature=temperature
            )
            # Base tools
            self.tools = tools + [
                StructuredTool.from_function(lambda text: summarize(text, llm=self.llm), name="summarize", description="Summarize a long text chunk.")
            ]
            # RAG components
            self.docs: List[Any] = []
            self.retriever: Optional[BM25Retriever] = None
        except Exception as e:
            print(f"Error initializing LLM: {e}")
            raise

    def add_files(self, file_paths: List[str]):
        """
        Load and index documents for RAG based on file extensions or URLs.
        Supports: PDF, CSV, Excel, JSONL, images, audio (mp3/wav), and YouTube URLs.
        """
        for path in file_paths:
            ext = Path(path).suffix.lower()
            loaded_docs = []
            try:
                if ext == ".csv":
                    loader = CSVLoader(path)
                    loaded_docs = loader.load()
                elif ext == ".pdf":
                    loader = PyPDFLoader(path)
                    loaded_docs = loader.load()
                elif ext in [".xlsx", ".xls"]:
                    import pandas as pd
                    df = pd.read_excel(path)
                    text_content = df.to_string()
                    loaded_docs = [Document(page_content=text_content)]
                elif ext == ".jsonl":
                    with open(path, 'r', encoding='utf-8') as file:
                        content = [json.loads(line) for line in file]
                        text_content = json.dumps(content, indent=2)
                        loaded_docs = [Document(page_content=text_content)]
                elif ext in [".png", ".jpg", ".jpeg"]:
                    text = pytesseract.image_to_string(Image.open(path))
                    if text.strip():
                        loaded_docs = [Document(page_content=text)]
                elif ext in [".mp3", ".wav"]:
                    loader = AssemblyAIAudioTranscriptLoader(file_path=path)
                    loaded_docs = loader.load()
                elif "youtube" in path:
                    loader = YoutubeLoader.from_youtube_url(path)
                    loaded_docs = loader.load()
                else:
                    print(f"Unsupported file type: {ext}")
                    continue
            except Exception as e:
                print(f"Error loading {path}: {e}")
                continue
            # Chunk every loaded doc
            text_splitter = RecursiveCharacterTextSplitter(chunk_size=800, chunk_overlap=100)
            for doc in loaded_docs:
                chunks = text_splitter.split_text(doc.page_content)
                for i, chunk in enumerate(chunks):
                    self.docs.append(Document(
                        page_content=chunk,
                        metadata={**getattr(doc, 'metadata', {}), "chunk": i, "source": path}
                    ))

    def build_retriever(self):
        """
        Create BM25Retriever over the loaded documents and register rag_search tool.
        """
        if not self.docs:
            return
            
        # Build retriever
        try:
            self.retriever = BM25Retriever.from_documents(self.docs)
            
            # Define tool with proper error handling
            @tool(name="rag_search")
            def rag_search(query: str) -> str:
                """Search loaded documents for relevant information."""
                if not self.retriever:
                    return "No documents loaded."
                docs = self.retriever.get_relevant_documents(query)
                if not docs:
                    return "No relevant information found."
                return "\n\n".join(f"{doc.metadata.get('source', '')}: {doc.page_content[:500]}" for doc in docs[:3])

            # Remove existing rag_search if present to prevent duplicates
            self.tools = [t for t in self.tools if t.name != "rag_search"]
            self.tools.append(rag_search)
        except Exception as e:
            print(f"Error building retriever: {e}")

    def __call__(
        self,
        question: str,
        file_paths: Optional[List[str]] = None
    ) -> str:
        try:
            state: Dict[str, Any] = {"messages": [], "input_file": None, "rag_used": False}
            tool_desc = "\n".join(f"{t.name}: {t.description}" for t in self.tools)
            rag_prompt = """
            If the question seems to be about any loaded documents, ALWAYS:
            1. Use the rag_search tool first to find relevant information
            2. Base your answer on the retrieved content
            3. If no relevant content is found, say so
            """
            sys_msg = SystemMessage(content=f"{SYSTEM_PROMPT}\n\n{rag_prompt if file_paths else ''}\n\nTools:\n{tool_desc}")
            state["messages"] = [sys_msg]
            if file_paths and all(isinstance(p, str) for p in file_paths):
                try:
                    self.add_files(file_paths)
                    self.build_retriever()
                except Exception as file_err:
                    print(f"Warning: Error loading files: {file_err}")
            state["messages"].append(HumanMessage(content=question))
            if file_paths:
                state["input_file"] = file_paths
            builder = StateGraph(dict)
            builder.add_node("assistant", self._assistant_node)
            # Add the tools node BEFORE adding edges
            def tool_node_with_rag_flag(state):
                state = ToolNode(self.tools).invoke(state)
                if state.get("input_file") and not state.get("rag_used", False):
                    state["rag_used"] = True
                return state
            builder.add_node("tools", tool_node_with_rag_flag)
            builder.add_edge(START, "assistant")
            # Graph flow: force rag_search if files loaded and not yet used, then use tools_condition
            def route(state):
                last_msg = state["messages"][-1] if state.get("messages") else None
                
                # Check if this is a math question that doesn't need RAG
                is_math_question = re.search(r'(calculate|compute|what is|solve|find the value|evaluate)', 
                                          state["messages"][-2].content.lower()) if len(state["messages"]) > 1 else False
                
                # Only force RAG if we have files AND it's not a pure math question AND RAG hasn't been used
                if (state.get("input_file") and not state.get("rag_used", False) and not is_math_question):
                    return "tools"
                    
                # Regular tool routing logic
                if last_msg and isinstance(last_msg, AIMessage):
                    if getattr(last_msg, "tool_calls", None):
                        return "tools"
                    if getattr(last_msg, "additional_kwargs", {}).get("tool_calls"):
                        return "tools"
                return END
            builder.add_conditional_edges("assistant", route, {"tools": "tools", END: END})
            builder.add_edge("tools", "assistant")
            # Instead of builder.update_node, define a custom tool node with rag flag logic
            graph = builder.compile()
            out = graph.invoke(state, {"recursion_limit": 10})
            last_message = out["messages"][-1].content if out.get("messages") else ""
            match = re.search(r"FINAL ANSWER[:\s]*([^\n]*)", last_message, re.IGNORECASE)
            if match:
                return match.group(1).strip()
            return last_message.strip()
        except Exception as e:
            return f"Error processing question: {e}"
    
    def run(self, question: str, file_paths: Optional[List[str]] = None) -> str:
        return self(question, file_paths)
    
    def _assistant_node(self, state: dict) -> dict:
        """Process messages with the LLM."""
        try:
            # Check if messages exist and ensure proper format
            if not state.get("messages") or len(state["messages"]) == 0:
                # Add a system message if empty
                state["messages"] = [SystemMessage(content=SYSTEM_PROMPT)]
        
            # Ensure we have at least a system and user message
            has_system = any(isinstance(m, SystemMessage) for m in state["messages"])
            has_human = any(isinstance(m, HumanMessage) for m in state["messages"])
        
            if not has_system:
                state["messages"].insert(0, SystemMessage(content=SYSTEM_PROMPT))
        
            if not has_human:
                state["messages"].append(HumanMessage(content="Hello"))
            
            # Invoke the chat model with our BaseMessage list
            resp = self.llm.invoke(state["messages"])
            state["messages"].append(resp)
            return state
        except Exception as e:
            error_msg = f"Error calling LLM: {str(e)}"
            print(error_msg)
            print(f"Message count: {len(state.get('messages', []))}")
            if state.get("messages"):
                print(f"Message types: {[type(m).__name__ for m in state['messages']]}")
            return state