Large Language Models & RAG Systems

The RAG Architecture

RAG (Retrieval-Augmented Generation) ensures your AI Tutor provides accurate, verifiable answers by combining LLM capabilities with your specific knowledge base.

from langchain.embeddings import HuggingFaceEmbeddings
from langchain.vectorstores import Chroma
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.chains import RetrievalQA
from langchain.llms import LlamaCpp
from langchain.callbacks.manager import CallbackManager
from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler
import chromadb
from typing import List, Dict, Optional
import json

class PacketCodersRAGTutor:
    """Production-ready RAG system for AI Tutor"""
    
    def __init__(self, model_path: str = "./models/llama-3-8b-instruct.gguf"):
        # Initialize embeddings model
        self.embeddings = HuggingFaceEmbeddings(
            model_name="sentence-transformers/all-mpnet-base-v2",
            model_kwargs={'device': 'cuda'},
            encode_kwargs={'normalize_embeddings': True}
        )
        
        # Initialize vector store
        self.vector_store = Chroma(
            collection_name="packetcoders_knowledge",
            embedding_function=self.embeddings,
            persist_directory="./chroma_db"
        )
        
        # Text splitter for optimal chunking
        self.text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=500,
            chunk_overlap=50,
            length_function=len,
            separators=["\n\n", "\n", ".", "!", "?", ",", " ", ""]
        )
        
        # Callback manager for streaming responses
        callback_manager = CallbackManager([StreamingStdOutCallbackHandler()])
        
        # Initialize Llama 3 with optimized settings
        self.llm = LlamaCpp(
            model_path=model_path,
            temperature=0.3,  # Lower temperature for more factual responses
            max_tokens=512,
            n_ctx=4096,
            n_batch=8,
            n_gpu_layers=35,
            callback_manager=callback_manager,
            verbose=False
        )
        
        # Create retrieval chain
        self.qa_chain = RetrievalQA.from_chain_type(
            llm=self.llm,
            chain_type="stuff",
            retriever=self.vector_store.as_retriever(
                search_kwargs={"k": 5}
            ),
            return_source_documents=True
        )
    
    def index_course_content(self, documents: List[Dict[str, str]]):
        """Index course materials into vector database"""
        all_texts = []
        all_metadatas = []
        
        for doc in documents:
            # Split document into chunks
            texts = self.text_splitter.split_text(doc['content'])
            
            # Create metadata for each chunk
            metadatas = [
                {
                    'source': doc['source'],
                    'course': doc.get('course', 'general'),
                    'module': doc.get('module', 'unknown'),
                    'type': doc.get('type', 'text')
                } for _ in texts
            ]
            
            all_texts.extend(texts)
            all_metadatas.extend(metadatas)
        
        # Add to vector store
        self.vector_store.add_texts(
            texts=all_texts,
            metadatas=all_metadatas
        )
        
        # Persist to disk
        self.vector_store.persist()
        
        return len(all_texts)
    
    def create_prompt_template(self) -> str:
        """Create the prompt template for the tutor"""
        return """You are the PacketCoders AI Tutor, specializing in network automation and Python programming.

Use the following context to answer the question. If you cannot answer based on the context, say so.
Always be helpful, clear, and provide practical examples when relevant.

Context: {context}

Question: {question}

Helpful Answer:"""
    
    def answer_question(self, question: str, include_sources: bool = True) -> Dict:
        """Generate answer with sources"""
        # Get response from chain
        result = self.qa_chain({"query": question})
        
        # Extract answer and sources
        answer = result['result']
        source_docs = result.get('source_documents', [])
        
        # Format response
        response = {
            'answer': answer,
            'confidence': self._calculate_confidence(source_docs),
            'sources': []
        }
        
        if include_sources and source_docs:
            seen_sources = set()
            for doc in source_docs:
                source = doc.metadata.get('source', 'unknown')
                if source not in seen_sources:
                    response['sources'].append({
                        'source': source,
                        'module': doc.metadata.get('module', 'unknown'),
                        'relevance_score': doc.metadata.get('score', 0.0)
                    })
                    seen_sources.add(source)
        
        return response
    
    def _calculate_confidence(self, source_docs: List) -> float:
        """Calculate confidence score based on retrieved documents"""
        if not source_docs:
            return 0.0
        
        # Simple confidence calculation based on number and quality of sources
        base_confidence = min(len(source_docs) * 0.2, 1.0)
        
        # Adjust based on relevance scores if available
        scores = [doc.metadata.get('score', 0.5) for doc in source_docs]
        avg_score = sum(scores) / len(scores) if scores else 0.5
        
        return min(base_confidence * avg_score * 1.5, 1.0)

