app.py

import gradio as gr
from transformers import AutoModelForCausalLM, AutoTokenizer, pipeline
from diffusers import StableDiffusionPipeline
import torch
import os
import logging

# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Read the Hugging Face access token from the environment variable
read_token = os.getenv('AccToken')
if not read_token:
    raise ValueError("Hugging Face access token not found. Please set the AccToken environment variable.")
from huggingface_hub import login
login(read_token)

# Set device to GPU if available
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
logger.info(f"Device set to use {device}")

# Define a dictionary of conversational models
conversational_models = {
    "Qwen": "Qwen/QwQ-32B",
    "DeepSeek R1": "deepseek-ai/DeepSeek-R1",
    "Perplexity (R1 Post-trained)": "perplexity-ai/r1-1776",
    "Llama-Instruct by Meta": "meta-llama/Llama-3.2-3B-Instruct",
    "Mistral": "mistralai/Mistral-7B-v0.1",
    "Gemma": "google/gemma-2-2b-it",
}

# Define a dictionary of Text-to-Image models
text_to_image_models = {
    "Stable Diffusion 3.5 Large": "stabilityai/stable-diffusion-3.5-large",
    "Stable Diffusion 1.4": "CompVis/stable-diffusion-v1-4",
    "Flux Dev": "black-forest-labs/FLUX.1-dev",
}

# Define a dictionary of Text-to-Speech models
text_to_speech_models = {
    "Spark TTS": "SparkAudio/Spark-TTS-0.5B",
}

# Initialize tokenizers and models for conversational AI
conversational_tokenizers = {}
conversational_models_loaded = {}

# Initialize pipelines for Text-to-Image
text_to_image_pipelines = {}

# Initialize pipelines for Text-to-Speech
text_to_speech_pipelines = {}

# Initialize pipelines for other tasks
visual_qa_pipeline = pipeline("visual-question-answering", model="dandelin/vilt-b32-finetuned-vqa", device=device)
document_qa_pipeline = pipeline("question-answering", model="deepset/roberta-base-squad2", device=device)
image_classification_pipeline = pipeline("image-classification", model="facebook/deit-base-distilled-patch16-224", device=device)
object_detection_pipeline = pipeline("object-detection", model="facebook/detr-resnet-50", device=device)
video_classification_pipeline = pipeline("video-classification", model="facebook/timesformer-base-finetuned-k400", device=device)
summarization_pipeline = pipeline("summarization", model="facebook/bart-large-cnn", device=device)

# Load speaker embeddings for text-to-audio
def load_speaker_embeddings(model_name):
    if model_name == "microsoft/speecht5_tts":
        logger.info("Loading speaker embeddings for SpeechT5")
        from datasets import load_dataset
        dataset = load_dataset("Matthijs/cmu-arctic-xvectors", split="validation")
        speaker_embeddings = torch.tensor(dataset[7306]["xvector"]).unsqueeze(0).to(device)  # Example speaker
        return speaker_embeddings
    return None

# Use a different model for text-to-audio if stabilityai/stable-audio-open-1.0 is not supported
try:
    text_to_audio_pipeline = pipeline("text-to-audio", model="stabilityai/stable-audio-open-1.0", device=device)
except ValueError as e:
    logger.error(f"Error loading stabilityai/stable-audio-open-1.0: {e}")
    logger.info("Falling back to a different text-to-audio model.")
    text_to_audio_pipeline = pipeline("text-to-audio", model="microsoft/speecht5_tts", device=device)
    speaker_embeddings = load_speaker_embeddings("microsoft/speecht5_tts")

audio_classification_pipeline = pipeline("audio-classification", model="facebook/wav2vec2-base", device=device)

def load_conversational_model(model_name):
    if model_name not in conversational_models_loaded:
        logger.info(f"Loading conversational model: {model_name}")
        tokenizer = AutoTokenizer.from_pretrained(conversational_models[model_name], use_auth_token=read_token)
        model = AutoModelForCausalLM.from_pretrained(conversational_models[model_name], use_auth_token=read_token).to(device)
        conversational_tokenizers[model_name] = tokenizer
        conversational_models_loaded[model_name] = model
    return conversational_tokenizers[model_name], conversational_models_loaded[model_name]

def chat(model_name, user_input, history=[]):
    tokenizer, model = load_conversational_model(model_name)
    
    # Encode the input
    input_ids = tokenizer.encode(user_input + tokenizer.eos_token, return_tensors="pt").to(device)
    
    # Generate a response
    with torch.no_grad():
        output = model.generate(input_ids, max_length=150, pad_token_id=tokenizer.eos_token_id)
    
    response = tokenizer.decode(output[0], skip_special_tokens=True)
    
    # Clean up the response to remove the user input part
    response = response[len(user_input):].strip()
    
    # Append to chat history
    history.append((user_input, response))
    
    return history, history

def generate_image(model_name, prompt):
    if model_name not in text_to_image_pipelines:
        logger.info(f"Loading text-to-image model: {model_name}")
        text_to_image_pipelines[model_name] = StableDiffusionPipeline.from_pretrained(
            text_to_image_models[model_name], use_auth_token=read_token, torch_dtype=torch.float16, device_map="auto"
        )
    pipeline = text_to_image_pipelines[model_name]
    image = pipeline(prompt).images[0]
    return image

def generate_speech(model_name, text):
    if model_name not in text_to_speech_pipelines:
        logger.info(f"Loading text-to-speech model: {model_name}")
        text_to_speech_pipelines[model_name] = pipeline(
            "text-to-speech", model=text_to_speech_models[model_name], use_auth_token=read_token, device=device
        )
    pipeline = text_to_speech_pipelines[model_name]
    audio = pipeline(text, speaker_embeddings=speaker_embeddings)
    return audio["audio"]



