import os import sys import gradio as gr sys.path.append("./ctm") from ctm.ctms.ctm_base import BaseConsciousnessTuringMachine ctm = BaseConsciousnessTuringMachine() ctm.add_supervisor("gpt4_supervisor") DEPLOYED = os.getenv("DEPLOYED", "true").lower() == "true" def convert_base64(image_array): image = Image.fromarray(image_array) buffer = io.BytesIO() image.save(buffer, format="PNG") byte_data = buffer.getvalue() base64_string = base64.b64encode(byte_data).decode("utf-8") return base64_string def introduction(): with gr.Column(scale=2): gr.Image("images/CTM-AI.png", elem_id="banner-image", show_label=False) with gr.Column(scale=5): gr.Markdown( """Consciousness Turing Machine Demo """ ) def add_processor(processor_name, display_name, state): print("add processor ", processor_name) ctm.add_processor(processor_name) print(ctm.processor_group_map) print(len(ctm.processor_list)) return display_name + " (added)" def processor_tab(): # Categorized model names text_processors = [ "gpt4_text_emotion_processor", "gpt4_text_summary_processor", "gpt4_speaker_intent_processor", "roberta_text_sentiment_processor", ] vision_processors = [ "gpt4v_cloth_fashion_processor", "gpt4v_face_emotion_processor", "gpt4v_ocr_processor", "gpt4v_posture_processor", "gpt4v_scene_location_processor", ] with gr.Blocks(): with gr.Row(): with gr.Column(scale=1): gr.Markdown("### Text Processors") for model_name in text_processors: display_name = ( model_name.replace("processor", "") .replace("_", " ") .title() ) button = gr.Button(display_name) processor_name = gr.Textbox( value=model_name, visible=False ) display_name = gr.Textbox( value=display_name, visible=False ) button.click( fn=add_processor, inputs=[processor_name, display_name, gr.State()], outputs=[button], ) with gr.Column(scale=1): gr.Markdown("### Vision Processors") for model_name in vision_processors: display_name = ( model_name.replace("processor", "") .replace("_", " ") .title() ) button = gr.Button(display_name) processor_name = gr.Textbox( value=model_name, visible=False ) display_name = gr.Textbox( value=display_name, visible=False ) button.click( fn=add_processor, inputs=[processor_name, display_name, gr.State()], outputs=[button], ) def forward(query, content, image, state): state["question"] = query ask_processors_output_info, state = ask_processors( query, content, image, state ) uptree_competition_output_info, state = uptree_competition(state) ask_supervisor_output_info, state = ask_supervisor(state) ctm.downtree_broadcast(state["winning_output"]) ctm.link_form(state["processor_output"]) return ( ask_processors_output_info, uptree_competition_output_info, ask_supervisor_output_info, state, ) def ask_processors(query, text, image, state): # Simulate processing here processor_output = ctm.ask_processors( query=query, text=text, image=image, ) output_info = "" for name, info in processor_output.items(): output_info += f"{name}: {info['gist']}\n" state["processor_output"] = processor_output return output_info, state def uptree_competition(state): winning_output = ctm.uptree_competition(state["processor_output"]) state["winning_output"] = winning_output output_info = ( "The winning processor is: {}\nThe winning gist is: {}\n".format( winning_output["name"], winning_output["gist"] ) ) return output_info, state def ask_supervisor(state): question = state["question"] winning_output = state["winning_output"] answer, score = ctm.ask_supervisor(question, winning_output) output_info = f'The answer to the query "{question}" is: {answer}\nThe confidence for answering is: {score}\n' state["answer"] = answer state["score"] = score return output_info, state def interface_tab(): with gr.Blocks(): state = gr.State({}) # State to hold and pass values with gr.Column(): # Inputs text = gr.Textbox(label="Enter your text here") query = gr.Textbox(label="Enter your query here") image = gr.Image(label="Upload your image") # audio = gr.Audio(label="Upload or Record Audio") # video = gr.Video(label="Upload or Record Video") # Processing buttons forward_button = gr.Button("Start CTM forward process") # Outputs processors_output = gr.Textbox( label="Processors Output", visible=True ) competition_output = gr.Textbox( label="Up-tree Competition Output", visible=True ) supervisor_output = gr.Textbox( label="Supervisor Output", visible=True ) # Set up button to start or continue processing forward_button.click( fn=forward, inputs=[query, text, image, state], outputs=[ processors_output, competition_output, supervisor_output, state, ], ) return interface_tab def main(): with gr.Blocks( css="""#chat_container {height: 820px; width: 1000px; margin-left: auto; margin-right: auto;} #chatbot {height: 600px; overflow: auto;} #create_container {height: 750px; margin-left: 0px; margin-right: 0px;} #tokenizer_renderer span {white-space: pre-wrap} """ ) as demo: with gr.Row(): introduction() with gr.Row(): processor_tab() with gr.Row(): interface_tab() return demo def start_demo(): demo = main() if DEPLOYED: demo.queue(api_open=False).launch(show_api=False) else: demo.queue() demo.launch(share=False, server_name="0.0.0.0") if __name__ == "__main__": start_demo()