# Implement safety guardrails
class SafetyGuardrails:
    """Safety measures for LLM outputs"""
    
    def __init__(self):
        self.dangerous_patterns = [
            # System commands
            r'rm\s+-rf', r'format\s+c:', r'del\s+/s',
            # Database operations
            r'drop\s+(database|table)', r'truncate\s+table',
            # Network commands that could cause issues
            r'shutdown', r'reload', r'write\s+erase'
        ]
        
        self.sensitive_patterns = [
            r'password\s*[:=]', r'api[_-]?key\s*[:=]',
            r'secret\s*[:=]', r'token\s*[:=]'
        ]
    
    def validate_output(self, text: str) -> tuple[bool, List[str]]:
        """Validate LLM output for safety"""
        import re
        issues = []
        
        # Check for dangerous commands
        for pattern in self.dangerous_patterns:
            if re.search(pattern, text, re.IGNORECASE):
                issues.append(f"Dangerous pattern detected: {pattern}")
        
        # Check for sensitive information
        for pattern in self.sensitive_patterns:
            if re.search(pattern, text, re.IGNORECASE):
                issues.append(f"Sensitive information pattern detected")
        
        return len(issues) == 0, issues
    
    def sanitize_output(self, text: str) -> str:
        """Sanitize output to remove sensitive information"""
        import re
        
        # Mask IP addresses (but keep format visible)
        text = re.sub(
            r'\b(?:[0-9]{1,3}\.){3}[0-9]{1,3}\b',
            lambda m: '.'.join(['xxx' if i > 0 else p for i, p in enumerate(m.group().split('.'))]),
            text
        )
        
        # Mask potential passwords
        text = re.sub(
            r'(password|passwd|pwd)\s*[:=]\s*\S+',
            r'\1: [REDACTED]',
            text,
            flags=re.IGNORECASE
        )
        
        return text
                

LoRA Fine-tuning Workshop

Fine-tune open-source models on your network automation data for domain expertise.

from transformers import (
    AutoModelForCausalLM,
    AutoTokenizer,
    TrainingArguments,
    Trainer,
    DataCollatorForLanguageModeling
)
from peft import LoraConfig, get_peft_model, TaskType
import torch
from datasets import Dataset
import json

