app.py

"""
SIKE Gradio Web Interface
A beautiful web UI for the Self-Improving Knowledge Engine
Run with: python gradio_app.py
"""

import gradio as gr
import json
import asyncio
from datetime import datetime
from typing import List, Tuple, Dict, Any
import sqlite3
from pathlib import Path

# Import the SIKE server
from sike_server import SIKEServer


class SIKEGradioApp:
    """Gradio web interface for SIKE"""
    
    def __init__(self):
        self.server = SIKEServer(storage_path="./sike_data")
        self.conversation_history: List[Dict[str, str]] = []
    
    # ======================
    # KNOWLEDGE MANAGEMENT
    # ======================
    
    def store_knowledge(self, topic: str, content: str, metadata: str) -> str:
        """Store new knowledge"""
        try:
            metadata_dict = json.loads(metadata) if metadata else {}
        except json.JSONDecodeError:
            return "❌ Error: Invalid JSON in metadata field"
        
        result = asyncio.run(self.server._store_knowledge({
            "topic": topic,
            "content": content,
            "metadata": metadata_dict
        }))
        
        return result[0].text
    
    def search_knowledge(self, query: str, limit: int) -> str:
        """Search knowledge base"""
        if not query:
            return "❌ Please enter a search query"
        
        result = asyncio.run(self.server._search_knowledge({
            "query": query,
            "limit": limit
        }))
        
        return result[0].text
    
    def get_all_knowledge(self) -> str:
        """Get all knowledge entries"""
        cursor = self.server.conn.cursor()
        cursor.execute("SELECT id, topic, content, timestamp FROM knowledge ORDER BY timestamp DESC LIMIT 50")
        entries = cursor.fetchall()
        
        if not entries:
            return "No knowledge entries found."
        
        output = "# 📚 All Knowledge Entries\n\n"
        for entry_id, topic, content, timestamp in entries:
            output += f"## ID: {entry_id} | Topic: {topic}\n"
            output += f"**Content:** {content}\n"
            output += f"**Stored:** {timestamp}\n\n"
            output += "---\n\n"
        
        return output
    
    # ======================
    # EXPERIENCE LOGGING
    # ======================
    
    def log_experience(self, context: str, attempt: str, result: str, 
                       lesson: str, success: bool) -> str:
        """Log an experience"""
        if not all([context, attempt, result, lesson]):
            return "❌ Please fill in all fields"
        
        result_obj = asyncio.run(self.server._log_experience({
            "context": context,
            "attempt": attempt,
            "result": result,
            "lesson_learned": lesson,
            "success": success
        }))
        
        return result_obj[0].text
    
    def search_experiences(self, query: str, limit: int) -> str:
        """Search past experiences"""
        if not query:
            return "❌ Please enter a search query"
        
        result = asyncio.run(self.server._search_experiences({
            "query": query,
            "limit": limit
        }))
        
        return result[0].text
    
    def get_all_experiences(self) -> str:
        """Get all experience logs"""
        cursor = self.server.conn.cursor()
        cursor.execute("""
            SELECT id, context, attempt, result, lesson_learned, success, timestamp 
            FROM experiences 
            ORDER BY timestamp DESC 
            LIMIT 50
        """)
        experiences = cursor.fetchall()
        
        if not experiences:
            return "No experiences logged yet."
        
        output = "# 📝 Experience Log\n\n"
        for exp_id, context, attempt, result, lesson, success, timestamp in experiences:
            status = "✅ Success" if success else "❌ Failure"
            output += f"## {status} | ID: {exp_id}\n"
            output += f"**Context:** {context}\n"
            output += f"**Attempt:** {attempt}\n"
            output += f"**Result:** {result}\n"
            output += f"**Lesson Learned:** {lesson}\n"
            output += f"**Time:** {timestamp}\n\n"
            output += "---\n\n"
        
        return output
    
    # ======================
    # SKILL PATCHES
    # ======================
    
    def create_skill_patch(self, name: str, rule: str, examples: str, 
                          priority: int) -> str:
        """Create a skill patch"""
        if not all([name, rule, examples]):
            return "❌ Please fill in all fields"
        
        try:
            examples_list = [ex.strip() for ex in examples.split("\n") if ex.strip()]
        except:
            return "❌ Error: Examples should be one per line"
        
        result = asyncio.run(self.server._create_skill_patch({
            "name": name,
            "rule": rule,
            "examples": examples_list,
            "priority": priority
        }))
        
        return result[0].text
    
    def apply_skill_patches(self, context: str) -> str:
        """Apply skill patches to a context"""
        if not context:
            return "❌ Please enter a context"
        
        result = asyncio.run(self.server._apply_skill_patches({
            "context": context
        }))
        
        return result[0].text
    
    def get_all_skill_patches(self) -> str:
        """Get all skill patches"""
        if not self.server.skill_patches:
            return "No skill patches created yet."
        
        output = "# 🎯 Skill Patches\n\n"
        sorted_patches = sorted(self.server.skill_patches, 
                               key=lambda p: p.priority, 
                               reverse=True)
        
        for i, patch in enumerate(sorted_patches, 1):
            output += f"## {i}. {patch.name}\n"
            output += f"**Priority:** {patch.priority}/10\n"
            output += f"**Rule:** {patch.rule}\n"
            output += f"**Usage Count:** {patch.usage_count}\n"
            output += f"**Success Rate:** {patch.success_rate:.1%}\n"
            output += f"**Examples:**\n"
            for ex in patch.examples:
                output += f"- {ex}\n"
            output += "\n---\n\n"
        
        return output
    
    # ======================
    # SELF-REVIEW
    # ======================
    
    def review_output(self, input_text: str, output_text: str, 
                     critique: str, improvements: str, rating: int) -> str:
        """Review an output"""
        if not all([input_text, output_text, critique]):
            return "❌ Please fill in required fields"
        
        result = asyncio.run(self.server._review_output({
            "input_text": input_text,
            "output_text": output_text,
            "critique": critique,
            "improvements": improvements,
            "rating": rating
        }))
        
        return result[0].text
    
    def get_all_reviews(self) -> str:
        """Get all output reviews"""
        cursor = self.server.conn.cursor()
        cursor.execute("""
            SELECT id, input_text, output_text, critique, improvements, rating, timestamp 
            FROM output_reviews 
            ORDER BY timestamp DESC 
            LIMIT 50
        """)
        reviews = cursor.fetchall()
        
        if not reviews:
            return "No reviews yet."
        
        output = "# 📊 Output Reviews\n\n"
        for review_id, inp, out, critique, improvements, rating, timestamp in reviews:
            output += f"## Review #{review_id} | Rating: {rating or 'N/A'}/10\n"
            output += f"**Input:** {inp}\n"
            output += f"**Output:** {out}\n"
            output += f"**Critique:** {critique}\n"
            if improvements:
                output += f"**Improvements:** {improvements}\n"
            output += f"**Time:** {timestamp}\n\n"
            output += "---\n\n"
        
        return output
    
    # ======================
    # STATISTICS
    # ======================
    
    def get_statistics(self) -> str:
        """Get learning statistics"""
        result = asyncio.run(self.server._get_learning_stats({}))
        return result[0].text
    
    def get_dashboard_stats(self) -> Tuple[str, str, str, str]:
        """Get stats for dashboard cards"""
        cursor = self.server.conn.cursor()
        
        # Knowledge count
        cursor.execute("SELECT COUNT(*) FROM knowledge")
        knowledge_count = cursor.fetchone()[0]
        
        # Experiences
        cursor.execute("SELECT COUNT(*) FROM experiences")
        total_exp = cursor.fetchone()[0]
        cursor.execute("SELECT COUNT(*) FROM experiences WHERE success = 1")
        success_exp = cursor.fetchone()[0]
        success_rate = (success_exp / total_exp * 100) if total_exp > 0 else 0
        
        # Skill patches
        patch_count = len(self.server.skill_patches)
        
        # Reviews
        cursor.execute("SELECT AVG(rating) FROM output_reviews WHERE rating IS NOT NULL")
        avg_rating = cursor.fetchone()[0] or 0
        
        return (
            f"📚 {knowledge_count}",
            f"📝 {total_exp} ({success_rate:.1f}% success)",
            f"🎯 {patch_count} patches",
            f"📊 {avg_rating:.1f}/10"
        )
    
    # ======================
    # CHAT INTERFACE
    # ======================
    
    def chat(self, message: str, history: List[Tuple[str, str]]) -> Tuple[str, List[Tuple[str, str]]]:
        """Chat interface with SIKE integration"""
        if not message:
            return "", history
        
        # Parse commands
        if message.startswith("/"):
            response = self.handle_command(message)
        else:
            # Regular chat - demonstrate knowledge search
            response = self.generate_response(message)
        
        history.append((message, response))
        return "", history
    
    def handle_command(self, command: str) -> str:
        """Handle special commands"""
        parts = command[1:].split(maxsplit=1)
        cmd = parts[0].lower()
        
        if cmd == "help":
            return """
**Available Commands:**

- `/search <query>` - Search knowledge base
- `/learn <topic>: <content>` - Store knowledge
- `/stats` - Show statistics
- `/patches` - Show skill patches
- `/experiences` - Show recent experiences
- `/help` - Show this help

Or just chat normally!
"""
        
        elif cmd == "search" and len(parts) > 1:
            query = parts[1]
            result = asyncio.run(self.server._search_knowledge({
                "query": query,
                "limit": 3
            }))
            return result[0].text
        
        elif cmd == "learn" and len(parts) > 1:
            try:
                topic, content = parts[1].split(":", 1)
                result = asyncio.run(self.server._store_knowledge({
                    "topic": topic.strip(),
                    "content": content.strip(),
                    "metadata": {}
                }))
                return result[0].text
            except:
                return "❌ Format: /learn <topic>: <content>"
        
        elif cmd == "stats":
            return self.get_statistics()
        
        elif cmd == "patches":
            return self.get_all_skill_patches()
        
        elif cmd == "experiences":
            cursor = self.server.conn.cursor()
            cursor.execute("""
                SELECT context, success, lesson_learned 
                FROM experiences 
                ORDER BY timestamp DESC 
                LIMIT 5
            """)
            experiences = cursor.fetchall()
            
            if not experiences:
                return "No experiences logged yet."
            
            output = "**Recent Experiences:**\n\n"
            for context, success, lesson in experiences:
                status = "✅" if success else "❌"
                output += f"{status} {context}\n💡 Lesson: {lesson}\n\n"
            return output
        
        else:
            return "❌ Unknown command. Type `/help` for available commands."
    
    def generate_response(self, message: str) -> str:
        """Generate response with knowledge integration"""
        # Search for relevant knowledge
        search_result = asyncio.run(self.server._search_knowledge({
            "query": message,
            "limit": 2
        }))
        
        # Check if we found relevant knowledge
        if "No relevant knowledge" not in search_result[0].text:
            return f"💡 **Based on my accumulated knowledge:**\n\n{search_result[0].text}\n\n*(This response uses SIKE's knowledge base)*"
        else:
            return f"I don't have specific knowledge about '{message}' yet. You can:\n\n- Use `/learn <topic>: <content>` to teach me\n- Use `/search` to search existing knowledge\n- Use `/help` for more commands"


