Spaces:
Runtime error
Runtime error
File size: 6,380 Bytes
7a01de3 f7e23b0 bf33043 7a01de3 bf33043 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 |
import os
import asyncio
import copy
import inspect
import warnings
import json
import logging
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.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, FluxPipeline
import torch
import gradio as gr
# Corrected code with closed parenthesis and explicit token handling
image_model = FluxPipeline.from_pretrained(
"black-forest-labs/FLUX.1-dev",
torch_dtype=torch.bfloat16,
use_auth_token=os.getenv("HUGGINGFACE_TOKEN")
)
image_model.enable_model_cpu_offload()
# Define data models for Hugging Face
class FileDataDict(BaseModel):
path: str
url: Optional[str] = None
size: Optional[int] = None
orig_name: Optional[str] = None
mime_type: Optional[str] = None
is_stream: Optional[bool] = False
class Config:
arbitrary_types_allowed = True
class MessageDict(BaseModel):
content: Union[str, FileDataDict, tuple, Component]
role: Literal["user", "assistant", "system"]
metadata: Optional[dict] = None
options: Optional[List[dict]] = None
class Config:
arbitrary_types_allowed = True
class ChatMessage(GradioModel):
role: Literal["user", "assistant", "system"]
content: Union[str, FileData, Component]
metadata: dict = Field(default_factory=dict)
options: Optional[List[dict]] = None
class Config:
arbitrary_types_allowed = True
class ChatbotDataMessages(GradioRootModel):
root: List[ChatMessage]
# Universal Reasoning Aggregator
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
# Load models with explicit truncation
self.deepseek_model = pipeline(
"text-classification",
model="distilbert-base-uncased-finetuned-sst-2-english",
truncation=True
) # Updated model
self.davinci_model = pipeline(
"text2text-generation",
model="t5-small",
truncation=True
) # Replacing text-davinci with T5
self.additional_model = pipeline(
"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
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)
responses = [
f"Sentiment score: {sentiment_score}",
f"DeepSeek Response: {deepseek_response}",
f"T5 Response: {davinci_response}",
f"Additional Model Response: {additional_response}"
]
return "\n\n".join(responses)
def generate_image(self, prompt: str):
image = self.image_model(
prompt,
height=1024,
width=1024,
guidance_scale=3.5,
num_inference_steps=50,
max_sequence_length=512,
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
logging.info(f"Sentiment analysis result: {sentiment_score}")
return sentiment_score
# Main Multimodal Chatbot Component
class MultimodalChatbot(Component):
def __init__(
self,
value: Optional[List[MessageDict]] = None,
label: Optional[str] = None,
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
self.render = render
self.data_model = ChatbotDataMessages
self.universal_reasoning = UniversalReasoning({})
def preprocess(self, payload: Optional[ChatbotDataMessages]) -> List[MessageDict]:
# Handle None payload gracefully
if payload is None:
return []
return payload.root
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
class HuggingFaceChatbot:
def __init__(self):
# Initialize MultimodalChatbot with a default empty list
self.chatbot = MultimodalChatbot(value=[])
def setup_interface(self):
async def chatbot_logic(input_text: str) -> str:
return await self.chatbot.universal_reasoning.generate_response(input_text)
def image_logic(prompt: str):
return self.chatbot.universal_reasoning.generate_image(prompt)
interface = Interface(
fn=chatbot_logic,
inputs="text",
outputs="text",
title="Hugging Face Multimodal Chatbot",
)
image_interface = Interface(
fn=image_logic,
inputs="text",
outputs="image",
title="Image Generator",
)
return Blocks([interface, image_interface])
def launch(self):
interface = self.setup_interface()
interface.launch()
# If running as standalone
if __name__ == "__main__":
logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s')
chatbot = HuggingFaceChatbot()
chatbot.launch() |