class NetworkLLMFineTuner:
    """Fine-tune LLMs for network automation expertise"""
    
    def __init__(self, base_model: str = "meta-llama/Llama-2-7b-hf"):
        self.base_model_name = base_model
        
        # Load model and tokenizer
        self.model = AutoModelForCausalLM.from_pretrained(
            base_model,
            torch_dtype=torch.float16,
            device_map="auto",
            load_in_8bit=True  # Use 8-bit quantization
        )
        
        self.tokenizer = AutoTokenizer.from_pretrained(base_model)
        self.tokenizer.pad_token = self.tokenizer.eos_token
        
        # Configure LoRA
        self.lora_config = LoraConfig(
            r=16,  # Rank
            lora_alpha=32,
            lora_dropout=0.1,
            bias="none",
            task_type=TaskType.CAUSAL_LM,
            target_modules=["q_proj", "v_proj", "k_proj", "o_proj"]
        )
        
        # Apply LoRA to model
        self.model = get_peft_model(self.model, self.lora_config)
        self.model.print_trainable_parameters()
    
    def prepare_network_dataset(self, data_path: str) -> Dataset:
        """Prepare network configuration dataset"""
        with open(data_path, 'r') as f:
            raw_data = json.load(f)
        
        # Format data for instruction tuning
        formatted_data = []
        for item in raw_data:
            # Create instruction-response pairs
            text = f"""### Instruction:
{item['instruction']}

### Input:
{item.get('input', '')}

### Response:
{item['output']}"""
            
            formatted_data.append({'text': text})
        
        # Create dataset
        dataset = Dataset.from_list(formatted_data)
        
        # Tokenize
        def tokenize_function(examples):
            return self.tokenizer(
                examples['text'],
                truncation=True,
                padding='max_length',
                max_length=512
            )
        
        tokenized_dataset = dataset.map(tokenize_function, batched=True)
        return tokenized_dataset
    
    def create_training_args(self, output_dir: str = "./fine_tuned_model"):
        """Create optimized training arguments"""
        return TrainingArguments(
            output_dir=output_dir,
            num_train_epochs=3,
            per_device_train_batch_size=4,
            gradient_accumulation_steps=4,
            warmup_steps=100,
            learning_rate=2e-4,
            fp16=True,
            logging_steps=10,
            save_strategy="epoch",
            evaluation_strategy="epoch",
            load_best_model_at_end=True,
            report_to="tensorboard",
            remove_unused_columns=False,
            gradient_checkpointing=True,
        )
    
    def train(self, train_dataset, eval_dataset=None):
        """Fine-tune the model"""
        training_args = self.create_training_args()
        
        # Create trainer
        trainer = Trainer(
            model=self.model,
            args=training_args,
            train_dataset=train_dataset,
            eval_dataset=eval_dataset,
            tokenizer=self.tokenizer,
            data_collator=DataCollatorForLanguageModeling(
                tokenizer=self.tokenizer,
                mlm=False
            )
        )
        
        # Start training
        trainer.train()
        
        # Save the fine-tuned model
        trainer.save_model()
        self.tokenizer.save_pretrained(training_args.output_dir)
        
        return trainer
    
    def generate_network_config(self, prompt: str) -> str:
        """Generate network configuration using fine-tuned model"""
        inputs = self.tokenizer(prompt, return_tensors="pt")
        
        with torch.no_grad():
            outputs = self.model.generate(
                **inputs,
                max_new_tokens=256,
                temperature=0.7,
                do_sample=True,
                top_p=0.9,
                repetition_penalty=1.2
            )
        
        response = self.tokenizer.decode(outputs[0], skip_special_tokens=True)
        return response

# Example training data for network automation
sample_training_data = [
    {
        "instruction": "Configure a VLAN on a Cisco switch",
        "input": "VLAN 100 for Engineering department",
        "output": """configure terminal
vlan 100
name Engineering
exit
interface range gigabitEthernet 0/1-10
switchport mode access
switchport access vlan 100
exit
show vlan brief"""
    },
    {
        "instruction": "Set up OSPF routing",
        "input": "Area 0, network 192.168.1.0/24",
        "output": """configure terminal
router ospf 1
network 192.168.1.0 0.0.0.255 area 0
passive-interface default
no passive-interface gigabitEthernet 0/1
exit
show ip ospf neighbor"""
    }
]
                

PacketCoders AI Tutor v1.0

from fastapi import FastAPI, HTTPException, WebSocket
from pydantic import BaseModel
from typing import Optional, List, Dict
import asyncio
from datetime import datetime
import uuid

app = FastAPI(title="PacketCoders AI Tutor")

class Question(BaseModel):
    text: str
    student_id: Optional[str] = None
    course_context: Optional[str] = None
    
class TutorResponse(BaseModel):
    answer: str
    sources: List[Dict]
    confidence: float
    follow_up_questions: List[str]

