Update langgraph_agent.py

langgraph_agent.py

@@ -4,50 +4,44 @@ import contextlib
 import pandas as pd
 from typing import Dict, List, Union
 
-# New imports for image and audio processing
-from PIL import Image as PILImage # Used for type checking/potential future local processing
+from PIL import Image as PILImage
 from huggingface_hub import InferenceClient
 
 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
 
+from langchain_google_community.tools.Google Search import GoogleSearchResults
+
 @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()
         if not docs:
@@ -61,23 +55,21 @@ 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)}"}
 
+Google Search_tool = GoogleSearchResults(
+    api_key=os.getenv("GOOGLE_API_KEY"),
+    engine_id=os.getenv("GOOGLE_CSE_ID")
+)
 @tool
-def web_search(query: str) -> dict:
-    """Perform a web search (via Tavily) and return up to 3 results."""
+def google_web_search(query: str) -> dict:
     try:
-        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}
+        docs = Google Search_tool.invoke(query)
+        return {"google_web_results": docs}
     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)}"}
+        print(f"Error in google_web_search tool: {e}")
+        return {"google_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]}'
@@ -85,7 +77,6 @@ def arvix_search(query: str) -> dict:
     )
     return {"arvix_results": formatted}
 
-# Initialize Hugging Face Inference Client
 HF_API_TOKEN = os.getenv("HF_API_TOKEN")
 HF_INFERENCE_CLIENT = None
 if HF_API_TOKEN:
@@ -95,10 +86,6 @@ else:
 
 @tool
 def read_file_content(file_path: str) -> Dict[str, str]:
-    """
-    Reads the content of a file and returns its primary information.
-    For text/code/excel, returns content. For media, returns a prompt to use specific tools.
-    """
     try:
         _, file_extension = os.path.splitext(file_path)
         file_extension = file_extension.lower()
@@ -108,14 +95,12 @@ def read_file_content(file_path: str) -> Dict[str, str]:
                 content = f.read()
             return {"file_type": "text/code", "file_name": file_path, "file_content": content}
         elif file_extension == ".xlsx":
-             df = pd.read_excel(file_path)
-             content = df.to_string()
-             return {"file_type": "excel", "file_name": file_path, "file_content": content}
+            df = pd.read_excel(file_path)
+            content = df.to_string()
+            return {"file_type": "excel", "file_name": file_path, "file_content": content}
         elif file_extension in (".jpeg", ".jpg", ".png"):
-            # Indicate that it's an image and needs to be described by a specific tool
             return {"file_type": "image", "file_name": file_path, "file_content": f"Image file '{file_path}' detected. Use 'describe_image' tool to get a textual description."}
         elif file_extension == ".mp3":
-            # Indicate that it's an audio file and needs to be transcribed by a specific tool
             return {"file_type": "audio", "file_name": file_path, "file_content": f"Audio file '{file_path}' detected. Use 'transcribe_audio' tool to get the text transcription."}
         else:
             return {"file_type": "unsupported", "file_name": file_path, "file_content": f"Unsupported file type: {file_extension}. Only .txt, .py, .xlsx, .jpeg, .jpg, .png, .mp3 files are recognized."}
@@ -126,10 +111,6 @@ def read_file_content(file_path: str) -> Dict[str, str]:
 
 @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()
     with contextlib.redirect_stdout(old_stdout):
         try:
@@ -143,10 +124,6 @@ def python_interpreter(code: str) -> Dict[str, str]:
 
 @tool
 def describe_image(image_path: str) -> Dict[str, str]:
-    """
-    Generates a textual description for an image file (JPEG, JPG, PNG) using an image-to-text model
-    from the Hugging Face Inference API. Requires HF_API_TOKEN environment variable to be set.
-    """
     if not HF_INFERENCE_CLIENT:
         return {"error": "Hugging Face API token not configured for image description. Cannot use this tool."}
     try:
@@ -159,38 +136,48 @@ def describe_image(image_path: str) -> Dict[str, str]:
     except Exception as e:
         return {"error": f"Error describing image {image_path}: {str(e)}"}
 
-
+@tool
+def transcribe_audio(audio_path: str) -> Dict[str, str]:
+    if not HF_INFERENCE_CLIENT:
+        return {"error": "Hugging Face API token not configured for audio transcription. Cannot use this tool."}
+    try:
+        with open(audio_path, "rb") as f:
+            audio_bytes = f.read()
+        transcription = HF_INFERENCE_CLIENT.automatic_speech_recognition(audio_bytes)
+        return {"audio_transcription": transcription, "audio_path": audio_path}
+    except FileNotFoundError:
+        return {"error": f"Audio file not found: {audio_path}. Please ensure the file exists."}
+    except Exception as e:
+        return {"error": f"Error transcribing audio {audio_path}: {str(e)}"}
 
 API_KEY = os.getenv("GEMINI_API_KEY")
 HF_API_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,
+    wiki_search,
+    google_web_search,
+    arvix_search,
     read_file_content,
     python_interpreter,
     describe_image,
+    transcribe_audio,
 ]
 
-
 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-pro-preview-05-06", # Reverted model to gemini-2.5-pro-preview-05-06
-        temperature=1.0,
-        max_retries=2,
-        api_key=GEMINI_API_KEY,
-        max_tokens=5000
-)
-
+            model="gemini-2.5-pro-preview-05-06",
+            temperature=1.0,
+            max_retries=2,
+            api_key=GEMINI_API_KEY,
+            max_tokens=5000
+        )
     elif provider == "huggingface":
         llm = ChatHuggingFace(
             llm=HuggingFaceEndpoint(
@@ -204,10 +191,10 @@ def build_graph(provider: str = "gemini"):
     llm_with_tools = llm.bind_tools(tools)
 
     def assistant(state: MessagesState):
-    messages_to_send = [sys_msg] + state["messages"]
-    llm_response = llm_with_tools.invoke(messages_to_send)
-    print(f"LLM Raw Response: {llm_response}") # Add this line
-    return {"messages": [llm_response]}
+        messages_to_send = [sys_msg] + state["messages"]
+        llm_response = llm_with_tools.invoke(messages_to_send)
+        print(f"LLM Raw Response: {llm_response}")
+        return {"messages": [llm_response]}
 
     builder = StateGraph(MessagesState)
     builder.add_node("assistant", assistant)
@@ -219,6 +206,4 @@ def build_graph(provider: str = "gemini"):
     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
+    pass