Update langgraph_agent.py

langgraph_agent.py

@@ -1,4 +1,9 @@
 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
@@ -36,7 +41,6 @@ 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."""
@@ -54,16 +58,19 @@ def wiki_search(query: str) -> dict:
         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."""
-    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}
+    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:
@@ -75,6 +82,55 @@ def arvix_search(query: str) -> dict:
     )
     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")
@@ -83,6 +139,8 @@ 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
 ]
 
 
@@ -99,10 +157,9 @@ def build_graph(provider: str = "gemini"):
         temperature=1.0,
         max_retries=2,
         api_key=GEMINI_API_KEY,
-        max_tokens=5000    
+        max_tokens=5000
 )
 
-        
     elif provider == "huggingface":
         llm = ChatHuggingFace(
             llm=HuggingFaceEndpoint(
@@ -129,7 +186,7 @@ def build_graph(provider: str = "gemini"):
     return builder.compile()
 
 if __name__ == "__main__":
-    graph = build_graph()
-    msgs = graph.invoke({"messages":[ HumanMessage(content="What’s the capital of France?") ]})
-    for m in msgs["messages"]:
-        m.pretty_print()
+    # 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
+