streamlit UI
This commit is contained in:
Madhu
parent
7357ca8dbd
commit
14040aac48
2 changed files with 191 additions and 156 deletions
|
|
@ -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,
|
||||
)
|
||||
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.")
|
||||
|
|
@ -1,5 +1,5 @@
|
|||
autogen
|
||||
numpy
|
||||
openai
|
||||
streamlit
|
||||
chess
|
||||
python-chess
|
||||
autogen
|
||||
cairosvg
|
||||
pillow
|
||||
|
|
|
|||
Loading…
Reference in a new issue