edge-customer-service/
├── src/
│   ├── __init__.py
│   ├── config.py
│   ├── models/
│   │   ├── __init__.py
│   │   ├── slm_engine.py        # Local SLM inference
│   │   ├── quantization.py      # Model optimization
│   │   └── model_loader.py      # GGUF model loading
│   ├── rag/
│   │   ├── __init__.py
│   │   ├── local_vectordb.py    # SQLite-based vectors
│   │   ├── embeddings.py        # Local embeddings
│   │   └── retriever.py         # Local retrieval
│   ├── routing/
│   │   ├── __init__.py
│   │   ├── classifier.py        # Query classification
│   │   └── escalation.py        # Escalation logic
│   ├── context/
│   │   ├── __init__.py
│   │   ├── memory.py            # Conversation memory
│   │   └── personalization.py   # User preferences
│   ├── privacy/
│   │   ├── __init__.py
│   │   ├── pii_filter.py        # Local PII detection
│   │   └── data_retention.py    # Data lifecycle
│   └── app/
│       ├── __init__.py
│       ├── main.py              # FastAPI/Desktop app
│       └── ui.py                # Chat interface
├── models/                       # Downloaded GGUF models
├── data/                        # Local knowledge base
└── requirements.txt

# src/config.py
from pydantic_settings import BaseSettings
from pathlib import Path
from typing import Optional, List

class Settings(BaseSettings):
    # Model Settings
    model_path: Path = Path("./models/phi-3-mini-4k-instruct.Q4_K_M.gguf")
    context_length: int = 4096
    max_tokens: int = 512
    temperature: float = 0.7

    # Hardware
    n_gpu_layers: int = 0  # CPU only for privacy
    n_threads: int = 4
    n_batch: int = 512

    # Embeddings
    embedding_model: str = "all-MiniLM-L6-v2"
    embedding_dim: int = 384

    # Vector Store
    db_path: Path = Path("./data/vectors.db")
    chunk_size: int = 256
    chunk_overlap: int = 50

    # Privacy
    enable_pii_filter: bool = True
    data_retention_days: int = 30
    enable_analytics: bool = False  # No cloud telemetry

    # Escalation
    confidence_threshold: float = 0.7
    max_fallback_attempts: int = 2

    # Supported query types
    supported_intents: List[str] = [
        "faq",
        "troubleshooting",
        "account_info",
        "product_info",
        "returns",
        "shipping"
    ]

    class Config:
        env_file = ".env"

settings = Settings()

# src/models/slm_engine.py
from typing import Generator, Optional, Dict, List
from llama_cpp import Llama
from dataclasses import dataclass
from ..config import settings

@dataclass
class GenerationConfig:
    max_tokens: int = 512
    temperature: float = 0.7
    top_p: float = 0.9
    top_k: int = 40
    repeat_penalty: float = 1.1
    stop: List[str] = None

class LocalSLMEngine:
    """Local small language model inference engine."""

    def __init__(self, model_path: str = None):
        self.model_path = model_path or str(settings.model_path)
        self.llm = None
        self._load_model()

    def _load_model(self):
        """Load the GGUF model."""
        self.llm = Llama(
            model_path=self.model_path,
            n_ctx=settings.context_length,
            n_threads=settings.n_threads,
            n_gpu_layers=settings.n_gpu_layers,
            n_batch=settings.n_batch,
            verbose=False
        )
        print(f"Loaded model: {self.model_path}")

    def generate(
        self,
        prompt: str,
        config: GenerationConfig = None
    ) -> str:
        """Generate response synchronously."""
        config = config or GenerationConfig()

        response = self.llm(
            prompt,
            max_tokens=config.max_tokens,
            temperature=config.temperature,
            top_p=config.top_p,
            top_k=config.top_k,
            repeat_penalty=config.repeat_penalty,
            stop=config.stop or ["</s>", "User:", "Human:"]
        )

        return response["choices"][0]["text"].strip()

    def stream(
        self,
        prompt: str,
        config: GenerationConfig = None
    ) -> Generator[str, None, None]:
        """Stream response tokens."""
        config = config or GenerationConfig()

        for token in self.llm(
            prompt,
            max_tokens=config.max_tokens,
            temperature=config.temperature,
            top_p=config.top_p,
            top_k=config.top_k,
            repeat_penalty=config.repeat_penalty,
            stop=config.stop or ["</s>", "User:", "Human:"],
            stream=True
        ):
            yield token["choices"][0]["text"]

    def get_prompt(
        self,
        system: str,
        user: str,
        history: List[Dict] = None
    ) -> str:
        """Format prompt for the model."""
        # Phi-3 chat format
        prompt = f"<|system|>\n{system}<|end|>\n"

        if history:
            for msg in history[-5:]:  # Last 5 turns
                if msg["role"] == "user":
                    prompt += f"<|user|>\n{msg['content']}<|end|>\n"
                else:
                    prompt += f"<|assistant|>\n{msg['content']}<|end|>\n"

        prompt += f"<|user|>\n{user}<|end|>\n<|assistant|>\n"
        return prompt

    def get_stats(self) -> Dict:
        """Get model statistics."""
        return {
            "model_path": self.model_path,
            "context_length": settings.context_length,
            "vocab_size": self.llm.n_vocab(),
        }

