Spaces:
Running
Running
| 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 = """<h1 align="center">Suspicion-Agent Demo</h1>""" | |
| #display message for themes feature | |
| theme_addon_msg = """<center>This is an official Demo for <b>Suspicion-Agent: Playing Imperfect Information Games with Theory of Mind Aware GPT4</b>. Check out our paper for more details <a href="https://arxiv.org/abs/2309.17277" target="_blank">here</a>! Some Notes: In the initial games in the demo, Suspicion Agent typically exhibits aggressive play. This is because of the nature of GPT-4 and it is also observed in <a href="https://arxiv.org/abs/2308.12466" target="_blank">here</a>. </center> | |
| """ | |
| #Using info to add additional information about System message in GPT4 | |
| system_msg_info = """A conversation could begin with a system message to gently instruct the assistant. | |
| System message helps set the behavior of the AI Assistant. For example, the assistant could be instructed with 'You are a helpful assistant.'""" | |
| #Modifying existing Gradio Theme | |
| theme = gr.themes.Soft(primary_hue="zinc", secondary_hue="blue", neutral_hue="blue", | |
| text_size=gr.themes.sizes.text_lg) | |
| with gr.Blocks(css = """#col_container { margin-left: auto; margin-right: auto;} #chatbot {height: 520px; overflow: auto;}""", | |
| theme=theme) as demo: | |
| gr.HTML(title) | |
| gr.HTML("""<h3 align="center">🔥This Huggingface Gradio Demo provides a variety of game matches against a GPT4 agent. Please note that you would be needing an OPENAI API key for GPT4 access</h1>""") | |
| gr.HTML(theme_addon_msg) | |
| gr.HTML("""<center><a href="https://arxiv.org/abs/2309.17277" target="_blank">Github Codes</a></center>""") | |
| gr.HTML('''<center><a href="https://huggingface.co/spaces/cr7-gjx/Suspicion-Agent-Demo?duplicate=true"><img src="https://bit.ly/3gLdBN6" alt="Duplicate Space"></a>Duplicate the Space and run securely with your OpenAI API Key</center>''') | |
| initial_state = { | |
| 'verified': False, | |
| 'settings': None, | |
| 'env': None, | |
| 'ctx': None, | |
| 'bot_long_memory': [], | |
| 'bot_short_memory': [], | |
| 'agents_num': 2, | |
| 'chips': [50,50], | |
| 'user_index': 1, | |
| 'game_idx': 0, | |
| 'mode': 'first_tom', | |
| } | |
| with gr.Column(elem_id = "col_container"): | |
| #Users need to provide their own GPT4 API key, it is no longer provided by Huggingface | |
| with gr.Row(): | |
| with gr.Column(): | |
| openai_gpt4_key = gr.Textbox(label="OpenAI GPT4 Key", value="", type="password", placeholder="sk..", info = "You have to provide your own GPT4 keys for this app to function properly",) | |
| with gr.Column(): | |
| game_selection = gr.Dropdown( | |
| ["leduc-holdem", "limit-holdem","coup"], label="Game Selections", info="Select the game to play from the dropdown" | |
| ) | |
| #with gr.Accordion(label="System message:", open=False): | |
| # system_msg = gr.Textbox(label="Instruct the AI Assistant to set its beaviour", info = system_msg_info, value="",placeholder="Type here..") | |
| # accordion_msg = gr.HTML(value="🚧 To set System message you will have to refresh the app", visible=False) | |
| with gr.Row(): | |
| instruction_panel = gr.Textbox(label='Game Instructions') | |
| with gr.Row(): | |
| with gr.Column(): | |
| system_chatbot = gr.Chatbot(label='Game Status', elem_id="system_chatbot") | |
| with gr.Column(): | |
| dialogue_chatbot = gr.Chatbot(label='Dialogue with GPT4', elem_id="dialogue_chatbot") | |
| #chatbot = gr.Chatbot(label='GPT4', elem_id="chatbot") | |
| #action = gr.Radio(['call', 'raise', 'fold'], label="Actions", info="Select the action to play") | |
| action = gr.Dropdown(placeholder="", label="Select an action.", info="") | |
| inputs = gr.Textbox(placeholder="", label="Type a message for the opponent. Messages are optional.") | |
| state = gr.State(initial_state) | |
| with gr.Row(): | |
| with gr.Column(scale=7): | |
| b1 = gr.Button().style(full_width=True) | |
| with gr.Column(scale=3): | |
| server_status_code = gr.Textbox(label="Status code from OpenAI server", ) | |
| #top_p, temperature | |
| with gr.Accordion("Parameters", open=False): | |
| top_p = gr.Slider(minimum=-0, maximum=1.0, value=1.0, step=0.05, interactive=True, label="Top-p (nucleus sampling)",) | |
| temperature = gr.Slider(minimum=-0, maximum=5.0, value=1.0, step=0.1, interactive=True, label="Temperature",) | |
| chat_counter = gr.Number(value=0, visible=False, precision=0) | |
| #Event handling | |
| inputs.submit(predict, [openai_gpt4_key, game_selection, action, inputs, top_p, temperature, chat_counter, dialogue_chatbot, system_chatbot, state], [system_chatbot, dialogue_chatbot, action, state, chat_counter, server_status_code],) #openai_api_key | |
| b1.click(predict, [openai_gpt4_key, game_selection, action, inputs, top_p, temperature, chat_counter, dialogue_chatbot, system_chatbot, state], [system_chatbot, dialogue_chatbot, action, state, chat_counter, server_status_code],) #openai_api_key | |
| #inputs.submit(set_visible_false, [], [system_msg]) | |
| #b1.click(set_visible_false, [], [system_msg]) | |
| #inputs.submit(set_visible_true, [], [accordion_msg]) | |
| #b1.click(set_visible_true, [], [accordion_msg]) | |
| game_selection.select(update_instruction, [game_selection], [instruction_panel]) | |
| b1.click(reset_textbox, [], [inputs]) | |
| inputs.submit(reset_textbox, [], [inputs]) | |
| #b1.click(set_valid_actions, [], [valid_actions]) | |
| #inputs.submit(set_valid_actions, [], [valid_actions]) | |
| demo.queue(max_size=99, concurrency_count=20).launch(debug=True) | |