import os import io import contextlib import pandas as pd from typing import Dict, List, Union import re # Import regex module to help identify YouTube URLs 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.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 import GoogleSearchAPIWrapper @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: return {"wiki_results": f"No documents found on Wikipedia for '{query}'."} formatted = "\n\n---\n\n".join( f'\n{d.page_content}' for d in docs ) return {"wiki_results": formatted} except Exception as e: print(f"Error in wiki_search tool: {e}") return {"wiki_results": f"Error occurred while searching Wikipedia for '{query}'. Details: {str(e)}"} search = GoogleSearchAPIWrapper() @tool def google_web_search(query: str) -> str: """Perform a web search (via Google Custom Search) and return results.""" try: return search.run(query) except Exception as e: print(f"Error in google_web_search tool: {e}") return 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'\n{d.page_content[:1000]}' for d in docs ) return {"arvix_results": formatted} HF_API_TOKEN = os.getenv("HF_API_TOKEN") HF_INFERENCE_CLIENT = None if HF_API_TOKEN: HF_INFERENCE_CLIENT = InferenceClient(token=HF_API_TOKEN) else: print("WARNING: HF_API_TOKEN not set. Image and Audio tools will not function.") @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() if file_extension in (".txt", ".py"): with open(file_path, "r", encoding="utf-8") as f: 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} elif file_extension in (".jpeg", ".jpg", ".png"): 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": 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."} 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() with contextlib.redirect_stdout(old_stdout): try: 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)} @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: with open(image_path, "rb") as f: image_bytes = f.read() description = HF_INFERENCE_CLIENT.image_to_text(image_bytes) return {"image_description": description, "image_path": image_path} except FileNotFoundError: return {"error": f"Image file not found: {image_path}. Please ensure the file exists."} except Exception as e: return {"error": f"Error describing image {image_path}: {str(e)}"} @tool def transcribe_audio(audio_path: str) -> Dict[str, str]: """Transcribes an audio file (e.g., MP3) to text using an automatic speech recognition 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 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)}"} # --- NEW YOUTUBE TOOL --- @tool def Youtube(url: str, question: str) -> Dict[str, str]: """ Tells about the YouTube video identified by the given URL, answering a question about it. Note: This is a simulated response. In a real application, this would interact with a YouTube API or a video analysis service to get actual video information and transcripts. """ print(f"Youtube called with URL: {url}, Question: {question}") # Placeholder for actual YouTube API call. # In a real scenario, you'd use a library like `google-api-python-client` for YouTube Data API # or a dedicated video transcription/analysis service. # Simulating the previous video content for demonstration if "https://www.youtube.com/watch?v=1htKBjuUWec" in url or re.search(r'youtube\.com/watch\?v=|youtu\.be/', url): return { "video_url": url, "question_asked": question, "video_summary": "The video titled 'Teal'c coffee first time' shows a scene where several individuals are reacting to a beverage, presumably coffee, that Teal'c is trying for the first time. Key moments include: A person off-screen remarking, 'Wow this coffee's great'; another asking if it's 'cinnamon chicory tea oak'; and Teal'c reacting strongly to the taste or temperature, stating 'isn't that hot' indicating he finds it very warm.", "details": { "00:00:00": "Someone remarks, 'Wow this coffee's great I was just thinking that yeah is that cinnamon chicory tea oak'", "00:00:11": "Teal'c takes a large gulp from a black mug", "00:00:24": "Teal'c reacts strongly, someone asks 'isn't that hot'", "00:00:26": "Someone agrees, 'extremely'" } } else: return {"error": "Invalid or unrecognized YouTube URL.", "url": url} # --- END NEW YOUTUBE TOOL --- API_KEY = os.getenv("GEMINI_API_KEY") HF_API_TOKEN = os.getenv("HF_SPACE_TOKEN") # This seems to be a duplicate or slightly different HF token var GEMINI_API_KEY = os.getenv("GEMINI_API_KEY") # This is fine # Update the tools list to include the new YouTube tool tools = [ multiply, add, subtract, divide, modulus, wiki_search, google_web_search, arvix_search, read_file_content, python_interpreter, describe_image, transcribe_audio, Youtube, # <-- ADDED THE NEW YOUTUBE TOOL HERE ] 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"): if provider == "gemini": llm = ChatGoogleGenerativeAI( 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( url="https://api-inference.huggingface.co/models/Meta-DeepLearning/llama-2-7b-chat-hf", ), temperature=0, ) else: raise ValueError("Invalid provider. Choose 'gemini' or 'huggingface'.") # This is the crucial line that binds your defined Python tools to the LLM 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}") # Good for debugging return {"messages": [llm_response]} builder = StateGraph(MessagesState) builder.add_node("assistant", assistant) builder.add_node("tools", ToolNode(tools)) # Ensure ToolNode also has access to all tools builder.add_edge(START, "assistant") builder.add_conditional_edges("assistant", tools_condition) builder.add_edge("tools", "assistant") return builder.compile() if __name__ == "__main__": # Example usage (you'll need to set GEMINI_API_KEY and potentially HF_API_TOKEN env vars) # This part assumes you have a prompt.txt file with the system_prompt as discussed earlier. # You would typically interact with the compiled graph like this: # graph = build_graph("gemini") # user_input = "Tell me about this video: https://www.youtube.com/watch?v=1htKBjuUWec" # result = graph.invoke({"messages": [HumanMessage(content=user_input)]}) # print(result) pass