Add YoutubeUnderstandingTool for analyzing YouTube videos

tools/youtube_understanding.py

@@ -0,0 +1,60 @@
+from typing import Any, Optional
+from langchain_core.callbacks import CallbackManagerForToolRun
+from langchain_core.tools import BaseTool
+from langchain_core.tools.base import ArgsSchema
+from pydantic import BaseModel, Field
+from google.genai.types import Tool, ToolCodeExecution, GenerateContentConfig, Content, Part, FileData
+from google import genai
+
+class YoutubeUnderstandingInput(BaseModel):
+    video_url: str = Field(description="The URL of the YouTube video to understand")
+    task: str = Field(description="The task to perform on the video")
+
+class YoutubeUnderstandingTool(BaseTool):
+    name: str = "youtube_understanding"
+    description: str = "Useful for understanding youtube videos"
+    args_schema: Optional[ArgsSchema] = YoutubeUnderstandingInput
+    client: Any = None
+    model_id: str = "gemini-2.0-flash"
+    return_direct: bool = False
+
+    def __init__(self, api_key: Optional[str] = None, **kwargs):
+        super().__init__(**kwargs)
+        self.client = genai.Client(api_key=api_key)
+
+    def _run(self, video_url: str, task: str, run_manager: Optional[CallbackManagerForToolRun]=None) -> str:
+        """"""
+        try:
+            response = self.client.models.generate_content(
+                model=self.model_id,
+                contents=Content(
+                    parts=[
+                        Part(
+                            file_data=FileData(file_uri=video_url)
+                        ),
+                        Part(text=task)
+                    ]
+                )
+            )
+
+            return response.text
+        except Exception as e:
+            return f"Error performing code execution: {str(e)}"
+
+    async def _arun(self, video_url: str, task: str, run_manager: Optional[CallbackManagerForToolRun]=None) -> str:
+        """Run the code execution tool asynchronously."""
+        return self._run(video_url, task, run_manager=run_manager.get_sync())
+
+# Example usage
+if __name__ == "__main__":
+    from dotenv import load_dotenv
+
+    # Load API key from environment variables
+    load_dotenv()
+
+    # Create the Google Search runnable
+    yt_understanding= YoutubeUnderstandingTool()
+
+    # Run a search query
+    result = yt_understanding.invoke({"task": "Examine the video. What does Teal'c say in response to the question \"Isn't that hot?\"", "video_url": "https://www.youtube.com/watch?v=1htKBjuUWec"})
+    print(result)