def create_gradio_interface():
    """Create the Gradio interface"""
    app = SIKEGradioApp()
    
    # Custom CSS
    custom_css = """
    .stat-card {
        background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
        color: white;
        padding: 20px;
        border-radius: 10px;
        text-align: center;
        font-size: 24px;
        font-weight: bold;
    }
    """
    
    with gr.Blocks(title="SIKE - Self-Improving Knowledge Engine") as demo:
        
        gr.Markdown("""
        # 🧠 SIKE - Self-Improving Knowledge Engine
        ### Give your AI a brain that learns from experience
        
        SIKE enables persistent learning through knowledge storage, experience logging, 
        and self-improving skill patches. Every interaction makes it smarter!
        """)
        
        # Dashboard
        with gr.Tab("📊 Dashboard"):
            gr.Markdown("## Learning Statistics Overview")
            
            with gr.Row():
                knowledge_stat = gr.Textbox(label="Knowledge Base", 
                                           elem_classes="stat-card", interactive=False)
                experience_stat = gr.Textbox(label="Experiences", 
                                            elem_classes="stat-card", interactive=False)
                patches_stat = gr.Textbox(label="Skill Patches", 
                                         elem_classes="stat-card", interactive=False)
                rating_stat = gr.Textbox(label="Avg Rating", 
                                        elem_classes="stat-card", interactive=False)
            
            refresh_btn = gr.Button("🔄 Refresh Stats", variant="primary")
            
            detailed_stats = gr.Markdown()
            
            def update_dashboard():
                stats = app.get_dashboard_stats()
                detailed = app.get_statistics()
                return stats[0], stats[1], stats[2], stats[3], detailed
            
            refresh_btn.click(
                update_dashboard,
                outputs=[knowledge_stat, experience_stat, patches_stat, 
                        rating_stat, detailed_stats]
            )
            
            # Initial load
            demo.load(
                update_dashboard,
                outputs=[knowledge_stat, experience_stat, patches_stat, 
                        rating_stat, detailed_stats]
            )
        
        # Chat Interface
        with gr.Tab("💬 Chat with SIKE"):
            gr.Markdown("""
            Chat with SIKE! It will search its knowledge base to answer questions.
            
            **Try these commands:**
            - `/help` - Show available commands
            - `/search <query>` - Search knowledge
            - `/learn <topic>: <content>` - Teach SIKE something
            - `/stats` - View statistics
            """)
            
            chatbot = gr.Chatbot(height=500)
            msg = gr.Textbox(label="Message", placeholder="Ask a question or type /help...")
            clear = gr.Button("Clear Chat")
            
            msg.submit(app.chat, [msg, chatbot], [msg, chatbot])
            clear.click(lambda: [], None, chatbot)
        
        # Knowledge Management
        with gr.Tab("📚 Knowledge Base"):
            with gr.Row():
                with gr.Column():
                    gr.Markdown("### Store New Knowledge")
                    topic_input = gr.Textbox(label="Topic", placeholder="e.g., Python Best Practices")
                    content_input = gr.Textbox(label="Content", 
                                               placeholder="e.g., Always use type hints", 
                                               lines=3)
                    metadata_input = gr.Textbox(label="Metadata (JSON)", 
                                               placeholder='{"category": "programming"}',
                                               lines=2)
                    store_btn = gr.Button("💾 Store Knowledge", variant="primary")
                    store_output = gr.Textbox(label="Result", lines=3)
                    
                    store_btn.click(
                        app.store_knowledge,
                        inputs=[topic_input, content_input, metadata_input],
                        outputs=store_output
                    )
                
                with gr.Column():
                    gr.Markdown("### Search Knowledge")
                    search_input = gr.Textbox(label="Search Query", 
                                             placeholder="e.g., Python type hints")
                    search_limit = gr.Slider(1, 10, value=5, step=1, 
                                           label="Max Results")
                    search_btn = gr.Button("🔍 Search", variant="primary")
                    search_output = gr.Markdown()
                    
                    search_btn.click(
                        app.search_knowledge,
                        inputs=[search_input, search_limit],
                        outputs=search_output
                    )
            
            gr.Markdown("### All Knowledge Entries")
            view_knowledge_btn = gr.Button("📖 View All Knowledge")
            all_knowledge_output = gr.Markdown()
            
            view_knowledge_btn.click(
                app.get_all_knowledge,
                outputs=all_knowledge_output
            )
        
        # Experience Logging
        with gr.Tab("📝 Experience Log"):
            with gr.Row():
                with gr.Column():
                    gr.Markdown("### Log New Experience")
                    context_input = gr.Textbox(label="Context", 
                                              placeholder="What was happening?")
                    attempt_input = gr.Textbox(label="Attempt", 
                                              placeholder="What did you try?")
                    result_input = gr.Textbox(label="Result", 
                                             placeholder="What happened?")
                    lesson_input = gr.Textbox(label="Lesson Learned", 
                                            placeholder="What did you learn?")
                    success_input = gr.Checkbox(label="Was it successful?", value=True)
                    log_btn = gr.Button("📝 Log Experience", variant="primary")
                    log_output = gr.Textbox(label="Result", lines=3)
                    
                    log_btn.click(
                        app.log_experience,
                        inputs=[context_input, attempt_input, result_input, 
                               lesson_input, success_input],
                        outputs=log_output
                    )
                
                with gr.Column():
                    gr.Markdown("### Search Experiences")
                    exp_search_input = gr.Textbox(label="Search Query")
                    exp_search_limit = gr.Slider(1, 10, value=5, step=1, 
                                                label="Max Results")
                    exp_search_btn = gr.Button("🔍 Search Experiences", variant="primary")
                    exp_search_output = gr.Markdown()
                    
                    exp_search_btn.click(
                        app.search_experiences,
                        inputs=[exp_search_input, exp_search_limit],
                        outputs=exp_search_output
                    )
            
            gr.Markdown("### All Experiences")
            view_exp_btn = gr.Button("📋 View All Experiences")
            all_exp_output = gr.Markdown()
            
            view_exp_btn.click(
                app.get_all_experiences,
                outputs=all_exp_output
            )
        
        # Skill Patches
        with gr.Tab("🎯 Skill Patches"):
            with gr.Row():
                with gr.Column():
                    gr.Markdown("### Create Skill Patch")
                    patch_name = gr.Textbox(label="Patch Name", 
                                           placeholder="e.g., code_comments_rule")
                    patch_rule = gr.Textbox(label="Rule", 
                                           placeholder="e.g., Always include comments in code",
                                           lines=2)
                    patch_examples = gr.Textbox(label="Examples (one per line)", 
                                               placeholder="# Initialize variable\n# Process data",
                                               lines=4)
                    patch_priority = gr.Slider(0, 10, value=5, step=1, 
                                              label="Priority")
                    create_patch_btn = gr.Button("🎯 Create Patch", variant="primary")
                    create_patch_output = gr.Textbox(label="Result", lines=3)
                    
                    create_patch_btn.click(
                        app.create_skill_patch,
                        inputs=[patch_name, patch_rule, patch_examples, patch_priority],
                        outputs=create_patch_output
                    )
                
                with gr.Column():
                    gr.Markdown("### Apply Patches")
                    apply_context = gr.Textbox(label="Context", 
                                              placeholder="e.g., Writing Python code for beginners",
                                              lines=3)
                    apply_btn = gr.Button("⚡ Apply Patches", variant="primary")
                    apply_output = gr.Markdown()
                    
                    apply_btn.click(
                        app.apply_skill_patches,
                        inputs=apply_context,
                        outputs=apply_output
                    )
            
            gr.Markdown("### All Skill Patches")
            view_patches_btn = gr.Button("📜 View All Patches")
            all_patches_output = gr.Markdown()
            
            view_patches_btn.click(
                app.get_all_skill_patches,
                outputs=all_patches_output
            )
        
        # Self-Review
        with gr.Tab("🔍 Self-Review"):
            gr.Markdown("### Review Output Quality")
            review_input = gr.Textbox(label="Input/Question", 
                                     placeholder="What was asked?")
            review_output = gr.Textbox(label="Output/Answer", 
                                      placeholder="What was provided?", 
                                      lines=3)
            review_critique = gr.Textbox(label="Critique", 
                                        placeholder="What could be better?", 
                                        lines=3)
            review_improvements = gr.Textbox(label="Improvements", 
                                           placeholder="Specific suggestions", 
                                           lines=3)
            review_rating = gr.Slider(1, 10, value=5, step=1, label="Rating")
            review_btn = gr.Button("📊 Submit Review", variant="primary")
            review_result = gr.Textbox(label="Result", lines=2)
            
            review_btn.click(
                app.review_output,
                inputs=[review_input, review_output, review_critique, 
                       review_improvements, review_rating],
                outputs=review_result
            )
            
            gr.Markdown("### All Reviews")
            view_reviews_btn = gr.Button("📋 View All Reviews")
            all_reviews_output = gr.Markdown()
            
            view_reviews_btn.click(
                app.get_all_reviews,
                outputs=all_reviews_output
            )
        
        # About
        with gr.Tab("ℹ️ About"):
            gr.Markdown("""
            ## About SIKE
            
            **Self-Improving Knowledge Engine** is an MCP server that enables persistent 
            learning and self-improvement for AI agents.
            
            ### Key Features
            
            - **📚 Knowledge Base**: Store and search information with semantic search
            - **📝 Experience Logging**: Track what works and what doesn't
            - **🎯 Skill Patches**: Create behavioral rules that improve responses
            - **🔍 Self-Review**: Agents can critique and improve their own outputs
            - **📊 Statistics**: Track learning progress over time
            
            ### How It Works
            
            1. **Store Knowledge**: Add information to the persistent knowledge base
            2. **Log Experiences**: Record attempts, results, and lessons learned
            3. **Create Patches**: Define rules that modify future behavior
            4. **Apply Patches**: Automatically improve responses using learned patterns
            5. **Self-Evaluate**: Review outputs and create improvements
            
            ### Technology Stack
            
            - **ChromaDB**: Vector search for semantic knowledge retrieval
            - **SQLite**: Structured data storage
            - **Gradio**: Beautiful web interface
            - **Python**: Core implementation
            
            ### Created For
            
            HuggingFace MCP Hackathon 2024
            
            ### Links
            
            - [GitHub Repository](#)
            - [Documentation](#)
            - [MCP Protocol](https://modelcontextprotocol.io)
            
            ---
            
            Made with 🧠 by [Your Name]
            """)
    
    return demo


if __name__ == "__main__":
    print("🚀 Starting SIKE Gradio Interface...")
    print("📝 Data will be stored in: ./sike_data/")
    print("🌐 Opening web interface...\n")
    
    demo = create_gradio_interface()
    demo.launch(
        server_name="0.0.0.0",
        server_port=7860,
        share=False,  # Set to True for public link
        show_error=True
    )