Upload 3 files

agents/tools/ai_tools.py

@@ -0,0 +1,535 @@
+# References:
+
+# https://docs.crewai.com/introduction
+# https://ai.google.dev/gemini-api/docs
+
+import base64, chess, os
+from agents.models.llms import (
+    LLM_WEB_SEARCH,
+    LLM_WEB_BROWSER,
+    LLM_IMAGE_ANALYSIS,
+    LLM_AUDIO_ANALYSIS,
+    LLM_VIDEO_ANALYSIS,
+    LLM_YOUTUBE_ANALYSIS,
+    LLM_DOCUMENT_ANALYSIS,
+    LLM_CODE_GENERATION,
+    LLM_CODE_EXECUTION,
+    LLM_IMAGE_TO_FEN,
+    LLM_ALGEBRAIC_NOTATION,
+    LLM_FINAL_ANSWER,
+
+    THINKING_LEVEL_WEB_SEARCH,
+    THINKING_LEVEL_MEDIA_ANALYSIS,
+    THINKING_LEVEL_YOUTUBE_ANALYSIS,
+    THINKING_LEVEL_DOCUMENT_ANALYSIS,
+    THINKING_LEVEL_CODE_GENERATION,
+    THINKING_LEVEL_CODE_EXECUTION,
+    THINKING_LEVEL_IMAGE_TO_FEN,
+    THINKING_LEVEL_ALGEBRAIC_NOTATION,
+    THINKING_LEVEL_FINAL_ANSWER
+)
+from agents.models.prompts import (
+    PROMPT_IMG_TO_FEN,
+    PROMPT_ALGEBRAIC_NOTATION,
+    PROMPT_FINAL_ANSWER
+)
+from crewai.tools import tool
+from crewai_tools import StagehandTool
+from google import genai
+from google.genai import types
+from utils.utils import (
+    read_docx_text,
+    read_pptx_text,
+    is_ext
+)
+
+class AITools():
+    def _get_client():
+        return genai.Client(api_key=os.environ["GEMINI_API_KEY"])
+
+    def _media_analysis_tool(tool_name: str, model: str, question: str, file_path: str) -> str:
+        print(f"🛠️ AITools: {tool_name}: question={question}, file_path={file_path}")
+        
+        try:
+            client = AITools._get_client()
+            
+            file = client.files.upload(file=file_path)
+    
+            response = client.models.generate_content(
+                model=model,
+                contents=[file, question],
+                config=types.GenerateContentConfig(
+                    thinking_config=types.ThinkingConfig(
+                        thinking_level=THINKING_LEVEL_MEDIA_ANALYSIS
+                    )
+                )
+            )
+
+            result = response.text.strip()
+            print(f"🛠️ AITools: {tool_name}: result={result}")
+            return result
+        except Exception as e:
+            print(f"⚠️ AITools: {tool_name}: exception={str(e)}")
+            raise RuntimeError(f"Processing failed: {str(e)}")
+
+    def _extract_execution_result(response):
+        for part in response.candidates[0].content.parts:
+            if part.code_execution_result is not None:
+                return part.code_execution_result.output.strip()
+
+        return None
+
+    @tool("Web Search Tool")
+    def web_search_tool(question: str) -> str:
+        """Given a question only, search the web to answer the question.
+    
+        Args:
+            question (str): Question to answer
+            
+        Returns:
+            str: Answer to the question
+            
+        Raises:
+            RuntimeError: If processing fails
+        """
+        print(f"🛠️ AITools: web_search_tool: question={question}")
+
+        try:
+            client = AITools._get_client()
+            
+            response = client.models.generate_content(
+                model=LLM_WEB_SEARCH,
+                contents=question,
+                config=types.GenerateContentConfig(
+                    tools=[types.Tool(google_search=types.GoogleSearch())],
+                    thinking_config=types.ThinkingConfig(
+                        thinking_level=THINKING_LEVEL_WEB_SEARCH
+                    )
+                )
+            )
+
+            result = response.text.strip()
+            print(f"🛠️ AITools: web_search_tool: result={result}")
+            return result
+        except Exception as e:
+            print(f"⚠️ AITools: web_search_tool: exception={str(e)}")
+            raise RuntimeError(f"Processing failed: {str(e)}")
+
+    @tool("Web Browser Tool")
+    def web_browser_tool(question: str, url: str) -> str:
+        """Given a question and URL, load the URL and act, extract, or observe to answer the question.
+    
+        Args:
+            question (str): Question about a URL
+            url (str): The target URL (must be http/https). "http://"/"https://" will be auto-added if missing.
+            
+        Returns:
+            str: Answer to the question
+            
+        Raises:
+            RuntimeError: If processing fails
+        """
+        print(f"🛠️ AITools: web_browser_tool: question={question}, url={url}")
+        
+        try:
+            url_str = url.strip()
+            
+            if not url_str.lower().startswith(("http://", "https://")):
+                url_str = f"https://{url_str}"
+
+            with StagehandTool(
+                api_key=os.environ["BROWSERBASE_API_KEY"],
+                project_id=os.environ["BROWSERBASE_PROJECT_ID"],
+                model_api_key=os.environ["BROWSERBASE_MODEL_API_KEY"],
+                model_name=LLM_WEB_BROWSER,
+                dom_settle_timeout_ms=5000,
+                headless=True,
+                self_heal=True,
+                wait_for_captcha_solves=True,
+                verbose=3
+            ) as stagehand_tool:
+                result = stagehand_tool.run(
+                    instruction=question,
+                    url=url_str,
+                    command_type="act" # TODO: act, extract, observe
+                ).strip()
+
+                print(f"🛠️ AITools: web_browser_tool: result={result}")
+                return result
+        except Exception as e:
+            print(f"⚠️ AITools: web_browser_tool: exception={str(e)}")
+            raise RuntimeError(f"Processing failed: {str(e)}")
+
+    @tool("Image Analysis Tool")
+    def image_analysis_tool(question: str, file_path: str) -> str:
+        """Given a question and image file, analyze the image to answer the question.
+    
+        Args:
+            question (str): Question about an image file
+            file_path (str): The image file path
+            
+        Returns:
+            str: Answer to the question about the image file
+            
+        Raises:
+            RuntimeError: If processing fails
+        """
+        return AITools._media_analysis_tool("image_analysis_tool", LLM_IMAGE_ANALYSIS, question, file_path)
+    
+    @tool("Audio Analysis Tool")
+    def audio_analysis_tool(question: str, file_path: str) -> str:
+        """Given a question and audio file, analyze the audio to answer the question.
+
+        Args:
+            question (str): Question about an audio file
+            file_path (str): The audio file path
+            
+        Returns:
+            str: Answer to the question about the audio file
+            
+        Raises:
+            RuntimeError: If processing fails
+        """
+        return AITools._media_analysis_tool("audio_analysis_tool", LLM_AUDIO_ANALYSIS, question, file_path)
+    
+    @tool("Video Analysis Tool")
+    def video_analysis_tool(question: str, file_path: str) -> str:
+        """Given a question and video file, analyze the video to answer the question.
+    
+        Args:
+            question (str): Question about a video file
+            file_path (str): The video file path
+            
+        Returns:
+            str: Answer to the question about the video file
+            
+        Raises:
+            RuntimeError: If processing fails
+        """
+        return AITools._media_analysis_tool("video_analysis_tool", LLM_VIDEO_ANALYSIS, question, file_path)
+            
+    @tool("YouTube Analysis Tool")
+    def youtube_analysis_tool(question: str, url: str) -> str:
+        """Given a question and YouTube URL, analyze the video to answer the question.
+    
+        Args:
+            question (str): Question about a YouTube video
+            url (str): The YouTube URL
+            
+        Returns:
+            str: Answer to the question about the YouTube video
+            
+        Raises:
+            RuntimeError: If processing fails
+        """
+        print(f"🛠️ AITools: youtube_analysis_tool: question={question}, url={url}")
+        
+        try:
+            client = AITools._get_client()
+    
+            result = client.models.generate_content(
+                model=LLM_YOUTUBE_ANALYSIS,
+                contents=types.Content(
+                    parts=[types.Part(file_data=types.FileData(file_uri=url)),
+                           types.Part(text=question)]
+                ),
+                config=types.GenerateContentConfig(
+                    thinking_config=types.ThinkingConfig(
+                        thinking_level=THINKING_LEVEL_YOUTUBE_ANALYSIS
+                    )
+                )
+            ).strip()
+
+            print(f"🛠️ AITools: youtube_analysis_tool: result={result}")
+            return result
+        except Exception as e:
+            print(f"⚠️ AITools: youtube_analysis_tool: exception={str(e)}")
+            raise RuntimeError(f"Processing failed: {str(e)}")
+    
+    @tool("Document Analysis Tool")
+    def document_analysis_tool(question: str, file_path: str) -> str:
+        """Given a question and document file, analyze the document to answer the question.
+    
+        Args:
+            question (str): Question about a document file
+            file_path (str): The document file path
+            
+        Returns:
+            str: Answer to the question about the document file
+            
+        Raises:
+            RuntimeError: If processing fails
+        """
+        print(f"🛠️ AITools: document_analysis_tool: question={question}, file_path={file_path}")
+
+        try:
+            client = AITools._get_client()
+    
+            contents = []
+            
+            if is_ext(file_path, ".docx"):
+                text_data = read_docx_text(file_path)
+                contents = [f"{question}\n{text_data}"]
+                print(f"🛠️ Text data:\n{text_data}")
+            elif is_ext(file_path, ".pptx"):
+                text_data = read_pptx_text(file_path)
+                contents = [f"{question}\n{text_data}"]
+                print(f"🛠️ Text data:\n{text_data}")
+            else:
+                file = client.files.upload(file=file_path)
+                contents = [file, question]
+            
+            response = client.models.generate_content(
+                model=LLM_DOCUMENT_ANALYSIS,
+                contents=contents,
+                config=types.GenerateContentConfig(
+                    thinking_config=types.ThinkingConfig(
+                        thinking_level=THINKING_LEVEL_DOCUMENT_ANALYSIS
+                    )
+                )
+            )
+          
+            result = response.text.strip()
+            print(f"🛠️ AITools: document_analysis_tool: result={result}")
+            return result
+        except Exception as e:
+            print(f"⚠️ AITools: document_analysis_tool: exception={str(e)}")
+            raise RuntimeError(f"Processing failed: {str(e)}")
+    
+    @tool("Code Generation and Execution Tool")
+    def code_generation_and_execution_tool(question: str, json_data: str) -> str:
+        """Given a question and JSON data, generate and execute code to answer the question.
+        Args:
+            question (str): Question to answer
+             file_path (str): The JSON data
+
+        Returns:
+            str: Answer to the question
+           
+        Raises:
+            RuntimeError: If processing fails
+        """
+        print(f"🛠️ AITools: code_generation_and_execution_tool: question={question}, json_data={json_data}")
+
+        try:
+            client = AITools._get_client()
+                    
+            response = client.models.generate_content(
+                model=LLM_CODE_GENERATION,
+                contents=[f"{question}\n{json_data}"],
+                config=types.GenerateContentConfig(
+                    tools=[types.Tool(code_execution=types.ToolCodeExecution)],
+                    thinking_config=types.ThinkingConfig(
+                        thinking_level=THINKING_LEVEL_CODE_GENERATION
+                    )
+                ),
+            )
+
+            result = AITools._extract_execution_result(response)
+
+            print(f"🛠️ AITools: code_generation_and_execution_tool: result={result}")
+            return result
+        except Exception as e:
+            print(f"⚠️ AITools: code_generation_and_execution_tool: exception={str(e)}")
+            raise RuntimeError(f"Processing failed: {str(e)}")
+
+    @tool("Code Execution Tool")
+    def code_execution_tool(question: str, file_path: str) -> str:
+        """Given a question and Python file, execute the file to answer the question.
+    
+        Args:
+            question (str): Question to answer
+            file_path (str): The Python file path
+            
+        Returns:
+            str: Answer to the question
+            
+        Raises:
+            RuntimeError: If processing fails
+        """
+        print(f"🛠️ AITools: code_execution_tool: question={question}, file_path={file_path}")
+
+        try:
+            client = AITools._get_client()
+    
+            file = client.files.upload(file=file_path)
+
+            response = client.models.generate_content(
+                model=LLM_CODE_EXECUTION,
+                contents=[file, question],
+                config=types.GenerateContentConfig(
+                    tools=[types.Tool(code_execution=types.ToolCodeExecution)],
+                    thinking_config=types.ThinkingConfig(
+                        thinking_level=THINKING_LEVEL_CODE_EXECUTION
+                    )
+                ),
+            )
+
+            result = AITools._extract_execution_result(response)
+
+            print(f"🛠️ AITools: code_execution_tool: result={result}")
+            return result
+        except Exception as e:
+            print(f"⚠️ AITools: code_execution_tool: exception={str(e)}")
+            raise RuntimeError(f"Processing failed: {str(e)}")
+
+    @tool("Image to FEN Tool")
+    def img_to_fen_tool(question: str, file_path: str, active_color: str) -> str:
+        """Given a chess question, image file, and active color, return the FEN.
+    
+        Args:
+            question (str): The chess question
+            file_path (str): The image file path
+            active_color (str): The active color
+            
+        Returns:
+            str: FEN of the chess position
+            
+        Raises:
+            RuntimeError: If processing fails
+        """
+        print(f"🛠️ AITools: img_to_fen_tool: question={question}, file_path={file_path}, active_color={active_color}")
+
+        try:         
+            client = AITools._get_client()
+
+            with open(file_path, "rb") as f:
+                img_bytes = f.read()
+                img_b64 = base64.b64encode(img_bytes).decode("ascii")
+            
+            prompt = PROMPT_IMG_TO_FEN.format(question=question, active_color=active_color)
+
+            content = types.Content(
+                parts=[
+                    types.Part(text=prompt),
+                    types.Part(
+                        inline_data=types.Blob(
+                            mime_type="image/png",
+                            data=base64.b64decode(img_b64),
+                        )
+                    )
+                ]
+            )
+
+            response = client.models.generate_content(
+                model=LLM_IMAGE_TO_FEN,
+                contents=[content],
+                config=types.GenerateContentConfig(
+                    thinking_config=types.ThinkingConfig(
+                        thinking_level=THINKING_LEVEL_IMAGE_TO_FEN
+                    )
+                )
+            )
+
+            result = None
+
+            for part in response.parts:
+                if part.text is not None:
+                    result = part.text.strip()
+                    break
+
+            board = chess.Board(result) # FEN validation
+
+            print(f"🛠️ AITools: img_to_fen_tool: result={result}")
+            return result
+        except Exception as e:
+            print(f"⚠️ AITools: img_to_fen_tool: exception={str(e)}")
+            raise RuntimeError(f"Processing failed: {str(e)}")
+
+    @tool("Algebraic Notation Tool")
+    def algebraic_notation_tool(question: str, file_path: str, position_evaluation: str) -> str:
+        """Given a chess question, image file, and position evaluation in UCI notation, answer the question in algebraic notation.
+    
+        Args:
+            question (str): The chess question
+            file_path (str): The image file path
+            position_evaluation (str): The position evaluation in UCI notation
+            
+        Returns:
+            str: Answer to the question in algebraic notation
+            
+        Raises:
+            RuntimeError: If processing fails
+        """
+        print(f"🛠️ AITools: algebraic_notation_tool: question={question}, file_path={file_path}, position_evaluation={position_evaluation}")
+
+        try:
+            client = AITools._get_client()
+
+            with open(file_path, "rb") as f:
+                img_bytes = f.read()
+                img_b64 = base64.b64encode(img_bytes).decode("ascii")
+            
+            prompt = PROMPT_ALGEBRAIC_NOTATION.format(question=question, position_evaluation=position_evaluation)
+
+            content = types.Content(
+                parts=[
+                    types.Part(text=prompt),
+                    types.Part(
+                        inline_data=types.Blob(
+                            mime_type="image/png",
+                            data=base64.b64decode(img_b64),
+                        )
+                    )
+                ]
+            )
+            
+            response = client.models.generate_content(
+                model=LLM_ALGEBRAIC_NOTATION,
+                contents=[content],
+                config=types.GenerateContentConfig(
+                    thinking_config=types.ThinkingConfig(
+                        thinking_level=THINKING_LEVEL_ALGEBRAIC_NOTATION
+                    )
+                )
+            )
+
+            result = None
+            
+            for part in response.parts:
+                if part.text is not None:
+                    result = part.text.strip()
+                    break
+
+            print(f"🛠️ AITools: algebraic_notation_tool: result={result}")
+            return result
+        except Exception as e:
+            print(f"⚠️ AITools: algebraic_notation_tool: exception={str(e)}")
+            raise RuntimeError(f"Processing failed: {str(e)}")
+
+    def final_answer_tool(question: str, answer: str) -> str:
+        """Given a question and initial answer, get the final answer.
+    
+        Args:
+            question (str): The question
+            answer (str): The initial answer
+            
+        Returns:
+            str: Final answer
+            
+        Raises:
+            RuntimeError: If processing fails
+        """    
+        print(f"🛠️ AITools: final_answer_tool: question={question}, answer={answer}")
+
+        try:
+            client = AITools._get_client()
+        
+            prompt = PROMPT_FINAL_ANSWER.format(question=question, answer=answer)
+
+            response = client.models.generate_content(
+                model=LLM_FINAL_ANSWER, 
+                contents=[prompt],
+                config=types.GenerateContentConfig(
+                    thinking_config=types.ThinkingConfig(
+                        thinking_level=THINKING_LEVEL_FINAL_ANSWER
+                    )
+                )
+            )
+    
+            result = response.text.strip()
+            print(f"🛠️ AITools: final_answer_tool: result={result}")
+            return result
+        except Exception as e:
+            print(f"⚠️ AITools: final_answer_tool: exception={str(e)}")
+            raise RuntimeError(f"Processing failed: {str(e)}")