def visual_qa(image, question):
    result = visual_qa_pipeline(image, question)
    return result["answer"]

def document_qa(document, question):
    result = document_qa_pipeline(question=question, context=document)
    return result["answer"]

def image_classification(image):
    result = image_classification_pipeline(image)
    return result

def object_detection(image):
    result = object_detection_pipeline(image)
    return result

def video_classification(video):
    result = video_classification_pipeline(video)
    return result

def summarize_text(text):
    result = summarization_pipeline(text)
    return result[0]["summary_text"]

def text_to_audio(text):
    global speaker_embeddings
    result = text_to_audio_pipeline(text, speaker_embeddings=speaker_embeddings)
    return result["audio"]

def audio_classification(audio):
    result = audio_classification_pipeline(audio)
    return result

# Define the Gradio interface
with gr.Blocks() as demo:
    gr.Markdown("## Versatile AI Chatbot and Text-to-X Tasks")
    
    with gr.Tab("Conversational AI"):
        conversational_model_choice = gr.Dropdown(list(conversational_models.keys()), label="Choose a Conversational Model")
        conversational_chatbot = gr.Chatbot(label="Chat")
        conversational_message = gr.Textbox(label="Message")
        conversational_submit = gr.Button("Submit")
        
        conversational_submit.click(chat, inputs=[conversational_model_choice, conversational_message, conversational_chatbot], outputs=[conversational_chatbot, conversational_chatbot])
        conversational_message.submit(chat, inputs=[conversational_model_choice, conversational_message, conversational_chatbot], outputs=[conversational_chatbot, conversational_chatbot])
    
    with gr.Tab("Text-to-Image"):
        text_to_image_model_choice = gr.Dropdown(list(text_to_image_models.keys()), label="Choose a Text-to-Image Model")
        text_to_image_prompt = gr.Textbox(label="Prompt")
        text_to_image_generate = gr.Button("Generate Image")
        text_to_image_output = gr.Image(label="Generated Image")
        
        text_to_image_generate.click(generate_image, inputs=[text_to_image_model_choice, text_to_image_prompt], outputs=text_to_image_output)
    
    with gr.Tab("Text-to-Speech"):
        text_to_speech_model_choice = gr.Dropdown(list(text_to_speech_models.keys()), label="Choose a Text-to-Speech Model")
        text_to_speech_text = gr.Textbox(label="Text")
        text_to_speech_generate = gr.Button("Generate Speech")
        text_to_speech_output = gr.Audio(label="Generated Speech")
        
        text_to_speech_generate.click(generate_speech, inputs=[text_to_speech_model_choice, text_to_speech_text], outputs=text_to_speech_output)
    
    with gr.Tab("Visual Question Answering"):
        visual_qa_image = gr.Image(label="Upload Image")
        visual_qa_question = gr.Textbox(label="Question")
        visual_qa_generate = gr.Button("Answer")
        visual_qa_output = gr.Textbox(label="Answer")
        
        visual_qa_generate.click(visual_qa, inputs=[visual_qa_image, visual_qa_question], outputs=visual_qa_output)
    
    with gr.Tab("Document Question Answering"):
        document_qa_document = gr.Textbox(label="Document Text")
        document_qa_question = gr.Textbox(label="Question")
        document_qa_generate = gr.Button("Answer")
        document_qa_output = gr.Textbox(label="Answer")
        
        document_qa_generate.click(document_qa, inputs=[document_qa_document, document_qa_question], outputs=document_qa_output)
    
    with gr.Tab("Image Classification"):
        image_classification_image = gr.Image(label="Upload Image")
        image_classification_generate = gr.Button("Classify")
        image_classification_output = gr.Textbox(label="Classification Result")
        
        image_classification_generate.click(image_classification, inputs=image_classification_image, outputs=image_classification_output)
    
    with gr.Tab("Object Detection"):
        object_detection_image = gr.Image(label="Upload Image")
        object_detection_generate = gr.Button("Detect")
        object_detection_output = gr.Image(label="Detection Result")
        
        object_detection_generate.click(object_detection, inputs=object_detection_image, outputs=object_detection_output)
    
    with gr.Tab("Video Classification"):
        video_classification_video = gr.Video(label="Upload Video")
        video_classification_generate = gr.Button("Classify")
        video_classification_output = gr.Textbox(label="Classification Result")
        
        video_classification_generate.click(video_classification, inputs=video_classification_video, outputs=video_classification_output)
    
    with gr.Tab("Summarization"):
        summarize_text_text = gr.Textbox(label="Text")
        summarize_text_generate = gr.Button("Summarize")
        summarize_text_output = gr.Textbox(label="Summary")
        
        summarize_text_generate.click(summarize_text, inputs=summarize_text_text, outputs=summarize_text_output)
    
    with gr.Tab("Text-to-Audio"):
        text_to_audio_text = gr.Textbox(label="Text")
        text_to_audio_generate = gr.Button("Generate Audio")
        text_to_audio_output = gr.Audio(label="Generated Audio")
        
        text_to_audio_generate.click(text_to_audio, inputs=text_to_audio_text, outputs=text_to_audio_output)

    with gr.Tab("Audio Classification"):
        audio_classification_audio = gr.Audio(label="Upload Audio")
        audio_classification_generate = gr.Button("Classify")
        audio_classification_output = gr.Textbox(label="Classification Result")
        
        audio_classification_generate.click(audio_classification, inputs=audio_classification_audio, outputs=audio_classification_output)

# Launch the Gradio interface
demo.launch()