Multi-Agent实践第3期：看智能体下五子棋

import agentscope
agentscope.init(model_configs="./model_configs.json")
# 替换成自己的model config name
YOUR_MODEL_CONFIGURATION_NAME = "{YOUR_MODEL_CONFIGURATION_NAME}"

import numpy as np
from agentscope.message import Msg
from typing import Tuple
from agentscope.agents import AgentBase
CURRENT_BOARD_PROMPT_TEMPLATE = """The current board is as follows:
{board}
{player}, it's your turn.
)"""
NAME_BLACK = "Alice"
NAME_WHITE = "Bob"
# The mapping from name to piece
NAME_TO_PIECE = {
    NAME_BLACK: "o",
    NAME_WHITE: "x",
}
EMPTY_PIECE = "0"
piece_black = NAME_TO_PIECE[NAME_BLACK]
piece_white = NAME_TO_PIECE[NAME_WHITE]

class BoardAgent(AgentBase):
    
    def __init__(self, name):
        super().__init__(name=name, use_memory=False)
        
        # Init the board
        self.size = 15
        self.board = np.full((self.size, self.size), 0)
        
        # Record the status of the game
        self.game_end = False
        
    def reply(self, x: dict = None) -> dict:
        if x is None:
            # 开始游戏
            content = "Welcome to the Gomoku game! Black player goes first. Please make your move." 
        else:
          # 在游戏过程中
          # 1. 检查落子是否合法；
          # 2. 检查落子是否能带来胜利或平局
          # 3. 若游戏继续，更新棋盘信息
          # 4. 打印当前棋局 （依赖完整代码中的board2img函数）
          # 代码参见文章开头链接
          # ...
    # 是否合法落子
    def assert_valid_move(self, x: int, y: int) -> bool:
        return 0 <= x < self.size and 0 <= y < self.size and self.board[x, y] is None
    #检查胜利与否
    def check_win(self, move: Tuple[int, int], piece: str) -> bool:
        # 代码参见代码参见文章开头链接
        # ...
    #检查平局
    def check_draw(self) -> bool:
        return np.all(self.board != None)
    #棋盘状态to文字
    def board2text(self) -> str:
        return "\n".join([str(_)[1:-1].replace(", ", " ") for _ in self.board])

import json
from agentscope.models import ModelResponse
from typing import Optional
# 说明游戏规则和agent回答规则 （样例）
SYS_PROMPT_TEMPLATE = """
You're a skillful Gomoku player. You should play against your opponent according to the following rules:
Game Rules:
1. This Gomoku board is a 15*15 grid. Moves are made by specifying row and column indexes, with [0, 0] marking the top-left corner and [14, 14] indicating the bottom-right corner.
2. The goal is to be the first player to form an unbroken line of your pieces horizontally, vertically, or diagonally. 
3. If the board is completely filled with pieces and no player has formed a row of five, the game is declared a draw.
Note: 
1. Your pieces are represented by '{}', your opponent's by '{}'. 0 represents an empty spot on the board. 
2. You should think carefully about your strategy and moves, considering both your and your opponent's subsequent moves. 
3. Make sure you don't place your piece on a spot that has already been occupied. 
4. Only an unbroken line of five same pieces will win the game. For example, "xxxoxx" won't be considered a win. 
5. Note the unbroken line can be formed in any direction: horizontal, vertical, or diagonal.
"""
# 提示词（样例）
HINT_PROMPT = """You should respond in the following format, which can be loaded by json.loads in Python:
{{
    "thought": "what you thought",
    "move": [row, col]
}}
"""
# 提取大模型回答中的格式化/有效信息
def parse_func(response: ModelResponse) -> ModelResponse:
    res_dict = json.loads(response.text)
    if "move" in res_dict and "thought" in res_dict:
        return ModelResponse(raw=res_dict)
    else:
        raise ValueError(f"Invalid response format in parse_func with response: {response.text}")
# 下五子棋的智能体
class GomokuAgent(AgentBase):
    """A Gomoku agent that can play the game with another agent."""
    
    def __init__(self, name, sys_prompt, model_config_name):
        super().__init__(name=name, sys_prompt=sys_prompt, model_config_name=model_config_name)
        
        self.memory.add(Msg("system", sys_prompt, role="system"))
        
    def reply(self, x: Optional[dict] = None) -> dict:
        if self.memory:
            self.memory.add(x)
        
        msg_hint = Msg("system", HINT_PROMPT, role="system")
        # 组织prompt
        prompt = self.model.format(
            self.memory.get_memory(),
            msg_hint,     
        )
        
        response = self.model(
            prompt,
            parse_func=parse_func,
            max_retries=3,
        ).raw
        
        self.speak(Msg(self.name, json.dumps(response, indent=4, ensure_ascii=False), role="assistant"))
        
        if self.memory:
            self.memory.add(Msg(self.name, response, role="assistant"))
        
        # 隐去想法（thought）
        return Msg(self.name, response["move"], "assistant")

black = GomokuAgent(
    NAME_BLACK,
    model_config_name=YOUR_MODEL_CONFIGURATION_NAME,
    sys_prompt=SYS_PROMPT_TEMPLATE.format(piece_black, piece_white),
)
                    
white = GomokuAgent(
    NAME_WHITE,
    model_config_name=YOUR_MODEL_CONFIGURATION_NAME,
    sys_prompt=SYS_PROMPT_TEMPLATE.format(piece_white, piece_black),
)
board = BoardAgent(name="Host")

from agentscope import msghub
MAX_STEPS = 10
msg = None
i = 0
# Use a msg hub to share conversation between two players, e.g. white player can hear what black player says to the board
with msghub(participants=[black, white, board]):
    while not board.game_end and i < MAX_STEPS:
        for player in [black, white]:
            # receive the move from the player, judge if the game ends and remind the player to make a move
            msg = board(msg)
            
            # end the game if draw or win 
            if board.game_end:
                break
                
            # make a move
            msg = player(msg)
            
            i += 1