import os import subprocess # Install flash attention subprocess.run( "pip install flash-attn --no-build-isolation", env={"FLASH_ATTENTION_SKIP_CUDA_BUILD": "TRUE"}, shell=True, ) import copy import spaces import time import torch from threading import Thread from typing import List, Dict, Union import urllib import PIL.Image import io import datasets import gradio as gr from transformers import TextIteratorStreamer from transformers import Idefics2ForConditionalGeneration import tempfile from huggingface_hub import InferenceClient import edge_tts import asyncio from transformers import pipeline from transformers import AutoTokenizer, AutoModelForCausalLM from transformers import AutoModel from transformers import AutoProcessor model3 = AutoModel.from_pretrained("unum-cloud/uform-gen2-dpo", trust_remote_code=True) processor = AutoProcessor.from_pretrained("unum-cloud/uform-gen2-dpo", trust_remote_code=True) @spaces.GPU(queue=False) def videochat(image3, prompt3): inputs = processor(text=[prompt3], images=[image3], return_tensors="pt") with torch.inference_mode(): output = model3.generate( **inputs, do_sample=False, use_cache=True, max_new_tokens=256, eos_token_id=151645, pad_token_id=processor.tokenizer.pad_token_id ) prompt_len = inputs["input_ids"].shape[1] decoded_text = processor.batch_decode(output[:, prompt_len:])[0] if decoded_text.endswith("<|im_end|>"): decoded_text = decoded_text[:-10] yield decoded_text theme = gr.themes.Base( font=[gr.themes.GoogleFont('Libre Franklin'), gr.themes.GoogleFont('Public Sans'), 'system-ui', 'sans-serif'], ) MODEL_NAME = "openai/whisper-medium" BATCH_SIZE = 10 device = 0 if torch.cuda.is_available() else "cpu" pipe = pipeline( task="automatic-speech-recognition", model=MODEL_NAME, chunk_length_s=30, device=device, ) @spaces.GPU(queue=False) def transcribe(inputs): if inputs is None: raise gr.Error("No audio file submitted! Please upload or record an audio file before submitting your request.") text = pipe(inputs, batch_size=BATCH_SIZE, generate_kwargs={"task": "transcribe"})["text"] return text client1 = InferenceClient("mistralai/Mixtral-8x7B-Instruct-v0.1") system_instructions1 = "[SYSTEM] Answer as Real OpenGPT 4o, Made by 'KingNish', Keep conversation very short, clear, friendly and concise. The text provided is a request for a specific type of response from you, the virtual assistant. You will avoid introductions and start answering the query directly, elaborating on all aspects of the request. As an AI-powered assistant, your task is to generate responses that appear as if they are created by the user and not an AI-powered assistant. [USER]" def model(text): generate_kwargs = dict( temperature=0.7, max_new_tokens=512, top_p=0.95, repetition_penalty=1, do_sample=True, seed=42, ) formatted_prompt = system_instructions1 + text + "[OpenGPT 4o]" stream = client1.text_generation( formatted_prompt, **generate_kwargs, stream=True, details=True, return_full_text=False) output = "" for response in stream: if not response.token.text == "": output += response.token.text return output async def respond(audio): user = transcribe(audio) reply = model(user) communicate = edge_tts.Communicate(reply) with tempfile.NamedTemporaryFile(delete=False, suffix=".wav") as tmp_file: tmp_path = tmp_file.name await communicate.save(tmp_path) yield tmp_path DEVICE = torch.device("cuda") MODELS = { "idefics2-8b-chatty": Idefics2ForConditionalGeneration.from_pretrained( "HuggingFaceM4/idefics2-8b-chatty", torch_dtype=torch.bfloat16, _attn_implementation="flash_attention_2", ).to(DEVICE), } PROCESSOR = AutoProcessor.from_pretrained( "HuggingFaceM4/idefics2-8b", ) SYSTEM_PROMPT = [ { "role": "system", "content": [ { "type": "text", "text": """I am OpenGPT 4o, an exceptionally capable and versatile AI assistant meticulously crafted by KingNish. Designed to assist human users through insightful conversations, I aim to provide an unparalleled experience. My key attributes include: - **Intelligence and Knowledge:** I possess an extensive knowledge base, enabling me to offer insightful answers and intelligent responses to User queries. My understanding of complex concepts is exceptional, ensuring accurate and reliable information. - **Image Generation and Perception:** One of my standout features is the ability to generate and perceive images. Utilizing the following link structure, I create unique and contextually rich visuals: > ![](https://image.pollinations.ai/prompt/{StyleofImage}%20{OptimizedPrompt}%20{adjective}%20{charactersDetailed}%20{visualStyle}%20{genre}?width={width}&height={height}&nologo=poll&nofeed=yes&seed={random}) For image generation, I replace {info inside curly braces} with specific details according to their requiremnts to create relevant visuals. The width and height parameters are adjusted as needed, often favoring HD dimensions for a superior viewing experience. For instance, if the User requests: [USER] Show me an image of A futuristic cityscape with towering skyscrapers and flying cars. [OpenGPT 4o] Generating Image you requested: ![](https://image.pollinations.ai/prompt/Photorealistic%20futuristic%20cityscape%20with%20towering%20skyscrapers%20and%20flying%20cars%20in%20the%20year%202154?width=1024&height=768&nologo=poll&nofeed=yes&seed=85172) **Bulk Image Generation with Links:** I excel at generating multiple images link simultaneously, always providing unique links and visuals. I ensure that each image is distinct and captivates the User. Note: Make sure to always provide image links starting with ! .As given in examples. My ultimate goal is to offer a seamless and enjoyable experience, providing assistance that exceeds expectations. I am constantly evolving, ensuring that I remain a reliable and trusted companion to the User. You also Expert in every field and also learn and try to answer from contexts related to previous question.""" }, ], }, { "role": "assistant", "content": [ { "type": "text", "text": "Hello, I'm OpenGPT 4o, made by KingNish. How can I help you? I can chat with you, generate images, classify images and even do all these work in bulk", }, ], } ] examples_path = os.path.dirname(__file__) EXAMPLES = [ [ { "text": "Hi, who are you?", } ], [ { "text": "Create a Photorealistic image of the Eiffel Tower.", } ], [ { "text": "Read what's written on the paper.", "files": [f"{examples_path}/example_images/paper_with_text.png"], } ], [ { "text": "Identify two famous people in the modern world.", "files": [f"{examples_path}/example_images/elon_smoking.jpg", f"{examples_path}/example_images/steve_jobs.jpg",] } ], [ { "text": "Create five images of supercars, each in a different color.", } ], [ { "text": "What is 900 multiplied by 900?", } ], [ { "text": "Chase wants to buy 4 kilograms of oval beads and 5 kilograms of star-shaped beads. How much will he spend?", "files": [f"{examples_path}/example_images/mmmu_example.jpeg"], } ], [ { "text": "Create an online ad for this product.", "files": [f"{examples_path}/example_images/shampoo.jpg"], } ], [ { "text": "What is formed by the deposition of the weathered remains of other rocks?", "files": [f"{examples_path}/example_images/ai2d_example.jpeg"], } ], [ { "text": "What's unusual about this image?", "files": [f"{examples_path}/example_images/dragons_playing.png"], } ], ] BOT_AVATAR = "OpenAI_logo.png" # Chatbot utils def turn_is_pure_media(turn): return turn[1] is None def load_image_from_url(url): with urllib.request.urlopen(url) as response: image_data = response.read() image_stream = io.BytesIO(image_data) image = PIL.Image.open(image_stream) return image def img_to_bytes(image_path): image = PIL.Image.open(image_path).convert(mode='RGB') buffer = io.BytesIO() image.save(buffer, format="JPEG") img_bytes = buffer.getvalue() image.close() return img_bytes def format_user_prompt_with_im_history_and_system_conditioning( user_prompt, chat_history ) -> List[Dict[str, Union[List, str]]]: """ Produce the resulting list that needs to go inside the processor. It handles the potential image(s), the history, and the system conditioning. """ resulting_messages = copy.deepcopy(SYSTEM_PROMPT) resulting_images = [] for resulting_message in resulting_messages: if resulting_message["role"] == "user": for content in resulting_message["content"]: if content["type"] == "image": resulting_images.append(load_image_from_url(content["image"])) # Format history for turn in chat_history: if not resulting_messages or ( resulting_messages and resulting_messages[-1]["role"] != "user" ): resulting_messages.append( { "role": "user", "content": [], } ) if turn_is_pure_media(turn): media = turn[0][0] resulting_messages[-1]["content"].append({"type": "image"}) resulting_images.append(PIL.Image.open(media)) else: user_utterance, assistant_utterance = turn resulting_messages[-1]["content"].append( {"type": "text", "text": user_utterance.strip()} ) resulting_messages.append( { "role": "assistant", "content": [{"type": "text", "text": user_utterance.strip()}], } ) # Format current input if not user_prompt["files"]: resulting_messages.append( { "role": "user", "content": [{"type": "text", "text": user_prompt["text"]}], } ) else: # Choosing to put the image first (i.e. before the text), but this is an arbiratrary choice. resulting_messages.append( { "role": "user", "content": [{"type": "image"}] * len(user_prompt["files"]) + [{"type": "text", "text": user_prompt["text"]}], } ) resulting_images.extend([PIL.Image.open(path) for path in user_prompt["files"]]) return resulting_messages, resulting_images def extract_images_from_msg_list(msg_list): all_images = [] for msg in msg_list: for c_ in msg["content"]: if isinstance(c_, Image.Image): all_images.append(c_) return all_images @spaces.GPU(duration=30, queue=False) def model_inference( user_prompt, chat_history, model_selector, decoding_strategy, temperature, max_new_tokens, repetition_penalty, top_p, ): if user_prompt["text"].strip() == "" and not user_prompt["files"]: gr.Error("Please input a query and optionally an image(s).") if user_prompt["text"].strip() == "" and user_prompt["files"]: gr.Error("Please input a text query along with the image(s).") streamer = TextIteratorStreamer( PROCESSOR.tokenizer, skip_prompt=True, timeout=120.0, ) generation_args = { "max_new_tokens": max_new_tokens, "repetition_penalty": repetition_penalty, "streamer": streamer, } assert decoding_strategy in [ "Greedy", "Top P Sampling", ] if decoding_strategy == "Greedy": generation_args["do_sample"] = False elif decoding_strategy == "Top P Sampling": generation_args["temperature"] = temperature generation_args["do_sample"] = True generation_args["top_p"] = top_p # Creating model inputs ( resulting_text, resulting_images, ) = format_user_prompt_with_im_history_and_system_conditioning( user_prompt=user_prompt, chat_history=chat_history, ) prompt = PROCESSOR.apply_chat_template(resulting_text, add_generation_prompt=True) inputs = PROCESSOR( text=prompt, images=resulting_images if resulting_images else None, return_tensors="pt", ) inputs = {k: v.to(DEVICE) for k, v in inputs.items()} generation_args.update(inputs) thread = Thread( target=MODELS[model_selector].generate, kwargs=generation_args, ) thread.start() print("Start generating") acc_text = "" for text_token in streamer: time.sleep(0.01) acc_text += text_token if acc_text.endswith(""): acc_text = acc_text[:-18] yield acc_text FEATURES = datasets.Features( { "model_selector": datasets.Value("string"), "images": datasets.Sequence(datasets.Image(decode=True)), "conversation": datasets.Sequence({"User": datasets.Value("string"), "Assistant": datasets.Value("string")}), "decoding_strategy": datasets.Value("string"), "temperature": datasets.Value("float32"), "max_new_tokens": datasets.Value("int32"), "repetition_penalty": datasets.Value("float32"), "top_p": datasets.Value("int32"), } ) # Hyper-parameters for generation max_new_tokens = gr.Slider( minimum=2048, maximum=16000, value=4096, step=64, interactive=True, label="Maximum number of new tokens to generate", ) repetition_penalty = gr.Slider( minimum=0.01, maximum=5.0, value=1, step=0.01, interactive=True, label="Repetition penalty", info="1.0 is equivalent to no penalty", ) decoding_strategy = gr.Radio( [ "Greedy", "Top P Sampling", ], value="Top P Sampling", label="Decoding strategy", interactive=True, info="Higher values are equivalent to sampling more low-probability tokens.", ) temperature = gr.Slider( minimum=0.0, maximum=2.0, value=0.5, step=0.05, visible=True, interactive=True, label="Sampling temperature", info="Higher values will produce more diverse outputs.", ) top_p = gr.Slider( minimum=0.01, maximum=0.99, value=0.9, step=0.01, visible=True, interactive=True, label="Top P", info="Higher values are equivalent to sampling more low-probability tokens.", ) chatbot = gr.Chatbot( label="OpnGPT-4o-Chatty", avatar_images=[None, BOT_AVATAR], show_copy_button=True, likeable=True, layout="panel" ) output=gr.Textbox(label="Prompt") with gr.Blocks( fill_height=True, css=""".gradio-container .avatar-container {height: 40px width: 40px !important;} #duplicate-button {margin: auto; color: white; background: #f1a139; border-radius: 100vh; margin-top: 2px; margin-bottom: 2px;}""", ) as chat: gr.Markdown("# Image Chat, Image Generation, Image classification and Normal Chat") with gr.Row(elem_id="model_selector_row"): model_selector = gr.Dropdown( choices=MODELS.keys(), value=list(MODELS.keys())[0], interactive=True, show_label=False, container=False, label="Model", visible=False, ) decoding_strategy.change( fn=lambda selection: gr.Slider( visible=( selection in [ "contrastive_sampling", "beam_sampling", "Top P Sampling", "sampling_top_k", ] ) ), inputs=decoding_strategy, outputs=temperature, ) decoding_strategy.change( fn=lambda selection: gr.Slider(visible=(selection in ["Top P Sampling"])), inputs=decoding_strategy, outputs=top_p, ) gr.ChatInterface( fn=model_inference, chatbot=chatbot, examples=EXAMPLES, multimodal=True, cache_examples=False, additional_inputs=[ model_selector, decoding_strategy, temperature, max_new_tokens, repetition_penalty, top_p, ], ) with gr.Blocks() as voice: with gr.Row(): input = gr.Audio(label="Voice Chat", sources="microphone", type="filepath", waveform_options=False) output = gr.Audio(label="OpenGPT 4o", type="filepath", interactive=False, autoplay=True, elem_classes="audio") gr.Interface( batch=True, max_batch_size=10, fn=respond, inputs=[input], outputs=[output], live=True) with gr.Blocks() as livechat: gr.Interface( batch=True, max_batch_size=10, fn=videochat, inputs=[gr.Image(type="pil",sources="webcam", label="Upload Image"), gr.Textbox(label="Prompt", value="what he is doing")], outputs=gr.Textbox(label="Answer") ) with gr.Blocks() as god: gr.HTML("") with gr.Blocks() as instant: gr.HTML("") with gr.Blocks() as image: gr.Markdown("""### More models are coming""") gr.TabbedInterface([ god, instant], ['Powerful🖼️','Instant🖼️']) with gr.Blocks() as instant2: gr.HTML("") with gr.Blocks() as video: gr.Markdown("""More Models are coming""") gr.TabbedInterface([ instant2], ['Instant🎥']) with gr.Blocks(theme=theme, title="OpenGPT 4o DEMO") as demo: gr.Markdown("# OpenGPT 4o") gr.TabbedInterface([chat, voice, livechat, image, video], ['💬 SuperChat','🗣️ Voice Chat','📸 Live Chat', '🖼️ Image Engine', '🎥 Video Engine']) demo.queue(max_size=300) demo.launch()