Spaces:
Running
Running
Reality123b
commited on
Update app.py
Browse files
app.py
CHANGED
@@ -21,24 +21,24 @@ class XylariaChat:
|
|
21 |
self.hf_token = os.getenv("HF_TOKEN")
|
22 |
if not self.hf_token:
|
23 |
raise ValueError("HuggingFace token not found in environment variables")
|
24 |
-
|
25 |
# Initialize the inference client with the Qwen model
|
26 |
self.client = InferenceClient(
|
27 |
-
model="Qwen/QwQ-32B-Preview",
|
28 |
api_key=self.hf_token
|
29 |
)
|
30 |
-
|
31 |
# Image captioning API setup
|
32 |
-
self.image_api_url = "https://api-inference.huggingface.co/models/
|
33 |
self.image_api_headers = {"Authorization": f"Bearer {self.hf_token}"}
|
34 |
-
|
35 |
# Initialize conversation history and persistent memory
|
36 |
self.conversation_history = []
|
37 |
self.persistent_memory = {}
|
38 |
-
|
39 |
# System prompt with more detailed instructions
|
40 |
self.system_prompt = """You are a helpful and harmless assistant. You are Xylaria developed by Sk Md Saad Amin(india). You should think step-by-step."""
|
41 |
-
|
42 |
def store_information(self, key, value):
|
43 |
"""Store important information in persistent memory"""
|
44 |
self.persistent_memory[key] = value
|
@@ -50,31 +50,31 @@ class XylariaChat:
|
|
50 |
|
51 |
def reset_conversation(self):
|
52 |
"""
|
53 |
-
Completely reset the conversation history, persistent memory,
|
54 |
and clear API-side memory
|
55 |
"""
|
56 |
# Clear local memory
|
57 |
self.conversation_history = []
|
58 |
self.persistent_memory.clear()
|
59 |
-
|
60 |
-
# Reinitialize the client
|
61 |
try:
|
62 |
self.client = InferenceClient(
|
63 |
-
model="Qwen/QwQ-32B-Preview",
|
64 |
api_key=self.hf_token
|
65 |
)
|
66 |
except Exception as e:
|
67 |
print(f"Error resetting API client: {e}")
|
68 |
-
|
69 |
return None # To clear the chatbot interface
|
70 |
|
71 |
def caption_image(self, image):
|
72 |
"""
|
73 |
Caption an uploaded image using Hugging Face API
|
74 |
-
|
75 |
Args:
|
76 |
image (str): Base64 encoded image or file path
|
77 |
-
|
78 |
Returns:
|
79 |
str: Image caption or error message
|
80 |
"""
|
@@ -89,83 +89,83 @@ class XylariaChat:
|
|
89 |
if image.startswith('data:image'):
|
90 |
image = image.split(',')[1]
|
91 |
data = base64.b64decode(image)
|
92 |
-
# If image is a file-like object
|
93 |
else:
|
94 |
data = image.read()
|
95 |
-
|
96 |
# Send request to Hugging Face API
|
97 |
response = requests.post(
|
98 |
-
self.image_api_url,
|
99 |
-
headers=self.image_api_headers,
|
100 |
data=data
|
101 |
)
|
102 |
-
|
103 |
# Check response
|
104 |
if response.status_code == 200:
|
105 |
caption = response.json()[0].get('generated_text', 'No caption generated')
|
106 |
return caption
|
107 |
else:
|
108 |
-
return f"Error captioning image: {response.text}"
|
109 |
-
|
110 |
except Exception as e:
|
111 |
return f"Error processing image: {str(e)}"
|
112 |
|
113 |
def get_response(self, user_input, image=None):
|
114 |
"""
|
115 |
Generate a response using chat completions with improved error handling
|
116 |
-
|
117 |
Args:
|
118 |
user_input (str): User's message
|
119 |
image (optional): Uploaded image
|
120 |
-
|
121 |
Returns:
|
122 |
Stream of chat completions or error message
|
123 |
"""
|
124 |
try:
|
125 |
# Prepare messages with conversation context and persistent memory
|
126 |
messages = []
|
127 |
-
|
128 |
# Add system prompt as first message
|
129 |
messages.append(ChatMessage(
|
130 |
-
role="system",
|
131 |
content=self.system_prompt
|
132 |
-
).to_dict())
|
133 |
-
|
134 |
# Add persistent memory context if available
|
135 |
if self.persistent_memory:
|
136 |
memory_context = "Remembered Information:\n" + "\n".join(
|
137 |
[f"{k}: {v}" for k, v in self.persistent_memory.items()]
|
138 |
)
|
139 |
messages.append(ChatMessage(
|
140 |
-
role="system",
|
141 |
content=memory_context
|
142 |
-
).to_dict())
|
143 |
-
|
144 |
# Convert existing conversation history to ChatMessage objects and then to dictionaries
|
145 |
for msg in self.conversation_history:
|
146 |
messages.append(ChatMessage(
|
147 |
-
role=msg['role'],
|
148 |
content=msg['content']
|
149 |
-
).to_dict())
|
150 |
-
|
151 |
# Process image if uploaded
|
152 |
if image:
|
153 |
image_caption = self.caption_image(image)
|
154 |
user_input = f"Image description: {image_caption}\n\nUser's message: {user_input}"
|
155 |
-
|
156 |
# Add user input
|
157 |
messages.append(ChatMessage(
|
158 |
-
role="user",
|
159 |
content=user_input
|
160 |
-
).to_dict())
|
161 |
-
|
162 |
# Generate response with streaming
|
163 |
-
stream = self.client.
|
|
|
164 |
model="Qwen/QwQ-32B-Preview",
|
165 |
-
|
166 |
-
|
167 |
-
|
168 |
-
top_p=0.7,
|
169 |
stream=True
|
170 |
)
|
171 |
|
@@ -175,6 +175,25 @@ class XylariaChat:
|
|
175 |
print(f"Detailed error in get_response: {e}")
|
176 |
return f"Error generating response: {str(e)}"
|
177 |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
178 |
def create_interface(self):
|
179 |
def streaming_response(message, chat_history, image_filepath):
|
180 |
# Check if an image was actually uploaded
|
@@ -200,7 +219,7 @@ class XylariaChat:
|
|
200 |
if chunk.choices and chunk.choices[0].delta and chunk.choices[0].delta.content:
|
201 |
chunk_content = chunk.choices[0].delta.content
|
202 |
full_response += chunk_content
|
203 |
-
|
204 |
# Update the last message in chat history with partial response
|
205 |
updated_history[-1][1] = full_response
|
206 |
yield "", updated_history, None
|
@@ -232,8 +251,8 @@ class XylariaChat:
|
|
232 |
.chatbot-container .message {
|
233 |
font-family: 'Inter', sans-serif !important;
|
234 |
}
|
235 |
-
.gradio-container input,
|
236 |
-
.gradio-container textarea,
|
237 |
.gradio-container button {
|
238 |
font-family: 'Inter', sans-serif !important;
|
239 |
}
|
|
|
21 |
self.hf_token = os.getenv("HF_TOKEN")
|
22 |
if not self.hf_token:
|
23 |
raise ValueError("HuggingFace token not found in environment variables")
|
24 |
+
|
25 |
# Initialize the inference client with the Qwen model
|
26 |
self.client = InferenceClient(
|
27 |
+
model="Qwen/QwQ-32B-Preview", # Using the specified model
|
28 |
api_key=self.hf_token
|
29 |
)
|
30 |
+
|
31 |
# Image captioning API setup
|
32 |
+
self.image_api_url = "https://api-inference.huggingface.co/models/Salesforce/blip-image-captioning-large"
|
33 |
self.image_api_headers = {"Authorization": f"Bearer {self.hf_token}"}
|
34 |
+
|
35 |
# Initialize conversation history and persistent memory
|
36 |
self.conversation_history = []
|
37 |
self.persistent_memory = {}
|
38 |
+
|
39 |
# System prompt with more detailed instructions
|
40 |
self.system_prompt = """You are a helpful and harmless assistant. You are Xylaria developed by Sk Md Saad Amin(india). You should think step-by-step."""
|
41 |
+
|
42 |
def store_information(self, key, value):
|
43 |
"""Store important information in persistent memory"""
|
44 |
self.persistent_memory[key] = value
|
|
|
50 |
|
51 |
def reset_conversation(self):
|
52 |
"""
|
53 |
+
Completely reset the conversation history, persistent memory,
|
54 |
and clear API-side memory
|
55 |
"""
|
56 |
# Clear local memory
|
57 |
self.conversation_history = []
|
58 |
self.persistent_memory.clear()
|
59 |
+
|
60 |
+
# Reinitialize the client (not strictly necessary for the API, but can help with local state)
|
61 |
try:
|
62 |
self.client = InferenceClient(
|
63 |
+
model="Qwen/QwQ-32B-Preview",
|
64 |
api_key=self.hf_token
|
65 |
)
|
66 |
except Exception as e:
|
67 |
print(f"Error resetting API client: {e}")
|
68 |
+
|
69 |
return None # To clear the chatbot interface
|
70 |
|
71 |
def caption_image(self, image):
|
72 |
"""
|
73 |
Caption an uploaded image using Hugging Face API
|
74 |
+
|
75 |
Args:
|
76 |
image (str): Base64 encoded image or file path
|
77 |
+
|
78 |
Returns:
|
79 |
str: Image caption or error message
|
80 |
"""
|
|
|
89 |
if image.startswith('data:image'):
|
90 |
image = image.split(',')[1]
|
91 |
data = base64.b64decode(image)
|
92 |
+
# If image is a file-like object (unlikely with Gradio, but good to have)
|
93 |
else:
|
94 |
data = image.read()
|
95 |
+
|
96 |
# Send request to Hugging Face API
|
97 |
response = requests.post(
|
98 |
+
self.image_api_url,
|
99 |
+
headers=self.image_api_headers,
|
100 |
data=data
|
101 |
)
|
102 |
+
|
103 |
# Check response
|
104 |
if response.status_code == 200:
|
105 |
caption = response.json()[0].get('generated_text', 'No caption generated')
|
106 |
return caption
|
107 |
else:
|
108 |
+
return f"Error captioning image: {response.status_code} - {response.text}"
|
109 |
+
|
110 |
except Exception as e:
|
111 |
return f"Error processing image: {str(e)}"
|
112 |
|
113 |
def get_response(self, user_input, image=None):
|
114 |
"""
|
115 |
Generate a response using chat completions with improved error handling
|
116 |
+
|
117 |
Args:
|
118 |
user_input (str): User's message
|
119 |
image (optional): Uploaded image
|
120 |
+
|
121 |
Returns:
|
122 |
Stream of chat completions or error message
|
123 |
"""
|
124 |
try:
|
125 |
# Prepare messages with conversation context and persistent memory
|
126 |
messages = []
|
127 |
+
|
128 |
# Add system prompt as first message
|
129 |
messages.append(ChatMessage(
|
130 |
+
role="system",
|
131 |
content=self.system_prompt
|
132 |
+
).to_dict())
|
133 |
+
|
134 |
# Add persistent memory context if available
|
135 |
if self.persistent_memory:
|
136 |
memory_context = "Remembered Information:\n" + "\n".join(
|
137 |
[f"{k}: {v}" for k, v in self.persistent_memory.items()]
|
138 |
)
|
139 |
messages.append(ChatMessage(
|
140 |
+
role="system",
|
141 |
content=memory_context
|
142 |
+
).to_dict())
|
143 |
+
|
144 |
# Convert existing conversation history to ChatMessage objects and then to dictionaries
|
145 |
for msg in self.conversation_history:
|
146 |
messages.append(ChatMessage(
|
147 |
+
role=msg['role'],
|
148 |
content=msg['content']
|
149 |
+
).to_dict())
|
150 |
+
|
151 |
# Process image if uploaded
|
152 |
if image:
|
153 |
image_caption = self.caption_image(image)
|
154 |
user_input = f"Image description: {image_caption}\n\nUser's message: {user_input}"
|
155 |
+
|
156 |
# Add user input
|
157 |
messages.append(ChatMessage(
|
158 |
+
role="user",
|
159 |
content=user_input
|
160 |
+
).to_dict())
|
161 |
+
|
162 |
# Generate response with streaming
|
163 |
+
stream = self.client.chat_completion(
|
164 |
+
messages=messages,
|
165 |
model="Qwen/QwQ-32B-Preview",
|
166 |
+
temperature=0.7,
|
167 |
+
max_tokens=16384,
|
168 |
+
top_p=0.9,
|
|
|
169 |
stream=True
|
170 |
)
|
171 |
|
|
|
175 |
print(f"Detailed error in get_response: {e}")
|
176 |
return f"Error generating response: {str(e)}"
|
177 |
|
178 |
+
def messages_to_prompt(self, messages):
|
179 |
+
"""
|
180 |
+
Convert a list of ChatMessage dictionaries to a single prompt string.
|
181 |
+
|
182 |
+
This is a simple implementation and you might need to adjust it
|
183 |
+
based on the specific requirements of the model you are using.
|
184 |
+
"""
|
185 |
+
prompt = ""
|
186 |
+
for msg in messages:
|
187 |
+
if msg["role"] == "system":
|
188 |
+
prompt += f"<|system|>\n{msg['content']}<|end|>\n"
|
189 |
+
elif msg["role"] == "user":
|
190 |
+
prompt += f"<|user|>\n{msg['content']}<|end|>\n"
|
191 |
+
elif msg["role"] == "assistant":
|
192 |
+
prompt += f"<|assistant|>\n{msg['content']}<|end|>\n"
|
193 |
+
prompt += "<|assistant|>\n" # Start of assistant's turn
|
194 |
+
return prompt
|
195 |
+
|
196 |
+
|
197 |
def create_interface(self):
|
198 |
def streaming_response(message, chat_history, image_filepath):
|
199 |
# Check if an image was actually uploaded
|
|
|
219 |
if chunk.choices and chunk.choices[0].delta and chunk.choices[0].delta.content:
|
220 |
chunk_content = chunk.choices[0].delta.content
|
221 |
full_response += chunk_content
|
222 |
+
|
223 |
# Update the last message in chat history with partial response
|
224 |
updated_history[-1][1] = full_response
|
225 |
yield "", updated_history, None
|
|
|
251 |
.chatbot-container .message {
|
252 |
font-family: 'Inter', sans-serif !important;
|
253 |
}
|
254 |
+
.gradio-container input,
|
255 |
+
.gradio-container textarea,
|
256 |
.gradio-container button {
|
257 |
font-family: 'Inter', sans-serif !important;
|
258 |
}
|