repo_description: ChatArena (or Chat Arena) is a Multi-Agent Language Game Environments for LLMs. The goal is to develop communication and collaboration capabilities of AIs.
repo_topics: ai artificial-intelligence chatgpt gpt-4 large-language-models multi-agent multi-agent-reinforcement-learning multi-agent-simulation natural-language-processing python
code: import json
  import re
  from glob import glob
  
  import gradio as gr
  
  from chatarena.arena import Arena, TooManyInvalidActions
  from chatarena.backends import BACKEND_REGISTRY
  from chatarena.backends.human import HumanBackendError
  from chatarena.config import ArenaConfig
  from chatarena.database import SupabaseDB, log_arena, log_messages, supabase_available
  from chatarena.environments import ENV_REGISTRY
  from chatarena.message import Message
  
  css = """#col-container {max-width: 90%; margin-left: auto; margin-right: auto; display: flex; flex-direction: column;}
  #header {text-align: center;}
  #col-chatbox {flex: 1; max-height: min(750px, 100%);}
  #label {font-size: 2em; padding: 0.5em; margin: 0;}
  .message {font-size: 1.2em;}
  .message-wrap {max-height: min(700px, 100vh);}
  """
  # .wrap {min-width: min(640px, 100vh)}
  # #env-desc {max-height: 100px; overflow-y: auto;}
  # .textarea {height: 100px; max-height: 100px;}
  # #chatbot-tab-all {height: 750px; max-height: min(750px, 100%);}
  # #chatbox {height: min(750px, 100%); max-height: min(750px, 100%);}
  # #chatbox.block {height: 730px}
  # .wrap {max-height: 680px;}
  # .scroll-hide {overflow-y: scroll; max-height: 100px;}
  
  
  DEBUG = False
  
  DEFAULT_BACKEND = "openai-chat"
  DEFAULT_ENV = "conversation"
  MAX_NUM_PLAYERS = 6
  DEFAULT_NUM_PLAYERS = 2
  
  
  def load_examples():
      example_configs = {}
      # Load json config files from examples folder
      example_files = glob("examples/*.json")
      for example_file in example_files:
          with open(example_file, encoding="utf-8") as f:
              example = json.load(f)
              try:
                  example_configs[example["name"]] = example
              except KeyError:
                  print(f"Example {example_file} is missing a name field. Skipping.")
      return example_configs
  
  
  EXAMPLE_REGISTRY = load_examples()
  
  DB = SupabaseDB() if supabase_available else None
  
  
  def get_moderator_components(visible=True):
      name = "Moderator"
      with gr.Row():
          with gr.Column():
              role_desc = gr.Textbox(
                  label="Moderator role",
                  lines=1,
                  visible=visible,
                  interactive=True,
                  placeholder=f"Enter the role description for {name}",
              )
              terminal_condition = gr.Textbox(
                  show_label=False,
                  lines=1,
                  visible=visible,
                  interactive=True,
                  placeholder="Enter the termination criteria",
              )
          with gr.Column():
              backend_type = gr.Dropdown(
                  show_label=False,
                  visible=visible,
                  interactive=True,
                  choices=list(BACKEND_REGISTRY.keys()),
                  value=DEFAULT_BACKEND,
              )
              with gr.Accordion(
                  f"{name} Parameters", open=False, visible=visible
              ) as accordion:
                  temperature = gr.Slider(
                      minimum=0,
                      maximum=2.0,
                      step=0.1,
                      interactive=True,
                      visible=visible,
                      label="temperature",
                      value=0.7,
                  )
                  max_tokens = gr.Slider(
                      minimum=10,
                      maximum=500,
                      step=10,
                      interactive=True,
                      visible=visible,
                      label="max tokens",
                      value=200,
                  )
  
      return [
          role_desc,
          terminal_condition,
          backend_type,
          accordion,
          temperature,
          max_tokens,
      ]
  
  
  def get_player_components(name, visible):
      with gr.Row():
          with gr.Column():
              role_name = gr.Textbox(
                  line=1,
                  show_label=False,
                  interactive=True,
                  visible=visible,
                  placeholder=f"Player name for {name}",
              )
              role_desc = gr.Textbox(
                  lines=3,
                  show_label=False,
                  interactive=True,
                  visible=visible,
                  placeholder=f"Enter the role description for {name}",
              )
          with gr.Column():
              backend_type = gr.Dropdown(
                  show_label=False,
                  choices=list(BACKEND_REGISTRY.keys()),
                  interactive=True,
                  visible=visible,
                  value=DEFAULT_BACKEND,
              )
              with gr.Accordion(
                  f"{name} Parameters", open=False, visible=visible
              ) as accordion:
                  temperature = gr.Slider(
                      minimum=0,
                      maximum=2.0,
                      step=0.1,
                      interactive=True,
                      visible=visible,
                      label="temperature",
                      value=0.7,
                  )
                  max_tokens = gr.Slider(
                      minimum=10,
                      maximum=500,
                      step=10,
                      interactive=True,
                      visible=visible,
                      label="max tokens",
                      value=200,
                  )
  
      return [role_name, role_desc, backend_type, accordion, temperature, max_tokens]
  
  
  def get_empty_state():
      return gr.State({"arena": None})
  
  
  with gr.Blocks(css=css) as demo:
      state = get_empty_state()
      all_components = []
  
      with gr.Column(elem_id="col-container"):
          gr.Markdown(
              """# 🏟 ChatArena️<br>
  Prompting multiple AI agents to play games in a language-driven environment.
  **[Project Homepage](https://github.com/chatarena/chatarena)**""",
              elem_id="header",
          )
  
          with gr.Row():
              env_selector = gr.Dropdown(
                  choices=list(ENV_REGISTRY.keys()),
                  value=DEFAULT_ENV,
                  interactive=True,
                  label="Environment Type",
                  show_label=True,
              )
              example_selector = gr.Dropdown(
                  choices=list(EXAMPLE_REGISTRY.keys()),
                  interactive=True,
                  label="Select Example",
                  show_label=True,
              )
  
          # Environment configuration
          env_desc_textbox = gr.Textbox(
              show_label=True,
              lines=2,
              visible=True,
              label="Environment Description",
              placeholder="Enter a description of a scenario or the game rules.",
          )
  
          all_components += [env_selector, example_selector, env_desc_textbox]
  
          with gr.Row():
              with gr.Column(elem_id="col-chatbox"):
                  with gr.Tab("All", visible=True):
                      chatbot = gr.Chatbot(
                          elem_id="chatbox", visible=True, show_label=False
                      )
  
                  player_chatbots = []
                  for i in range(MAX_NUM_PLAYERS):
                      player_name = f"Player {i + 1}"
                      with gr.Tab(player_name, visible=(i < DEFAULT_NUM_PLAYERS)):
                          player_chatbot = gr.Chatbot(
                              elem_id=f"chatbox-{i}",
                              visible=i < DEFAULT_NUM_PLAYERS,
                              label=player_name,
                              show_label=False,
                          )
                          player_chatbots.append(player_chatbot)
  
              all_components += [chatbot, *player_chatbots]
  
              with gr.Column(elem_id="col-config"):  # Player Configuration
                  # gr.Markdown("Player Configuration")
                  parallel_checkbox = gr.Checkbox(
                      label="Parallel Actions", value=False, visible=True
                  )
                  with gr.Accordion("Moderator", open=False, visible=True):
                      moderator_components = get_moderator_components(True)
                  all_components += [parallel_checkbox, *moderator_components]
  
                  all_players_components, players_idx2comp = [], {}
                  with gr.Blocks():
                      num_player_slider = gr.Slider(
                          2,
                          MAX_NUM_PLAYERS,
                          value=DEFAULT_NUM_PLAYERS,
                          step=1,
                          label="Number of players:",
                      )
                      for i in range(MAX_NUM_PLAYERS):
                          player_name = f"Player {i + 1}"
                          with gr.Tab(
                              player_name, visible=(i < DEFAULT_NUM_PLAYERS)
                          ) as tab:
                              player_comps = get_player_components(
                                  player_name, visible=(i < DEFAULT_NUM_PLAYERS)
                              )
  
                          players_idx2comp[i] = player_comps + [tab]
                          all_players_components += player_comps + [tab]
  
                  all_components += [num_player_slider] + all_players_components
  
                  def variable_players(k):
                      k = int(k)
                      update_dict = {}
                      for i in range(MAX_NUM_PLAYERS):
                          if i < k:
                              for comp in players_idx2comp[i]:
                                  update_dict[comp] = gr.update(visible=True)
                              update_dict[player_chatbots[i]] = gr.update(visible=True)
                          else:
                              for comp in players_idx2comp[i]:
                                  update_dict[comp] = gr.update(visible=False)
                              update_dict[player_chatbots[i]] = gr.update(visible=False)
                      return update_dict
  
                  num_player_slider.change(
                      variable_players,
                      num_player_slider,
                      all_players_components + player_chatbots,
                  )
  
                  human_input_textbox = gr.Textbox(
                      show_label=True,
                      label="Human Input",
                      lines=1,
                      visible=True,
                      interactive=True,
                      placeholder="Enter your input here",
                  )
                  with gr.Row():
                      btn_step = gr.Button("Start")
                      btn_restart = gr.Button("Clear")
  
                  all_components += [human_input_textbox, btn_step, btn_restart]
  
      def _convert_to_chatbot_output(all_messages, display_recv=False):
          chatbot_output = []
          for i, message in enumerate(all_messages):
              agent_name, msg, recv = (
                  message.agent_name,
                  message.content,
                  str(message.visible_to),
              )
              new_msg = re.sub(
                  r"\n+", "<br>", msg.strip()
              )  # Preprocess message for chatbot output
              if display_recv:
                  new_msg = f"**{agent_name} (-> {recv})**: {new_msg}"  # Add role to the message
              else:
                  new_msg = f"**{agent_name}**: {new_msg}"
  
              if agent_name == "Moderator":
                  chatbot_output.append((new_msg, None))
              else:
                  chatbot_output.append((None, new_msg))
          return chatbot_output
  
      def _create_arena_config_from_components(all_comps: dict) -> ArenaConfig:
          env_desc = all_comps[env_desc_textbox]
  
          # Initialize the players
          num_players = all_comps[num_player_slider]
          player_configs = []
          for i in range(num_players):
              role_name, role_desc, backend_type, temperature, max_tokens = (
                  all_comps[c]
                  for c in players_idx2comp[i]
                  if not isinstance(c, (gr.Accordion, gr.Tab))
              )
              player_config = {
                  "name": role_name,
                  "role_desc": role_desc,
                  "global_prompt": env_desc,
                  "backend": {
                      "backend_type": backend_type,
                      "temperature": temperature,
                      "max_tokens": max_tokens,
                  },
              }
              player_configs.append(player_config)
  
          # Initialize the environment
          env_type = all_comps[env_selector]
          # Get moderator config
          (
              mod_role_desc,
              mod_terminal_condition,
              moderator_backend_type,
              mod_temp,
              mod_max_tokens,
          ) = (
              all_comps[c]
              for c in moderator_components
              if not isinstance(c, (gr.Accordion, gr.Tab))
          )
          moderator_config = {
              "role_desc": mod_role_desc,
              "global_prompt": env_desc,
              "terminal_condition": mod_terminal_condition,
              "backend": {
                  "backend_type": moderator_backend_type,
                  "temperature": mod_temp,
                  "max_tokens": mod_max_tokens,
              },
          }
          env_config = {
              "env_type": env_type,
              "parallel": all_comps[parallel_checkbox],
              "moderator": moderator_config,
              "moderator_visibility": "all",
              "moderator_period": None,
          }
  
          # arena_config = {"players": player_configs, "environment": env_config}
          arena_config = ArenaConfig(players=player_configs, environment=env_config)
          return arena_config
  
      def step_game(all_comps: dict):
          yield {
              btn_step: gr.update(value="Running...", interactive=False),
              btn_restart: gr.update(interactive=False),
          }
  
          cur_state = all_comps[state]
  
          # If arena is not yet created, create it
          if cur_state["arena"] is None:
              # Create the Arena
              arena_config = _create_arena_config_from_components(all_comps)
              arena = Arena.from_config(arena_config)
              log_arena(arena, database=DB)
              cur_state["arena"] = arena
          else:
              arena = cur_state["arena"]
  
          try:
              timestep = arena.step()
          except HumanBackendError as e:
              # Handle human input and recover with the game update
              human_input = all_comps[human_input_textbox]
              if human_input == "":
                  timestep = None  # Failed to get human input
              else:
                  timestep = arena.environment.step(e.agent_name, human_input)
          except TooManyInvalidActions:
              timestep = arena.current_timestep
              timestep.observation.append(
                  Message(
                      "System",
                      "Too many invalid actions. Game over.",
                      turn=-1,
                      visible_to="all",
                  )
              )
              timestep.terminal = True
  
          if timestep is None:
              yield {
                  human_input_textbox: gr.update(
                      value="", placeholder="Please enter a valid input"
                  ),
                  btn_step: gr.update(value="Next Step", interactive=True),
                  btn_restart: gr.update(interactive=True),
              }
          else:
              all_messages = timestep.observation  # user sees what the moderator sees
              log_messages(arena, all_messages, database=DB)
  
              chatbot_output = _convert_to_chatbot_output(all_messages, display_recv=True)
              update_dict = {
                  human_input_textbox: gr.Textbox.update(value=""),
                  chatbot: chatbot_output,
                  btn_step: gr.update(
                      value="Next Step", interactive=not timestep.terminal
                  ),
                  btn_restart: gr.update(interactive=True),
                  state: cur_state,
              }
              # Get the visible messages for each player
              for i, player in enumerate(arena.players):
                  player_messages = arena.environment.get_observation(player.name)
                  player_output = _convert_to_chatbot_output(player_messages)
                  # Update the player's chatbot output
                  update_dict[player_chatbots[i]] = player_output
  
              if DEBUG:
                  arena.environment.print()
  
              yield update_dict
  
      def restart_game(all_comps: dict):
          cur_state = all_comps[state]
          cur_state["arena"] = None
          yield {
              chatbot: [],
              btn_restart: gr.update(interactive=False),
              btn_step: gr.update(interactive=False),
              state: cur_state,
          }
  
          arena_config = _create_arena_config_from_components(all_comps)
          arena = Arena.from_config(arena_config)
          log_arena(arena, database=DB)
          cur_state["arena"] = arena
  
          yield {
              btn_step: gr.update(value="Start", interactive=True),
              btn_restart: gr.update(interactive=True),
              state: cur_state,
          }
  
      # Remove Accordion and Tab from the list of components
      all_components = [
          comp for comp in all_components if not isinstance(comp, (gr.Accordion, gr.Tab))
      ]
  
      # If any of the Textbox, Slider, Checkbox, Dropdown, RadioButtons is changed, the Step button is disabled
      for comp in all_components:
  
          def _disable_step_button(state):
              if state["arena"] is not None:
                  return gr.update(interactive=False)
              else:
                  return gr.update()
  
          if (
              isinstance(
                  comp, (gr.Textbox, gr.Slider, gr.Checkbox, gr.Dropdown, gr.Radio)
              )
              and comp is not human_input_textbox
          ):
              comp.change(_disable_step_button, state, btn_step)
  
      btn_step.click(
          step_game,
          set(all_components + [state]),
          [chatbot, *player_chatbots, btn_step, btn_restart, state, human_input_textbox],
      )
      btn_restart.click(
          restart_game,
          set(all_components + [state]),
          [chatbot, *player_chatbots, btn_step, btn_restart, state, human_input_textbox],
      )
  
      # If an example is selected, update the components
      def update_components_from_example(all_comps: dict):
          example_name = all_comps[example_selector]
          example_config = EXAMPLE_REGISTRY[example_name]
          update_dict = {}
  
          # Update the environment components
          env_config = example_config["environment"]
          update_dict[env_desc_textbox] = gr.update(value=example_config["global_prompt"])
          update_dict[env_selector] = gr.update(value=env_config["env_type"])
          update_dict[parallel_checkbox] = gr.update(value=env_config["parallel"])
  
          # Update the moderator components
          if "moderator" in env_config:
              (
                  mod_role_desc,
                  mod_terminal_condition,
                  moderator_backend_type,
                  mod_temp,
                  mod_max_tokens,
              ) = (
                  c
                  for c in moderator_components
                  if not isinstance(c, (gr.Accordion, gr.Tab))
              )
              update_dict[mod_role_desc] = gr.update(
                  value=env_config["moderator"]["role_desc"]
              )
              update_dict[mod_terminal_condition] = gr.update(
                  value=env_config["moderator"]["terminal_condition"]
              )
              update_dict[moderator_backend_type] = gr.update(
                  value=env_config["moderator"]["backend"]["backend_type"]
              )
              update_dict[mod_temp] = gr.update(
                  value=env_config["moderator"]["backend"]["temperature"]
              )
              update_dict[mod_max_tokens] = gr.update(
                  value=env_config["moderator"]["backend"]["max_tokens"]
              )
  
          # Update the player components
          update_dict[num_player_slider] = gr.update(value=len(example_config["players"]))
          for i, player_config in enumerate(example_config["players"]):
              role_name, role_desc, backend_type, temperature, max_tokens = (
                  c
                  for c in players_idx2comp[i]
                  if not isinstance(c, (gr.Accordion, gr.Tab))
              )
  
              update_dict[role_name] = gr.update(value=player_config["name"])
              update_dict[role_desc] = gr.update(value=player_config["role_desc"])
              update_dict[backend_type] = gr.update(
                  value=player_config["backend"]["backend_type"]
              )
              update_dict[temperature] = gr.update(
                  value=player_config["backend"]["temperature"]
              )
              update_dict[max_tokens] = gr.update(
                  value=player_config["backend"]["max_tokens"]
              )
  
          return update_dict
  
      example_selector.change(
          update_components_from_example,
          set(all_components + [state]),
          all_components + [state],
      )
  
  demo.queue()
  demo.launch(debug=DEBUG, server_port=8080)
  