diff --git a/ai_agent_tutorials/ai_chess_game/ai_chess_agents.py b/ai_agent_tutorials/ai_chess_game/ai_chess_agents.py index 99f3d7a..f4d4e8e 100644 --- a/ai_agent_tutorials/ai_chess_game/ai_chess_agents.py +++ b/ai_agent_tutorials/ai_chess_game/ai_chess_agents.py @@ -1,170 +1,205 @@ import os -from typing import List import chess import chess.svg -from IPython.display import display -from typing_extensions import Annotated -player_white_config_list = [ - { - "model": "gpt-4-turbo-preview", - "api_key": os.environ.get("OPENAI_API_KEY"), - }, -] - -player_black_config_list = [ - { - "model": "gpt-4-turbo-preview", - "api_key": os.environ.get("OPENAI_API_KEY"), - }, -] - - -# Initialize the board. -board = chess.Board() - -# Keep track of whether a move has been made. -made_move = False - -def get_legal_moves() -> Annotated[str, "A list of legal moves in UCI format"]: - return "Possible moves are: " + ",".join([str(move) for move in board.legal_moves]) - -def make_move(move: Annotated[str, "A move in UCI format."]) -> Annotated[str, "Result of the move."]: - move = chess.Move.from_uci(move) - board.push_uci(str(move)) - global made_move - made_move = True - # Display the board. - display( - chess.svg.board(board, arrows=[(move.from_square, move.to_square)], fill={move.from_square: "gray"}, size=200) - ) - # Get the piece name. - piece = board.piece_at(move.to_square) - piece_symbol = piece.unicode_symbol() - piece_name = ( - chess.piece_name(piece.piece_type).capitalize() - if piece_symbol.isupper() - else chess.piece_name(piece.piece_type) - ) - - result_msg = f"Moved {piece_name} ({piece_symbol}) from {chess.SQUARE_NAMES[move.from_square]} to {chess.SQUARE_NAMES[move.to_square]}." - - # Add game state information - if board.is_checkmate(): - result_msg += f"\nCheckmate! {'White' if board.turn == chess.BLACK else 'Black'} wins!" - elif board.is_stalemate(): - result_msg += "\nGame ended in stalemate!" - elif board.is_insufficient_material(): - result_msg += "\nGame ended - insufficient material to checkmate!" - elif board.is_check(): - result_msg += "\nCheck!" - - return result_msg - +import streamlit as st +from typing import List, Annotated +from IPython.display import display, SVG from autogen import ConversableAgent, register_function -player_white = ConversableAgent( - name="Player_White", # Updated name - system_message="You are a chess player and you play as white. " - "First call get_legal_moves() first, to get list of legal moves. " - "Then call make_move(move) to make a move.", - llm_config={"config_list": player_white_config_list, "cache_seed": None}, -) +# Initialize session state for the OpenAI API key and game state +if "openai_api_key" not in st.session_state: + st.session_state.openai_api_key = None +if "board" not in st.session_state: + st.session_state.board = chess.Board() +if "made_move" not in st.session_state: + st.session_state.made_move = False +if "board_svg" not in st.session_state: + st.session_state.board_svg = None -player_black = ConversableAgent( - name="Player_Black", # Updated name - system_message="You are a chess player and you play as black. " - "First call get_legal_moves() first, to get list of legal moves. " - "Then call make_move(move) to make a move.", - llm_config={"config_list": player_black_config_list, "cache_seed": None}, -) +# Streamlit sidebar for OpenAI API key input +st.sidebar.title("Chess Agent Configuration") +openai_api_key = st.sidebar.text_input("Enter your OpenAI API key:", type="password") +if openai_api_key: + st.session_state.openai_api_key = openai_api_key + st.sidebar.success("API key saved!") + +# Function to get legal moves +def get_legal_moves() -> Annotated[str, "A list of legal moves in UCI format"]: + legal_moves = [str(move) for move in st.session_state.board.legal_moves] + return "Possible moves are: " + ",".join(legal_moves) + +# Function to make a move +def make_move(move: Annotated[str, "A move in UCI format."]) -> Annotated[str, "Result of the move."]: + try: + chess_move = chess.Move.from_uci(move) + if chess_move not in st.session_state.board.legal_moves: + return f"Invalid move: {move}. Please call get_legal_moves() to see valid moves." + + st.session_state.board.push(chess_move) + st.session_state.made_move = True + + # Render board visualization + board_svg = chess.svg.board( + st.session_state.board, + arrows=[(chess_move.from_square, chess_move.to_square)], + fill={chess_move.from_square: "gray"}, + size=400 + ) + st.session_state.board_svg = board_svg # Save SVG to session state + + # Get moved piece details + moved_piece = st.session_state.board.piece_at(chess_move.to_square) + piece_unicode = moved_piece.unicode_symbol() + piece_type_name = chess.piece_name(moved_piece.piece_type) + piece_name = piece_type_name.capitalize() if piece_unicode.isupper() else piece_type_name + + # Build move description + move_desc = f"Moved {piece_name} ({piece_unicode}) from {chess.SQUARE_NAMES[chess_move.from_square]} to {chess.SQUARE_NAMES[chess_move.to_square]}." + + # Check game state + if st.session_state.board.is_checkmate(): + winner = 'White' if st.session_state.board.turn == chess.BLACK else 'Black' + move_desc += f"\nCheckmate! {winner} wins!" + elif st.session_state.board.is_stalemate(): + move_desc += "\nGame ended in stalemate!" + elif st.session_state.board.is_insufficient_material(): + move_desc += "\nGame ended - insufficient material to checkmate!" + elif st.session_state.board.is_check(): + move_desc += "\nCheck!" + + return move_desc + except ValueError: + return f"Invalid move format: {move}. Please use UCI format (e.g., 'e2e4')." # Check if the player has made a move, and reset the flag if move is made. def check_made_move(msg): - global made_move - if made_move: - made_move = False + if st.session_state.made_move: + st.session_state.made_move = False return True else: return False -board_proxy = ConversableAgent( - name="Board_Proxy", # Updated name - llm_config=False, - # The board proxy will only terminate the conversation if the player has made a move. - is_termination_msg=check_made_move, - # The auto reply message is set to keep the player agent retrying until a move is made. - default_auto_reply="Please make a move.", - human_input_mode="NEVER", -) - -register_function( - make_move, - caller=player_white, - executor=board_proxy, - name="make_move", - description="Call this tool to make a move.", -) - -register_function( - get_legal_moves, - caller=player_white, - executor=board_proxy, - name="get_legal_moves", - description="Get legal moves.", -) - -register_function( - make_move, - caller=player_black, - executor=board_proxy, - name="make_move", - description="Call this tool to make a move.", -) - -register_function( - get_legal_moves, - caller=player_black, - executor=board_proxy, - name="get_legal_moves", - description="Get legal moves.", -) - -player_black.llm_config["tools"] - -player_white.register_nested_chats( - trigger=player_black, - chat_queue=[ +# Initialize players and proxy agent if API key is provided +if st.session_state.openai_api_key: + player_white_config_list = [ { - # The initial message is the one received by the player agent from - # the other player agent. - "sender": board_proxy, - "recipient": player_white, - # The final message is sent to the player agent. - "summary_method": "last_msg", - } - ], -) + "model": "gpt-4-turbo-preview", + "api_key": st.session_state.openai_api_key, + }, + ] -player_black.register_nested_chats( - trigger=player_white, - chat_queue=[ + player_black_config_list = [ { - # The initial message is the one received by the player agent from - # the other player agent. - "sender": board_proxy, - "recipient": player_black, - # The final message is sent to the player agent. - "summary_method": "last_msg", - } - ], -) + "model": "gpt-4-turbo-preview", + "api_key": st.session_state.openai_api_key, + }, + ] -# Clear the board. -board = chess.Board() + player_white = ConversableAgent( + name="Player_White", + system_message="You are a professional chess player and you play as white. " + "First call get_legal_moves() first, to get list of legal moves. " + "Then call make_move(move) to make a move.", + llm_config={"config_list": player_white_config_list, "cache_seed": None}, + ) -# Remove max_turns to let the game continue until completion -chat_result = player_black.initiate_chat( - player_white, - message="Let's play chess! Your move.", - max_turns=10, -) \ No newline at end of file + player_black = ConversableAgent( + name="Player_Black", + system_message="You are a professional chess player and you play as black. " + "First call get_legal_moves() first, to get list of legal moves. " + "Then call make_move(move) to make a move.", + llm_config={"config_list": player_black_config_list, "cache_seed": None}, + ) + + # Proxy agent to manage the board and validate moves + board_proxy = ConversableAgent( + name="Board_Proxy", + llm_config=False, + is_termination_msg=check_made_move, + default_auto_reply="Please make a move.", + human_input_mode="NEVER", + ) + + # Register functions for both players + register_function( + make_move, + caller=player_white, + executor=board_proxy, + name="make_move", + description="Call this tool to make a move.", + ) + + register_function( + get_legal_moves, + caller=player_white, + executor=board_proxy, + name="get_legal_moves", + description="Get legal moves.", + ) + + register_function( + make_move, + caller=player_black, + executor=board_proxy, + name="make_move", + description="Call this tool to make a move.", + ) + + register_function( + get_legal_moves, + caller=player_black, + executor=board_proxy, + name="get_legal_moves", + description="Get legal moves.", + ) + + # Register nested chats for both players + player_white.register_nested_chats( + trigger=player_black, + chat_queue=[ + { + # The initial message is the one received by the player agent from + # the other player agent. + "sender": board_proxy, + "recipient": player_white, + # The final message is sent to the player agent. + "summary_method": "last_msg", + } + ], + ) + + player_black.register_nested_chats( + trigger=player_white, + chat_queue=[ + { + # The initial message is the one received by the player agent from + # the other player agent. + "sender": board_proxy, + "recipient": player_black, + # The final message is sent to the player agent. + "summary_method": "last_msg", + } + ], + ) + + # Streamlit UI for playing the game + st.title("Chess Agent Game") + if st.button("Start Game"): + st.session_state.board.reset() + st.session_state.made_move = False + st.session_state.board_svg = None + st.write("Game started! White's turn.") + + # Initiate the chat between Player_White and Board_Proxy + chat_result = player_black.initiate_chat( + player_white, + message="Let's play chess! You go first, its your move.", + max_turns=100, # Set a high enough number to allow the game to complete + ) + st.write(chat_result) + + # Display the chessboard SVG + if st.session_state.board_svg: + st.image(st.session_state.board_svg, caption="Chess Board", use_column_width=True) +else: + st.warning("Please enter your OpenAI API key in the sidebar to start the game.") \ No newline at end of file diff --git a/ai_agent_tutorials/ai_chess_game/requirements.txt b/ai_agent_tutorials/ai_chess_game/requirements.txt index e8130a5..5e78602 100644 --- a/ai_agent_tutorials/ai_chess_game/requirements.txt +++ b/ai_agent_tutorials/ai_chess_game/requirements.txt @@ -1,5 +1,5 @@ -autogen -numpy -openai streamlit -chess \ No newline at end of file +python-chess +autogen +cairosvg +pillow