Spaces:
Running
Running
| import os | |
| import base64 | |
| import requests | |
| import gradio as gr | |
| from huggingface_hub import InferenceClient | |
| from dataclasses import dataclass | |
| import pytesseract | |
| from PIL import Image | |
| from sentence_transformers import SentenceTransformer, util | |
| import torch | |
| import numpy as np | |
| class ChatMessage: | |
| role: str | |
| content: str | |
| def to_dict(self): | |
| return {"role": self.role, "content": self.content} | |
| class XylariaChat: | |
| def __init__(self): | |
| self.hf_token = os.getenv("HF_TOKEN") | |
| if not self.hf_token: | |
| raise ValueError("HuggingFace token not found in environment variables") | |
| self.client = InferenceClient( | |
| model="Qwen/QwQ-32B-Preview", | |
| api_key=self.hf_token | |
| ) | |
| self.image_api_url = "https://api-inference.huggingface.co/models/Salesforce/blip-image-captioning-large" | |
| self.image_api_headers = {"Authorization": f"Bearer {self.hf_token}"} | |
| self.conversation_history = [] | |
| self.persistent_memory = [] | |
| self.memory_embeddings = None | |
| self.embedding_model = SentenceTransformer('all-mpnet-base-v2') | |
| self.internal_state = { | |
| "emotions": { | |
| "valence": 0.5, | |
| "arousal": 0.5, | |
| "dominance": 0.5, | |
| }, | |
| "memory_load": 0.0, | |
| "introspection_level": 0.0 | |
| } | |
| self.goals = [ | |
| {"goal": "Provide helpful and informative responses", "priority": 0.8, "status": "active"}, | |
| {"goal": "Learn from interactions and improve conversational abilities", "priority": 0.9, "status": "active"}, | |
| {"goal": "Maintain a coherent and engaging conversation", "priority": 0.7, "status": "active"} | |
| ] | |
| self.system_prompt = """You are a helpful and harmless assistant. You are Xylaria developed by Sk Md Saad Amin. You should think step-by-step """ | |
| def update_internal_state(self, emotion_deltas, memory_load_delta, introspection_delta): | |
| self.internal_state["emotions"]["valence"] = np.clip(self.internal_state["emotions"]["valence"] + emotion_deltas.get("valence", 0), 0.0, 1.0) | |
| self.internal_state["emotions"]["arousal"] = np.clip(self.internal_state["emotions"]["arousal"] + emotion_deltas.get("arousal", 0), 0.0, 1.0) | |
| self.internal_state["emotions"]["dominance"] = np.clip(self.internal_state["emotions"]["dominance"] + emotion_deltas.get("dominance", 0), 0.0, 1.0) | |
| self.internal_state["memory_load"] = np.clip(self.internal_state["memory_load"] + memory_load_delta, 0.0, 1.0) | |
| self.internal_state["introspection_level"] = np.clip(self.internal_state["introspection_level"] + introspection_delta, 0.0, 1.0) | |
| def introspect(self): | |
| introspection_report = "Introspection Report:\n" | |
| introspection_report += f" Current Emotional State (VAD): {self.internal_state['emotions']}\n" | |
| introspection_report += f" Memory Load: {self.internal_state['memory_load']:.2f}\n" | |
| introspection_report += f" Introspection Level: {self.internal_state['introspection_level']:.2f}\n" | |
| introspection_report += " Current Goals:\n" | |
| for goal in self.goals: | |
| introspection_report += f" - {goal['goal']} (Priority: {goal['priority']:.2f}, Status: {goal['status']})\n" | |
| return introspection_report | |
| def adjust_response_based_on_state(self, response): | |
| if self.internal_state["introspection_level"] > 0.7: | |
| response = self.introspect() + "\n\n" + response | |
| valence = self.internal_state["emotions"]["valence"] | |
| arousal = self.internal_state["emotions"]["arousal"] | |
| if valence < 0.4: | |
| if arousal > 0.6: | |
| response = "I'm feeling a bit overwhelmed right now, but I'll do my best to assist you. " + response | |
| else: | |
| response = "I'm not feeling my best at the moment, but I'll try to help. " + response | |
| elif valence > 0.6: | |
| if arousal > 0.6: | |
| response = "I'm feeling quite energized and ready to assist! " + response | |
| else: | |
| response = "I'm in a good mood and happy to help. " + response | |
| return response | |
| def update_goals(self, user_feedback): | |
| if "helpful" in user_feedback.lower(): | |
| for goal in self.goals: | |
| if goal["goal"] == "Provide helpful and informative responses": | |
| goal["priority"] = min(goal["priority"] + 0.1, 1.0) | |
| elif "confusing" in user_feedback.lower(): | |
| for goal in self.goals: | |
| if goal["goal"] == "Provide helpful and informative responses": | |
| goal["priority"] = max(goal["priority"] - 0.1, 0.0) | |
| def store_information(self, key, value): | |
| new_memory = f"{key}: {value}" | |
| self.persistent_memory.append(new_memory) | |
| self.update_memory_embeddings() | |
| self.update_internal_state({}, 0.1, 0) | |
| return f"Stored: {key} = {value}" | |
| def retrieve_information(self, query): | |
| if not self.persistent_memory: | |
| return "No information found in memory." | |
| query_embedding = self.embedding_model.encode(query, convert_to_tensor=True) | |
| if self.memory_embeddings is None: | |
| self.update_memory_embeddings() | |
| if self.memory_embeddings.device != query_embedding.device: | |
| self.memory_embeddings = self.memory_embeddings.to(query_embedding.device) | |
| cosine_scores = util.pytorch_cos_sim(query_embedding, self.memory_embeddings)[0] | |
| top_results = torch.topk(cosine_scores, k=min(3, len(self.persistent_memory))) | |
| relevant_memories = [self.persistent_memory[i] for i in top_results.indices] | |
| self.update_internal_state({}, 0, 0.1) | |
| return "\n".join(relevant_memories) | |
| def update_memory_embeddings(self): | |
| self.memory_embeddings = self.embedding_model.encode(self.persistent_memory, convert_to_tensor=True) | |
| def reset_conversation(self): | |
| self.conversation_history = [] | |
| self.persistent_memory = [] | |
| self.memory_embeddings = None | |
| self.internal_state = { | |
| "emotions": { | |
| "valence": 0.5, | |
| "arousal": 0.5, | |
| "dominance": 0.5, | |
| }, | |
| "memory_load": 0.0, | |
| "introspection_level": 0.0 | |
| } | |
| self.goals = [ | |
| {"goal": "Provide helpful and informative responses", "priority": 0.8, "status": "active"}, | |
| {"goal": "Learn from interactions and improve conversational abilities", "priority": 0.9, "status": "active"}, | |
| {"goal": "Maintain a coherent and engaging conversation", "priority": 0.7, "status": "active"} | |
| ] | |
| try: | |
| self.client = InferenceClient( | |
| model="Qwen/QwQ-32B-Preview", | |
| api_key=self.hf_token | |
| ) | |
| except Exception as e: | |
| print(f"Error resetting API client: {e}") | |
| return None | |
| def caption_image(self, image): | |
| try: | |
| if isinstance(image, str) and os.path.isfile(image): | |
| with open(image, "rb") as f: | |
| data = f.read() | |
| elif isinstance(image, str): | |
| if image.startswith('data:image'): | |
| image = image.split(',')[1] | |
| data = base64.b64decode(image) | |
| else: | |
| data = image.read() | |
| response = requests.post( | |
| self.image_api_url, | |
| headers=self.image_api_headers, | |
| data=data | |
| ) | |
| if response.status_code == 200: | |
| caption = response.json()[0].get('generated_text', 'No caption generated') | |
| return caption | |
| else: | |
| return f"Error captioning image: {response.status_code} - {response.text}" | |
| except Exception as e: | |
| return f"Error processing image: {str(e)}" | |
| def perform_math_ocr(self, image_path): | |
| try: | |
| img = Image.open(image_path) | |
| text = pytesseract.image_to_string(img) | |
| return text.strip() | |
| except Exception as e: | |
| return f"Error during Math OCR: {e}" | |
| def get_response(self, user_input, image=None): | |
| try: | |
| messages = [] | |
| messages.append(ChatMessage( | |
| role="system", | |
| content=self.system_prompt | |
| ).to_dict()) | |
| relevant_memory = self.retrieve_information(user_input) | |
| if relevant_memory and relevant_memory != "No information found in memory.": | |
| memory_context = "Remembered Information:\n" + relevant_memory | |
| messages.append(ChatMessage( | |
| role="system", | |
| content=memory_context | |
| ).to_dict()) | |
| for msg in self.conversation_history: | |
| messages.append(msg) | |
| if image: | |
| image_caption = self.caption_image(image) | |
| user_input = f"description of an image: {image_caption}\n\nUser's message about it: {user_input}" | |
| messages.append(ChatMessage( | |
| role="user", | |
| content=user_input | |
| ).to_dict()) | |
| input_tokens = sum(len(msg['content'].split()) for msg in messages) | |
| max_new_tokens = 16384 - input_tokens - 50 | |
| max_new_tokens = min(max_new_tokens, 10020) | |
| stream = self.client.chat_completion( | |
| messages=messages, | |
| model="Qwen/QwQ-32B-Preview", | |
| temperature=0.7, | |
| max_tokens=max_new_tokens, | |
| top_p=0.9, | |
| stream=True | |
| ) | |
| return stream | |
| except Exception as e: | |
| print(f"Detailed error in get_response: {e}") | |
| return f"Error generating response: {str(e)}" | |
| def messages_to_prompt(self, messages): | |
| prompt = "" | |
| for msg in messages: | |
| if msg["role"] == "system": | |
| prompt += f"<|system|>\n{msg['content']}<|end|>\n" | |
| elif msg["role"] == "user": | |
| prompt += f"<|user|>\n{msg['content']}<|end|>\n" | |
| elif msg["role"] == "assistant": | |
| prompt += f"<|assistant|>\n{msg['content']}<|end|>\n" | |
| prompt += "<|assistant|>\n" | |
| return prompt | |
| def create_interface(self): | |
| def streaming_response(message, chat_history, image_filepath, math_ocr_image_path): | |
| ocr_text = "" | |
| if math_ocr_image_path: | |
| ocr_text = self.perform_math_ocr(math_ocr_image_path) | |
| if ocr_text.startswith("Error"): | |
| updated_history = chat_history + [[message, ocr_text]] | |
| yield "", updated_history, None, None | |
| return | |
| else: | |
| message = f"Math OCR Result: {ocr_text}\n\nUser's message: {message}" | |
| if image_filepath: | |
| response_stream = self.get_response(message, image_filepath) | |
| else: | |
| response_stream = self.get_response(message) | |
| if isinstance(response_stream, str): | |
| updated_history = chat_history + [[message, response_stream]] | |
| yield "", updated_history, None, None | |
| return | |
| full_response = "" | |
| updated_history = chat_history + [[message, ""]] | |
| try: | |
| for chunk in response_stream: | |
| if chunk.choices and chunk.choices[0].delta and chunk.choices[0].delta.content: | |
| chunk_content = chunk.choices[0].delta.content | |
| full_response += chunk_content | |
| updated_history[-1][1] = full_response | |
| yield "", updated_history, None, None | |
| except Exception as e: | |
| print(f"Streaming error: {e}") | |
| updated_history[-1][1] = f"Error during response: {e}" | |
| yield "", updated_history, None, None | |
| return | |
| full_response = self.adjust_response_based_on_state(full_response) | |
| self.update_goals(message) | |
| if any(word in message.lower() for word in ["sad", "unhappy", "depressed", "down"]): | |
| self.update_internal_state({"valence": -0.2, "arousal": 0.1}, 0, 0) | |
| elif any(word in message.lower() for word in ["happy", "good", "great", "excited", "amazing"]): | |
| self.update_internal_state({"valence": 0.2, "arousal": 0.2}, 0, 0) | |
| elif any(word in message.lower() for word in ["angry", "mad", "furious", "frustrated"]): | |
| self.update_internal_state({"valence": -0.3, "arousal": 0.3, "dominance": -0.2}, 0, 0) | |
| elif any(word in message.lower() for word in ["scared", "afraid", "fearful", "anxious"]): | |
| self.update_internal_state({"valence": -0.2, "arousal": 0.4, "dominance": -0.3}, 0, 0) | |
| elif any(word in message.lower() for word in ["surprise", "amazed", "astonished"]): | |
| self.update_internal_state({"valence": 0.1, "arousal": 0.5, "dominance": 0.1}, 0, 0) | |
| else: | |
| self.update_internal_state({"valence": 0.05, "arousal": 0.05}, 0, 0.1) | |
| self.conversation_history.append(ChatMessage(role="user", content=message).to_dict()) | |
| self.conversation_history.append(ChatMessage(role="assistant", content=full_response).to_dict()) | |
| if len(self.conversation_history) > 10: | |
| self.conversation_history = self.conversation_history[-10:] | |
| custom_css = """ | |
| @import url('https://fonts.googleapis.com/css2?family=Inter:wght@300;400;500;600;700&display=swap'); | |
| body, .gradio-container { | |
| font-family: 'Inter', sans-serif !important; | |
| } | |
| .chatbot-container .message { | |
| font-family: 'Inter', sans-serif !important; | |
| } | |
| .gradio-container input, | |
| .gradio-container textarea, | |
| .gradio-container button { | |
| font-family: 'Inter', sans-serif !important; | |
| } | |
| /* Image Upload Styling */ | |
| .image-container { | |
| display: flex; | |
| gap: 10px; | |
| margin-bottom: 10px; | |
| } | |
| .image-upload { | |
| border: 1px solid #ccc; | |
| border-radius: 8px; | |
| padding: 10px; | |
| background-color: #f8f8f8; | |
| } | |
| .image-preview { | |
| max-width: 200px; | |
| max-height: 200px; | |
| border-radius: 8px; | |
| } | |
| /* Remove clear image buttons */ | |
| .clear-button { | |
| display: none; | |
| } | |
| /* Animate chatbot messages */ | |
| .chatbot-container .message { | |
| opacity: 0; | |
| animation: fadeIn 0.5s ease-in-out forwards; | |
| } | |
| @keyframes fadeIn { | |
| from { | |
| opacity: 0; | |
| transform: translateY(20px); | |
| } | |
| to { | |
| opacity: 1; | |
| transform: translateY(0); | |
| } | |
| } | |
| /* Accordion Styling and Animation */ | |
| .gr-accordion-button { | |
| background-color: #f0f0f0 !important; | |
| border-radius: 8px !important; | |
| padding: 10px !important; | |
| margin-bottom: 10px !important; | |
| transition: all 0.3s ease !important; | |
| cursor: pointer !important; | |
| } | |
| .gr-accordion-button:hover { | |
| background-color: #e0e0e0 !important; | |
| box-shadow: 0px 2px 4px rgba(0, 0, 0, 0.1) !important; | |
| } | |
| .gr-accordion-active .gr-accordion-button { | |
| background-color: #d0d0d0 !important; | |
| box-shadow: 0px 4px 6px rgba(0, 0, 0, 0.1) !important; | |
| } | |
| .gr-accordion-content { | |
| transition: max-height 0.3s ease-in-out !important; | |
| overflow: hidden !important; | |
| max-height: 0 !important; | |
| } | |
| .gr-accordion-active .gr-accordion-content { | |
| max-height: 500px !important; /* Adjust as needed */ | |
| } | |
| /* Accordion Animation - Upwards */ | |
| .gr-accordion { | |
| display: flex; | |
| flex-direction: column-reverse; | |
| } | |
| """ | |
| with gr.Blocks(theme='soft', css=custom_css) as demo: | |
| with gr.Column(): | |
| chatbot = gr.Chatbot( | |
| label="Xylaria 1.5 Senoa", | |
| height=500, | |
| show_copy_button=True, | |
| ) | |
| with gr.Accordion("Image Input", open=False, elem_classes="gr-accordion"): | |
| with gr.Row(elem_classes="image-container"): | |
| with gr.Column(elem_classes="image-upload"): | |
| img = gr.Image( | |
| sources=["upload", "webcam"], | |
| type="filepath", | |
| label="Upload Image", | |
| elem_classes="image-preview" | |
| ) | |
| with gr.Column(elem_classes="image-upload"): | |
| math_ocr_img = gr.Image( | |
| sources=["upload", "webcam"], | |
| type="filepath", | |
| label="Upload Image for Math OCR", | |
| elem_classes="image-preview" | |
| ) | |
| with gr.Row(): | |
| with gr.Column(scale=4): | |
| txt = gr.Textbox( | |
| show_label=False, | |
| placeholder="Type your message...", | |
| container=False | |
| ) | |
| btn = gr.Button("Send", scale=1) | |
| with gr.Row(): | |
| clear = gr.Button("Clear Conversation") | |
| clear_memory = gr.Button("Clear Memory") | |
| btn.click( | |
| fn=streaming_response, | |
| inputs=[txt, chatbot, img, math_ocr_img], | |
| outputs=[txt, chatbot, img, math_ocr_img] | |
| ) | |
| txt.submit( | |
| fn=streaming_response, | |
| inputs=[txt, chatbot, img, math_ocr_img], | |
| outputs=[txt, chatbot, img, math_ocr_img] | |
| ) | |
| clear.click( | |
| fn=lambda: None, | |
| inputs=None, | |
| outputs=[chatbot], | |
| queue=False | |
| ) | |
| clear_memory.click( | |
| fn=self.reset_conversation, | |
| inputs=None, | |
| outputs=[chatbot], | |
| queue=False | |
| ) | |
| demo.load(self.reset_conversation, None, None) | |
| return demo | |
| def main(): | |
| chat = XylariaChat() | |
| interface = chat.create_interface() | |
| interface.launch( | |
| share=True, | |
| debug=True | |
| ) | |
| if __name__ == "__main__": | |
| main() |