# src/rag/local_vectordb.py
import sqlite3
import json
import numpy as np
from typing import List, Tuple, Dict
from sentence_transformers import SentenceTransformer
from ..config import settings

class LocalVectorDB:
    """SQLite-based vector database for offline RAG."""

    def __init__(self, db_path: str = None):
        self.db_path = db_path or str(settings.db_path)
        self.embedding_model = SentenceTransformer(settings.embedding_model)
        self._init_db()

    def _init_db(self):
        """Initialize database schema."""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()

        cursor.execute("""
            CREATE TABLE IF NOT EXISTS documents (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                content TEXT NOT NULL,
                embedding BLOB NOT NULL,
                metadata TEXT,
                source TEXT,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)

        cursor.execute("""
            CREATE INDEX IF NOT EXISTS idx_source ON documents(source)
        """)

        conn.commit()
        conn.close()

    def add_documents(
        self,
        documents: List[str],
        metadata: List[Dict] = None,
        source: str = "unknown"
    ):
        """Add documents to the vector store."""
        # Generate embeddings locally
        embeddings = self.embedding_model.encode(documents)

        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()

        for i, (doc, emb) in enumerate(zip(documents, embeddings)):
            meta = json.dumps(metadata[i]) if metadata and i < len(metadata) else "{}"
            cursor.execute(
                "INSERT INTO documents (content, embedding, metadata, source) VALUES (?, ?, ?, ?)",
                (doc, emb.tobytes(), meta, source)
            )

        conn.commit()
        conn.close()

    def search(
        self,
        query: str,
        k: int = 5,
        source_filter: str = None
    ) -> List[Tuple[str, float, Dict]]:
        """Search for similar documents."""
        # Embed query locally
        query_embedding = self.embedding_model.encode([query])[0]

        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()

        if source_filter:
            cursor.execute(
                "SELECT id, content, embedding, metadata FROM documents WHERE source = ?",
                (source_filter,)
            )
        else:
            cursor.execute(
                "SELECT id, content, embedding, metadata FROM documents"
            )

        results = []
        for row in cursor.fetchall():
            doc_id, content, emb_bytes, meta_str = row
            doc_embedding = np.frombuffer(emb_bytes, dtype=np.float32)

            # Cosine similarity
            similarity = np.dot(query_embedding, doc_embedding) / (
                np.linalg.norm(query_embedding) * np.linalg.norm(doc_embedding)
            )

            results.append((content, float(similarity), json.loads(meta_str)))

        conn.close()

        # Sort by similarity and return top k
        results.sort(key=lambda x: x[1], reverse=True)
        return results[:k]

    def delete_by_source(self, source: str):
        """Delete documents by source."""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute("DELETE FROM documents WHERE source = ?", (source,))
        conn.commit()
        conn.close()

    def get_stats(self) -> Dict:
        """Get database statistics."""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()

        cursor.execute("SELECT COUNT(*) FROM documents")
        total = cursor.fetchone()[0]

        cursor.execute("SELECT source, COUNT(*) FROM documents GROUP BY source")
        by_source = dict(cursor.fetchall())

        conn.close()

        return {"total_documents": total, "by_source": by_source}

# src/routing/classifier.py
from typing import Tuple, List
from enum import Enum
from dataclasses import dataclass
from ..models.slm_engine import LocalSLMEngine, GenerationConfig
from ..config import settings

class QueryIntent(str, Enum):
    FAQ = "faq"
    TROUBLESHOOTING = "troubleshooting"
    ACCOUNT = "account_info"
    PRODUCT = "product_info"
    RETURNS = "returns"
    SHIPPING = "shipping"
    ESCALATE = "escalate"
    CHITCHAT = "chitchat"

@dataclass
class ClassificationResult:
    intent: QueryIntent
    confidence: float
    reasoning: str
    requires_knowledge: bool

class QueryClassifier:
    """Classifies customer queries using local SLM."""

    def __init__(self, slm: LocalSLMEngine):
        self.slm = slm

    def classify(self, query: str) -> ClassificationResult:
        """Classify a customer query."""
        prompt = self.slm.get_prompt(
            system="""You are a query classifier for customer service.
Classify the query into one of these categories:
- faq: General questions about the company/product
- troubleshooting: Technical issues or problems
- account_info: Account-related queries
- product_info: Product details or comparisons
- returns: Return or refund requests
- shipping: Shipping or delivery questions
- escalate: Complex issues needing human help
- chitchat: Casual conversation

Respond with JSON: {"intent": "category", "confidence": 0.0-1.0, "needs_knowledge": true/false}""",
            user=f"Classify this query: {query}"
        )

        response = self.slm.generate(
            prompt,
            GenerationConfig(max_tokens=100, temperature=0.1)
        )

        # Parse response
        try:
            import json
            result = json.loads(response)
            return ClassificationResult(
                intent=QueryIntent(result.get("intent", "escalate")),
                confidence=float(result.get("confidence", 0.5)),
                reasoning="",
                requires_knowledge=result.get("needs_knowledge", True)
            )
        except:
            # Fallback to escalation if parsing fails
            return ClassificationResult(
                intent=QueryIntent.ESCALATE,
                confidence=0.3,
                reasoning="Failed to parse classification",
                requires_knowledge=True
            )


class EscalationManager:
    """Manages escalation to cloud or human agents."""

    def __init__(self):
        self.escalation_triggers = [
            "speak to human",
            "talk to agent",
            "real person",
            "supervisor",
            "manager",
            "complaint",
            "legal",
            "lawyer"
        ]

    def should_escalate(
        self,
        query: str,
        classification: ClassificationResult,
        attempt_count: int
    ) -> Tuple[bool, str]:
        """Determine if query should be escalated."""
        # Check for explicit escalation triggers
        query_lower = query.lower()
        for trigger in self.escalation_triggers:
            if trigger in query_lower:
                return True, "User requested human assistance"

        # Check classification confidence
        if classification.confidence < settings.confidence_threshold:
            return True, "Low confidence in automated response"

        # Check attempt count
        if attempt_count >= settings.max_fallback_attempts:
            return True, "Maximum retry attempts exceeded"

        # Classification-based escalation
        if classification.intent == QueryIntent.ESCALATE:
            return True, "Query classified as requiring escalation"

        return False, ""

# src/privacy/pii_filter.py
import re
from typing import List, Tuple, Dict
from dataclasses import dataclass
from enum import Enum

class PIIType(str, Enum):
    EMAIL = "email"
    PHONE = "phone"
    SSN = "ssn"
    CREDIT_CARD = "credit_card"
    ADDRESS = "address"
    NAME = "name"

@dataclass
class PIIMatch:
    pii_type: PIIType
    value: str
    start: int
    end: int
    masked: str

class LocalPIIFilter:
    """Local PII detection and masking - no cloud calls."""

    def __init__(self):
        self.patterns = {
            PIIType.EMAIL: r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b',
            PIIType.PHONE: r'\b(?:\+1[-.\s]?)?(?:\(?\d{3}\)?[-.\s]?)?\d{3}[-.\s]?\d{4}\b',
            PIIType.SSN: r'\b\d{3}[-\s]?\d{2}[-\s]?\d{4}\b',
            PIIType.CREDIT_CARD: r'\b(?:\d{4}[-\s]?){3}\d{4}\b',
        }

        self.masks = {
            PIIType.EMAIL: "[EMAIL]",
            PIIType.PHONE: "[PHONE]",
            PIIType.SSN: "[SSN]",
            PIIType.CREDIT_CARD: "[CARD]",
            PIIType.ADDRESS: "[ADDRESS]",
            PIIType.NAME: "[NAME]"
        }

    def detect(self, text: str) -> List[PIIMatch]:
        """Detect PII in text."""
        matches = []

        for pii_type, pattern in self.patterns.items():
            for match in re.finditer(pattern, text, re.IGNORECASE):
                matches.append(PIIMatch(
                    pii_type=pii_type,
                    value=match.group(),
                    start=match.start(),
                    end=match.end(),
                    masked=self.masks[pii_type]
                ))

        return sorted(matches, key=lambda x: x.start)

    def mask(self, text: str) -> Tuple[str, List[PIIMatch]]:
        """Detect and mask PII in text."""
        matches = self.detect(text)

        if not matches:
            return text, []

        # Mask from end to start to preserve indices
        masked_text = text
        for match in reversed(matches):
            masked_text = (
                masked_text[:match.start] +
                match.masked +
                masked_text[match.end:]
            )

        return masked_text, matches

    def get_stats(self, text: str) -> Dict[str, int]:
        """Get PII statistics without exposing values."""
        matches = self.detect(text)
        stats = {}
        for match in matches:
            pii_type = match.pii_type.value
            stats[pii_type] = stats.get(pii_type, 0) + 1
        return stats

# src/app/agent.py
from typing import Generator, Dict, List, Optional
from dataclasses import dataclass
from ..models.slm_engine import LocalSLMEngine, GenerationConfig
from ..rag.local_vectordb import LocalVectorDB
from ..routing.classifier import QueryClassifier, EscalationManager, QueryIntent
from ..privacy.pii_filter import LocalPIIFilter
from ..config import settings

@dataclass
class AgentResponse:
    content: str
    intent: str
    confidence: float
    sources: List[str]
    escalated: bool
    escalation_reason: str = ""

class CustomerServiceAgent:
    """Local customer service agent using SLM."""

    def __init__(self):
        self.slm = LocalSLMEngine()
        self.vectordb = LocalVectorDB()
        self.classifier = QueryClassifier(self.slm)
        self.escalation = EscalationManager()
        self.pii_filter = LocalPIIFilter()
        self.conversation_history: List[Dict] = []

    def process(
        self,
        user_message: str,
        attempt_count: int = 0
    ) -> AgentResponse:
        """Process a customer message."""
        # Filter PII before processing
        masked_message, pii_matches = self.pii_filter.mask(user_message)

        # Classify the query
        classification = self.classifier.classify(masked_message)

        # Check for escalation
        should_escalate, reason = self.escalation.should_escalate(
            masked_message,
            classification,
            attempt_count
        )

        if should_escalate:
            return AgentResponse(
                content="I'll connect you with a human agent who can better assist you. Please hold.",
                intent=classification.intent.value,
                confidence=classification.confidence,
                sources=[],
                escalated=True,
                escalation_reason=reason
            )

        # Retrieve relevant context if needed
        sources = []
        context = ""
        if classification.requires_knowledge:
            results = self.vectordb.search(
                masked_message,
                k=3,
                source_filter=self._get_source_for_intent(classification.intent)
            )
            context = "\n\n".join([r[0] for r in results])
            sources = [r[2].get("source", "knowledge base") for r in results]

        # Generate response
        response = self._generate_response(
            masked_message,
            classification.intent,
            context
        )

        # Store in history
        self.conversation_history.append({
            "role": "user",
            "content": masked_message
        })
        self.conversation_history.append({
            "role": "assistant",
            "content": response
        })

        return AgentResponse(
            content=response,
            intent=classification.intent.value,
            confidence=classification.confidence,
            sources=sources,
            escalated=False
        )

    def stream(
        self,
        user_message: str
    ) -> Generator[str, None, None]:
        """Stream response tokens."""
        masked_message, _ = self.pii_filter.mask(user_message)
        classification = self.classifier.classify(masked_message)

        # Get context
        context = ""
        if classification.requires_knowledge:
            results = self.vectordb.search(masked_message, k=3)
            context = "\n\n".join([r[0] for r in results])

        # Build prompt
        prompt = self._build_prompt(masked_message, classification.intent, context)

        # Stream response
        for token in self.slm.stream(prompt):
            yield token

    def _generate_response(
        self,
        query: str,
        intent: QueryIntent,
        context: str
    ) -> str:
        """Generate a response using the SLM."""
        prompt = self._build_prompt(query, intent, context)
        return self.slm.generate(
            prompt,
            GenerationConfig(max_tokens=settings.max_tokens)
        )

    def _build_prompt(
        self,
        query: str,
        intent: QueryIntent,
        context: str
    ) -> str:
        """Build the prompt for the SLM."""
        system = f"""You are a helpful customer service assistant.
Your role is to help customers with {intent.value} questions.
Be concise, friendly, and helpful. If you're unsure, say so.
Never make up information - only use the provided context."""

        if context:
            system += f"\n\nRelevant information:\n{context}"

        return self.slm.get_prompt(
            system=system,
            user=query,
            history=self.conversation_history[-4:]  # Last 2 turns
        )

    def _get_source_for_intent(self, intent: QueryIntent) -> Optional[str]:
        """Map intent to knowledge base source."""
        mapping = {
            QueryIntent.FAQ: "faq",
            QueryIntent.TROUBLESHOOTING: "troubleshooting",
            QueryIntent.PRODUCT: "products",
            QueryIntent.RETURNS: "policies",
            QueryIntent.SHIPPING: "shipping"
        }
        return mapping.get(intent)

    def clear_history(self):
        """Clear conversation history."""
        self.conversation_history = []

# src/app/ui.py
import gradio as gr
from .agent import CustomerServiceAgent

def create_chat_interface():
    """Create Gradio chat interface."""
    agent = CustomerServiceAgent()

    def respond(message, history):
        """Handle chat response."""
        response = agent.process(message)

        # Format response with metadata
        output = response.content

        if response.sources:
            output += f"\n\n_Sources: {', '.join(response.sources)}_"

        if response.escalated:
            output = f"🔄 **Escalating to human agent**\n\n{output}"

        return output

    def clear():
        """Clear conversation."""
        agent.clear_history()
        return []

    # Create interface
    with gr.Blocks(title="Customer Service Assistant") as demo:
        gr.Markdown("# 🤖 Customer Service Assistant")
        gr.Markdown("_Powered by local AI - Your data stays on your device_")

        chatbot = gr.Chatbot(height=400)
        msg = gr.Textbox(
            placeholder="How can I help you today?",
            label="Your message"
        )
        clear_btn = gr.Button("Clear conversation")

        msg.submit(respond, [msg, chatbot], [chatbot])
        msg.submit(lambda: "", None, [msg])
        clear_btn.click(clear, None, [chatbot])

        gr.Markdown("""
        ### Privacy Notice
        - All processing happens locally on your device
        - No data is sent to external servers
        - Conversation history is stored only in memory
        """)

    return demo

if __name__ == "__main__":
    demo = create_chat_interface()
    demo.launch(server_name="0.0.0.0", server_port=7860)

# build_desktop.py
"""Build standalone desktop application."""
import PyInstaller.__main__
import os

PyInstaller.__main__.run([
    'src/app/ui.py',
    '--name=CustomerServiceAI',
    '--onefile',
    '--windowed',
    '--add-data=models:models',
    '--add-data=data:data',
    '--hidden-import=llama_cpp',
    '--hidden-import=sentence_transformers',
])

FROM python:3.11-slim

WORKDIR /app

# Install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy application
COPY src/ ./src/
COPY models/ ./models/
COPY data/ ./data/

# Expose port
EXPOSE 7860

# Run application
CMD ["python", "-m", "src.app.ui"]