用python写一个有AI的斗地主游戏——简述后端代码和思路
Posted EricFrenzy
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了用python写一个有AI的斗地主游戏——简述后端代码和思路相关的知识，希望对你有一定的参考价值。
源码请看我的Github页面。
这是我一个课程的学术项目，请不要抄袭，引用时请注明出处。
本专栏系列旨在帮助小白从零开始开发一个项目，同时分享自己写代码时的感想。
请大佬们为我的拙见留情，有不规范之处烦请多多包涵！
文章目录
开场白
在上一篇博客里，已经介绍了开始前的一些准备。这篇博客讲简要介绍游戏开发中后端代码结构的思路。当然，也是博主自己琢磨的，有遗漏或不足之处请指教！
逻辑
游戏的实现大概可以分为两层（就像网页开发一样）：前端和后端。后端负责储存和管理游戏的逻辑（比如现在是谁的回合，谁手里都有什么牌，我能不能出这个对子等等），而前端负责与用户的交互（比如在窗口里显示自己的手牌，获取输入并更新游戏数据等等）。用类来表示前端和后端有种种好处，其中对新手最有帮助的就是能够以更加“人类”的方式来组织和看待程序内容。比如，用类的你可以这样想：“我想要游戏里玩家1打出这些牌然后玩家2出牌，那么我可以调用Game这个游戏逻辑类里的makePlay这个出牌的函数。”以这种方式思考和解决问题需要一些时间磨合，但是对长远的开发习惯和效率都有极大好处。以下是游戏后端大致逻辑：
| 步骤 | 人类逻辑/游戏流程 | 程序逻辑 |
|---|---|---|
| 1 | 三名玩家进入房间/上桌 | 初始化游戏类，包括玩家名称，场上顺序，等等 |
| 2 | 洗牌并给每个玩家发初始的17张牌（留出3张地主牌） | 用游戏类给每个玩家类添加手牌，并在游戏类里维持地主牌的记录 |
| 3 | 叫地主 | 按照顺序给每个玩家叫地主的机会，安排出牌顺序、给地主牌、改变玩家身份 |
| 4 | 按照顺序和规则出牌或过牌 | 按照出牌顺序，玩家出牌时调用相关函数进行可行性检测和模拟出牌 |
| 5 | 如果有人出完了牌，结束游戏 | 在每一次出牌后检查游戏是否结束，没结束的话继续上一步 |
这些游戏逻辑将在前端被构造和模拟，这里的后端代码只是提供实现这些逻辑的基本工具。还有AI出牌的部分也会放到前端部分详细讲解。
后端代码和思路
斗地主的后端逻辑和井字棋比起来还是有一定复杂度的。博主设计的游戏后端逻辑主要放到了gameEngine.py里，其中各包含更加细分的类。思来想去，我们需要实现以下功能：
'''
GameEngine.py描述
Game Class: 一个用来表达游戏状态的类（保存所有游戏信息）
__init__: 用来构造类，需要游玩的三个玩家的id，初始化类
sortHelper: 目前不重要，用来帮助排序卡的大小（因为卡的表达方式机器不易读）
shuffleDeck: 创建并打乱初始排队
dealCard: 将每个玩家17张卡随机分发，并选择3张随机地主牌
chooseLandlord: 把一个玩家的身份变为地主
assignPlayOrder: 根据玩家身份调整出牌顺序
whichPattern: 返回选择牌的种类（比如单张，对子，顺子，三代二等等）以及它们的大小
isValidPlay: 返回选择的牌是否为可以按照规则打的牌
makePlay: 模拟现实中的玩家出牌，即出牌后轮到下一家
checkWin: 检查游戏状态（返回0代表游戏继续，1代表地主获胜，2代表农民获胜）
createAI: 用AI玩家代替人类玩家
AIMakePlay: AI出牌
player Class: 一个用来表达玩家状态的类（包括玩家名字/id，手牌，和是否为地主）
__init__: 用玩家id构造类
playCard: 从玩家手牌中移除选择的卡
AI Class:
__init__: 用AI玩家id构造类，继承player类的方法
getAllMoves: AI根据手牌生成所有可以出的牌
'''
接下来我们就一个一个实现了。
gameEngine.py
首先是Game类。构建它的时候用到了以下内容：
class Game:
def __init__(self, p1id, p2id, p3id):
# 用来生成和表达卡的一些常量
self.colors = ['heart', 'spade', 'diamond', 'club'] # 扑克牌的四个色
self.nums = ['A', '2', '3', '4', '5', '6',
'7', '8', '9', '10', 'J', 'Q', 'K'] # 扑克牌的数字大小
self.specials = ['X', 'D'] # 小王和大王
self.cardOrder = '3': 1, '4': 2, '5': 3, '6': 4, '7': 5, '8': 6, '9': 7, '10': 8,
'J': 9, 'Q': 10, 'K': 11, 'A': 12, '2': 13, 'X': 14, 'D': 15 # 用来比较牌大小的字典
# 创建游戏内的玩家（player类）和一个用玩家名称指向player对象的字典
self.p1 = player(p1id)
self.p2 = player(p2id)
self.p3 = player(p3id)
self.playerDict = p1id: self.p1, p2id: self.p2, p3id: self.p3
# 一些重要的游戏状态
self.currentPlayer = '' # 当前玩家id
self.prevPlayer = '' # 上一个玩家id
self.prevPlay = ['', []] # 上一次出牌，包括出牌者id和出的牌
self.playOrder = [] # 出牌顺序，包含三个玩家的id
self.landLordCards = [] # 地主牌
在我们完成构造函数后，就要开始写功能性函数了。由于部分比较多，请小伙伴们挑选自己感兴趣的部分阅读，比如“欸这个whichPattern好像有点技术含量，了解了解”。具体如下：
# 帮助给卡排序的函数，输入卡的名称如‘club 2’，根据之前定义的self.cardOrder返回它的虚拟大小/排名
def sortHelper(self, x):
if x[-1] == '0': # 如果卡以0结尾，那么它是个10
return self.cardOrder['10']
return self.cardOrder[x[-1]]
# 创建牌堆并洗牌，放到self.deck即游戏牌堆里
def shuffleDeck(self):
self.deck = []
for i in self.colors: # 生成所有排列组合
for j in self.nums:
self.deck.append(i+' '+j) # “花色 数字”
self.deck.append(self.specials[0]) # 放入小王和大王
self.deck.append(self.specials[1])
random.shuffle(self.deck) # 打乱列表元素顺序（洗牌），别忘了先import random
# 发每个玩家初始的17张手牌和3张地主牌
def dealCard(self):
# 这里介绍下，如果牌堆是完全随机的，发牌顺序对游戏公平性和体验不会有影响
# 我们自己洗牌一般都不会洗的太散，所以要一个人一个人发避免炸弹太多
self.landLordCards = [] # 发地主牌
for i in range(3):
choice = random.choice(self.deck) # 随机选一张
self.landLordCards.append(choice) # 放到地主牌里
self.deck.remove(choice) # 从牌堆里移除它
self.p1Card = [] # 发玩家1的牌
for i in range(17):
choice = random.choice(self.deck) # 随机选一张
self.p1Card.append(choice) # 放到该玩家手牌中
self.deck.remove(choice) # 从牌堆里移除它
self.p2Card = [] # 发玩家2的牌
for i in range(17):
choice = random.choice(self.deck)
self.p2Card.append(choice)
self.deck.remove(choice)
self.p3Card = [] # 发玩家3的牌
for i in range(17):
choice = random.choice(self.deck)
self.p3Card.append(choice)
self.deck.remove(choice)
self.p1.cards = self.p1Card # 把每个玩家的牌放到他们的类里
self.p2.cards = self.p2Card
self.p3.cards = self.p3Card
self.p1.cards.sort(key=lambda x: self.sortHelper(x)) # 给他们排序，模拟了玩的时候自己理牌
self.p2.cards.sort(key=lambda x: self.sortHelper(x))
self.p3.cards.sort(key=lambda x: self.sortHelper(x))
# 根据输入的玩家名称/id，选择地主身份
def chooseLandlord(self, name):
self.playerDict[name].identity = 'p' # 把这个玩家身份改为p，农民为s
for card in self.landLordCards: # 把地主牌放到这个玩家手牌里
self.playerDict[name].cards.append(card)
self.playerDict[name].cards.sort(key=lambda x: self.sortHelper(x)) # 给手牌排序
# 出牌顺序
def assignPlayOrder(self):
if self.p1.identity == 'p': # 地主第一个出，剩下两个玩家随机打乱
self.playOrder = [self.p2.name, self.p3.name]
random.shuffle(self.playOrder)
self.playOrder.insert(0, self.p1.name)
elif self.p2.identity == 'p':
self.playOrder = [self.p1.name, self.p3.name]
random.shuffle(self.playOrder)
self.playOrder.insert(0, self.p2.name)
elif self.p3.identity == 'p':
self.playOrder = [self.p1.name, self.p2.name]
random.shuffle(self.playOrder)
self.playOrder.insert(0, self.p3.name)
else: # 还没选地主的情况，三个玩家顺序随机（等待叫地主）
self.playOrder = [self.p1.name, self.p2.name, self.p3.name]
random.shuffle(self.playOrder)
self.currentPlayer = self.playOrder[0] # 当前玩家为第一个玩家
# 获取选择卡的种类和大小
def whichPattern(self, selectedCards):
cardValues = []
for i in selectedCards: # 把所有选择的卡的值放到列表里
if i[-1] == '0': # 如果0结尾那么它是个10
cardValues.append(self.cardOrder['10'])
else:
cardValues.append(self.cardOrder[i[-1]])
# 这里获取卡的种类和大小。utils是我写的另一个文件，借鉴了DouZero的utils库
# DouZero的项目很有趣，AI打斗地主很智能，感兴趣的话请看https://github.com/kwai/DouZero
# 我代码里的utils.py里有着各种各样的工具，为了方便管理就都放到了一个文件里，后面会展开介绍
# get_move_type(cardValues)输入选择的卡（的值），返回整数代表的卡的种类和卡的大小
pattern = utils.get_move_type(cardValues)
return pattern # 返回一个字典type: 1, value: 5，有卡的种类和大小
# 检查是否为可以打的牌
def isValidPlay(self, selected):
selectedCards = sorted(selected, key=lambda x: self.sortHelper(x)) # 先排序，方便比较
if self.prevPlay[0] == self.currentPlayer:
return True # 即上两家都过牌，又轮到自己了，出啥都行
pattern1 = self.whichPattern(self.prevPlay[1]) # 上一个出牌的种类和大小
pattern2 = self.whichPattern(selectedCards) # 当前选择牌的种类和大小
if pattern2['type'] == 15 or pattern1['type'] == 5:
return False # 即上家出牌是王炸或者当前选择牌非法
elif pattern2['type'] == 5 or pattern1['type'] == 0:
return True # 即当前出牌是王炸或者前一家过牌
else:
if pattern1['type'] == pattern2['type'] and\\
pattern1['rank'] < pattern2['rank']:
try: # 看看是不是三带一
if pattern1['len'] == pattern2['len']:
return True
return False
except:
return True # 如果种类一样并且选择的要更大，可以出
else:
return False
# 模拟游戏出牌
def makePlay(self, selectedCards):
if selectedCards == [] and self.prevPlay != []:
pass # 过牌，什么都不做
else: # 有牌打的话那就打牌
self.playerDict[self.currentPlayer].playCard(selectedCards) # 当前玩家出牌
self.prevPlay = [self.currentPlayer, selectedCards] # 记录本次出牌
self.prevPlayer = self.currentPlayer # 顺序轮换
playerIndex = self.playOrder.index(self.currentPlayer) # 当前玩家位置
if playerIndex == 2: # 如果是列表里最后一个，循环到第一个
self.currentPlayer = self.playOrder[0]
else: # 否则转到下一个玩家
self.currentPlayer = self.playOrder[playerIndex+1]
# 检查游戏是否结束
def checkWin(self):
if self.playerDict[self.prevPlayer].cards == []: # 游戏在当前玩家出完牌后没有手牌时结束
if self.playerDict[self.prevPlayer].identity == 'p':
return 1 # 地主赢
else:
return 2 # 农民赢
else:
return 0 # 游戏还在进行
Game类里还有个很重要的部分，那就是用AI出牌。具体如下：
# 创建AI玩家
def createAI(self, name2, name3):
self.p2 = AI(name2) # AI是player的子类
self.p3 = AI(name3)
self.playerDict[name2] = self.p2 # 把对应的玩家改成AI
self.playerDict[name3] = self.p3
# AI出牌。这里的逻辑很简单，即从能出的牌里随便选一个出。之后可以进行提升
def AIMakePlay(self, name, chosenLandLord):
AIplayer = self.playerDict[name]
if chosenLandLord: # 如果叫了地主正在出牌
moves = AIplayer.getAllMoves() # 生成可以出的牌
possibleMoves = [] # 根据场上情况实际可以打的牌牌
for move in moves:
realcards = []
for card in move: # 先把生成的牌（无花色，只有大小）变成实际的牌（有花色和大小）
for hand in AIplayer.cards:
if hand[-1] == card[-1] and hand not in realcards:
realcards.append(hand)
if self.isValidPlay(realcards): # 如果可以打，加到选项里
possibleMoves.append(realcards)
possibleMoves.append([]) # 即允许过牌
move = random.choice(possibleMoves) # 随机选一个出牌方式出牌
self.makePlay(move)
else: # 还没人叫地主的话就自己叫地主
self.chooseLandlord(name)
self.assignPlayOrder()
在Game类里的时候我们用到了player对象来表达玩家的相关信息，这里我们定义下player类：
class player:
def __init__(self, name):
self.name = name # 玩家名称/id
self.identity = 's' # s代表农民，p代表地主
self.cards = [] # 初始手牌
# 出牌/从手牌中移除某些牌
def playCard(self, selectedCards):
for i in selectedCards:
self.cards.remove(i)
还有刚用到的AI玩家类。具体如下：
class AI(player): # 这里用到了继承
def __init__(self,name):
super().__init__(name) # 继承了player类的构造函数
# 因为用到了继承，所以AI也继承了player类的playCard函数，所以无需再定义
# 生成所有可以出的牌
def getAllMoves(self):
envmoves = utils.MovesGener(self.cards).gen_moves() # 这里又用到了utils.py，后面会介绍
# MovesGener是一个出牌生成器，gen_moves()生成可以出的牌
EnvCard2RealCard = 3: '3', 4: '4', 5: '5', 6: '6', 7: '7',
8: '8', 9: '9', 10: '10', 11: 'J', 12: 'Q',
13: 'K', 14: 'A', 17: '2', 20: 'X', 30: 'D'
realmoves = []
for move in envmoves:
realmove = [] # 把生成的虚拟数字变回扑克无花色数字
for card in move:
realmove.append(EnvCard2RealCard[card])
realmoves.append(realmove)
return realmoves
到这里，游戏引擎/后端逻辑最主要的部分就写好了。
utils.py
刚才提到了很多工具都来自utils.py。以下是它的代码以及介绍：
'''
重要声明：以下打星号（*）的函数均借鉴于DouZero，请感兴趣的同学前往https://github.com/kwai/DouZero查看他们的代码
utils.py描述
is_contiuous_seq: 输入出的牌，返回它是否连续（类似顺子，但没限制长度） *
get_move_type: 输入出的牌，返回它的种类（比如顺子，单张，三带一）和它的大小（比如三带一的大小就是那三张的大小） *
select: 输入一些牌和一个代表长度的数字，生成该长度这些牌的不同组合 *
MovesGener Class: 用来生成可行出牌的类 *
'''
这里是对于DouZero这部分代码的引用：
@InProceedingspmlr-v139-zha21a,
title = DouZero: Mastering DouDizhu with Self-Play Deep Reinforcement Learning,
author = Zha, Daochen and Xie, Jingru and Ma, Wenye and Zhang, Sheng and Lian, Xiangru and Hu, Xia and Liu, Ji,
booktitle = Proceedings of the 38th International Conference on Machine Learning,
pages = 12333--12344,
year = 2021,
editor = Meila, Marina and Zhang, Tong,
volume = 139,
series = Proceedings of Machine Learning Research,
month = 18--24 Jul,
publisher = PMLR,
pdf = http://proceedings.mlr.press/v139/zha21a/zha21a.pdf[教你做小游戏] 用86行代码写一个联机五子棋WebSocket后端
背景
上篇文章《用177行代码写个体验超好的五子棋》，我们一起用177行代码实现了一个本地对战的五子棋游戏。
现在，如果我们要做一个联机五子棋，怎么办呢？
需求分析
首先，我们需要一个后端服务。2个不同的玩家，一起连接这个后端服务，把要下的棋告诉后端，后端再转发给另一个玩家即可。当然，如果有观战的，也要把当前期局转发给观战者。
此外，为了让2个玩家联机，还需要有「房间号」的概念，只有同一个房间的人才能联机对战。不同房间的人互不影响，允许同时有多个房间的人同时玩游戏。
流程
整个通信流程是这样的：
- 玩家A请求进入房间1。玩家A会执黑棋。
- 玩家B请求进入房间1。玩家B会执白棋。此时人已满，其他人进入将观战。
- 玩家C请求进入房间1。玩家C是观战者。
- 玩家A请求下棋，告诉坐标给服务器。
- 服务器通知玩家B、玩家C，告诉大家A下棋的坐标。
- 玩家B请求下棋，告诉坐标给服务器。
- 服务器通知玩家A、玩家C，告诉大家B下棋的坐标。
之后循环4-7步骤。
为了简化后端逻辑，把逻辑判断都放在前端。例如在前端判断是否游戏结束（五联珠），如果游戏结束，前端不允许再发任何请求。
技术选型
协议与方案
因为涉及到服务器主动给用户发送数据，所以有几种可选方案：
- Http轮询：若在等待对方下棋，则前端每隔1s就发送一条请求，看看对方是否下棋。
- Http长轮询：若在等待对方下棋，则前端每隔1s就发送一条请求，看看对方是否下棋。但是后台不会立即返回结果，要等到接口超过某个时间才返回结果。
- WebSocket：建立好浏览器、服务器的连接，可随时主动向浏览器推送数据。
这里我们选择WebSocket，因为这种场景下Http协议确实有很大的资源浪费。而WebSocket虽然实现起来有点难度，但是节约了资源。
具体实现方案
只要某个编程语言/框架可以支持WebSocket就可以。
因为我以前经常用Django，用过Channels，对它的底层依赖daphne有所了解，所以我直接选择了daphne。它是ASGI标准的一种实现。
daphne是一个非常轻量的选择，不像Django+Channels这套框架提供了很重的解决方案。daphne只提供了基础的ASGI实现，没有其它冗余的功能。就好比：我开发五子棋前端时，使用了SVG + Dom API，没有用React框架一样。
开发
基础知识
daphne要求我们以这样的格式定义一个服务：
# server.py
async def application(scope, receive, send):
# 处理websocket协议
if scope[type] == websocket:
# 先接收第一个包，必须是建立连接的包(connect)，否则拒绝服务
event = await receive()
if event[type] != websocket.connect:
return
# 校验通过，发送accept，表明建立ws连接成功
await send(type: websocket.accept)
# 此后双方可以互相随时发消息。开启个无限循环
while True:
# 接收一个包
event = await receive()
# 如果是断开连接的请求，就结束循环
if event[type] == websocket.disconnect:
break
# 这种方式可以读取包的文本内容
data = event[text]
# 这种方式可以发送一个包给浏览器，这里是把浏览器发来的包原封不动传回去
await send(type: websocket.send, text: data)
运行方法：
pip install daphne
daphne -b 0.0.0.0 -p 8001 server:application
业务开发
我们需要定义一个房间集合，称之为house
house =
编写玩家初次连接（进入房间）的逻辑：
import json
async def application(scope, receive, send):
if scope[type] == websocket:
event = await receive()
if event[type] != websocket.connect:
return
await send(type: websocket.accept)
# 建立连接后，要求前端发送一个EnterRoom事件，以json格式提供用户id和房间号room
event = await receive()
data = json.loads(event[text])
if data[type] != EnterRoom or not data[id] or not data[room]:
# 若前端发送的第一个事件不是这个，就报错，断开连接
await send(type: websocket.close, code: 403)
return
room_id = data[room]
user_id = data[id]
# 看看房间号是否在house内，不在则创建一个room
if room_id not in house:
house[room_id] =
black: None,
white: None,
pieces: [],
sends: [],
users: [],
room = house[room_id]
old = False # 看玩家是不是老玩家（断线重连进来的）
if room[black] == user_id or room[white] == user_id:
old = True
if user_id in room[users]:
old_send = room[sends][room[users].index(user_id)]
room[sends].remove(old_send)
room[users].remove(user_id)
await old_send(type: websocket.close, code: 4000)
else: # 说明玩家是第一次进，给他拿黑棋或白棋
if room[black] is None:
room[black] = user_id
elif room[white] is None:
room[white] = user_id
# 如果玩家没拿到黑棋也没拿到白旗，就是观战者
visiting = room[black] != user_id and room[white] != user_id
# 把玩家的send函数存到room里，方便其他玩家下棋时调用，从而广播下棋事件
room[sends].append(send)
# 把玩家ID存进去
room[users].append(user_id)
玩家进入房间后，我们需要给他通知一下这个房间的基本信息，例如是否已经开始了？当前场上的期局是怎样的？
await send(type: websocket.send, text: json.dumps(
type: InitializeRoomState,
pieces: room[pieces], # 场上棋子情况
visiting: visiting, # 你是否是观战者
black: room[black] == user_id if not visiting else bool(len(room[pieces]) % 2), # 如果你在下棋：黑棋是你吗？如果你是观战者：黑棋是谁？
ready: bool(room[black] and room[white]), # 房间是否准备好开局了？只要有2个人同时在，就可以开了
))
# 因为有人进入了房间，所以需要广播一下这个消息。
if not old and (room[black] == user_id or room[white] == user_id):
for _send in room[sends]:
if _send == send:
continue
await _send(type: websocket.send, text: json.dumps(
type: AddPlayer,
ready: bool(room[black] and room[white]),
))
while True:
event = await receive()
# 有人断线了，处理一下。若房间空了，还要删掉房间，以防内存占用无限增大
if event[type] == websocket.disconnect:
if send in room[sends]:
room[sends].remove(send)
room[users].remove(user_id)
if len(room[pieces]) == 0 and len(room[sends]) == 0:
del house[room_id]
break
# 有人发送了事件，接收一下
data = json.loads(event[text])
# 如果是下棋事件，就改一下room的pieces数据，并广播给大家
if data[type] == DropPiece:
room[pieces].append((data[x], data[y]))
for _send in room[sends]:
if _send == send: # 不需要给自己通知，所以跳过自己
continue
await _send(type: websocket.send, text: json.dumps(
type: DropPiece,
x: data[x],
y: data[y],
))
当然，写好这些后，还需要测试，最好直接写好前端一起联调。我们下篇文章把前端的WebSocket逻辑补充一下。
完整源码
包含了前后端源码（总共不到400行）: https://github.com/HullQin/gobang
是一个非常值得学习的关于WebSocket的demo。
写在最后
我是HullQin，公众号线下聚会游戏的作者（欢迎关注公众号，发送加微信，交个朋友），转发本文前需获得作者HullQin授权。我独立开发了《联机桌游合集》，是个网页，可以很方便的跟朋友联机玩斗地主、五子棋等游戏，不收费没广告。还独立开发了《合成大西瓜重制版》。还开发了《Dice Crush》参加Game Jam 2022。喜欢可以关注我 HullQin 噢～我有空了会分享做游戏的相关技术。
以上是关于用python写一个有AI的斗地主游戏——简述后端代码和思路的主要内容，如果未能解决你的问题，请参考以下文章
自己写的一部分斗地主的程序，没有去写界面，临时是用黑框来显示的