File size: 4,341 Bytes
ed306c8 722929d ed306c8 722929d ed306c8 722929d ed306c8 722929d ed306c8 722929d ed306c8 722929d ed306c8 722929d ed306c8 722929d ed306c8 722929d ed306c8 722929d ed306c8 722929d |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 |
import chess, chess.svg, math
from autogen import ConversableAgent, register_function
from typing_extensions import Annotated
board = None
board_svgs = None
made_move = None
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
board_svgs.append(chess.svg.board(
board,
arrows=[(move.from_square, move.to_square)],
fill={move.from_square: "gray"},
size=250
))
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)
)
return f"Moved {piece_name} ({piece_symbol}) from "\
f"{chess.SQUARE_NAMES[move.from_square]} to "\
f"{chess.SQUARE_NAMES[move.to_square]}."
def check_made_move(msg):
global made_move
if made_move:
made_move = False
return True
else:
return False
def get_num_turns(num_moves):
# Each turn includes two moves (one by each player)
# The first move by player black kicks off the chat
# The first move by player white starts the game
num_turns = math.ceil(num_moves / 2)
if num_moves % 2 == 0:
num_turns += 1
return num_turns
def initialize():
global board, board_svgs, made_move
board = chess.Board()
board_svgs = []
made_move = False
def run_multi_agent(llm, task):
#initialize()
llm_config = {"model": llm}
user_proxy = autogen.ConversableAgent(
name="Admin",
system_message="Give the task, and send "
"instructions to writer to refine the blog post.",
code_execution_config=False,
llm_config=llm_config,
human_input_mode="ALWAYS",
)
planner = autogen.ConversableAgent(
name="Planner",
system_message="Given a task, please determine "
"what information is needed to complete the task. "
"Please note that the information will all be retrieved using"
" Python code. Please only suggest information that can be "
"retrieved using Python code. "
"After each step is done by others, check the progress and "
"instruct the remaining steps. If a step fails, try to "
"workaround",
description="Planner. Given a task, determine what "
"information is needed to complete the task. "
"After each step is done by others, check the progress and "
"instruct the remaining steps",
llm_config=llm_config,
)
engineer = autogen.AssistantAgent(
name="Engineer",
llm_config=llm_config,
description="An engineer that writes code based on the plan "
"provided by the planner.",
)
executor = autogen.ConversableAgent(
name="Executor",
system_message="Execute the code written by the "
"engineer and report the result.",
human_input_mode="NEVER",
code_execution_config={
"last_n_messages": 3,
"work_dir": "coding",
"use_docker": False,
},
)
writer = autogen.ConversableAgent(
name="Writer",
llm_config=llm_config,
system_message="Writer."
"Please write blogs in markdown format (with relevant titles)"
" and put the content in pseudo ```md``` code block. "
"You take feedback from the admin and refine your blog.",
description="Writer."
"Write blogs based on the code execution results and take "
"feedback from the admin to refine the blog."
)
groupchat = autogen.GroupChat(
agents=[user_proxy, engineer, writer, executor, planner],
messages=[],
max_round=10,
)
manager = autogen.GroupChatManager(
groupchat=groupchat, llm_config=llm_config
)
groupchat_result = user_proxy.initiate_chat(
manager,
message=task,
)
return groupchat_result |