agents/tools/deterministic_tools.py

@@ -0,0 +1,98 @@
+# References:
+
+# https://docs.crewai.com/introduction
+
+from crewai.tools import tool
+
+class DeterministicTools():
+    # Arithmetic
+
+    @tool("Add Tool")
+    def add_tool(a: float, b: float) -> float:
+        """Add two numbers.
+        
+        Args:
+            a (float): First number
+            b (float): Second number
+
+        Returns:
+            number: Result
+        """
+        print(f"🤖 DeterministicTools: add_tool: a={a}, b={b}")
+
+        result = a + b
+        print(f"🤖 DeterministicTools: add_tool: result={result}")
+        return result
+    
+    @tool("Subtract Tool")
+    def subtract_tool(a: float, b: float) -> float:
+        """Subtract two numbers.
+        
+        Args:
+            a (float): First number
+            b (float): Second number
+
+        Returns:
+            number: Result
+        """
+        print(f"🤖 DeterministicTools: subtract_tool: a={a}, b={b}")
+
+        result = a - b
+        print(f"🤖 DeterministicTools: subtract_tool: result={result}")
+        return result
+    
+    @tool("Multiply Tool")
+    def multiply_tool(a: float, b: float) -> float:
+        """Multiply two numbers.
+        Args:
+            a (float): First number
+            b (float): Second number
+
+        Returns:
+            number: Result
+        """
+        print(f"🤖 DeterministicTools: multiply_tool: a={a}, b={b}")
+
+        result = a * b
+        print(f"🤖 DeterministicTools: multiply_tool: result={result}")
+        return result
+    
+    @tool("Divide Tool")
+    def divide_tool(a: float, b: float) -> float:
+        """Divide two numbers.
+        
+        Args:
+            a (float): First number
+            b (float): Second number
+
+        Returns:
+            number: Result
+
+        Raises:
+            RuntimeError: If processing fails
+        """
+        print(f"🤖 DeterministicTools: divide_tool: a={a}, b={b}")
+
+        if b == 0:
+            raise RuntimeError("Cannot divide by zero.")
+
+        result = a / b
+        print(f"🤖 DeterministicTools: divide_tool: result={result}")
+        return result
+    
+    @tool("Modulus Tool")
+    def modulus_tool(a: float, b: float) -> float:
+        """Get the modulus of two numbers.
+        
+        Args:
+            a (float): First number
+            b (float): Second number
+
+        Returns:
+            number: Result
+        """
+        print(f"🤖 DeterministicTools: modulus_tool: a={a}, b={b}")
+
+        result = a % b
+        print(f"🤖 DeterministicTools: modulus_tool: result={result}")
+        return result

