Update app.py
Browse files
app.py
CHANGED
@@ -4,22 +4,27 @@ import logging
|
|
4 |
from typing import Optional, List, Union, Literal
|
5 |
from pathlib import Path
|
6 |
from pydantic import BaseModel, Field
|
7 |
-
from gradio import Interface, Blocks
|
|
|
8 |
from gradio.data_classes import FileData, GradioModel, GradioRootModel
|
9 |
from transformers import pipeline
|
10 |
from diffusers import DiffusionPipeline
|
11 |
import torch
|
12 |
import gradio as gr
|
13 |
|
14 |
-
# Load gated image model
|
|
|
|
|
|
|
|
|
15 |
image_model = DiffusionPipeline.from_pretrained(
|
16 |
"black-forest-labs/FLUX.1-dev",
|
17 |
torch_dtype=torch.bfloat16 if torch.cuda.is_available() else torch.float32,
|
18 |
-
use_auth_token=
|
19 |
)
|
20 |
image_model.enable_model_cpu_offload()
|
21 |
|
22 |
-
#
|
23 |
class FileDataDict(BaseModel):
|
24 |
path: str
|
25 |
url: Optional[str] = None
|
@@ -31,7 +36,7 @@ class FileDataDict(BaseModel):
|
|
31 |
arbitrary_types_allowed = True
|
32 |
|
33 |
class MessageDict(BaseModel):
|
34 |
-
content: Union[str, FileDataDict, tuple,
|
35 |
role: Literal["user", "assistant", "system"]
|
36 |
metadata: Optional[dict] = None
|
37 |
options: Optional[List[dict]] = None
|
@@ -40,7 +45,7 @@ class MessageDict(BaseModel):
|
|
40 |
|
41 |
class ChatMessage(GradioModel):
|
42 |
role: Literal["user", "assistant", "system"]
|
43 |
-
content: Union[str, FileData,
|
44 |
metadata: dict = Field(default_factory=dict)
|
45 |
options: Optional[List[dict]] = None
|
46 |
class Config:
|
@@ -49,31 +54,15 @@ class ChatMessage(GradioModel):
|
|
49 |
class ChatbotDataMessages(GradioRootModel):
|
50 |
root: List[ChatMessage]
|
51 |
|
52 |
-
#
|
53 |
class UniversalReasoning:
|
54 |
def __init__(self, config):
|
55 |
self.config = config
|
56 |
self.context_history = []
|
57 |
self.sentiment_analyzer = pipeline("sentiment-analysis")
|
58 |
-
|
59 |
-
self.
|
60 |
-
|
61 |
-
model="distilbert-base-uncased-finetuned-sst-2-english",
|
62 |
-
truncation=True
|
63 |
-
)
|
64 |
-
|
65 |
-
self.davinci_model = pipeline(
|
66 |
-
"text2text-generation",
|
67 |
-
model="t5-small",
|
68 |
-
truncation=True
|
69 |
-
)
|
70 |
-
|
71 |
-
self.additional_model = pipeline(
|
72 |
-
"text-generation",
|
73 |
-
model="EleutherAI/gpt-neo-125M",
|
74 |
-
truncation=True
|
75 |
-
)
|
76 |
-
|
77 |
self.image_model = image_model
|
78 |
|
79 |
async def generate_response(self, question: str) -> str:
|
@@ -87,7 +76,7 @@ class UniversalReasoning:
|
|
87 |
f"Sentiment score: {sentiment_score}",
|
88 |
f"DeepSeek Response: {deepseek_response}",
|
89 |
f"T5 Response: {davinci_response}",
|
90 |
-
f"
|
91 |
]
|
92 |
return "\n\n".join(responses)
|
93 |
|
@@ -98,7 +87,6 @@ class UniversalReasoning:
|
|
98 |
width=1024,
|
99 |
guidance_scale=3.5,
|
100 |
num_inference_steps=50,
|
101 |
-
max_sequence_length=512,
|
102 |
generator=torch.Generator('cpu').manual_seed(0)
|
103 |
).images[0]
|
104 |
image.save("flux-dev.png")
|
@@ -109,63 +97,39 @@ class UniversalReasoning:
|
|
109 |
logging.info(f"Sentiment analysis result: {sentiment_score}")
|
110 |
return sentiment_score
|
111 |
|
112 |
-
# Main
|
113 |
-
class MultimodalChatbot(Component):
|
114 |
-
def __init__(
|
115 |
-
self,
|
116 |
-
value: Optional[List[MessageDict]] = None,
|
117 |
-
label: Optional[str] = None,
|
118 |
-
render: bool = True,
|
119 |
-
log_file: Optional[Path] = None,
|
120 |
-
):
|
121 |
-
value = value or []
|
122 |
-
super().__init__(label=label, value=value)
|
123 |
-
self.log_file = log_file
|
124 |
-
self.render = render
|
125 |
-
self.data_model = ChatbotDataMessages
|
126 |
-
self.universal_reasoning = UniversalReasoning({})
|
127 |
-
|
128 |
-
def preprocess(self, payload: Optional[ChatbotDataMessages]) -> List[MessageDict]:
|
129 |
-
return payload.root if payload else []
|
130 |
-
|
131 |
-
def postprocess(self, messages: Optional[List[MessageDict]]) -> ChatbotDataMessages:
|
132 |
-
messages = messages or []
|
133 |
-
return ChatbotDataMessages(root=messages)
|
134 |
-
|
135 |
-
# Gradio Interface
|
136 |
class HuggingFaceChatbot:
|
137 |
def __init__(self):
|
138 |
-
self.
|
139 |
|
140 |
def setup_interface(self):
|
141 |
async def chatbot_logic(input_text: str) -> str:
|
142 |
-
return await self.
|
143 |
|
144 |
def image_logic(prompt: str):
|
145 |
-
return self.
|
146 |
|
147 |
-
|
148 |
fn=chatbot_logic,
|
149 |
-
inputs="
|
150 |
-
outputs="
|
151 |
-
title="
|
152 |
)
|
153 |
|
154 |
image_interface = Interface(
|
155 |
fn=image_logic,
|
156 |
-
inputs="
|
157 |
-
outputs="
|
158 |
-
title="Image Generator"
|
159 |
)
|
160 |
|
161 |
-
return Blocks([
|
162 |
|
163 |
def launch(self):
|
164 |
-
|
165 |
-
|
166 |
|
167 |
-
#
|
168 |
if __name__ == "__main__":
|
169 |
-
logging.basicConfig(level=logging.
|
170 |
-
|
171 |
-
chatbot.launch()
|
|
|
4 |
from typing import Optional, List, Union, Literal
|
5 |
from pathlib import Path
|
6 |
from pydantic import BaseModel, Field
|
7 |
+
from gradio import Interface, Blocks
|
8 |
+
from gradio.components import Textbox, Image
|
9 |
from gradio.data_classes import FileData, GradioModel, GradioRootModel
|
10 |
from transformers import pipeline
|
11 |
from diffusers import DiffusionPipeline
|
12 |
import torch
|
13 |
import gradio as gr
|
14 |
|
15 |
+
# Load gated image model securely
|
16 |
+
hf_token = os.getenv("HUGGINGFACE_TOKEN")
|
17 |
+
if not hf_token:
|
18 |
+
raise RuntimeError("Missing HUGGINGFACE_TOKEN env var for gated model access.")
|
19 |
+
|
20 |
image_model = DiffusionPipeline.from_pretrained(
|
21 |
"black-forest-labs/FLUX.1-dev",
|
22 |
torch_dtype=torch.bfloat16 if torch.cuda.is_available() else torch.float32,
|
23 |
+
use_auth_token=hf_token
|
24 |
)
|
25 |
image_model.enable_model_cpu_offload()
|
26 |
|
27 |
+
# Data models
|
28 |
class FileDataDict(BaseModel):
|
29 |
path: str
|
30 |
url: Optional[str] = None
|
|
|
36 |
arbitrary_types_allowed = True
|
37 |
|
38 |
class MessageDict(BaseModel):
|
39 |
+
content: Union[str, FileDataDict, tuple, str]
|
40 |
role: Literal["user", "assistant", "system"]
|
41 |
metadata: Optional[dict] = None
|
42 |
options: Optional[List[dict]] = None
|
|
|
45 |
|
46 |
class ChatMessage(GradioModel):
|
47 |
role: Literal["user", "assistant", "system"]
|
48 |
+
content: Union[str, FileData, str]
|
49 |
metadata: dict = Field(default_factory=dict)
|
50 |
options: Optional[List[dict]] = None
|
51 |
class Config:
|
|
|
54 |
class ChatbotDataMessages(GradioRootModel):
|
55 |
root: List[ChatMessage]
|
56 |
|
57 |
+
# Reasoning Engine
|
58 |
class UniversalReasoning:
|
59 |
def __init__(self, config):
|
60 |
self.config = config
|
61 |
self.context_history = []
|
62 |
self.sentiment_analyzer = pipeline("sentiment-analysis")
|
63 |
+
self.deepseek_model = pipeline("text-classification", model="distilbert-base-uncased-finetuned-sst-2-english")
|
64 |
+
self.davinci_model = pipeline("text2text-generation", model="t5-small")
|
65 |
+
self.additional_model = pipeline("text-generation", model="EleutherAI/gpt-neo-125M")
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
66 |
self.image_model = image_model
|
67 |
|
68 |
async def generate_response(self, question: str) -> str:
|
|
|
76 |
f"Sentiment score: {sentiment_score}",
|
77 |
f"DeepSeek Response: {deepseek_response}",
|
78 |
f"T5 Response: {davinci_response}",
|
79 |
+
f"GPT-Neo Response: {additional_response}"
|
80 |
]
|
81 |
return "\n\n".join(responses)
|
82 |
|
|
|
87 |
width=1024,
|
88 |
guidance_scale=3.5,
|
89 |
num_inference_steps=50,
|
|
|
90 |
generator=torch.Generator('cpu').manual_seed(0)
|
91 |
).images[0]
|
92 |
image.save("flux-dev.png")
|
|
|
97 |
logging.info(f"Sentiment analysis result: {sentiment_score}")
|
98 |
return sentiment_score
|
99 |
|
100 |
+
# Main Gradio App
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
101 |
class HuggingFaceChatbot:
|
102 |
def __init__(self):
|
103 |
+
self.universal_reasoning = UniversalReasoning(config={})
|
104 |
|
105 |
def setup_interface(self):
|
106 |
async def chatbot_logic(input_text: str) -> str:
|
107 |
+
return await self.universal_reasoning.generate_response(input_text)
|
108 |
|
109 |
def image_logic(prompt: str):
|
110 |
+
return self.universal_reasoning.generate_image(prompt)
|
111 |
|
112 |
+
text_interface = Interface(
|
113 |
fn=chatbot_logic,
|
114 |
+
inputs=Textbox(label="Ask anything"),
|
115 |
+
outputs=Textbox(label="Reasoned Answer"),
|
116 |
+
title="🧠 Codettes-BlackForest Chatbot"
|
117 |
)
|
118 |
|
119 |
image_interface = Interface(
|
120 |
fn=image_logic,
|
121 |
+
inputs=Textbox(label="Describe an image"),
|
122 |
+
outputs=Image(label="Generated Image"),
|
123 |
+
title="🎨 Image Generator (FLUX.1-dev)"
|
124 |
)
|
125 |
|
126 |
+
return Blocks([text_interface, image_interface])
|
127 |
|
128 |
def launch(self):
|
129 |
+
app = self.setup_interface()
|
130 |
+
app.launch()
|
131 |
|
132 |
+
# Launch the app
|
133 |
if __name__ == "__main__":
|
134 |
+
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
|
135 |
+
HuggingFaceChatbot().launch()
|
|