class AITutor:
    """Complete AI Tutor with personality and pedagogical features"""
    
    def __init__(self):
        self.rag_system = PacketCodersRAGTutor()
        self.safety_guard = SafetyGuardrails()
        self.conversation_history = {}
        
    async def process_question(self, question: Question) -> TutorResponse:
        """Process student question with full pipeline"""
        
        # Get conversation context
        history = self.conversation_history.get(
            question.student_id, []
        ) if question.student_id else []
        
        # Check if question needs clarification
        if self._needs_clarification(question.text):
            return self._request_clarification(question.text)
        
        # Get RAG response
        rag_response = self.rag_system.answer_question(
            question.text,
            include_sources=True
        )
        
        # Validate safety
        is_safe, issues = self.safety_guard.validate_output(
            rag_response['answer']
        )
        
        if not is_safe:
            rag_response['answer'] = self._generate_safe_alternative(
                question.text, issues
            )
        
        # Add pedagogical elements
        response = TutorResponse(
            answer=self._add_teaching_elements(rag_response['answer']),
            sources=rag_response['sources'],
            confidence=rag_response['confidence'],
            follow_up_questions=self._generate_follow_ups(question.text)
        )
        
        # Update history
        if question.student_id:
            self._update_history(question.student_id, question.text, response)
        
        return response
    
    def _needs_clarification(self, question: str) -> bool:
        """Check if question is too vague"""
        vague_indicators = ['it', 'that', 'this thing', 'the error']
        question_lower = question.lower()
        
        if len(question.split()) < 3:
            return True
            
        return any(indicator in question_lower for indicator in vague_indicators)
    
    def _request_clarification(self, question: str) -> TutorResponse:
        """Generate clarification request"""
        return TutorResponse(
            answer="I'd love to help! Could you provide more specific details? For example, what specific technology, command, or concept are you asking about?",
            sources=[],
            confidence=1.0,
            follow_up_questions=[
                "Which specific network protocol are you working with?",
                "Can you share the exact error message?",
                "What have you already tried?"
            ]
        )
    
    def _add_teaching_elements(self, answer: str) -> str:
        """Add pedagogical elements to response"""
        # Add encouragement
        encouragements = [
            "Great question! ",
            "This is an important concept. ",
            "You're on the right track thinking about this. "
        ]
        
        import random
        prefix = random.choice(encouragements)
        
        # Add practical tip
        suffix = "\n\n💡 Pro tip: Practice this concept in the lab environment to reinforce your understanding!"
        
        return prefix + answer + suffix
    
    def _generate_follow_ups(self, question: str) -> List[str]:
        """Generate educational follow-up questions"""
        # In production, use LLM to generate contextual follow-ups
        return [
            "Would you like to see a practical example?",
            "How would you apply this in a production environment?",
            "What challenges might you face implementing this?"
        ]
    
    def _generate_safe_alternative(self, question: str, issues: List[str]) -> str:
        """Generate safe alternative response"""
        return f"""I notice your question might involve potentially risky operations. 
        
Instead of providing direct commands that could be dangerous, let me explain the concept and best practices:

1. Always test commands in a lab environment first
2. Have backups before making configuration changes
3. Use version control for your configurations

Would you like me to explain the safe approach to what you're trying to accomplish?"""
    
    def _update_history(self, student_id: str, question: str, response: TutorResponse):
        """Update conversation history"""
        if student_id not in self.conversation_history:
            self.conversation_history[student_id] = []
        
        self.conversation_history[student_id].append({
            'timestamp': datetime.now().isoformat(),
            'question': question,
            'answer': response.answer,
            'confidence': response.confidence
        })
        
        # Keep only last 10 interactions
        self.conversation_history[student_id] = self.conversation_history[student_id][-10:]

# Initialize tutor
tutor = AITutor()

@app.post("/ask", response_model=TutorResponse)
async def ask_question(question: Question):
    """Endpoint for asking questions"""
    try:
        response = await tutor.process_question(question)
        return response
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.websocket("/chat")
async def websocket_endpoint(websocket: WebSocket):
    """WebSocket for real-time chat"""
    await websocket.accept()
    session_id = str(uuid.uuid4())
    
    try:
        while True:
            # Receive question
            data = await websocket.receive_text()
            question = Question(text=data, student_id=session_id)
            
            # Process and send response
            response = await tutor.process_question(question)
            await websocket.send_json(response.dict())
            
    except Exception as e:
        await websocket.send_json({"error": str(e)})
        await websocket.close()

@app.get("/health")
def health_check():
    return {"status": "healthy", "service": "AI Tutor"}