omni_ui.py

# Import required libraries
from gradio import Interface, Textbox, HTML
import threading
import os
import glob
import base64
from langchain.llms import OpenAIChat
from swarms.agents import OmniModalAgent

# Function to convert image to base64


def image_to_base64(image_path):
    with open(image_path, "rb") as image_file:
        return base64.b64encode(image_file.read()).decode()

# Function to get the most recently created image in the directory


def get_latest_image():
    list_of_files = glob.glob('./*.png')  # Replace with your image file type
    if not list_of_files:
        return None
    latest_file = max(list_of_files, key=os.path.getctime)
    return latest_file


# Initialize your OmniModalAgent
# Replace with your actual initialization
llm = OpenAIChat(model_name="gpt-4", openai_api_key="OPENAI_API_KEY")
agent = OmniModalAgent(llm)  # Replace with your actual initialization

# Global variable to store chat history
chat_history = []

# Function to update chat


def update_chat(user_input):
    global chat_history
    chat_history.append({"type": "user", "content": user_input})

    # Get agent response
    agent_response = agent.run(user_input)

    # Handle the case where agent_response is not in the expected dictionary format
    if not isinstance(agent_response, dict):
        agent_response = {"type": "text", "content": str(agent_response)}

    chat_history.append(agent_response)

    # Check for the most recently created image and add it to the chat history
    latest_image = get_latest_image()
    if latest_image:
        chat_history.append({"type": "image", "content": latest_image})
    return render_chat(chat_history)

# Function to render chat as HTML


def render_chat(chat_history):
    chat_str = "<div style='max-height:400px;overflow-y:scroll;'>"
    for message in chat_history:
        if message['type'] == 'user':
            chat_str += f"<p><strong>User:</strong> {message['content']}</p>"
        elif message['type'] == 'text':
            chat_str += f"<p><strong>Agent:</strong> {message['content']}</p>"
        elif message['type'] == 'image':
            img_path = os.path.join(".", message['content'])
            base64_img = image_to_base64(img_path)
            chat_str += f"<p><strong>Agent:</strong> <img src='data:image/png;base64,{base64_img}' alt='image' width='200'/></p>"
    chat_str += "</div>"
    return chat_str


# Define Gradio interface
iface = Interface(
    fn=update_chat,
    inputs=Textbox(label="Your Message", type="text"),
    outputs=HTML(label="Chat History"),
    live=False
)

# Function to update the chat display


def update_display():
    global chat_history
    while True:
        iface.update(render_chat(chat_history))


# Run the update_display function in a separate thread
threading.Thread(target=update_display).start()

# Run Gradio interface
iface.launch()