slm-runtime-platform/ ├── models/ │ ├── embeddings/ │ │ └── all-MiniLM-L6-v2/ # Frozen tiny LLM │ ├── classifiers/ │ │ └── intent_head.pkl # Learnable NN head │ └── base_slm/ │ └── phi-3-mini/ # Frozen response model ├── src/ │ ├── intent_detector.py # Two-stage intent system │ ├── state_manager.py # Conversation state │ ├── decision_engine.py # Orchestrator │ ├── retriever.py # RAG system │ └── response_generator.py # SLM wrapper ├── data/ │ ├── conversations/ # Session logs │ ├── feedback/ # Learning data │ └── knowledge_base/ # RAG documents ├── config/ │ └── system_config.yaml └── main.py # Entry point

# Create virtual environment python -m venv venv source venv/bin/activate # On Windows: venv\Scripts\activate # Install dependencies pip install sentence-transformers # For tiny LLM embeddings pip install scikit-learn # For NN classifier head pip install chromadb # For RAG vector DB pip install ollama # For base SLM pip install fastapi uvicorn # For API (optional) # Download embedding model (one-time) python -c "from sentence_transformers import SentenceTransformer; SentenceTransformer('all-MiniLM-L6-v2')" # Pull base SLM (one-time) ollama pull phi3:mini

from sentence_transformers import SentenceTransformer from sklearn.neural_network import MLPClassifier import pickle import numpy as np class TwoStageIntentDetector: def __init__(self, model_path='models/embeddings/all-MiniLM-L6-v2'): # Stage 1: Frozen tiny LLM for embeddings print("Loading frozen embedding model...") self.embedding_model = SentenceTransformer('all-MiniLM-L6-v2') # Stage 2: Learnable classifier head self.classifier = MLPClassifier( hidden_layer_sizes=(128, 64), activation='relu', warm_start=True, max_iter=100, random_state=42 ) self.intent_classes = [ "ask_question", "request_data", "clarification", "correction", "confirmation", "end_conversation" ] self.is_trained = False def predict(self, user_text, return_all_probs=False): """Two-stage prediction""" # Stage 1: Get semantic embedding (frozen) embedding = self.embedding_model.encode(user_text) if not self.is_trained: return { "intent": "ask_question", # Default "confidence": 0.5, "status": "not_trained" } # Stage 2: Classify with learnable head probs = self.classifier.predict_proba([embedding])[0] intent_idx = probs.argmax() result = { "intent": self.intent_classes[intent_idx], "confidence": float(probs[intent_idx]), "embedding": embedding # Cache for learning } if return_all_probs: result["all_probs"] = dict(zip(self.intent_classes, probs)) return result def initial_train(self, training_data): """Initial training with small dataset""" texts = [item['text'] for item in training_data] labels = [item['intent'] for item in training_data] # Get embeddings from frozen model embeddings = self.embedding_model.encode(texts) # Train classifier head self.classifier.fit(embeddings, labels) self.is_trained = True print(f"✓ Trained on {len(training_data)} examples") def learn_online(self, user_text, correct_intent): """Online learning via partial_fit""" # Get embedding (frozen) embedding = self.embedding_model.encode(user_text) # Update only the classifier head self.classifier.partial_fit( [embedding], [correct_intent], classes=self.intent_classes ) print(f"✓ Online update: '{user_text[:30]}...' → {correct_intent}") def save(self, path='models/classifiers/intent_head.pkl'): """Save only the learnable head (embedding model stays frozen)""" with open(path, 'wb') as f: pickle.dump(self.classifier, f) print(f"✓ Saved classifier head to {path}") def load(self, path='models/classifiers/intent_head.pkl'): """Load saved classifier head""" with open(path, 'rb') as f: self.classifier = pickle.load(f) self.is_trained = True print(f"✓ Loaded classifier head from {path}")

import json from datetime import datetime class StateManager: def __init__(self): self.sessions = {} self.transition_history = [] def create_session(self, session_id): self.sessions[session_id] = { "session_id": session_id, "goal": None, "current_step": "initial", "filled_slots": {}, "missing_slots": [], "last_intent": None, "created_at": datetime.now().isoformat() } return self.sessions[session_id] def update_state(self, session_id, updates): if session_id not in self.sessions: self.create_session(session_id) self.sessions[session_id].update(updates) return self.sessions[session_id] def log_transition(self, state, action, outcome): """Learn from state transitions""" self.transition_history.append({ "state": state, "action": action, "outcome": outcome, "timestamp": datetime.now().isoformat() })

