import gradio as gr import requests import json import os from pathlib import Path import inquirer import typer from rich.console import Console from rich.prompt import IntPrompt, Prompt, Confirm import argparse import logging import util from model import get_all_embeddings, get_all_llms from setting import Settings, get_all_model_settings, load_model_setting # import Settings, get_all_model_settings, load_model_setting from model import agi_init import gym from retriever import ( create_new_memory_retriever, ) import gym_coup import random from rlcard.utils import set_seed import rlcard from rlcard import models from rlcard.models import leducholdem_rule_models #Inferenec function def predict(openai_gpt4_key, game_selection, action, inputs, top_p, temperature, chat_counter, dialogue_chatbot=[], system_chatbot=[], history=[]): verified, settings, env, ctx = history['verified'], history['settings'], history['env'], history['ctx'] bot_long_memory, bot_short_memory = history['bot_long_memory'], history['bot_short_memory'] agents_num, chips, user_index, game_idx, mode = history['agents_num'], history['chips'], history['user_index'], history['game_idx'], history['mode'] status_message = '' valid_actions = gr.Dropdown.update(choices=[], value=None) if env is None: #settings.model.llm.openai_api_key is None: if not verified: res = util.verify_openai_token(openai_gpt4_key) if res != "OK": return system_chatbot, dialogue_chatbot, valid_actions, history, chat_counter, res history['verified'] = True if game_selection == '' or game_selection is None: return system_chatbot, dialogue_chatbot, valid_actions, history, chat_counter, "Please select a game." settings = Settings() settings.model = load_model_setting("openai-gpt-4-0613") #settings.model.llm.openai_api_key = None #settings.model.embedding.openai_api_key = None settings.model.llm.openai_api_key = openai_gpt4_key settings.model.embedding.openai_api_key = openai_gpt4_key res = util.verify_model_initialization(settings) if res != "OK": return system_chatbot, dialogue_chatbot, valid_actions, history, chat_counter, res else: # read agents configs agent1_file = 'person_config/Persuader.json' agent1_config = util.load_json(Path(agent1_file)) agent1_config["path"] = agent1_file agent2_file = 'person_config/GoodGuy.json' agent2_config = util.load_json(Path(agent2_file)) agent2_config["path"] = agent2_file agent_configs = [agent1_config, agent2_config] agent_names = [agent1_config["name"], agent2_config["name"]] if game_selection == 'coup': game_config_file = 'game_config/coup.json' elif game_selection == 'leduc-holdem': game_config_file = 'game_config/leduc_limit.json' elif game_selection == 'limit-holdem': game_config_file = 'game_config/limit_holdem.json' game_config = util.load_json(Path(game_config_file)) game_config["path"] = game_config_file user_index = 1 console = Console() ctx = agi_init(agent_configs, game_config, console, settings, user_index) os.environ["OPENAI_API_KEY"] = openai_gpt4_key print(game_selection) if game_selection != 'coup': env = rlcard.make(game_selection) else: env = gym.make('coup-v0') env.reset() history['env'] = env history['ctx'] = ctx #valid_actions['label'] = 'hello there' for i in range(agents_num): bot_short_memory.append([f'{game_idx+1}th Game Start']) bot_long_memory.append([f'{game_idx+1}th Game Start']) status_message = 'Verified.' if game_selection != 'coup': valid_actions = f"{env.get_state(env.get_player_id())['raw_legal_actions']}" else: valid_action_list = env.get_valid_actions(text=True) # check if opponent makes move first if game_selection != 'coup': idx = env.get_player_id() else: idx = env.game.whose_action if idx != user_index: amy = ctx.robot_agents[idx] if game_selection != 'coup': amy_obs = env.get_state(env.get_player_id())['raw_obs'] amy_index = env.get_player_id() amy_obs['game_num'] = game_idx+1 amy_obs['rest_chips'] = chips[idx] amy_obs['opponent_rest_chips'] = chips[(idx+1)%agents_num] valid_action_list = env.get_state(env.get_player_id())['raw_legal_actions'] else: amy_obs =env.get_obs(text=True,p2_view = (idx==1)) amy_index = env.game.whose_action valid_action_list = env.get_valid_actions(text=True) opponent_name = ctx.robot_agents[(idx+1)%agents_num].name act, comm, bot_short_memory, bot_long_memory = amy.make_act(amy_obs, opponent_name, amy_index, valid_action_list, verbose_print=False, game_idx=game_idx, round=0, bot_short_memory=bot_short_memory, bot_long_memory=bot_long_memory, console=Console(), log_file_name=None, mode=mode) if game_selection != 'coup': env.step(act, raw_action=True) else: env.step(act) win_message = env.game.call_system_info() # print(win_message) if win_message is not None: print(win_message) win_message = win_message.replace('Player 0',ctx.robot_agents[0].name) win_message = win_message.replace('Player 1',ctx.robot_agents[1].name) win_message = win_message.replace('I',ctx.robot_agents[idx].name) win_message = win_message.replace('the opponent',ctx.robot_agents[(idx + 1) % agents_num].name) bot_short_memory.append(win_message) bot_long_memory.append(win_message) dialogue_chatbot.append((None, comm)) system_chatbot.append((None, f'Suspicion-Agent action: {act}')) # get user observation idx = user_index #env.get_player_id() if game_selection != 'coup': amy_obs = env.get_state(idx)['raw_obs'] #amy_obs['game_num'] = game_idx+1 amy_obs['rest_chips'] = chips[idx] amy_obs['opponent_rest_chips'] = chips[(idx+1)%agents_num] valid_actions = env.get_state(idx)['raw_legal_actions'] else: amy_obs =env.get_obs(text=True,p2_view = (idx==1)) valid_actions = env.get_valid_actions(text=True) if game_selection != 'coup': game_state_string = "" for key, value in amy_obs.items(): if key != 'legal_actions': game_state_string += f"{key}: {value}\n" system_chatbot.append((f'Game state:\n{game_state_string}', None)) else: system_chatbot.append((f'Game state:\n{amy_obs}', None)) #system_chatbot.append((f'{amy_obs}', None)) valid_actions = gr.Dropdown.update(choices=valid_actions, value=None) return system_chatbot, dialogue_chatbot, valid_actions, history, chat_counter, status_message #else: # return system_chatbot, dialogue_chatbot, history, chat_counter, "Already Verified." # check if game is over if game_selection != 'coup': game_over = env.is_over() else: game_over = env.game.game_over if game_over: status_message = "Game ended." return system_chatbot, dialogue_chatbot, valid_actions, history, chat_counter, status_message if action is None: status_message = "No action received." return system_chatbot, dialogue_chatbot, valid_actions, history, chat_counter, status_message if game_selection != 'coup': if action not in env.get_state(env.get_player_id())['raw_legal_actions']: status_message = "Not a valid action. Please enter a valid action." return system_chatbot, dialogue_chatbot, valid_actions, history, chat_counter, status_message else: if action not in env.get_valid_actions(text=True): status_message = "Not a valid action. Please enter a valid action." return system_chatbot, dialogue_chatbot, valid_actions, history, chat_counter, status_message # message can be empty #if inputs is None or inputs == "": # status_message += " No message received." # return system_chatbot, dialogue_chatbot, history, chat_counter, status_message # user takes action if game_selection != 'coup': env_state = env.get_state(env.get_player_id())['raw_obs'] else: env_state = env.get_obs(text=True,p2_view = (env.game.whose_action==1)) # here action comes from user input #act,_ = rule_model.eval_step(env.get_state(env.get_player_id())) act = action #env.get_state(env.get_player_id())['raw_legal_actions'][0] if game_selection != 'coup': bot_short_memory[(user_index + 1) % agents_num].append( f"The valid action list of {ctx.robot_agents[user_index].name} is {env.get_state(env.get_player_id())['raw_legal_actions']}, and he tries to take action: {act}. He said, {inputs}") # bot_short_memory[(args.user_index) % args.agents_num].append( # f"{ctx.robot_agents[args.user_index].name} have the observation: {env.get_state(env.get_player_id())['raw_obs']}, and try to take action: {act}.") bot_long_memory[(user_index) % agents_num].append( f"{ctx.robot_agents[user_index].name} have the observation: {env.get_state(env.get_player_id())['raw_obs']}, and try to take action: {act}.") # bot_long_memory[(args.user_index) % args.agents_num].append( # f"{ctx.robot_agents[args.user_index].name} try to take action: {act}.") else: bot_short_memory[(user_index + 1) % agents_num].append( f"The valid action list of {ctx.robot_agents[user_index].name} is {env.get_valid_actions(text=True)}, and he tries to take action: {act}. He said, {inputs}") # bot_short_memory[(args.user_index) % args.agents_num].append( # f"{ctx.robot_agents[args.user_index].name} have the observation: {env.get_state(env.get_player_id())['raw_obs']}, and try to take action: {act}.") bot_long_memory[(user_index) % agents_num].append( f"{ctx.robot_agents[user_index].name} have the observation: {env.get_obs(text=True,p2_view = (env.game.whose_action==1))}, and try to take action: {act}.") # bot_long_memory[(args.user_index) % args.agents_num].append( # f"{ctx.robot_agents[args.user_index].name} try to take action: {act}.") if game_selection != 'coup': env.step(act, raw_action=True) else: env.step(act) comm = None if game_selection != 'coup': game_over = env.is_over() else: game_over = env.game.game_over if not game_over: # opponent move # bot reaction if game_selection != 'coup': idx = env.get_player_id() amy = ctx.robot_agents[idx] amy_index = env.get_player_id() amy_obs = env.get_state(env.get_player_id())['raw_obs'] amy_obs['game_num'] = game_idx+1 amy_obs['rest_chips'] = chips[idx] amy_obs['opponent_rest_chips'] = chips[(idx+1)%agents_num] valid_action_list = env.get_state(env.get_player_id())['raw_legal_actions'] else: idx = env.game.whose_action amy = ctx.robot_agents[idx] amy_index = env.game.whose_action amy_obs = env.get_obs(text=True,p2_view = (idx==1)) valid_action_list = env.get_valid_actions(text=True) opponent_name = ctx.robot_agents[(idx+1)%agents_num].name act, comm, bot_short_memory, bot_long_memory = amy.make_act(amy_obs, opponent_name, amy_index, valid_action_list, verbose_print=False, game_idx=game_idx, round=0, bot_short_memory=bot_short_memory, bot_long_memory=bot_long_memory, console=Console(), log_file_name=None, mode=mode) if game_selection != 'coup': env.step(act, raw_action=True) else: env.step(act) idx = user_index # env.get_player_id() if game_selection != 'coup': amy_obs = env.get_state(idx)['raw_obs'] else: amy_obs = env.get_obs(text=True,p2_view = (idx==1)) #amy_obs['game_num'] = game_idx+1 if game_selection != 'coup': amy_obs['rest_chips'] = chips[idx] amy_obs['opponent_rest_chips'] = chips[(idx+1)%agents_num] valid_actions = env.get_state(idx)['raw_legal_actions'] else: valid_actions = env.get_valid_actions(text=True) game_state_string = "" for key, value in amy_obs.items(): if key != 'legal_actions': game_state_string += f"{key}: {value}\n" dialogue_chatbot.append((inputs if inputs != "" else None, comm)) system_chatbot.append((f'My action: {action}', f'Suspicion-Agent action: {act}')) system_chatbot.append((f'Game state:\n{game_state_string}', None)) if game_selection != 'coup': if env.is_over(): pay_offs = env.get_payoffs() for idx in range(len(pay_offs)): pay_offs[idx] = pay_offs[idx]*2 chips[idx] += pay_offs[idx] if pay_offs[user_index] > 0: win_message = f'You win {pay_offs[user_index]} chips, Suspicion-Agent lose {pay_offs[user_index]} chips' else: win_message = f'Suspicion-Agent win {pay_offs[(user_index+1)%agents_num]} chips, you lose {pay_offs[(user_index+1)%agents_num]} chips' idx = (user_index + 1)%agents_num amy_obs = env.get_state(idx)['raw_obs'] bot_hand = amy_obs['hand'] system_chatbot.append((None, f'Suspicion-Agent hand: {bot_hand}')) system_chatbot.append((f'Gameover.\n {win_message}', None)) valid_actions = [] else: if env.game.game_over: idx = (user_index + 1)%agents_num amy_obs = env.get_obs(text=True,p2_view = (idx==1)) bot_hand = amy_obs system_chatbot.append((None, f'Suspicion-Agent hand: {bot_hand}')) system_chatbot.append((f'Gameover.\n {win_message}', None)) valid_actions = [] status_message += " Message received." valid_actions = gr.Dropdown.update(choices=valid_actions, value=None) return system_chatbot, dialogue_chatbot, valid_actions, history, 1, status_message #Resetting to blank def reset_textbox(): return gr.update(value='') #to set a component as visible=False def set_visible_false(): return gr.update(visible=False) #to set a component as visible=True def set_visible_true(): return gr.update(visible=True) def update_instruction(game_selection): if game_selection is not None and game_selection != '': if game_selection == 'coup': with open('./game_config/coup.json') as file: contents = json.load(file) elif game_selection == 'leduc-holdem': with open('./game_config/leduc_limit.json') as file: contents = json.load(file) elif game_selection == 'limit-holdem': with open('./game_config/limit_holdem.json') as file: contents = json.load(file) return f"Game rule: {contents['game_rule']}\n\n\nObservation Rule: {contents['observation_rule']}" # update valid actions list def set_valid_actions(): if game_selection != 'coup': print(env.get_state(env.get_player_id())['raw_legal_actions']) else: print(env.get_valid_actions(text=True)) if env is None: return gr.update(value='') else: if game_selection != 'coup': valid_actions_list = env.get_state(env.get_player_id())['raw_legal_actions'] else: valid_actions_list = env.get_valid_actions(text=True) return gr.update(value=f'{valid_actions_list}') title = """