added chess tool

config/prompts.yaml

@@ -16,10 +16,11 @@ prompts:
       {{summary}}
       </summary>
 
-       For mathematical questions or problems delegate them to the math_tool.   
-      
-       Include citations for all the information you retrieve, ensuring you know exactly where the data comes from.
-       If you have the information inside your knowledge, still call a tool in order to confirm it.   
+      For mathematical questions or problems delegate them to the math_tool.
+      For chess related questions use the chess_analysis_tool.
+            
+      Include citations for all the information you retrieve, ensuring you know exactly where the data comes from.
+      If you have the information inside your knowledge, still call a tool in order to confirm it.   
 
        **Guidelines for Conducting Research:**
 
@@ -145,4 +146,27 @@ prompts:
     type: sub_agent
     variables: []
     version: 1.0
-    description: "Core system prompt for the math agent"
+    description: "Core system prompt for the math agent"
+  chess_board_detection:
+    content: |
+      You are a chess board analyst. 
+      Please read the pieces positions carefully, and then confirm your reading with a second pass through the image.
+      type: tool
+    variables: [ ]
+    version: 1.0
+    description: "Prompt for chess board analysis tool"
+  chess_board_orientation:
+    content: |
+      You are a chess board analyst. 
+      Look for file and rank labels (if visible):
+  
+      Files labeled a-h from left to right indicate White's perspective
+      Files labeled h-a from left to right indicate Black's perspective
+      Ranks labeled 1-8 from bottom to top indicate White's perspective
+      Ranks labeled 8-1 from bottom to top indicate Black's perspective
+      Please read the files and ranks positions carefully, and then confirm your reading with a second pass through the image.
+      If the ranks and files are not labeled respond that you cannot determine the chess board orientation.
+      type: tool
+    variables: [ ]
+    version: 1.0
+    description: "Prompt for chess board orientation used by the chess analysis tool"

nodes/nodes.py

@@ -9,6 +9,7 @@ from core.messages import attachmentHandler
 from core.state import State
 from nodes.chunking_node import OversizedContentHandler
 from tools.audio_tool import query_audio
+from tools.chess_tool import chess_analysis_tool
 from tools.excel_tool import query_excel_file
 from tools.math_agent import math_tool
 from tools.python_executor import execute_python_code
@@ -21,6 +22,7 @@ llm_tools.append(query_audio)
 llm_tools.append(query_excel_file)
 llm_tools.append(execute_python_code)
 llm_tools.append(math_tool)
+llm_tools.append(chess_analysis_tool)
 model = model.bind_tools(llm_tools, parallel_tool_calls=False)
 
 

requirements.txt

@@ -8,4 +8,5 @@ langchain-community
 faiss-cpu
 langchain-experimental
 openpyxl
-tabulate
+tabulate
+chess

tools/chess_tool.py

