YXHXianYu
commented
1 month ago We implemented a similar demo before and it's possible.
If I reset the environment with agents_count = 0 it fails with some key error. I assume it's not meant to reset without agents.
Yes, it's true. MineLand needs to load some data from server through at least 1 agent.
I was trying to use Mineland.add_agent() to add the agent as it does in the example to connect at any server.
It's possible. But this function isn't stable all the time. Use it carefully. (maybe it has some bugs)
Also I don't see where the instances of Alex are stored when they are agents in the server. How can I access to Alex.run method of each one?
The instances of Alex are stored in python side. In another word, alex can run without MineLand or Minecraft (although you need to provide some information (observation and etc.))
In fact, MineLand provide a gym-like API. You can refer to: https://github.com/Farama-Foundation/Gymnasium. So, MineLand and Agent should be indepentent.
We implemented a similar demo before. Because of the copyright, I cannot show the whole code. But I can show parts of it. The following is an example that combines MineLand and another agent (not Alex, but similar):
import sys
import mineland
import math
import time
from xbot import XBot
# class is convient
class Main:
def __init__(
self,
is_enable_use_last_action_as_trigger = False):
self.mland = mineland.MineLand(
server_host="xxx",
server_port=25565,
agents_count=1,
agents_config=[{'name': 'TheListener'}],
)
self.agents = {} # key: username, value: Agent
self.bot_id = {} # key: username, value: bot_id
self.bot_id_to_username = {} # key: bot_id, value: username
self.bot_name = {} # key: username, value: bot_name
self.bot_cnt = 1
self.idle_timer = {} # key: username, value: idle_timer (a float)
self.idle_timer_max = 4
self.is_enable_use_last_action_as_trigger = is_enable_use_last_action_as_trigger
if self.is_enable_use_last_action_as_trigger:
self.bot_last_response = {}
self.bot_name_prefix = 'XBot'
def action_follow_and_lookat(self, username: str):
assert username in self.agents, "No such username"
return '''some prompt'''
def generate_an_agent(self, username: str):
assert username not in self.agents, "Duplicated username"
self.bot_name[username] = f'XBot{self.bot_cnt}'
self.agents[username] = XBot(self.bot_name[username], username)
self.bot_id[username] = self.bot_cnt
self.bot_id_to_username[self.bot_cnt] = username
self.idle_timer[username] = 0
self.bot_cnt += 1
if self.is_enable_use_last_action_as_trigger:
self.bot_last_response[username] = {'type': 'idle'}
self.mland.add_an_agent({'name': self.bot_name[username]})
def disconnect_an_agent(self, username: str):
if username not in self.agents:
return
self.agents[username].stop()
self.mland.disconnect_an_agent(self.bot_name[username])
del self.agents[username]
del self.bot_id_to_username[self.bot_id[username]]
del self.bot_id[username]
del self.bot_name[username]
del self.idle_timer[username]
# be careful, we would not release the bot_cnt
if self.is_enable_use_last_action_as_trigger:
del self.bot_last_response[username]
def agent_control(self, events):
for event in events:
if event is None:
continue
if event['type'] == 'playerJoined' and event['player_name'] != 'TheListener' and event['player_name'].startswith(self.bot_name_prefix) == False:
self.generate_an_agent(event['player_name'])
elif event['type'] == 'playerLeft' and event['player_name'] != 'TheListener' and event['player_name'].startswith(self.bot_name_prefix) == False:
self.disconnect_an_agent(event['player_name'])
def code_check(self, code_info):
string = ''
for code_info_p in code_info:
if code_info_p is None or code_info_p['code_error'] is None or code_info_p['code_error'] == 'None':
continue
string += str(code_info_p) + '; '
if string != '':
print(string)
def is_agent_stop(self, obs):
if math.fabs(obs['location_stats']['vel'][0]) >= 0.02:
return False
if math.fabs(obs['location_stats']['vel'][2]) >= 0.02:
return False
return True
def append_obs_to_input(self, dict, obs, code_info, events):
dict['obs'] = obs
dict['code_info'] = code_info
dict['event'] = events
return dict
def collect_triggers(self, username, obs, code_info, events):
res = []
for event in events:
if event['type'] == 'chat' and event['username'] == username:
res.append({
'type': 'chat',
'message': event['only_message'],
})
elif event['type'] == 'entityHurt' and event['entity_type'] == 'self':
res.append({
'type': 'agent_hurt',
})
elif event['type'] == 'entityHurt' and event['entity_type'] == 'player' and event['entity_name'] == username:
res.append({
'type': 'user_hurt',
})
elif event['type'] == 'entitySpawn' and event['entity_type'] == 'hostile':
res.append({
'type': 'entity_spawn',
'entity_name': event['entity_name'],
})
if code_info['code_error'] is None or code_info['code_error'] == 'None':
res.append({
'type': 'code_error',
'error': code_info['code_error'],
})
for r in res:
r = self.append_obs_to_input(r, obs, code_info, events)
return res
# Entry for this class
def run(self):
obs = self.mland.reset()
assert self.bot_cnt == len(obs)
# Initialize
act = mineland.Action.no_op(self.bot_cnt)
obs, code_info, event, _, _ = self.mland.step(act)
# use ctrl+c to break
while(True):
# When a player connect to the server, generate a new agent that is only responds to the player
# When a player disconnect to the server, disconnect the corresponding agent
self.agent_control(event[0])
# self.code_check(code_info)
# Default actions
act = mineland.Action.no_op(self.bot_cnt)
# Collect the triggers, and add them to each XBot (collect the triggers from the player's chat message)
for [username, agent] in self.agents.items():
idx = self.bot_id[username]
if idx >= len(obs) or obs[idx] is None:
continue
results = self.collect_triggers(username, obs[idx], code_info[idx], event[idx])
for result in results:
print(username + "'s bot appending:", result['type'])
agent.add_to_input(result)
if self.is_enable_use_last_action_as_trigger:
if self.bot_last_response[username]['type'] == 'code' \
and self.bot_last_response[username]['code'] != 'await setIdle(bot);' \
and code_info[idx].is_ready: # check whether the last code is completed
agent.add_to_input(self.append_obs_to_input({
'type': 'last_code_feedback',
'code': self.bot_last_response[username]['code'],
}, obs[idx], code_info[idx], event[idx]))
self.bot_last_response[username] = {'type': 'idle'}
# Collect the output of each XBot, and add them to the act
for [username, agent] in self.agents.items():
idx = self.bot_id[username]
if idx >= len(obs) or obs[idx] is None:
continue
response = agent.get_from_output()
if response is None:
continue
print("Response:", response)
if response['type'] == 'chat':
response['message'] = response['message'].replace("'", "\\'")
act[idx] = mineland.Action(mineland.Action.NEW, "bot.chat('" + response['message'] + "')")
elif response['type'] == 'code':
act[idx] = mineland.Action(mineland.Action.NEW, response['code'])
elif response['type'] == 'idle':
act[idx] = mineland.Action(mineland.Action.RESUME, '')
else:
assert False, "Unknown response type"
if self.is_enable_use_last_action_as_trigger:
self.bot_last_response[username] = response
# When a bot is idle, let it follow and look at the target player.
for [username, agent] in self.agents.items():
# prevent deadlock sometimes which will show global error : TypeError: Cannot read properties of undefined (reading 'position')
time.sleep(3)
idx = self.bot_id[username]
if idx >= len(obs) or obs[idx] is None:
continue
# print("vel:", obs[idx]['location_stats']['vel'])
if code_info[idx].is_ready and act[idx].type == mineland.Action.RESUME and self.is_agent_stop(obs[idx]):
if self.idle_timer[username] > 0:
self.idle_timer[username] -= 1
if self.idle_timer[username] == 0:
act[idx] = mineland.Action(mineland.Action.NEW, self.action_follow_and_lookat(username))
else:
self.idle_timer[username] = self.idle_timer_max
# step (step execution will last for 200ms, by default)
obs, code_info, event, _, _ = self.mland.step(act)
def debug(self):
obs = self.mland.reset()
assert self.bot_cnt == len(obs)
# Initialize
act = mineland.Action.no_op(self.bot_cnt)
obs, code_info, event, _, _ = self.mland.step(act)
# use ctrl+c to break
while(True):
self.code_check(code_info)
act = mineland.Action.no_op(self.bot_cnt)
string = ''
tmp = input()
while tmp != '':
string += tmp + '\n'
tmp = input()
act[0] = mineland.Action(mineland.Action.NEW, string)
obs, code_info, event, _, _ = self.mland.step(act)
def add_skill(infos, path = None):
agent = XBot('SkillAgent', 'SkillAgent')
if path is not None:
agent.action_agent.skill_agent.ckpt_dir = path
for info in infos:
agent.action_agent.skill_agent.add_new_skill({
"program_name": info["program_name"],
"program_code": info["program_code"],
"program_arguments": info["program_arguments"],
})
Main(
is_enable_use_last_action_as_trigger=True,
).run()In the code, act represents the action that agents will do. response is the raw action of agent. response = agent.get_from_output() gets the action of agent.
If you dont use gym before, I recommend to run a simple demo of gymnasium. The design goal of MineLand is similar to gymnasium.
jbarrerobuch
I can run Alex in simulation, brain os working and receiving critics. Now I'd like to run it in my own server (lan game). I successfully ran the example script connect to any server.
So now I need a merge of both example scripts.
I'm trying to modify the connect to any server script to add just 1 bot and use the Alex.run method in a loop.
I'm having difficulties adding the bots to the server. If I reset the environment with agents_count = 0 it fails with some key error. I assume it's not meant to reset without agents. I was trying to use Mineland.add_agent() to add the agent as it does in the example to connect at any server.
Also I don't see where the instances of Alex are stored when they are agents in the server. How can I access to Alex.run method of each one?
Thanks!