langgraph_agent.py

import os
import io
import contextlib
import pandas as pd # Added for Excel file handling
from typing import Dict, List, Union # Added for type hinting

from langgraph.graph import START, StateGraph, MessagesState
from langgraph.prebuilt import tools_condition, ToolNode
from langchain_openai import ChatOpenAI
from langchain_huggingface import ChatHuggingFace, HuggingFaceEndpoint
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_community.document_loaders import WikipediaLoader, ArxivLoader
from langchain_core.messages import SystemMessage, HumanMessage
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_core.tools import tool

@tool
def multiply(a: int, b: int) -> int:
    """Multiply two integers."""
    return a * b

@tool
def add(a: int, b: int) -> int:
    """Add two integers."""
    return a + b

@tool
def subtract(a: int, b: int) -> int:
    """Subtract the second integer from the first."""
    return a - b

@tool
def divide(a: int, b: int) -> float:
    """Divide first integer by second; error if divisor is zero."""
    if b == 0:
        raise ValueError("Cannot divide by zero.")
    return a / b

@tool
def modulus(a: int, b: int) -> int:
    """Return the remainder of dividing first integer by second."""
    return a % b

@tool
def wiki_search(query: str) -> dict:
    """Search Wikipedia for a query and return up to 2 documents."""
    try:
        docs = WikipediaLoader(query=query, load_max_docs=2, lang="en").load() # Added lang="en" for clarity
        if not docs:
            return {"wiki_results": f"No documents found on Wikipedia for '{query}'."}
        formatted = "\n\n---\n\n".join(
            f'<Document source="{d.metadata.get("source", "N/A")}"/>\n{d.page_content}' # Added .get for safety
            for d in docs
        )
        return {"wiki_results": formatted}
    except Exception as e:
        # Log the full error for debugging if possible
        print(f"Error in wiki_search tool: {e}")
        return {"wiki_results": f"Error occurred while searching Wikipedia for '{query}'. Details: {str(e)}"}

@tool
def web_search(query: str) -> dict:
    """Perform a web search (via Tavily) and return up to 3 results."""
    try: # Added try-except block for robustness
        docs = TavilySearchResults(max_results=3).invoke(query=query)
        formatted = "\n\n---\n\n".join(
            f'<Document source="{d.metadata["source"]}"/>\n{d.page_content}'
            for d in docs
        )
        return {"web_results": formatted}
    except Exception as e:
        print(f"Error in web_search tool: {e}")
        return {"web_results": f"Error occurred while searching the web for '{query}'. Details: {str(e)}"}

@tool
def arvix_search(query: str) -> dict:
    """Search arXiv for a query and return up to 3 paper excerpts."""
    docs = ArxivLoader(query=query, load_max_docs=3).load()
    formatted = "\n\n---\n\n".join(
        f'<Document source="{d.metadata["source"]}"/>\n{d.page_content[:1000]}'
        for d in docs
    )
    return {"arvix_results": formatted}

@tool
def read_file_content(file_path: str) -> Dict[str, str]:
    """
    Reads the content of a file and returns it.
    Supports text (.txt), Python (.py), and Excel (.xlsx) files.
    For other file types, returns a message indicating limited support.
    """
    try:
        _, file_extension = os.path.splitext(file_path)
        content = ""
        if file_extension.lower() in (".txt", ".py"):
            with open(file_path, "r", encoding="utf-8") as f:
                content = f.read()
        elif file_extension.lower() == ".xlsx":
             # Ensure pandas is installed for this.
             df = pd.read_excel(file_path)
             content = df.to_string() # Convert Excel to string representation
        elif file_extension.lower() == ".mp3":
            content = "Audio file provided. Unable to directly process audio. Consider using a transcription service if available."
        elif file_extension.lower() == ".png":
            content = "Image file provided. Unable to directly process images. Consider using an OCR or image analysis service if available."
        else:
            content = f"Unsupported file type: {file_extension}. Only .txt, .py, and .xlsx files are fully supported for reading content."
        return {"file_content": content, "file_name": file_path}
    except FileNotFoundError:
        return {"file_error": f"File not found: {file_path}. Please ensure the file exists in the environment."}
    except Exception as e:
        return {"file_error": f"Error reading file {file_path}: {e}"}

@tool
def python_interpreter(code: str) -> Dict[str, str]:
    """
    Executes Python code and returns its standard output.
    If there's an error during execution, it returns the error message.
    """
    old_stdout = io.StringIO()
    # Redirect stdout to capture print statements
    with contextlib.redirect_stdout(old_stdout):
        try:
            # Create a dictionary to hold the execution scope for exec
            exec_globals = {}
            exec_locals = {}
            exec(code, exec_globals, exec_locals)
            output = old_stdout.getvalue()
            return {"execution_result": output.strip()}
        except Exception as e:
            return {"execution_error": str(e)}


API_KEY = os.getenv("GEMINI_API_KEY")
HF_SPACE_TOKEN = os.getenv("HF_SPACE_TOKEN")
GEMINI_API_KEY = os.getenv("GEMINI_API_KEY")


tools = [
    multiply, add, subtract, divide, modulus,
    wiki_search, web_search, arvix_search,
    read_file_content, # Added new tool
    python_interpreter, # Added new tool
]


with open("prompt.txt", "r", encoding="utf-8") as f:
    system_prompt = f.read()
sys_msg = SystemMessage(content=system_prompt)


def build_graph(provider: str = "gemini"):
    """Build the LangGraph agent with chosen LLM (default: Gemini)."""
    if provider == "gemini":
        llm = ChatGoogleGenerativeAI(
        model= "gemini-2.5-flash-preview-05-20",
        temperature=1.0,
        max_retries=2,
        api_key=GEMINI_API_KEY,
        max_tokens=5000
)

    elif provider == "huggingface":
        llm = ChatHuggingFace(
            llm=HuggingFaceEndpoint(
                url="https://api-inference.huggingface.co/models/Meta-DeepLearning/llama-2-7b-chat-hf",
            ),
            temperature=0,
        )
    else:
        raise ValueError("Invalid provider. Choose 'openai' or 'huggingface'.")

    llm_with_tools = llm.bind_tools(tools)

    def assistant(state: MessagesState):
        messages_to_send = [sys_msg] + state["messages"]
        return {"messages": [llm_with_tools.invoke(messages_to_send)]}

    builder = StateGraph(MessagesState)
    builder.add_node("assistant", assistant)
    builder.add_node("tools", ToolNode(tools))
    builder.add_edge(START, "assistant")
    builder.add_conditional_edges("assistant", tools_condition)
    builder.add_edge("tools", "assistant")

    return builder.compile()

if __name__ == "__main__":
    # This block is intentionally left empty as per user request to remove examples.
    # Your agent will interact with the graph by invoking it with messages.
    pass