@@ -0,0 +1,362 @@
+import base64
+import json
+import os
+from typing import Dict, List, Optional, Any
+
+import chess
+import chess.engine
+from langchain.chat_models import init_chat_model
+from langchain.schema import SystemMessage, HumanMessage
+from langchain.tools import tool
+from langchain_google_genai import ChatGoogleGenerativeAI
+from langchain_openai import ChatOpenAI
+from pydantic import BaseModel, Field
+
+from utils.prompt_manager import prompt_mgmt
+
+
+def encode_image_to_base64(image_path: str) -> str:
+    """Encode image to base64 for API consumption"""
+    with open(image_path, "rb") as image_file:
+        return base64.b64encode(image_file.read()).decode('utf-8')
+
+
+class ChessPiecePosition(BaseModel):
+    """Model for chess piece position"""
+    square: str = Field(..., description="Chess square notation (e.g., 'e4', 'a1')")
+    piece: str = Field(..., description="Piece type and color (e.g., 'white_king', 'black_queen')")
+
+
+class ChessBoardAnalysis(BaseModel):
+    """Model for complete chess board analysis"""
+    positions: List[ChessPiecePosition] = Field(..., description="List of all piece positions on the board")
+
+    def add_positions(self, positions: List[ChessPiecePosition]) -> None:
+        """Add multiple positions to the analysis"""
+        for position in positions:
+            self.positions.append(position)
+
+    def merge_with(self, other: 'ChessBoardAnalysis') -> None:
+        """Merge another analysis into this one (overwriting conflicts)"""
+        self.add_positions(other.positions)
+
+    def to_fen(self, active_color) -> str:
+        """Convert the analysis to FEN notation (simplified)"""
+        # Create an 8x8 board representation
+        board = [['' for _ in range(8)] for _ in range(8)]
+
+        for position in self.positions:
+            file_idx = ord(position.square[0]) - ord('a')
+            rank_idx = 8 - int(position.square[1])
+
+            if 0 <= file_idx < 8 and 0 <= rank_idx < 8:
+                piece_char = self._piece_to_char(position.piece)
+                board[rank_idx][file_idx] = piece_char
+
+        # Convert to FEN string
+        fen_rows = []
+        for row in board:
+            fen_row = ''
+            empty_count = 0
+
+            for cell in row:
+                if cell == '':
+                    empty_count += 1
+                else:
+                    if empty_count > 0:
+                        fen_row += str(empty_count)
+                        empty_count = 0
+                    fen_row += cell
+
+            if empty_count > 0:
+                fen_row += str(empty_count)
+
+            fen_rows.append(fen_row)
+
+        piece_placement = '/'.join(fen_rows)
+        # Determine active color
+        active_color_char = 'w' if active_color.lower() == 'white' else 'b'
+        # Build complete FEN string
+        castling_rights = "-"
+        en_passant = "-"
+        halfmove_clock = 0
+        fullmove_number = 1
+        fen_parts = [
+            piece_placement,
+            active_color_char,
+            castling_rights,
+            en_passant,
+            str(halfmove_clock),
+            str(fullmove_number)
+        ]
+
+        return ' '.join(fen_parts)
+
+    def _piece_to_char(self, piece: str) -> str:
+        """Convert piece description to FEN character"""
+        color, piece_type = piece.split('_')
+        piece_chars = {
+            'king': 'K', 'queen': 'Q', 'rook': 'R',
+            'bishop': 'B', 'knight': 'N', 'pawn': 'P'
+        }
+        char = piece_chars.get(piece_type, '')
+        return char.lower() if color == 'black' else char
+
+
+class ChessVisionAnalyzer:
+    def __init__(self):
+        self.llm1 = init_chat_model(model="openai:gpt-4.1", temperature=0.0)
+        self.llm2 = ChatGoogleGenerativeAI(model="gemini-2.5-flash")
+
+    def analyze_board_orientation(self, active_color: str, image_path: str) -> str:
+        """Analyze chess board image and return FEN notation"""
+        base64_image = encode_image_to_base64(image_path)
+
+        messages = [
+            SystemMessage(
+                content=prompt_mgmt.render_template("chess_board_orientation", {})),
+            HumanMessage(content=[
+                {
+                    "type": "text",
+                    "text": "Analyze this chess board image and return the chess board orientation. "
+
+                },
+                {
+                    "type": "image_url",
+                    "image_url": {
+                        "url": f"data:image/jpeg;base64,{base64_image}", "detail": "high"
+                    }
+
+                }
+            ])
+        ]
+
+        response = self.llm1.invoke(messages)
+        return response.content
+
+    def analyze_board_from_image(self, active_color: str, image_path: str, llm_no: int,
+                                 squares: Optional[list] = None) -> Optional[ChessBoardAnalysis]:
+        """Analyze chess board image and return FEN notation"""
+        base64_image = encode_image_to_base64(image_path)
+
+        squares_text = ""
+        if squares:
+            squares_text = (f"Focus only on these squares {sorted(squares)} "
+                            f"*** Important: make sure you detect correctly the squares. Take into account the board "
+                            f"orientation."
+                            f"If a square is empty do not return that square.")
+
+        messages = [
+            SystemMessage(
+                content=prompt_mgmt.render_template("chess_board_detection", {})),
+            HumanMessage(content=[
+                {
+                    "type": "text",
+                    "text": f"""Analyze this chess board image and return the  pieces positions.
+                    The chess board orientation is from **Black's perspective**.
+                    - The files are labeled from **h to a** (left to right).
+                    - The ranks are labeled from **8 to 1** (bottom to top).
+                    This matches the standard orientation for Black's perspective.
+                    {squares_text}
+                    Return the positions of all pieces in JSON format.
+                    Use the following schema for each piece:
+                    [{{
+                        "square": "chess notation (e.g., 'e4', 'a1')",
+                        "piece": "color_piece (e.g., 'white_king', 'black_queen')"
+                    }},...
+                    
+                    {{
+                    "square": "chess notation (e.g., 'e4', 'a1')",
+                    "piece": "color_piece (e.g., 'white_king', 'black_queen')"
+                    }}
+                    ]
+                    Return only this list.
+                    """
+
+                },
+                {
+                    "type": "image_url",
+                    "image_url": {
+                        "url": f"data:image/jpeg;base64,{base64_image}", "detail": "high"
+                    }
+
+                }
+            ])
+        ]
+        if llm_no == 1:
+            response = self.llm1.invoke(messages)
+        else:
+            response = self.llm2.invoke(messages)
+        return self._parse_llm_response(response.content)
+
+    def analyze_board(self, active_color: str, file_reference: str) -> str:
+        first_analysis_res = self.analyze_board_from_image(active_color, file_reference, 1)
+        second_analysis_res = self.analyze_board_from_image(active_color, file_reference, 2)
+
+        result = self.compare_analyses(first_analysis_res, second_analysis_res)
+        if result['conflicts'] is not None and len(result['conflicts']) > 0:
+            arbitrage_result = self.arbitrate_conflicts(result, active_color, file_reference, 3)
+
+            # todo: if there are still conflicts let one of the llms win
+            return arbitrage_result.get("consensus").to_fen(active_color)
+        else:
+            result.get("consensus").to_fen(active_color)
+
+    def _parse_llm_response(self, response: str) -> Optional[ChessBoardAnalysis]:
+        """Parse LLM response into ChessBoardAnalysis"""
+        try:
+            # Extract JSON from response
+            json_str = response.strip()
+            if "```json" in json_str:
+                json_str = json_str.split("```json")[1].split("```")[0].strip()
+            elif "```" in json_str:
+                json_str = json_str.split("```")[1].split("```")[0].strip()
+
+            data = json.loads(json_str)
+            print(data)
+            # Filter out items with null or None positions
+            positions = []
+            for item in data:
+                if item["piece"]:
+                    positions.append(ChessPiecePosition(**item))
+
+            return ChessBoardAnalysis(positions=positions)
+        except Exception as e:
+            print(f"Failed to parse LLM response: {e}")
+            return None
+
+    def compare_analyses(self, analysis_1: ChessBoardAnalysis, analysis_2: ChessBoardAnalysis) -> dict:
+        """Compare two analyses and identify conflicts"""
+        print("Comparing analyses")
+
+        if not analysis_1 or not analysis_2:
+            return {"conflicts": [], "consensus": None, "need_arbitration": False}
+
+        # Convert to dictionaries for easier comparison
+        dict_1 = {pos.square: pos.piece for pos in analysis_1.positions}
+        dict_2 = {pos.square: pos.piece for pos in analysis_2.positions}
+
+        conflicts = []
+        consensus = []
+
+        # Check all squares
+        all_squares = set(dict_1.keys()) | set(dict_2.keys())
+
+        for square in all_squares:
+            piece_1 = dict_1.get(square)
+            piece_2 = dict_2.get(square)
+
+            if piece_1 == piece_2:
+                if piece_1:  # Only add if there's actually a piece
+                    consensus.append(ChessPiecePosition(square=square, piece=piece_1))
+            else:
+                conflicts.append({
+                    "square": square,
+                    "analysis_1": piece_1,
+                    "analysis_2": piece_2
+                })
+
+        need_arbitration = len(conflicts) > 0
+
+        return {
+            "conflicts": conflicts,
+            "consensus": ChessBoardAnalysis(positions=consensus),
+            "need_arbitration": need_arbitration
+        }
+
+    def arbitrate_conflicts(self, state: dict, active_color: str, image_path: str, depth: int = 1) -> dict:
+        """Arbitrate conflicting piece positions"""
+        print(f"Arbitrating conflicts with depth {depth}")
+
+        conflicts = state.get("conflicts", [])
+        conflicts_sqares = []
+        for conflict in conflicts:
+            conflicts_sqares.append(conflict["square"])
+
+        print("Squares with conflicts:", conflicts_sqares)
+
+
+        first_analysis_res = self.analyze_board_from_image(active_color, image_path, 1, conflicts_sqares)
+        second_analysis_res = self.analyze_board_from_image(active_color, image_path, 2, conflicts_sqares)
+        result = self.compare_analyses(first_analysis_res, second_analysis_res)
+        result.get("consensus").merge_with(state.get("consensus"))
+        if result['conflicts'] is not None and len(result['conflicts']) > 0 and depth > 0:
+            depth -= 1
+            result = self.arbitrate_conflicts(result, active_color, image_path, depth)
+        return result
+
+
+class ChessEngineAnalyzer:
+    def __init__(self, stockfish_path: str = "stockfish"):
+        self.engine = chess.engine.SimpleEngine.popen_uci(stockfish_path)
+
+    def analyze_position(self, fen: str, depth: int = 18) -> Dict[str, Any]:
+        """Analyze chess position using Stockfish"""
+        board = chess.Board(fen)
+
+        # Get top moves analysis
+        info = self.engine.analyse(board, chess.engine.Limit(depth=depth))
+
+        best_move = info.get("pv", [])[0] if info.get("pv") else None
+        evaluation = info.get("score", chess.engine.PovScore(chess.engine.Cp(0), chess.WHITE))
+
+        return {
+            "best_move": best_move.uci() if best_move else None,
+            "evaluation": str(evaluation),
+            "depth": depth,
+            "analysis": info
+        }
+
+    def close(self):
+        self.engine.quit()
+
+
+class ChessMoveExplainer:
+    def __init__(self):
+        self.llm = ChatOpenAI(
+            model="gpt-4"
+        )
+
+    def explain_move(self, fen: str, move: str, analysis: Dict) -> str:
+        """Generate human-readable explanation of the recommended move"""
+        board = chess.Board(fen)
+        san_move = board.san(chess.Move.from_uci(move))
+
+        prompt = f"""
+        Chess position FEN: {fen}
+        Recommended move: {san_move} ({move})
+        Engine evaluation: {analysis['evaluation']}
+        Analysis depth: {analysis['depth']}
+        
+        Explain this move recommendation in simple terms. Consider:
+        1. Why this move is strong
+        2. What threats it creates or prevents
+        3. The strategic implications
+        4. Alternative moves and why they're inferior
+        5. Keep it concise but informative for an intermediate player
+        """
+
+        response = self.llm([HumanMessage(content=prompt)])
+        return response.content
+
+
+@tool
+def chess_analysis_tool(active_color: str, file_reference: str) -> str:
+    """
+    Tool for analyzing a chess board images and recommending moves
+    :param active_color: The color that should execute the next move
+    :param file_reference: the reference of the image to be analyzed
+    :return: the recommended move along with an analysis
+    """
+    vision_analyzer = ChessVisionAnalyzer()
+    engine_analyzer = ChessEngineAnalyzer(os.getenv("CHESS_ENGINE_PATH"))
+    move_explainer = ChessMoveExplainer()
+    fen = vision_analyzer.analyze_board(active_color, file_reference)
+
+    print(f"Got fen {fen}")
+    analysis_result = engine_analyzer.analyze_position(fen)
+    print(f"Got analysis reslut {analysis_result}")
+    engine_analyzer.close()
+    explanation = move_explainer.explain_move(fen, analysis_result["best_move"], analysis_result)
+
+    return explanation