agents/tools/mcp_tools.py

@@ -0,0 +1,64 @@
+# References:
+
+# https://docs.crewai.com/introduction
+
+import json, os
+from agents.mcp.mcp_client import (
+    call_mcp_tool,
+    MCP_SSE_URL_CHESS_POSITION_EVALUATION,
+    MCP_TOOL_CHESS_POSITION_EVALUATION
+)
+from crewai.tools import tool
+
+class MCPTools():           
+    # Chess position evaluation
+    
+    @tool("Best Move Tool")
+    def best_move_tool(fen: str) -> str:
+        """Get best move with continuation in UCI notation for chess position in FEN.
+        
+        Args:
+            fen (str): Chess position in FEN
+
+        Returns:
+            str: Best move with continuation in UCI notation (e.g., 'f7f2 e4e5 f2f1')
+
+        Raises:
+            RuntimeError: If processing fails
+
+        """
+        print(f"🛠️ MCPTools: best_move_tool: fen={fen}")
+        
+        try:
+            print(MCP_SSE_URL_CHESS_POSITION_EVALUATION)
+
+            mcp_url = os.getenv("MCP_SSE_URL", MCP_SSE_URL_CHESS_POSITION_EVALUATION)
+            
+            raw_result = call_mcp_tool(
+                mcp_url=mcp_url,
+                tool_name=MCP_TOOL_CHESS_POSITION_EVALUATION,
+                arguments={"fen": fen}
+            )
+
+            if isinstance(raw_result, str):
+                try:
+                    raw_result = eval(raw_result)
+                except:
+                    pass
+            
+            if isinstance(raw_result, tuple) and len(raw_result) > 0:
+                raw_result = raw_result[0]
+            
+            result = None
+
+            if isinstance(raw_result, dict) and "continuation" in raw_result:
+                result = raw_result["continuation"]
+            else:
+                result = raw_result
+
+            print(f"🛠️ MCPTools: best_move_tool: result={result}")
+    
+            return result
+        except Exception as e:
+            print(f"⚠️ MCPTools: best_move_tool: exception={str(e)}")
+            raise RuntimeError(f"Processing failed: {str(e)}")