Update app.py

app.py

@@ -1,26 +1,17 @@
 import os
 import asyncio
-import copy
-import inspect
-import warnings
-import json
 import logging
+from typing import Optional, List, Union, Literal
 from pathlib import Path
-from typing import Any, Literal, Optional, Union, List
-from cryptography.fernet import Fernet
 from pydantic import BaseModel, Field
-from gradio import Interface, Blocks
-from gradio.components import Component
+from gradio import Interface, Blocks, Component
 from gradio.data_classes import FileData, GradioModel, GradioRootModel
-from gradio.events import Events
-from gradio.exceptions import Error
-from gradio_client import utils as client_utils
 from transformers import pipeline
 from diffusers import DiffusionPipeline
 import torch
 import gradio as gr
 
-# Corrected code with closed parenthesis and explicit token handling
+# Load gated image model
 image_model = DiffusionPipeline.from_pretrained(
     "black-forest-labs/FLUX.1-dev",
     torch_dtype=torch.bfloat16 if torch.cuda.is_available() else torch.float32,
@@ -28,7 +19,7 @@ image_model = DiffusionPipeline.from_pretrained(
 )
 image_model.enable_model_cpu_offload()
 
-# Define data models for Hugging Face
+# Define data models
 class FileDataDict(BaseModel):
     path: str
     url: Optional[str] = None
@@ -36,7 +27,6 @@ class FileDataDict(BaseModel):
     orig_name: Optional[str] = None
     mime_type: Optional[str] = None
     is_stream: Optional[bool] = False
-
     class Config:
         arbitrary_types_allowed = True
 
@@ -45,7 +35,6 @@ class MessageDict(BaseModel):
     role: Literal["user", "assistant", "system"]
     metadata: Optional[dict] = None
     options: Optional[List[dict]] = None
-
     class Config:
         arbitrary_types_allowed = True
 
@@ -54,7 +43,6 @@ class ChatMessage(GradioModel):
     content: Union[str, FileData, Component]
     metadata: dict = Field(default_factory=dict)
     options: Optional[List[dict]] = None
-
     class Config:
         arbitrary_types_allowed = True
 
@@ -65,38 +53,35 @@ class ChatbotDataMessages(GradioRootModel):
 class UniversalReasoning:
     def __init__(self, config):
         self.config = config
-        self.sentiment_analyzer = pipeline("sentiment-analysis")  # Hugging Face sentiment analysis
-        self.context_history = []  # Maintain context history
+        self.context_history = []
+        self.sentiment_analyzer = pipeline("sentiment-analysis")
 
-        # Load models with explicit truncation
         self.deepseek_model = pipeline(
-            "text-classification", 
+            "text-classification",
             model="distilbert-base-uncased-finetuned-sst-2-english",
             truncation=True
-        )  # Updated model
+        )
 
         self.davinci_model = pipeline(
-            "text2text-generation", 
+            "text2text-generation",
             model="t5-small",
             truncation=True
-        )  # Replacing text-davinci with T5
+        )
 
         self.additional_model = pipeline(
-            "text-generation", 
+            "text-generation",
             model="EleutherAI/gpt-neo-125M",
             truncation=True
-        )  # Example GPT-Neo model
+        )
 
-        # Use earlier-defined image model
         self.image_model = image_model
 
     async def generate_response(self, question: str) -> str:
-        self.context_history.append(question)  # Add question to context history
+        self.context_history.append(question)
         sentiment_score = self.analyze_sentiment(question)
-
         deepseek_response = self.deepseek_model(question)
-        davinci_response = self.davinci_model(question, max_length=50, truncation=True)
-        additional_response = self.additional_model(question, max_length=100, truncation=True)
+        davinci_response = self.davinci_model(question, max_length=50)
+        additional_response = self.additional_model(question, max_length=100)
 
         responses = [
             f"Sentiment score: {sentiment_score}",
@@ -104,7 +89,6 @@ class UniversalReasoning:
             f"T5 Response: {davinci_response}",
             f"Additional Model Response: {additional_response}"
         ]
-
         return "\n\n".join(responses)
 
     def generate_image(self, prompt: str):
@@ -115,17 +99,17 @@ class UniversalReasoning:
             guidance_scale=3.5,
             num_inference_steps=50,
             max_sequence_length=512,
-            generator=torch.Generator("cpu").manual_seed(0)
+            generator=torch.Generator('cpu').manual_seed(0)
         ).images[0]
         image.save("flux-dev.png")
         return image
 
     def analyze_sentiment(self, text: str) -> list:
-        sentiment_score = self.sentiment_analyzer(text)  # Returns a list of dictionaries
+        sentiment_score = self.sentiment_analyzer(text)
         logging.info(f"Sentiment analysis result: {sentiment_score}")
         return sentiment_score
 
-# Main Multimodal Chatbot Component
+# Main Component
 class MultimodalChatbot(Component):
     def __init__(
         self,
@@ -134,7 +118,6 @@ class MultimodalChatbot(Component):
         render: bool = True,
         log_file: Optional[Path] = None,
     ):
-        # Ensure value is initialized as an empty list if None
         value = value or []
         super().__init__(label=label, value=value)
         self.log_file = log_file
@@ -143,20 +126,15 @@ class MultimodalChatbot(Component):
         self.universal_reasoning = UniversalReasoning({})
 
     def preprocess(self, payload: Optional[ChatbotDataMessages]) -> List[MessageDict]:
-        # Handle None payload gracefully
-        if payload is None:
-            return []
-        return payload.root
+        return payload.root if payload else []
 
     def postprocess(self, messages: Optional[List[MessageDict]]) -> ChatbotDataMessages:
-        # Ensure messages is a valid list
         messages = messages or []
         return ChatbotDataMessages(root=messages)
 
-# Hugging Face Integration Class
+# Gradio Interface
 class HuggingFaceChatbot:
     def __init__(self):
-        # Initialize MultimodalChatbot with a default empty list
         self.chatbot = MultimodalChatbot(value=[])
 
     def setup_interface(self):
@@ -186,7 +164,7 @@ class HuggingFaceChatbot:
         interface = self.setup_interface()
         interface.launch()
 
-# If running as standalone
+# Standalone launch
 if __name__ == "__main__":
     logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s')
     chatbot = HuggingFaceChatbot()