Mercury Notebook前端RAG实践

"""Mercury Notebook-to-Web Conversion and RAG Frontend Simulation==============================================================Simulates Mercury framework's core functionality:- YAML config parsing for widgets- Notebook cell to web component conversion- Interactive widget system (file upload, text input, slider, select)- Chat component with message flow- RAG pipeline (upload -> chunk -> retrieve -> generate -> display)- Docker deployment and kernel isolation""" import jsonimport hashlibimport mathimport refrom collections import OrderedDict try:    import numpy as np    HAS_NUMPY = Trueexcept ImportError:    HAS_NUMPY = False  # ============================================================# 1. Mercury YAML Config Parser# ============================================================class MercuryYAMLParser:    """Parse YAML-like configuration for Mercury widgets."""     def __init__(self):        self.widget_types = {            "text": self._parse_text_widget,            "textarea": self._parse_textarea_widget,            "slider": self._parse_slider_widget,            "select": self._parse_select_widget,            "checkbox": self._parse_checkbox_widget,            "file_upload": self._parse_file_upload_widget,            "numeric": self._parse_numeric_widget,        }     def parse_yaml_config(self, yaml_string):        """Parse YAML configuration string into widget definitions."""        widgets = OrderedDict()        current_widget = None        current_key = None         for line in yaml_string.strip().split("\n"):            stripped = line.strip()            if not stripped or stripped.startswith("#"):                continue             indent = len(line) - len(line.lstrip())             if indent == 0 and stripped.endswith(":"):                widget_name = stripped[:-1].strip()                current_widget = {"name": widget_name, "params": {}}                widgets[widget_name] = current_widget                current_key = None            elif indent > 0 and current_widget is not None:                if ":" in stripped:                    key, value = stripped.split(":", 1)                    key = key.strip()                    value = value.strip()                    current_widget["params"][key] = self._cast_value(value)                    current_key = key         validated = {}        for name, wdef in widgets.items():            wtype = wdef["params"].get("type", "text")            if wtype in self.widget_types:                validated[name] = self.widget_types[wtype](wdef["params"])            else:                validated[name] = self._parse_text_widget(wdef["params"])         return validated     def _cast_value(self, value):        """Cast string value to appropriate Python type."""        if value.lower() == "true":            return True        if value.lower() == "false":            return False        try:            return int(value)        except ValueError:            pass        try:            return float(value)        except ValueError:            pass        if value.startswith("[") and value.endswith("]"):            inner = value[1:-1]            return [v.strip().strip('"').strip("'") for v in inner.split(",") if v.strip()]        return value.strip('"').strip("'")     def _parse_text_widget(self, params):        return {            "type": "text",            "label": params.get("label", "Text Input"),            "value": params.get("value", ""),            "max": params.get("max", 100),        }     def _parse_textarea_widget(self, params):        return {            "type": "textarea",            "label": params.get("label", "Text Area"),            "value": params.get("value", ""),            "rows": params.get("rows", 5),        }     def _parse_slider_widget(self, params):        return {            "type": "slider",            "label": params.get("label", "Slider"),            "min": params.get("min", 0),            "max": params.get("max", 100),            "step": params.get("step", 1),            "value": params.get("value", 50),        }     def _parse_select_widget(self, params):        return {            "type": "select",            "label": params.get("label", "Select"),            "choices": params.get("choices", []),            "value": params.get("value", ""),            "multi": params.get("multi", False),        }     def _parse_checkbox_widget(self, params):        return {            "type": "checkbox",            "label": params.get("label", "Checkbox"),            "value": params.get("value", False),        }     def _parse_file_upload_widget(self, params):        return {            "type": "file_upload",            "label": params.get("label", "Upload File"),            "accept": params.get("accept", "*"),            "max_size": params.get("max_size", 10485760),        }     def _parse_numeric_widget(self, params):        return {            "type": "numeric",            "label": params.get("label", "Number"),            "value": params.get("value", 0),            "min": params.get("min", 0),            "max": params.get("max", 1000),            "step": params.get("step", 1),        }  # ============================================================# 2. Notebook-to-Web Converter# ============================================================class NotebookCell:    """Represents a single Jupyter notebook cell."""     def __init__(self, cell_type, source, metadata=None):        self.cell_type = cell_type  # "code" or "markdown"        self.source = source        self.metadata = metadata or {}        self.output = None        self.web_component = None     def to_web_component(self):        """Convert cell to a web component representation."""        if self.cell_type == "markdown":            self.web_component = {                "component": "MarkdownBlock",                "html": self._markdown_to_html(self.source),                "interactive": False,            }        elif self.cell_type == "code":            hide_code = self.metadata.get("hide_code", False)            hide_output = self.metadata.get("hide_output", False)            self.web_component = {                "component": "CodeBlock",                "code": self.source if not hide_code else None,                "output": self.output if not hide_output else None,                "interactive": True,                "cell_id": self.metadata.get("cell_id", "unknown"),            }        return self.web_component     def _markdown_to_html(self, md_text):        """Simple markdown to HTML conversion."""        html = md_text        html = re.sub(r"^### (.+)$", r"<h3>\1</h3>", html, flags=re.MULTILINE)        html = re.sub(r"^## (.+)$", r"<h2>\1</h2>", html, flags=re.MULTILINE)        html = re.sub(r"^# (.+)$", r"<h1>\1</h1>", html, flags=re.MULTILINE)        html = re.sub(r"\*\*(.+?)\*\*", r"<strong>\1</strong>", html)        html = re.sub(r"\*(.+?)\*", r"<em>\1</em>", html)        html = re.sub(r"`(.+?)`", r"<code>\1</code>", html)        return html  class NotebookToWebConverter:    """Convert a full Jupyter notebook to a Mercury web application."""     def __init__(self):        self.cells = []        self.yaml_parser = MercuryYAMLParser()        self.widgets = {}     def load_notebook(self, notebook_dict):        """Load notebook from dictionary representation."""        for cell_data in notebook_dict.get("cells", []):            cell = NotebookCell(                cell_type=cell_data.get("cell_type", "code"),                source="".join(cell_data.get("source", [])),                metadata=cell_data.get("metadata", {}),            )            self.cells.append(cell)        return len(self.cells)     def parse_widget_config(self, yaml_config):        """Parse widget configuration from YAML."""        self.widgets = self.yaml_parser.parse_yaml_config(yaml_config)        return self.widgets     def convert(self):        """Convert all cells to web components."""        components = []        for i, cell in enumerate(self.cells):            if cell.cell_type == "code":                cell.output = self._simulate_execution(cell.source)            component = cell.to_web_component()            component["order"] = i            components.append(component)         app_structure = {            "title": "Mercury Web App",            "components": components,            "widgets": self.widgets,            "total_cells": len(self.cells),            "interactive_cells": sum(1 for c in components if c.get("interactive")),        }        return app_structure     def _simulate_execution(self, code):        """Simulate code cell execution and return output."""        if "print(" in code:            match = re.search(r'print\(["\'](.+?)["\']\)', code)            if match:                return {"type": "stdout", "data": match.group(1)}        if "import" in code:            return {"type": "stdout", "data": "[Module imported successfully]"}        if "=" in code and "==" not in code:            return {"type": "stdout", "data": "[Variable assigned]"}        return {"type": "stdout", "data": "[Cell executed]"}  # ============================================================# 3. Widget System# ============================================================class WidgetManager:    """Manage interactive widgets and their state."""     def __init__(self):        self.widgets = {}        self.state_history = []     def register_widget(self, name, widget_config):        """Register a widget with its configuration."""        self.widgets[name] = {            "config": widget_config,            "current_value": widget_config.get("value", None),            "change_count": 0,        }     def update_value(self, name, new_value):        """Update widget value and record state change."""        if name not in self.widgets:            raise ValueError(f"Widget '{name}' not registered")         old_value = self.widgets[name]["current_value"]        self.widgets[name]["current_value"] = new_value        self.widgets[name]["change_count"] += 1         self.state_history.append({            "widget": name,            "old_value": old_value,            "new_value": new_value,            "timestamp": len(self.state_history),        })         return {"changed": old_value != new_value, "widget": name}     def get_value(self, name):        """Get current value of a widget."""        if name not in self.widgets:            raise ValueError(f"Widget '{name}' not registered")        return self.widgets[name]["current_value"]     def get_all_values(self):        """Get all widget values as a dictionary."""        return {name: w["current_value"] for name, w in self.widgets.items()}     def simulate_interaction(self, name, input_value):        """Simulate user interaction with a widget."""        widget = self.widgets.get(name)        if not widget:            return {"error": f"Widget '{name}' not found"}         wtype = widget["config"]["type"]        validated = self._validate_input(wtype, input_value, widget["config"])        result = self.update_value(name, validated)        result["trigger_rerun"] = True        return result     def _validate_input(self, widget_type, value, config):        """Validate and coerce input based on widget type."""        if widget_type == "slider":            val = float(value)            val = max(config.get("min", 0), min(config.get("max", 100), val))            return val        elif widget_type == "select":            choices = config.get("choices", [])            if value in choices:                return value            return choices[0] if choices else value        elif widget_type == "checkbox":            return bool(value)        elif widget_type == "file_upload":            max_size = config.get("max_size", 10485760)            if isinstance(value, dict):                if value.get("size", 0) > max_size:                    raise ValueError("File exceeds maximum size")                return value            return {"name": str(value), "size": 0}        return value  # ============================================================# 4. Chat Component# ============================================================class ChatMessage:    """Represents a single chat message."""     def __init__(self, role, content, metadata=None):        self.role = role  # "user", "assistant", "system"        self.content = content        self.metadata = metadata or {}        self.timestamp = hashlib.md5(f"{role}:{content}".encode()).hexdigest()[:8]     def to_dict(self):        return {            "role": self.role,            "content": self.content,            "metadata": self.metadata,            "msg_id": self.timestamp,        }  class MercuryChatComponent:    """Simulate Mercury chat component with message flow processing."""     def __init__(self, system_prompt=None):        self.messages = []        self.system_prompt = system_prompt or "You are a helpful assistant."        self.is_processing = False        self.message_handlers = []     def add_system_message(self):        """Initialize chat with system message."""        sys_msg = ChatMessage("system", self.system_prompt)        self.messages.append(sys_msg)        return sys_msg.to_dict()     def send_message(self, user_input, context=None):        """Process user message and generate response."""        user_msg = ChatMessage("user", user_input, {"context": context})        self.messages.append(user_msg)         self.is_processing = True        response_content = self._generate_response(user_input, context)        self.is_processing = False         assistant_msg = ChatMessage(            "assistant",            response_content,            {"sources": context.get("sources", []) if context else []},        )        self.messages.append(assistant_msg)         return {            "user_message": user_msg.to_dict(),            "assistant_message": assistant_msg.to_dict(),            "total_messages": len(self.messages),        }     def _generate_response(self, user_input, context):        """Generate simulated response based on input and context."""        if context and "documents" in context:            relevant = self._retrieve_relevant(user_input, context["documents"])            if relevant:                summary = " ".join([d["content"][:80] for d in relevant[:2]])                return f"Based on the retrieved documents: {summary}..."         if "hello" in user_input.lower() or "hi" in user_input.lower():            return "Hello! I am a Mercury chat assistant. How can I help you?"         return f"I received your message: '{user_input}'. This is a simulated response from the Mercury chat component."     def _retrieve_relevant(self, query, documents):        """Simple keyword-based retrieval simulation."""        query_terms = set(query.lower().split())        scored = []        for doc in documents:            doc_terms = set(doc["content"].lower().split())            overlap = len(query_terms & doc_terms)            if overlap > 0:                scored.append((overlap, doc))        scored.sort(key=lambda x: x[0], reverse=True)        return [doc for _, doc in scored[:3]]     def get_history(self):        """Return full message history."""        return [msg.to_dict() for msg in self.messages]     def register_handler(self, event_type, handler):        """Register a message event handler."""        self.message_handlers.append({"event": event_type, "handler": handler})  # ============================================================# 5. RAG Frontend Pipeline# ============================================================class DocumentChunker:    """Split documents into chunks for RAG processing."""     def __init__(self, chunk_size=200, chunk_overlap=50):        self.chunk_size = chunk_size        self.chunk_overlap = chunk_overlap     def chunk_document(self, text, doc_id="doc_0"):        """Split document text into overlapping chunks."""        chunks = []        start = 0        chunk_idx = 0         while start < len(text):            end = start + self.chunk_size            chunk_text = text[start:end]             if chunk_text.strip():                chunks.append({                    "chunk_id": f"{doc_id}_chunk_{chunk_idx}",                    "doc_id": doc_id,                    "content": chunk_text.strip(),                    "start_char": start,                    "end_char": min(end, len(text)),                    "index": chunk_idx,                })                chunk_idx += 1             start = end - self.chunk_overlap            if start >= len(text):                break         return chunks  class SimpleVectorStore:    """Simple vector store using numpy for similarity search."""     def __init__(self, dim=64):        self.dim = dim        self.vectors = []        self.metadata = []        if HAS_NUMPY:            self.rng = np.random.RandomState(42)     def _text_to_vector(self, text):        """Convert text to a simple vector representation."""        if HAS_NUMPY:            vec = np.zeros(self.dim)            for i, char in enumerate(text.lower()):                idx = ord(char) % self.dim                vec[idx] += 1.0 / (1.0 + i * 0.01)            norm = np.linalg.norm(vec)            if norm > 0:                vec = vec / norm            return vec        else:            vec = [0.0] * self.dim            for i, char in enumerate(text.lower()):                idx = ord(char) % self.dim                vec[idx] += 1.0 / (1.0 + i * 0.01)            norm = sum(v * v for v in vec) ** 0.5            if norm > 0:                vec = [v / norm for v in vec]            return vec     def add_chunks(self, chunks):        """Add document chunks to the vector store."""        for chunk in chunks:            vec = self._text_to_vector(chunk["content"])            self.vectors.append(vec)            self.metadata.append(chunk)        return len(chunks)     def search(self, query, top_k=3):        """Search for most similar chunks to the query."""        query_vec = self._text_to_vector(query)         scores = []        for i, doc_vec in enumerate(self.vectors):            if HAS_NUMPY:                similarity = float(np.dot(query_vec, doc_vec))            else:                similarity = sum(a * b for a, b in zip(query_vec, doc_vec))            scores.append((similarity, i))         scores.sort(key=lambda x: x[0], reverse=True)        results = []        for score, idx in scores[:top_k]:            result = dict(self.metadata[idx])            result["similarity_score"] = round(score, 4)            results.append(result)         return results  class RAGPipeline:    """Complete RAG pipeline: upload -> chunk -> retrieve -> generate -> display."""     def __init__(self):        self.chunker = DocumentChunker(chunk_size=200, chunk_overlap=50)        self.vector_store = SimpleVectorStore(dim=64)        self.documents = {}        self.query_history = []     def upload_document(self, doc_id, content, metadata=None):        """Upload and process a document into the RAG system."""        self.documents[doc_id] = {            "content": content,            "metadata": metadata or {},            "char_count": len(content),        }         chunks = self.chunker.chunk_document(content, doc_id)        added = self.vector_store.add_chunks(chunks)         return {            "doc_id": doc_id,            "char_count": len(content),            "chunks_created": added,            "status": "indexed",        }     def query(self, question, top_k=3):        """Process a RAG query: retrieve relevant chunks and generate answer."""        retrieved = self.vector_store.search(question, top_k=top_k)         context_parts = [r["content"] for r in retrieved]        context_text = " ".join(context_parts)         answer = self._generate_answer(question, context_text, retrieved)         result = {            "question": question,            "answer": answer,            "sources": [                {"chunk_id": r["chunk_id"], "score": r["similarity_score"], "content": r["content"][:100]}                for r in retrieved            ],            "total_chunks_searched": len(self.vector_store.metadata),        }         self.query_history.append(result)        return result     def _generate_answer(self, question, context, retrieved_chunks):        """Generate answer from retrieved context (simulated)."""        if not retrieved_chunks:            return "I could not find relevant information to answer your question."         source_refs = ", ".join([r["chunk_id"] for r in retrieved_chunks[:3]])        key_phrases = context[:200]         return (            f"Based on the retrieved context, here is the answer to '{question}': "            f"The relevant information indicates: {key_phrases}... "            f"[Sources: {source_refs}]"        )     def get_stats(self):        """Get RAG pipeline statistics."""        return {            "total_documents": len(self.documents),            "total_chunks": len(self.vector_store.metadata),            "total_queries": len(self.query_history),            "documents": {                doc_id: {"chars": info["char_count"]}                for doc_id, info in self.documents.items()            },        }  # ============================================================# 6. Deployment Manager# ============================================================class KernelSession:    """Represents an isolated kernel session for a user."""     def __init__(self, session_id, user_id):        self.session_id = session_id        self.user_id = user_id        self.variables = {}        self.is_active = True        self.execution_count = 0        self.memory_usage_mb = 0.0     def execute(self, code):        """Simulate code execution in isolated kernel."""        self.execution_count += 1        self.memory_usage_mb += 0.5         if "=" in code and "==" not in code:            parts = code.split("=", 1)            var_name = parts[0].strip()            var_value = parts[1].strip()            self.variables[var_name] = var_value         return {            "session_id": self.session_id,            "execution_count": self.execution_count,            "status": "ok",            "memory_mb": self.memory_usage_mb,        }     def get_state(self):        """Return current kernel state."""        return {            "session_id": self.session_id,            "user_id": self.user_id,            "is_active": self.is_active,            "variables": list(self.variables.keys()),            "execution_count": self.execution_count,            "memory_mb": self.memory_usage_mb,        }  class DeploymentManager:    """Simulate Docker deployment and multi-user management."""     def __init__(self, app_name="mercury-app"):        self.app_name = app_name        self.sessions = {}        self.deploy_config = {}        self.user_quota = {"max_sessions_per_user": 3, "max_memory_mb": 512}        self.is_deployed = False     def configure_deployment(self, config):        """Configure Docker deployment settings."""        self.deploy_config = {            "app_name": config.get("app_name", self.app_name),            "notebook_path": config.get("notebook_path", "app.ipynb"),            "port": config.get("port", 5006),            "workers": config.get("workers", 4),            "docker_image": config.get("docker_image", "mljar/mercury:latest"),            "enable_auth": config.get("enable_auth", False),            "ssl": config.get("ssl", False),        }        return self.deploy_config     def deploy(self):        """Simulate Docker deployment."""        if not self.deploy_config:            self.configure_deployment({})         self.is_deployed = True        return {            "status": "deployed",            "app_name": self.deploy_config["app_name"],            "url": f"http://localhost:{self.deploy_config['port']}",            "docker_image": self.deploy_config["docker_image"],            "workers": self.deploy_config["workers"],        }     def create_session(self, user_id):        """Create a new isolated kernel session for a user."""        user_sessions = [s for s in self.sessions.values() if s.user_id == user_id]        if len(user_sessions) >= self.user_quota["max_sessions_per_user"]:            return {"error": "Maximum sessions reached for user", "user_id": user_id}         session_id = hashlib.md5(f"{user_id}:{len(self.sessions)}".encode()).hexdigest()[:12]        session = KernelSession(session_id, user_id)        self.sessions[session_id] = session         return {            "session_id": session_id,            "user_id": user_id,            "status": "created",            "active_sessions": len(self.sessions),        }     def execute_in_session(self, session_id, code):        """Execute code in a specific kernel session."""        if session_id not in self.sessions:            return {"error": f"Session '{session_id}' not found"}         session = self.sessions[session_id]        if not session.is_active:            return {"error": "Session is no longer active"}         return session.execute(code)     def terminate_session(self, session_id):        """Terminate a kernel session."""        if session_id in self.sessions:            self.sessions[session_id].is_active = False            return {"session_id": session_id, "status": "terminated"}        return {"error": "Session not found"}     def get_deployment_status(self):        """Get current deployment status."""        active = sum(1 for s in self.sessions.values() if s.is_active)        total_memory = sum(s.memory_usage_mb for s in self.sessions.values())         return {            "app_name": self.app_name,            "is_deployed": self.is_deployed,            "config": self.deploy_config,            "total_sessions": len(self.sessions),            "active_sessions": active,            "total_memory_mb": round(total_memory, 2),            "users": list(set(s.user_id for s in self.sessions.values())),        }  # ============================================================# Main Execution - Full Pipeline Demo# ============================================================def main():    print("=" * 60)    print("Mercury Notebook-to-Web Conversion & RAG Frontend Demo")    print("=" * 60)     # --- Step 1: YAML Config Parsing ---    print("\n[Step 1] Mercury YAML Config Parser")    print("-" * 40)     yaml_config = """query_text:    type: text    label: Search Query    value: What is RAGtop_k_slider:    type: slider    label: Number of Results    min: 1    max: 10    step: 1    value: 3model_select:    type: select    label: LLM Model    choices: [gpt-4, gpt-3.5, llama2, mistral]    value: gpt-4file_input:    type: file_upload    label: Upload Document    accept: .pdf,.txt,.md    max_size: 5242880enable_cache:    type: checkbox    label: Enable Response Cache    value: true"""     parser = MercuryYAMLParser()    widgets = parser.parse_yaml_config(yaml_config)    print(f"Parsed {len(widgets)} widget definitions:")    for name, wdef in widgets.items():        print(f"  - {name}: type={wdef['type']}, label={wdef.get('label', 'N/A')}")     # --- Step 2: Notebook-to-Web Conversion ---    print("\n[Step 2] Notebook-to-Web Converter")    print("-" * 40)     notebook = {        "cells": [            {                "cell_type": "markdown",                "source": ["# Mercury RAG Application\n", "Interactive document Q&A system"],                "metadata": {},            },            {                "cell_type": "code",                "source": ["import numpy as np\n", "print('Modules loaded')"],                "metadata": {"hide_code": True, "cell_id": "cell_import"},            },            {                "cell_type": "code",                "source": ["query = 'What is retrieval augmented generation'\n", "top_k = 3"],                "metadata": {"hide_code": False, "cell_id": "cell_params"},            },            {                "cell_type": "code",                "source": ["results = search_documents(query, top_k)\n", "print('Search complete')"],                "metadata": {"hide_code": False, "cell_id": "cell_search"},            },        ]    }     converter = NotebookToWebConverter()    cell_count = converter.load_notebook(notebook)    converter.parse_widget_config(yaml_config)    app_structure = converter.convert()     print(f"Converted {cell_count} cells to web components")    print(f"Interactive cells: {app_structure['interactive_cells']}")    print(f"Total widgets: {len(app_structure['widgets'])}")     # --- Step 3: Widget System ---    print("\n[Step 3] Widget System - Interactive State Management")    print("-" * 40)     widget_mgr = WidgetManager()    for name, config in widgets.items():        widget_mgr.register_widget(name, config)     result = widget_mgr.simulate_interaction("top_k_slider", 5)    print(f"Slider interaction: {result}")     result = widget_mgr.simulate_interaction("model_select", "llama2")    print(f"Select interaction: {result}")     result = widget_mgr.simulate_interaction("query_text", "Explain vector embeddings")    print(f"Text interaction: {result}")     all_values = widget_mgr.get_all_values()    print(f"Current widget state: {json.dumps(all_values, indent=2, ensure_ascii=False)}")     # --- Step 4: Chat Component ---    print("\n[Step 4] Mercury Chat Component")    print("-" * 40)     chat = MercuryChatComponent(system_prompt="You are a RAG assistant for document Q&A.")    sys_msg = chat.add_system_message()    print(f"System message initialized: {sys_msg['role']}")     response1 = chat.send_message("Hello, what can you help me with?")    print(f"User: {response1['user_message']['content']}")    print(f"Assistant: {response1['assistant_message']['content']}")     response2 = chat.send_message(        "What is retrieval augmented generation?",        context={            "documents": [                {"content": "RAG combines retrieval with generation to produce grounded answers from documents"},                {"content": "Vector embeddings enable semantic search over document chunks"},            ],            "sources": ["doc1.pdf", "doc2.pdf"],        },    )    print(f"User: {response2['user_message']['content']}")    print(f"Assistant: {response2['assistant_message']['content'][:100]}...")    print(f"Total messages in history: {response2['total_messages']}")     # --- Step 5: RAG Pipeline ---    print("\n[Step 5] RAG Frontend Pipeline")    print("-" * 40)     rag = RAGPipeline()     doc1 = (        "Retrieval Augmented Generation (RAG) is a technique that combines information retrieval "        "with large language model generation. It first retrieves relevant document chunks from a "        "knowledge base, then uses these chunks as context for the language model to generate "        "accurate and grounded responses. RAG reduces hallucination by grounding generation in "        "retrieved facts."    )     doc2 = (        "Mercury is a Python framework that allows building web applications directly from "        "Jupyter Notebooks. It uses YAML configuration to define interactive widgets like sliders, "        "text inputs, and file uploads. When a user interacts with a widget, Mercury re-executes "        "the notebook with updated parameters, making it ideal for rapid prototyping of data "        "applications and RAG interfaces."    )     doc3 = (        "Docker deployment of Mercury applications provides kernel isolation between users. "        "Each user session runs in a separate kernel, ensuring that variable state and execution "        "do not interfere across users. The deployment manager handles session lifecycle, memory "        "limits, and concurrent access management for production deployments."    )     upload1 = rag.upload_document("rag_intro", doc1, {"source": "textbook"})    upload2 = rag.upload_document("mercury_guide", doc2, {"source": "tutorial"})    upload3 = rag.upload_document("deployment_guide", doc3, {"source": "manual"})    print(f"Document 1: {upload1['chunks_created']} chunks created")    print(f"Document 2: {upload2['chunks_created']} chunks created")    print(f"Document 3: {upload3['chunks_created']} chunks created")     query_result = rag.query("How does Mercury build web applications from notebooks?")    print(f"\nQuery: {query_result['question']}")    print(f"Answer: {query_result['answer'][:150]}...")    print(f"Sources found: {len(query_result['sources'])}")    for src in query_result["sources"]:        print(f"  - {src['chunk_id']} (score: {src['score']})")     stats = rag.get_stats()    print(f"\nRAG Stats: {stats['total_documents']} docs, {stats['total_chunks']} chunks, {stats['total_queries']} queries")     # --- Step 6: Deployment Manager ---    print("\n[Step 6] Deployment Manager - Docker & Multi-User")    print("-" * 40)     deployer = DeploymentManager(app_name="rag-mercury-app")    config = deployer.configure_deployment({        "app_name": "rag-mercury-app",        "notebook_path": "rag_app.ipynb",        "port": 8080,        "workers": 4,        "docker_image": "mljar/mercury:latest",        "enable_auth": True,        "ssl": True,    })    print(f"Deployment configured: {config['app_name']} on port {config['port']}")     deploy_result = deployer.deploy()    print(f"Deployed at: {deploy_result['url']}")    print(f"Docker image: {deploy_result['docker_image']}")     session1 = deployer.create_session("user_alice")    session2 = deployer.create_session("user_bob")    session3 = deployer.create_session("user_alice")    print(f"\nSession 1 (Alice): {session1['session_id']}")    print(f"Session 2 (Bob): {session2['session_id']}")    print(f"Session 3 (Alice): {session3['session_id']}")     exec_result = deployer.execute_in_session(session1["session_id"], "query = 'test query'")    print(f"\nExecution in Alice's session: count={exec_result['execution_count']}, memory={exec_result['memory_mb']}MB")     status = deployer.get_deployment_status()    print(f"\nDeployment Status:")    print(f"  App: {status['app_name']}")    print(f"  Deployed: {status['is_deployed']}")    print(f"  Active sessions: {status['active_sessions']}")    print(f"  Total memory: {status['total_memory_mb']}MB")    print(f"  Users: {status['users']}")     print("\n" + "=" * 60)    print("Demo complete. All Mercury components simulated successfully.")    print("=" * 60)  if __name__ == "__main__":    main()