from intent_detector import TwoStageIntentDetector from state_manager import StateManager import uuid class SLMRuntimeSystem: def __init__(self): print("Initializing SLM Runtime Learning Platform...") self.intent_detector = TwoStageIntentDetector() self.state_manager = StateManager() # Initial training data (minimal) self._bootstrap() def _bootstrap(self): """Minimal initial training""" training_data = [ {"text": "What is X?", "intent": "ask_question"}, {"text": "Show me the data", "intent": "request_data"}, {"text": "Can you clarify?", "intent": "clarification"}, {"text": "No I meant Y", "intent": "correction"}, {"text": "Yes that's right", "intent": "confirmation"}, {"text": "Goodbye", "intent": "end_conversation"}, ] self.intent_detector.initial_train(training_data) def process_message(self, user_text, session_id=None): if not session_id: session_id = str(uuid.uuid4()) # Step 1: Detect intent (two-stage) intent_result = self.intent_detector.predict(user_text) # Step 2: Update state state = self.state_manager.update_state(session_id, { "last_intent": intent_result["intent"] }) return { "intent": intent_result, "state": state, "session_id": session_id } # Usage if __name__ == "__main__": system = SLMRuntimeSystem() # Test result = system.process_message("I need my blood test results") print(result)

import torch from transformers import AutoModel, AutoTokenizer from sentence_transformers import SentenceTransformer import numpy as np class TinyLLMPruner: def __init__(self, base_model_name='distilbert-base-uncased'): self.model = AutoModel.from_pretrained(base_model_name) self.tokenizer = AutoTokenizer.from_pretrained(base_model_name) def calculate_head_importance(self, layer, sample_texts): """Calculate attention head importance scores""" importance_scores = [] with torch.no_grad(): for text in sample_texts: inputs = self.tokenizer(text, return_tensors='pt') outputs = layer(**inputs, output_attentions=True) # Average attention weights per head attn_weights = outputs.attentions[0] head_scores = attn_weights.mean(dim=(0, 2, 3)) importance_scores.append(head_scores) return torch.stack(importance_scores).mean(dim=0) def prune_model(self, domain_texts, prune_ratio=0.3): """Prune least important attention heads""" for layer_idx, layer in enumerate(self.model.transformer.layer): importance = self.calculate_head_importance(layer, domain_texts) # Keep top (1 - prune_ratio) heads num_keep = int(len(importance) * (1 - prune_ratio)) heads_to_keep = torch.topk(importance, num_keep).indices # Prune heads_to_prune = [i for i in range(len(importance)) if i not in heads_to_keep] layer.attention.prune_heads(heads_to_prune) print(f"Layer {layer_idx}: Pruned {len(heads_to_prune)} heads") def knowledge_distillation(self, teacher_model, student_texts, epochs=3): """Fine-tune pruned model to match teacher""" optimizer = torch.optim.AdamW(self.model.parameters(), lr=1e-4) for epoch in range(epochs): for text in student_texts: # Get teacher embeddings with torch.no_grad(): teacher_emb = teacher_model.encode(text) # Get student embeddings student_emb = self._get_embedding(text) # Cosine similarity loss loss = 1 - torch.nn.functional.cosine_similarity( teacher_emb, student_emb, dim=0 ) loss.backward() optimizer.step() optimizer.zero_grad() print(f"Epoch {epoch + 1}: Loss = {loss.item():.4f}") def save_pruned_model(self, output_path='models/pruned_tiny_llm'): self.model.save_pretrained(output_path) self.tokenizer.save_pretrained(output_path) print(f"✓ Saved pruned model to {output_path}") # Usage pruner = TinyLLMPruner('distilbert-base-uncased') # Your domain texts medical_texts = [ "Blood test results show elevated hemoglobin", "Patient reports chest pain and shortness of breath", # ... more domain examples ] pruner.prune_model(medical_texts, prune_ratio=0.3) pruner.save_pruned_model()