app.py

import os
import gradio as gr
from huggingface_hub import InferenceClient
import json

class XylariaChat:
    def __init__(self):
        # Securely load HuggingFace token
        self.hf_token = os.getenv("HF_TOKEN")
        if not self.hf_token:
            raise ValueError("HuggingFace token not found in environment variables")

        # Initialize the inference client
        self.client = InferenceClient(
            model="Qwen/Qwen-32B-Preview",
            api_key=self.hf_token
        )

        # Initialize conversation history and persistent memory
        self.conversation_history = []
        self.persistent_memory = {}
        self.chat_file_path = "chat_history.txt"  # File to save chats

        # System prompt with more detailed instructions
        self.system_prompt = """You are a helpful and harmless AI assistant you are Xylaria 1.4 Senoa, Made by Sk Md Saad Amin you think step by step
"""

    def store_information(self, key, value):
        """Store important information in persistent memory"""
        self.persistent_memory[key] = value

    def retrieve_information(self, key):
        """Retrieve information from persistent memory"""
        return self.persistent_memory.get(key)

    def save_chat(self):
        """Saves the current chat history to a text file."""
        try:
            with open(self.chat_file_path, "w") as f:
                chat_data = {
                    "conversation_history": self.conversation_history,
                    "persistent_memory": self.persistent_memory
                }
                json.dump(chat_data, f)
        except Exception as e:
            print(f"Error saving chat history: {e}")

    def load_chat(self):
        """Loads chat history from a text file."""
        try:
            with open(self.chat_file_path, "r") as f:
                chat_data = json.load(f)
                self.conversation_history = chat_data.get("conversation_history", [])
                self.persistent_memory = chat_data.get("persistent_memory", {})
                return self.conversation_history, self.persistent_memory
        except FileNotFoundError:
            print("Chat history file not found.")
            return [], {}
        except Exception as e:
            print(f"Error loading chat history: {e}")
            return [], {}

    def reset_conversation(self):
        """
        Completely reset the conversation history, persistent memory,
        and clear API-side memory
        """
        # Clear local memory
        self.conversation_history = []
        self.persistent_memory.clear()

        # Clear API-side memory by resetting the conversation
        try:
            # Attempt to clear any API-side session or context
            self.client = InferenceClient(
                model="Qwen/Qwen-32B-Preview",
                api_key=self.hf_token
            )
        except Exception as e:
            print(f"Error resetting API client: {e}")

        self.save_chat()  # Save the empty chat history
        return None  # To clear the chatbot interface

    def get_response(self, user_input):
        # Prepare messages with conversation context and persistent memory
        messages = [
            {"role": "system", "content": self.system_prompt},
            *self.conversation_history,
            {"role": "user", "content": user_input}
        ]

        # Add persistent memory context if available
        if self.persistent_memory:
            memory_context = "Remembered Information:\n" + "\n".join(
                [f"{k}: {v}" for k, v in self.persistent_memory.items()]
            )
            messages.insert(1, {"role": "system", "content": memory_context})

        # Generate response with streaming
        try:
            stream = self.client.chat.completions.create(
                messages=messages,
                temperature=0.5,
                max_tokens=10240,
                top_p=0.7,
                stream=True
            )

            return stream

        except Exception as e:
            return f"Error generating response: {str(e)}"

    def create_interface(self):
        def streaming_response(message, chat_history):
            # Clear input textbox
            response_stream = self.get_response(message)

            # If it's an error, return immediately
            if isinstance(response_stream, str):
                return "", chat_history + [[message, response_stream]]

            # Prepare for streaming response
            full_response = ""
            updated_history = chat_history + [[message, ""]]

            # Streaming output
            for chunk in response_stream:
                if chunk.choices[0].delta.content:
                    chunk_content = chunk.choices[0].delta.content
                    full_response += chunk_content

                    # Update the last message in chat history with partial response
                    updated_history[-1][1] = full_response
                    yield "", updated_history

            # Update conversation history
            self.conversation_history.append(
                {"role": "user", "content": message}
            )
            self.conversation_history.append(
                {"role": "assistant", "content": full_response}
            )

            # Limit conversation history to prevent token overflow
            if len(self.conversation_history) > 10:
                self.conversation_history = self.conversation_history[-10:]

            self.save_chat()

        def load_chat_interface():
            """Loads the chat history into the chatbot interface."""
            self.load_chat()
            return self.conversation_history

        # Custom CSS for Inter font and sidebar
        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;
        }

        /* Sidebar styling */
        #sidebar {
            background-color: #f2f2f2;
            border-right: 1px solid #ccc;
            padding: 10px;
            height: 100vh;
            overflow-y: auto;
        }

        #sidebar ul {
            list-style-type: none;
            padding: 0;
        }

        #sidebar li {
            margin-bottom: 5px;
        }
        /* Main chat area */
        #main-chat {
            padding: 20px;
        }
        """

        # Example prompts
        example_prompts = [
            "How do I get started with coding?",
            "Tell me a fun fact about science.",
            "What are some good books to read?"
        ]

        # Function to forward prompt to the textbox
        def forward_prompt(prompt):
            return prompt

        with gr.Blocks(theme='soft', css=custom_css) as demo:
            with gr.Row():
                # Sidebar for displaying chat history
                with gr.Column(elem_id="sidebar", scale=1):  # Added elem_id for CSS
                    gr.Markdown("### Chat History")
                    load_button = gr.Button("Load Chat History")
                    chat_list = gr.Markdown("No chat history found.")

                    load_button.click(
                        fn=lambda: gr.Markdown.update(value=self.format_chat_history()),
                        inputs=None,
                        outputs=[chat_list]
                    )

                # Main chat interface
                with gr.Column(elem_id="main-chat", scale=3):
                    # Show Xylaria and example prompts only on the first page/new chat
                    with gr.Column(visible=True) as start_page:
                        gr.Markdown("# Xylaria")
                        with gr.Row():
                          for prompt in example_prompts:
                            gr.Button(prompt).click(
                                fn=forward_prompt,
                                inputs=gr.State(prompt),
                                outputs=txt
                            )

                    with gr.Column(visible=False) as chat_page:
                        chatbot = gr.Chatbot(
                            label="Xylaria 1.4 Senoa",
                            height=500,
                            show_copy_button=True
                        )

                        # Input row with improved layout
                        with gr.Row():
                            txt = gr.Textbox(
                                show_label=False,
                                placeholder="Type your message...",
                                container=False,
                                scale=4
                            )
                            btn = gr.Button("Send", scale=1)

                        # Clear history and memory buttons
                        clear = gr.Button("Clear Conversation")
                        clear_memory = gr.Button("Clear Memory")

                    # Toggle between start page and chat page
                    def toggle_page(choice):
                        return gr.Column.update(visible=choice == "chat"), gr.Column.update(visible=choice == "start")

                    start_page.visible = True
                    chat_page.visible = False

                    # Submit functionality with streaming
                    btn.click(
                        fn=streaming_response,
                        inputs=[txt, chatbot],
                        outputs=[txt, chatbot]
                    ).then(
                        fn=lambda: toggle_page("chat"),
                        inputs=gr.State("chat"),
                        outputs=[chat_page, start_page]
                    )
                    txt.submit(
                        fn=streaming_response,
                        inputs=[txt, chatbot],
                        outputs=[txt, chatbot]
                    ).then(
                        fn=lambda: toggle_page("chat"),
                        inputs=gr.State("chat"),
                        outputs=[chat_page, start_page]
                    )

                    # Clear conversation history
                    clear.click(
                        fn=lambda: None,
                        inputs=None,
                        outputs=[chatbot],
                        queue=False
                    ).then(
                        fn=lambda: toggle_page("start"),
                        inputs=gr.State("start"),
                        outputs=[chat_page, start_page]
                    )

                    # Clear persistent memory and reset conversation
                    clear_memory.click(
                        fn=self.reset_conversation,
                        inputs=None,
                        outputs=[chatbot],
                        queue=False
                    ).then(
                        fn=lambda: toggle_page("start"),
                        inputs=gr.State("start"),
                        outputs=[chat_page, start_page]
                    )

                    # Ensure memory is cleared when the interface is closed
                    demo.load(self.reset_conversation, None, None)

        return demo

    def format_chat_history(self):
        """Formats the chat history for display in the sidebar."""
        self.load_chat()  # Load the chat history first
        if not self.conversation_history:
            return "No chat history found."

        formatted_history = ""
        for chat in self.conversation_history:
            if chat["role"] == "user":
                formatted_history += f"**You:** {chat['content']}\n\n"
            elif chat["role"] == "assistant":
                formatted_history += f"**Xylaria:** {chat['content']}\n\n"

        return formatted_history

# Launch the interface
def main():
    chat = XylariaChat()
    interface = chat.create_interface()
    interface.launch(
        share=True,  # Optional: create a public link
        debug=True   # Show detailed errors
    )

if __name__ == "__main__":
    main()