mirror of
https://github.com/Tencent/GameAISDK.git
synced 2026-10-20 22:26:02 +08:00
changqingai
430 lines
18 KiB
Markdown
430 lines
18 KiB
Markdown
## AI的二次开发
|
||
|
||
AI的二次开发中，有依赖AgentAPIMgr和ActionAPIMgr，AgentAPIMgr是Agent和GameReg进行交互的接口模块。ActionAPIMgr是和手机交互的动作接口模块。在本章我们先对这两部分进行介绍，再介绍AI插件。
|
||
|
||
## 1 AgentAPIMgr 介绍
|
||
|
||
### 1.1 整体流程
|
||
|
||
```
|
||
if __name__ == '__main__':
|
||
# step1: AgentAPIMgr为对外接口类，用于和识别模块交互。
|
||
agentAPI = AgentAPIMgr.AgentAPIMgr()
|
||
|
||
#step2:初始化：初始化tbus,加载识别任务配置文件如task.json，配置文件的介绍详见三章节
|
||
ret = agentAPI.Initialize("../cfg/task.json")
|
||
if not ret:
|
||
print('Agent API Init Failed')
|
||
return
|
||
|
||
#step3:发送识别命令到识别模块，参数介绍请详见二章节
|
||
ret = agentAPI.SendCmd(AgentAPIMgr.MSG_SEND_GROUP_ID, 1)
|
||
if not ret:
|
||
print('send message failed')
|
||
|
||
while True:
|
||
#step4:获取识别结果，参数介绍请详见第二章节
|
||
msgDict = agentAPI.GetInfo(AgentAPIMgr.GAME_RESULT_INFO)
|
||
|
||
#step5:资源释放，tbus资源释放
|
||
agentAPI.Release()
|
||
```
|
||
|
||
### 1.2 agentAPI的介绍
|
||
|
||
* **AgentAPI.Initialize(confFile，referFile)**
|
||
|
||
初始化操作，输入参数confFile为 json格式的任务配置文件。referFile为json格式的参考任务配置文件。此配置文件的详细说明请见第3章的场景识别配置中的[任务配置文件说明](../SDKTool/TaskConf.md)。初始化的返回结果为True,表明初始化成功;False初始化失败，可根据屏幕打印的错误信息，定位原因。
|
||
|
||
- **AgentAPI.SendCmd(cmdID, cmdValue)**
|
||
|
||
发送命令给识别模块，返回值为True,表示发送成功，False表示发送失败。识别模块会根据发送的命令，改变识别器。输入参数详情见下表：
|
||
|
||
| **cmdID** | **cmdValue** | **描述** |
|
||
| ------------------ | ------------------------------ | ------------------------------------------------------------ |
|
||
| MSG_SEND_GROUP_ID | groupID（python数字类型） | 发送任务groupID，识别模块接受此group下的任务列表信息，创建识别器（groupID的具体数值请参考对应json配置文件中的groupID字段），如发送groupID=1的任务：ret = agentAPI.SendCmd(AgentAPIMgr.MSG_SEND_GROUP_ID, 1) |
|
||
| MSG_SEND_TASK_FLAG | 由taskid和对应操作组成的字典。 | key值为taskID，为配置文件中对应group下的”taskID”项， value为True/False表示打开/关闭taskID对应的识别器。在后续的识别结果中，会增加或是减少此task的识别结果。如,打开taskID为1的任务，关闭taskID为2的任务。taskDict ={}taskDict[1] = TruetaskDict[2] = Falseret = agentAPI.SendCmd(AgentAPIMgr.MSG_SEND_TASK_FLAG, taskDict ) |
|
||
| MSG_SEND_ADD_TASK | task参数列表 | 增加一系列识别任务，识别模块接收到此消息后，根据参数列表创建一系列识别器。在后续的识别结果中，会增加此任务的识别结果。如增加taskID为10的任务：taskList =[]taskDict = dict()taskDict["taskID"] = 10taskDict["type"] = "fixed_obj"taskDict["elements"] = []elementDict = dict()elementDict["x"] = 946elementDict["y"] = 15elementDict["w"] = 220elementDict["h"] = 60elementDict["path"] = "../data/DNF/dnf_over.bmp"taskDict["elements"].append(elementDict)taskList.append(taskDict) agentAPI.SendCmd(AgentAPIMgr.MSG_SEND_ADD_TASK, taskList) |
|
||
| MSG_SEND_DEL_TASK | taskID 的 列表 | 删除一系列识别任务，识别模块接收到此消息后，根据task的设置，删除相应的识别器。在后续的识别结果中，不会存在此task识别结果。如：删除taskID为10的任务：taskList =[]taskList.append(10)ret = agentAPI.SendCmd(AgentAPIMgr.MSG_SEND_DEL_TASK, taskList) |
|
||
| MSG_SEND_CHG_TASK | task参数列表 | 改变一系列识别任务，识别模块接收到此消息后，根据新的任务参数，修改识别器。在后续的识别中，根据新识别器对图像做识别。如改变taskID为10的任务：taskList = []taskDict = dict()taskDict["taskID"] = 10taskDict["type"] = "fixed_obj"taskDict["elements"] = []elementDict = dict()elementDict["x"] = 1000elementDict["y"] = 0elementDict["w"] = 200elementDict["h"] = 200elementDict["path"] = "../data/DNF/dnf_playing.bmp"taskDict["elements"].append(elementDict)taskList.append(taskDict)ret = agentAPI.SendCmd(AgentAPIMgr.MSG_SEND_CHG_TASK, taskList) |
|
||
|
||
|
||
|
||
- **AgentAPI.GetInfo(type)**
|
||
|
||
获取结果，结果的类型type有以下几种。
|
||
|
||
| **type** | **Return** | **描述** |
|
||
| ------------------- | ------------------ | ------------------------------------------------------------ |
|
||
| CUR_GROUP_TASK_INFO | 当前组任务参数列表 | 返回值为 python中的列表类型(list)如：groupTaskInfo = agentAPI.GetInfo(AgentAPIMgr.CUR_GROUP_TASK_INFO)for task in groupTaskInfo:taskID = task.get('taskID')type = task.get('type')elements = task.get('elements') …… |
|
||
| CUR_GROUP_INFO | 当前组的参数 | 返回值为python中的字典类型(dict)如:curGroupInfo = agentAPI.GetInfo(AgentAPIMgr.CUR_GROUP_INFO)groupID = curGroupInfo.get('groupID')groupName = curGroupInfo.get('name')groupTask = curGroupInfo.get('task')…… |
|
||
| GAME_RESULT_INFO | 识别结果 | 返回值为python中的字典类型(dict)如:msgDict = agentAPI.GetInfo(AgentAPIMgr.GAME_RESULT_INFO)resultDic = msgDict['result'] groupID = msgDict['groupID']frameSeq = msgDict['frameSeq']deviceIndex = msgDict['deviceIndex']img = msgDict['image']..... |
|
||
| ALL_GROUP_INFO | 所有组的参数 | 返回值为python中列表(list)类型。如：allGroupInfo = agentAPI.GetInfo(AgentAPIMgr.ALL_GROUP_INFO)for groupInfo in allGroupInfo:groupID = groupInfo.get('groupID')groupName = groupInfo.get('name')groupTask = groupInfo.get('task') …… |
|
||
|
||
当调用AgentAPI.GetInfo(AgentAPIMgr.GAME_RESULT_INFO)接口去获取当前帧游戏图像的处理结果时，会返回每个任务(task)的结果项。不同task类型的结果说明，请见文档 [Task识别结果说明](TaskRecognitionResults.md) 。
|
||
|
||
- **AgentAPI.Release()**
|
||
|
||
在程序的最后，调用此函数，释放tbus资源。
|
||
|
||
## 2 ActionAPIMgr介绍
|
||
|
||
### 2.1 整体流程
|
||
|
||
```
|
||
if __name__ == '__main__':
|
||
#step1: ActionAPIMgr为对外接口类。
|
||
actionAPI = ActionAPIMgr.ActionAPIMgr()
|
||
|
||
#step2:初始化：初始化tbus
|
||
ret = actionAPI.Initialize()
|
||
if not ret:
|
||
print('Action API Init Failed')
|
||
return
|
||
|
||
#step3:发送动作
|
||
actionAPI.SendAction(actionID=12, actionData={}) # 发送自定义12号动作
|
||
actionAPI.Click(px=300, py=400, contact=0) # 发送点击动作，点击（300, 400）位置
|
||
|
||
#step4:资源释放，tbus资源释放
|
||
actionAPI.Finish()
|
||
```
|
||
|
||
### 2.2 ActionAPI的介绍
|
||
|
||
- ActionAPI.Initialize()
|
||
初始化操作，在其他方法之前调用。
|
||
- ActionAPI.Finish()
|
||
结束操作，在进程退出时调用释放资源
|
||
- ActionAPI.Reset(frameSeq=-1)
|
||
释放所有contact，frameSeq为帧序号
|
||
- ActionAPI.SendAction(actionID, actionData , frameSeq=-1)
|
||
发送自定义动作，actionID为动作ID，actionData为对应的动作参数（字典）
|
||
- ActionAPI.MovingInit(centerX, centerY, radius, contact=0, frameSeq=-1, waitTime=1000)
|
||
初始化移动摇杆中心点（centerX, centerY）和半径radius，触点ID为contact，之后等待时间waitTime毫秒
|
||
- ActionAPI.MovingFinish()
|
||
释放移动摇杆初始化时的触点
|
||
- ActionAPI.Moving(dirAngle，frameSeq=-1，durationMS=50)
|
||
控制移动摇杆向dirAngle方向滑动，滑动过程耗时durationMS毫秒
|
||
- ActionAPI.Move(px, py, contact=0, frameSeq=-1, waitTime=0)
|
||
移动触点contact到（px, py）位置，瞬间移动不补充过渡点，之后等待时间waitTime毫秒
|
||
- ActionAPI.Click(px, py, contact=0, frameSeq=-1, durationMS=-1)
|
||
点击触点contact在（px, py）位置，过程耗时durationMS毫秒
|
||
- ActionAPI.Down(px, py, contact=0, frameSeq=-1, waitTime=0)
|
||
按压触点contact在（px, py）位置，之后等待时间waitTime毫秒
|
||
- ActionAPI.Up(contact, frameSeq=-1, waitTime=0)
|
||
释放触点contact，之后等待时间waitTime毫秒
|
||
- ActionAPI.Swipe(sx, sy, ex, ey, contact=0, frameSeq=-1, durationMS=50, needUp=True)
|
||
滑动触点contact从（sx, sy）到（ex, ey），中间会补充过渡点，过程耗时durationMS毫秒，根据needUp决定最后是否释放触点
|
||
- ActionAPI.SwipeMove(px, py, contact=0, frameSeq=-1, durationMS=50)
|
||
移动触点contact到（px, py）位置，中间会补充过渡点，过程耗时durationMS毫秒
|
||
|
||
## 3. AI插件说明
|
||
|
||
### 3.1 AI插件介绍
|
||
|
||
- 平台中特定游戏的AI逻辑以AI插件的形式存在，AI插件主要包含两部分：Env插件和AI模型插件。
|
||
Env插件
|
||
Env插件通常实现与游戏业务相关的一些逻辑，如执行各种游戏动作输出、通过API设置图像识别任务并获取识别结果。
|
||
|
||
- AI模型插件
|
||
AI模型插件通常用来实现特定的游戏AI算法，如强化学习算法、模仿学习算法、行为树等算法等。根据Env插件得到的游戏状态信息（游戏图像、游戏中技能状态、血条、敌人位置等）决策游戏动作，并执行对应的游戏动作。
|
||
|
||
平台定义了Env插件和AI模型插件的接口，继承并实现对应的插件接口，即可实现自己的AI插件，并由平台AI框架加载运行AI插件。Env插件和AI模型插件不是一一对应的关系，一个AI模型插件可以对应不同的Env插件。如果不同游戏使用了同样的AI算法，但游戏的一些业务逻辑存在差异，这种情况下每个游戏可以实现不同的Env插件，但可以使用同一个AI模型插件。
|
||
|
||
平台内置了DQN强化学习和模仿学习插件，可以通过配置cfg/task/agent/AgentAI.ini文件来直接使用内置AI插件，也可以通过配置cfg/task/agent/AgentAI.ini文件来使用用户开发的AI插件。
|
||
|
||
### 3.2 用户AI插件开发
|
||
|
||
下面以一个Speed游戏为例，说明用户该如何开发AI插件。开发后的Speed AI插件存放在src/PlugIn/ai/目录下，目录如下：
|
||
|
||
```
|
||
game_ai_sdk
|
||
└─src
|
||
└─PlugIn
|
||
└─ai
|
||
└─speed
|
||
└─ SpeedAI.py
|
||
└─ SpeedEnv.py
|
||
```
|
||
|
||
其中SpeedEnv.py文件实现了Env插件，SpeedAI.py文件实现了AI模型插件。
|
||
|
||
其中SpeedEnv.py文件实现了Env插件，SpeedAI.py文件实现了AI模型插件。
|
||
|
||
修改cfg/task/agent/AgentAI.ini文件，让平台加载运行Speed AI插件，修改后的文件内容如下所示：
|
||
|
||
```
|
||
[AGENT_ENV]
|
||
UsePluginEnv = 1
|
||
EnvPackage = speed
|
||
EnvModule = SpeedEnv
|
||
EnvClass = SpeedEnv
|
||
|
||
[AI_MODEL]
|
||
UsePluginAIModel = 1
|
||
AIModelPackage = speed
|
||
AIModelModule = SpeedAI
|
||
AIModelClass = SpeedAI
|
||
|
||
[RUN_FUNCTION]
|
||
UseDefaultRunFunc = 1
|
||
;RunFuncPackage = MyPackage
|
||
;RunFuncModule = MyModule
|
||
;RunFuncName = MyRunFunc
|
||
```
|
||
|
||
- UsePluginEnv = 1 表示使用PlugIn目录下的Env插件，而非平台内置的Env插件。
|
||
|
||
- EnvPackage = speed 表示用户开发Env插件的Package名（Package名通常即为目录名）为speed
|
||
|
||
- EnvModule = SpeedEnv 表示用户开发Env插件的Module名（Module名通常即为文件名）为SpeedEnv
|
||
|
||
- EnvClass = SpeedEnv 表示用户开发的Env插件的Class名（Class名即为类名）为SpeedEnv
|
||
|
||
- UsePluginAIModel = 1 表示使用PlugIn目录下的AI模型插件
|
||
|
||
- AIModelPackage = speed 表示用户开发AI模型插件的Package名为speed
|
||
|
||
- AIModelModule = SpeedAI 表示用户开发AI模型插件的Module名为SpeedAI
|
||
|
||
- AIModelClass = SpeedAI 表示用户开发AI模型插件的Class名为SpeedAI
|
||
|
||
### 3.3 Env插件实现
|
||
|
||
Env插件需要继承平台中的GameEnv类，并实现GameEnv的接口。
|
||
SpeedEnv插件代码如下：
|
||
|
||
```
|
||
import time
|
||
import os
|
||
import cv2
|
||
|
||
from agentenv.GameEnv import GameEnv
|
||
from AgentAPI import AgentAPIMgr
|
||
from ActionAPI.ActionAPIMgr import ActionAPIMgr
|
||
from util import util
|
||
|
||
#游戏场景识别任务配置文件
|
||
TASK_CFG_FILE = 'cfg/task/gameReg/Task.json'
|
||
|
||
REG_GROUP_ID = 1
|
||
|
||
BEGIN_TASK_ID = 1 #游戏开始识别任务ID
|
||
WIN_TASK_ID = 2 #游戏结束识别任务ID
|
||
LOSE_TASK_ID = 3 #游戏失败识别任务ID
|
||
DATA_TASK_ID = 4 #游戏胜利识别任务ID
|
||
|
||
GAME_STATE_INVALID = 0
|
||
GAME_STATE_RUN = 1
|
||
GAME_STATE_WIN = 2
|
||
GAME_STATE_LOSE = 3
|
||
|
||
class SpeedEnv(GameEnv):
|
||
#构造函数
|
||
def __init__(self):
|
||
GameEnv.__init__(self)
|
||
self.__frameIndex = -1
|
||
self.__actionMgr = ActionAPIMgr() #创建执行动作对象
|
||
self.__agentAPI = AgentAPIMgr.AgentAPIMgr() #创建场景识别对象
|
||
self.Reset()
|
||
|
||
#初始化函数，通常初始化动作模块和识别模块
|
||
def Init(self):
|
||
taskCfgFile = util.ConvertToSDKFilePath(TASK_CFG_FILE)
|
||
self.__agentAPI.Initialize(taskCfgFile)
|
||
self.__agentAPI.SendCmd(AgentAPIMgr.MSG_SEND_GROUP_ID, REG_GROUP_ID)
|
||
self.__actionMgr.Initialize()
|
||
return True
|
||
|
||
#退出函数
|
||
def Finish(self):
|
||
self.__agentAPI.Release()
|
||
self.__actionMgr.Finish()
|
||
|
||
#游戏动作个数
|
||
def GetActionSpace(self):
|
||
return 3
|
||
|
||
#输出游戏动作
|
||
def DoAction(self, action):
|
||
if action == 0:
|
||
#执行点击动作1
|
||
self.__actionMgr.Click(140, 570, contact=0, frameSeq=self.__frameIndex, durationMS=100)
|
||
elif action == 1:
|
||
#执行点击动作2
|
||
self.__actionMgr.Click(300, 570, contact=0, frameSeq=self.__frameIndex, durationMS=100)
|
||
elif action == 2:
|
||
#不进行任何动作
|
||
pass
|
||
|
||
#根据识别模块API获取游戏状态信息，并返回状态信息
|
||
def GetState(self):
|
||
#get game data , image and state
|
||
gameInfo = self._GetGameInfo()
|
||
self.__frameIndex = gameInfo['frameSeq']
|
||
image = gameInfo['image']
|
||
state = self.__gameState
|
||
img = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
|
||
|
||
self.__isTerminal = True
|
||
if state == GAME_STATE_RUN:
|
||
self.__isTerminal = False
|
||
|
||
return img
|
||
|
||
#重置游戏状态，通常可以在每局游戏结束或开始时调用
|
||
def Reset(self):
|
||
self.__isTerminal = True
|
||
self.__gameState = GAME_STATE_INVALID
|
||
|
||
#根据识别模块获取的信息，判断游戏对局是否开始
|
||
def IsEpisodeStart(self):
|
||
gameInfo = self._GetGameInfo()
|
||
if self.__gameState == GAME_STATE_RUN:
|
||
self.__isTerminal = False
|
||
return True
|
||
|
||
return False
|
||
|
||
#根据识别模块获取的信息，判断游戏对局是否结束
|
||
def IsEpisodeOver(self):
|
||
return self.__isTerminal
|
||
|
||
#获取taskID对应的识别结果
|
||
def _GetTemplateState(self, resultDict, taskID):
|
||
state = False
|
||
px = -1
|
||
py = -1
|
||
regResults = resultDict.get(taskID)
|
||
if regResults is None:
|
||
return (state, px, py)
|
||
|
||
for item in regResults:
|
||
flag = item['flag']
|
||
if flag:
|
||
x = item['boxes'][0]['x']
|
||
y = item['boxes'][0]['y']
|
||
w = item['boxes'][0]['w']
|
||
h = item['boxes'][0]['h']
|
||
|
||
state = True
|
||
px = int(x + w/2)
|
||
py = int(y + h/2)
|
||
break
|
||
|
||
return (state, px, py)
|
||
|
||
#根据识别模块API获取识别的的游戏状态信息
|
||
def _GetGameInfo(self):
|
||
gameInfo = None
|
||
|
||
while True:
|
||
gameInfo = self.__agentAPI.GetInfo(AgentAPIMgr.GAME_RESULT_INFO)
|
||
if gameInfo is None:
|
||
time.sleep(0.002)
|
||
continue
|
||
|
||
result = gameInfo['result']
|
||
if result is None:
|
||
time.sleep(0.002)
|
||
continue
|
||
|
||
flag, _, _ = self._GetTemplateState(result, BEGIN_TASK_ID)
|
||
if flag is True:
|
||
self.__gameState = GAME_STATE_RUN
|
||
|
||
flag, _, _ = self._GetTemplateState(result, WIN_TASK_ID)
|
||
if flag is True:
|
||
self.__gameState = GAME_STATE_WIN
|
||
|
||
flag, _, _ = self._GetTemplateState(result, LOSE_TASK_ID)
|
||
if flag is True:
|
||
self.__gameState = GAME_STATE_LOSE
|
||
|
||
data = None
|
||
if result.get(DATA_TASK_ID) is not None:
|
||
data = result.get(DATA_TASK_ID)[0]
|
||
|
||
if data is None:
|
||
time.sleep(0.002)
|
||
continue
|
||
else:
|
||
break
|
||
|
||
return gameInfo
|
||
```
|
||
|
||
|
||
|
||
### 3.4 AI模型插件实现
|
||
|
||
AI模型插件需要继承平台中的AIModel类，并实现AIModel的接口。
|
||
实现QQSpeedAI AI模型插件举例：
|
||
|
||
```
|
||
import random
|
||
|
||
from aimodel.AIModel import AIModel
|
||
from agentenv.GameEnv import GameEnv
|
||
|
||
class SpeedAI(AIModel):
|
||
#构造函数
|
||
def __init__(self):
|
||
AIModel.__init__(self)
|
||
|
||
#初始化函数，参数agentEnv为 Env插件类实例对象
|
||
def Init(self, agentEnv):
|
||
self.__agentEnv = agentEnv
|
||
return True
|
||
|
||
#退出函数
|
||
def Finish(self):
|
||
pass
|
||
|
||
#检测到每一局游戏开始后，AI算法进行的操作可以在此处实现，如一些变量的重置等
|
||
def OnEpisodeStart(self):
|
||
pass
|
||
|
||
#检测到每一局游戏结束后，AI算法进行的操作可以在此处实现
|
||
def OnEpisodeOver(self):
|
||
self.__agentEnv.Reset()
|
||
|
||
#当加载进入游戏场景时，需要进行的操作可以在此处实现
|
||
def OnEnterEpisode(self):
|
||
pass
|
||
|
||
#当离开退出游戏场景时，需要进行的操作可以在此处实现
|
||
def OnLeaveEpisode(self):
|
||
pass
|
||
|
||
#训练AI操作的每一个step实现,通常强化学习算法需要实现此接口,基于规则的AI无需训练,不需要实现此接口
|
||
def TrainOneStep(self):
|
||
image = self.__agentEnv.GetState()
|
||
|
||
#AI测试的每一个step实现，通常实现为agentEnv获取游戏状态数据，然后根据AI算法输出对应的游戏操作
|
||
def TestOneStep(self):
|
||
image = self.__agentEnv.GetState()
|
||
#本样例采用随机动作输出
|
||
action = random.choice([0,1,2])
|
||
self.__agentEnv.DoAction(action)
|
||
```
|
||
|
||
|
||
|
||
### 3.5 编译运行AI插件
|
||
|
||
编译运行AI插件时，请确保在game_ai_sdk虚拟环境下。
|
||
进入game_ai_sdk/build目录，执行如下命令即可编译Speed AI插件：
|
||
|
||
```
|
||
(game_ai_sdk)./build_plugin.sh
|
||
```
|
||
|
||
进入game_ai_sdk/bin目录，运行如下命令即可运行Speed AI插件：
|
||
|
||
```
|
||
(game_ai_sdk)python agentai.py
|
||
```
|
||
|