import sys
sys.path.append("../../Gradio_Config")
from gradio_base import UIHelper, WebUI
import os
from gradio_base import WebUI, UIHelper, PORT, HOST, Client
from gradio_config import GradioConfig as gc
from typing import List, Tuple, Any
import gradio as gr
import time
class DebateUI(WebUI):
FORMAT = "{}\n<debate topic>\n{}\nAffirmative viewpoint:{}\nNegative viewpoint:{}\n<debate topic>{}"
AUDIENCE = "Audience"
cache = {}
all_agents_name = []
receive_server = None
@classmethod
def extract(cls, content):
topic = content.split("<debate topic>")[1].split("Affirmative viewpoint:")[0]
positive = content.split("<debate topic>")[1].split("Affirmative viewpoint:")[1].split("negative viewpoint:")[0]
negative = content.split("<debate topic>")[1].split("Affirmative viewpoint:")[1].split("negative viewpoint:")[1]
return topic.strip(), positive.strip(), negative.strip()
@classmethod
def merge(cls, theme, positive, negative, origin_content) -> str:
return cls.FORMAT.format(
origin_content.split("<debate topic>")[0],
theme, positive, negative,
origin_content.split("<debate topic>")[-1]
)
@classmethod
def convert2list4agentname(cls, sop):
only_name = []
agent_name = []
roles_to_names = sop.roles_to_names
for state_name,roles_names in roles_to_names.items():
for role,name in roles_names.items():
agent_name.append(f"{name}({role})")
only_name.append(name)
agent_name.append(cls.AUDIENCE)
agent_name = list(set(agent_name))
agent_name.sort()
return agent_name, only_name
def render_and_register_ui(self):
gc.add_agent(self.cache["only_name"])
def __init__(
self,
client_cmd: list,
socket_host: str = HOST,
socket_port: int = PORT,
bufsize: int = 1024,
ui_name: str = "DebateUI"
):
super(DebateUI, self).__init__(client_cmd, socket_host, socket_port, bufsize, ui_name)
self.first_recieve_from_client()
self.data_history = list()
self.caller = 0
def handle_message(self, history:list,
state, agent_name, token, node_name):
if state % 10 == 0:
self.data_history.append({agent_name: token})
elif state % 10 == 1:
# Same state. Need to add new bubble in same bubble.
if len(self.data_history) == 0:
self.data_history.append({agent_name:""})
self.data_history[-1][agent_name] += token
elif state % 10 == 2:
# New state. Need to add new bubble.
history.append([None, ""])
self.data_history.clear()
self.data_history.append({agent_name: token})
else:
assert False, "Invalid state."
render_data = self.render_bubble(history, self.data_history, node_name, render_node_name= True or state % 10 == 2)
return render_data
def start_button_when_click(self, theme, positive, negative, choose, mode, api_key):
"""
inputs=[self.text_theme, self.text_positive, self.text_negative, self.radio_choose],
outputs=[self.chatbot, self.btn_send]
"""
cosplay = None if choose == self.AUDIENCE else choose.split("(")[0]
message = dict(theme=theme, positive=positive, negative=negative, cosplay=cosplay, mode=mode, api_key=api_key)
self.send_start_cmd(message=message)
return gr.Chatbot.update(
visible=True
), gr.Button.update(visible=False)
def start_button_after_click(self, history):
"""
inputs=[self.chatbot],
outputs=[self.chatbot, self.text_user, self.btn_send, self.btn_reset, self.btn_next]
"""
if self.caller == 0:
# not single mode
self.data_history = list()
self.caller = 0
receive_server = self.receive_server
while True:
data_list: List = receive_server.send(None)
for item in data_list:
data = eval(item)
assert isinstance(data, list)
state, agent_name, token, node_name = data
assert isinstance(state, int)
if state == 30:
# user input
yield history,\
gr.Textbox.update(visible=True, interactive=True), \
gr.Button.update(visible=True, interactive=True),\
gr.Button.update(visible=True, interactive=True),\
gr.Button.update(visible=False)
return
elif state == 99:
# finish
yield history, gr.Textbox.update(visible=True, interactive=False, value="finish!"), \
gr.Button.update(visible=True, interactive=False, value="finish!"), gr.Button.update(visible=True, interactive=True),\
gr.Button.update(visible=False)
elif state == 98:
yield history, \
gr.Textbox.update(visible=False, interactive=False), \
gr.Button.update(visible=False, interactive=False),\
gr.Button.update(visible=False, interactive=False),\
gr.Button.update(visible=True, value=f"Next Agent: 🤖{agent_name} | Next Node: â•{node_name}")
return
else:
history = self.handle_message(history, state, agent_name, token, node_name)
yield history, \
gr.Textbox.update(visible=False, interactive=False), \
gr.Button.update(visible=False, interactive=False),\
gr.Button.update(visible=False, interactive=False),\
gr.Button.update(visible=False)
def send_button_when_click(self, text_user, history:list):
"""
inputs=[self.text_user, self.chatbot],
outputs=[self.text_user, self.btn_send, self.chatbot]
"""
history.append(
[UIHelper.wrap_css(text_user, "User"), None]
)
# print(f"server: send {text_user} to client")
self.send_message("<USER>"+text_user+self.SIGN["SPLIT"])
return gr.Textbox.update(value="", visible=False),\
gr.Button.update(visible=False), \
history,\
gr.Button.update(visible=False)
def reset_button_when_click(self, history, text_positive, text_negative, text_theme, text_user, btn_send, btn_start, btn_reset):
"""
self.chatbot,
self.text_positive,
self.text_negative,
self.text_theme,
self.text_user,
self.btn_send,
self.btn_start,
self.btn_reset
self.btn_next
"""
self.caller = 0
return None, \
"", \
"", \
"", \
"", \
gr.Button.update(value="Restarting...", interactive=False, visible=True),\
gr.Button.update(value="Restarting...", interactive=False, visible=True),\
gr.Button.update(value="Restarting...", interactive=False, visible=True),\
gr.Button.update(value="Restarting...", interactive=False, visible=False)
def reset_button_after_click(self, history, text_positive, text_negative, text_theme, text_user, btn_send, btn_start, btn_reset):
self.reset()
self.first_recieve_from_client(reset_mode=True)
return gr.Chatbot.update(value=None, visible=False),\
gr.Textbox.update(value=f"{self.cache['positive']}", interactive=True, visible=True),\
gr.Textbox.update(value=f"{self.cache['negative']}", interactive=True, visible=True),\
gr.Textbox.update(value=f"{self.cache['theme']}", interactive=True, visible=True),\
gr.Textbox.update(value=f"", interactive=True, visible=False),\
gr.Button.update(interactive=True, visible=False, value="Send"),\
gr.Button.update(interactive=True, visible=True, value="Start"),\
gr.Button.update(interactive=False, visible=False, value="Restart"),\
gr.Button.update(interactive=True, visible=False, value="Next Agent")
def btn_next_when_click(self):
yield gr.Button.update(visible=False)
self.send_message("nothing")
self.caller = 1 # will note clear the self.data_history
time.sleep(0.5)
return
def construct_ui(
self,
theme:str=None,
positive:str=None,
negative:str=None,
agents_name:List=None,
default_cos_play_id:int=None
):
theme = self.cache["theme"] if theme is None else theme
positive = self.cache["positive"] if positive is None else positive
negative = self.cache["negative"] if negative is None else negative
agents_name = self.cache["agents_name"] if agents_name is None else agents_name
default_cos_play_id = self.cache["default_cos_play_id"] if default_cos_play_id is None else default_cos_play_id
with gr.Blocks(css=gc.CSS) as demo:
gr.Markdown("""# Agents""")
gr.Markdown("""**Agents** is an open-source library/framework for building autonomous language agents.if you want to know more about **Agents**, please check our<a href="https://arxiv.org/pdf/2309.07870.pdf">📄 Paper</a> and<a href="http://www.aiwaves-agents.com/">📦 Github</a>. Here is a demo of **Agents**.""")
gr.Markdown("""<font size=5>If an error occurs or the queue is too long, please create your own demo by clicking <font color=red>Duplicate This Space</font> in the upper right corner.</font>""")
with gr.Row():
with gr.Column():
self.text_api = gr.Textbox(
value = self.cache["api_key"],
placeholder="openai key",
label="Please input valid openai key for gpt-3.5-turbo-16k."
)
self.radio_mode = gr.Radio(
[Client.SINGLE_MODE],
value=Client.SINGLE_MODE,
interactive=True,
label = Client.MODE_LABEL,
info = Client.MODE_INFO
)
self.text_theme = gr.Textbox(
label="Debate Topic:",
value=theme,
placeholder="Please input the Debate Topic"
)
self.text_positive = gr.Textbox(
label="Affirmative viewpoint:",
value=positive,
placeholder="Please input the Affirmative viewpoint"
)
self.text_negative = gr.Textbox(
label="Negative viewpoint:",
value=negative,
placeholder="Please input the Negative viewpoint"
)
self.radio_choose = gr.Radio(
agents_name,
value=agents_name[default_cos_play_id],
label="User'agent",
interactive=True
)
self.btn_start = gr.Button(
value="run"
)
VISIBLE = False
with gr.Column():
self.chatbot = gr.Chatbot(
height= 650,
elem_id="chatbot1",
label="Dialog",
visible=VISIBLE
)
self.btn_next = gr.Button(
value="Next Agent Start",
visible=False
)
self.text_user = gr.Textbox(
label="Input",
placeholder="Input here",
visible=VISIBLE
)
self.btn_send = gr.Button(
value="Send",
visible=VISIBLE
)
self.btn_reset = gr.Button(
value="Restart",
visible=VISIBLE
)
self.btn_start.click(
fn=self.start_button_when_click,
inputs=[self.text_theme, self.text_positive, self.text_negative, self.radio_choose, self.radio_mode, self.text_api],
outputs=[self.chatbot, self.btn_start]
).then(
fn=self.start_button_after_click,
inputs=[self.chatbot],
outputs=[self.chatbot, self.text_user, self.btn_send, self.btn_reset, self.btn_next]
)
self.btn_send.click(
fn=self.send_button_when_click,
inputs=[self.text_user, self.chatbot],
outputs=[self.text_user, self.btn_send, self.chatbot, self.btn_reset]
).then(
fn=self.start_button_after_click,
inputs=[self.chatbot],
outputs=[self.chatbot, self.text_user, self.btn_send, self.btn_reset, self.btn_next]
)
self.btn_reset.click(
fn=self.reset_button_when_click,
inputs=[
self.chatbot,
self.text_positive,
self.text_negative,
self.text_theme,
self.text_user,
self.btn_send,
self.btn_start,
self.btn_reset
],
outputs=[
self.chatbot,
self.text_positive,
self.text_negative,
self.text_theme,
self.text_user,
self.btn_send,
self.btn_start,
self.btn_reset,
self.btn_next
]
).then(
fn=self.reset_button_after_click,
inputs=[
self.chatbot,
self.text_positive,
self.text_negative,
self.text_theme,
self.text_user,
self.btn_send,
self.btn_start,
self.btn_reset
],
outputs=[
self.chatbot,
self.text_positive,
self.text_negative,
self.text_theme,
self.text_user,
self.btn_send,
self.btn_start,
self.btn_reset,
self.btn_next
]
)
self.btn_next.click(
fn=self.btn_next_when_click,
inputs=[],
outputs=[self.btn_next]
).then(
fn=self.start_button_after_click,
inputs=[self.chatbot],
outputs=[self.chatbot, self.text_user, self.btn_send, self.btn_reset, self.btn_next]
)
self.demo = demo
if __name__ == '__main__':
ui = DebateUI(client_cmd=["python","gradio_backend.py"])
ui.construct_ui()
ui.run()