From Machine Learning to LLMs – What Should You Learn Next?

# Skill transfer mapping: Classical ML -> LLM Development
# HIGH transfer = concept is directly applicable, only the tools change
# MEDIUM transfer = concept applies but requires significant adaptation
# LOW transfer = classical ML approach is rarely used in LLM pipelines

SKILL_TRANSFER = {
    "Feature Engineering": {
        "classical_ml": "Transform raw data into model-consumable numeric features",
        "llm_equivalent": "Prompt engineering — crafting inputs that elicit correct, "
                         "consistent, and well-formatted outputs from a language model",
        "transfer_level": "HIGH",
        "note": "Same principle: garbage in, garbage out. Better inputs produce better outputs."
    },
    "Train/Test Split Discipline": {
        "classical_ml": "Separate training data from evaluation data to measure "
                       "generalization, not memorization",
        "llm_equivalent": "Evaluation datasets with ground truth — never evaluate your "
                         "prompt on the same examples you used to design it",
        "transfer_level": "HIGH",
        "note": "Prompt overfitting is real. Testing on your design examples is cheating."
    },
    "Evaluation Metrics": {
        "classical_ml": "Precision, recall, F1, AUC, RMSE — objective metrics against labels",
        "llm_equivalent": "Faithfulness, relevance, correctness, hallucination rate — "
                         "measured against verified ground truth answers",
        "transfer_level": "HIGH",
        "note": "The principle is identical: systematic measurement against ground truth."
    },
    "Overfitting Detection": {
        "classical_ml": "Gap between training performance and held-out test performance",
        "llm_equivalent": "Prompt overfitting — pipeline works on your 10 hand-picked "
                         "test queries but fails on real user queries at scale",
        "transfer_level": "HIGH",
        "note": "Evaluate on diverse real user queries, not curated examples."
    },
    "Data Quality Thinking": {
        "classical_ml": "Clean, deduplicated, consistent, correctly labeled training data",
        "llm_equivalent": "Clean retrieval corpus — malformed chunks, duplicate documents, "
                         "and outdated content produce hallucinations and irrelevant answers",
        "transfer_level": "HIGH",
        "note": "Garbage in the vector store produces garbage answers. Same principle."
    },
    "Systematic Debugging": {
        "classical_ml": "Inspect misclassified examples to find patterns in model failures",
        "llm_equivalent": "Inspect hallucinated and incorrect answers to find prompt "
                         "or retrieval gaps that explain the failure",
        "transfer_level": "HIGH",
        "note": "Error analysis is error analysis regardless of model type."
    },
    "Model Training": {
        "classical_ml": "Gradient descent, hyperparameter tuning, cross-validation, "
                       "managing training runs and model weights",
        "llm_equivalent": "Rarely needed. Use pre-trained foundation models. "
                         "Fine-tuning is the exception, not the rule.",
        "transfer_level": "LOW",
        "note": "Most engineers spend zero time on model training in LLM pipelines."
    },
    "Hyperparameter Tuning": {
        "classical_ml": "Grid search, random search, Bayesian optimization over model parameters",
        "llm_equivalent": "Chunk size, overlap, top-k retrieval, temperature, "
                         "context window allocation — tuned on your eval dataset",
        "transfer_level": "MEDIUM",
        "note": "The mindset transfers but the parameters are completely different."
    }
}

for skill, mapping in SKILL_TRANSFER.items():
    level = mapping['transfer_level']
    print(f"[{level}] {skill}")
    print(f"  Classical ML : {mapping['classical_ml']}")
    print(f"  LLM Equivalent: {mapping['llm_equivalent']}")
    print(f"  Note: {mapping['note']}")
    print()

# The classical ML workflow vs the LLM orchestration workflow
# Both require engineering rigor — the surface changes, the depth does not.

CLASSICAL_ML_WORKFLOW = [
    "1. Collect and label training data",
    "2. Clean and preprocess features",
    "3. Split into train/validation/test",
    "4. Select and train model",
    "5. Tune hyperparameters on validation set",
    "6. Evaluate on held-out test set",
    "7. Deploy model serving endpoint",
    "8. Monitor predictions and retrain on data drift"
]

LLM_ORCHESTRATION_WORKFLOW = [
    "1. Collect and clean retrieval corpus (documents, policies, data)",
    "2. Chunk and embed documents into vector store",
    "3. Build evaluation dataset with verified ground truth answers",
    "4. Design and test retrieval pipeline (embedding model, chunk strategy, top-k)",
    "5. Design and test prompt (role, context, task, format, few-shot examples)",
    "6. Evaluate pipeline on eval dataset (faithfulness, relevance, correctness)",
    "7. Deploy RAG pipeline with monitoring on per-class metrics",
    "8. Add failing production queries to eval set weekly — iterate continuously"
]

print("Classical ML Workflow:")
for step in CLASSICAL_ML_WORKFLOW:
    print(f"  {step}")

print("\nLLM Orchestration Workflow:")
for step in LLM_ORCHESTRATION_WORKFLOW:
    print(f"  {step}")

# The key insight: steps 3, 6, and 8 are identical in principle.
# The evaluation discipline does not change — only the metrics and tools do.

import numpy as np
from typing import List, Dict, Any


class SimpleRAGPipeline:
    """Minimal RAG pipeline that illustrates the core pattern.

    This is not production code — it is a teaching implementation
    that makes the two phases explicit: retrieve, then generate.

    In production, use LangChain, LlamaIndex, or a purpose-built
    retrieval framework with proper error handling, caching, and
    observability.
    """

    def __init__(self, embedding_model, vector_store, llm_client):
        self.embedding_model = embedding_model
        self.vector_store = vector_store
        self.llm = llm_client

    # ---------------------------------------------------------------
    # PHASE 1: RETRIEVAL (This is classical ML territory)
    # ---------------------------------------------------------------
    def retrieve(self, query: str, top_k: int = 4) -> List[str]:
        """Embed the query and find the most similar document chunks.

        This is the same operation as k-nearest-neighbors in classical ML:
        compute the distance from the query vector to every stored vector
        and return the top-k closest matches.
        """
        # Convert the user query to the same vector space as the stored chunks
        query_embedding = self.embedding_model.encode(query)

        # Find the k most similar chunks by cosine similarity
        # The vector store handles this efficiently at scale (FAISS, Pinecone, Weaviate)
        results = self.vector_store.similarity_search(
            query_embedding, k=top_k
        )

        # Each result is a document chunk — typically 200-500 tokens
        return [r.page_content for r in results]

    # ---------------------------------------------------------------
    # PHASE 2: GENERATION (This is LLM territory)
    # ---------------------------------------------------------------
    def generate(self, query: str, context_chunks: List[str]) -> str:
        """Generate an answer grounded in retrieved context.

        The system prompt constrains the model to use only the
        provided context — this is what prevents hallucination.
        """
        context = "\n\n".join(
            [f"[Source {i+1}]: {chunk}"
             for i, chunk in enumerate(context_chunks)]
        )

        system_prompt = (
            "You are a helpful assistant. Answer the user's question "
            "using ONLY the information in the provided context. "
            "If the context does not contain the answer, respond with: "
            "'I don't have that information in my knowledge base.' "
            "Do not use your general knowledge — only the context."
        )

        response = self.llm.chat(
            system=system_prompt,
            user=f"Context:\n{context}\n\nQuestion: {query}"
        )
        return response

    # ---------------------------------------------------------------
    # FULL PIPELINE: Retrieve then generate
    # ---------------------------------------------------------------
    def answer(self, query: str, top_k: int = 4) -> Dict[str, Any]:
        """End-to-end RAG: retrieve relevant context, then generate."""
        # Phase 1: Retrieve
        chunks = self.retrieve(query, top_k=top_k)

        # Phase 2: Generate
        answer = self.generate(query, chunks)

        # Return both the answer and the sources for citation tracking
        return {
            "answer": answer,
            "sources": chunks,
            "retrieved_count": len(chunks)
        }


# ---------------------------------------------------------------
# INDEXING: What you do once, before any queries arrive
# ---------------------------------------------------------------
def build_index(documents: List[str], embedding_model, vector_store,
                chunk_size: int = 400, overlap: int = 50):
    """Chunk documents and store embeddings in the vector store.

    Chunking is data preprocessing — the same concept as creating
    feature windows in time series ML. Size matters enormously:
    - Too large: relevant signal is diluted by surrounding text
    - Too small: context is lost, answers lack coherence
    - 200-500 tokens with 50-token overlap is a safe starting point
    """
    chunks = []
    for doc in documents:
        # Naive fixed-size chunking for illustration
        # Production: use RecursiveCharacterTextSplitter or semantic chunking
        words = doc.split()
        for i in range(0, len(words), chunk_size - overlap):
            chunk = ' '.join(words[i:i + chunk_size])
            if chunk:
                chunks.append(chunk)

    print(f"Created {len(chunks)} chunks from {len(documents)} documents")

    # Embed all chunks and store in vector database
    embeddings = embedding_model.encode(chunks, batch_size=32, show_progress_bar=True)
    vector_store.add(chunks, embeddings)

    print(f"Indexed {len(chunks)} chunks. Ready for retrieval.")
    return vector_store

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
from langchain_openai import ChatOpenAI
from langchain_community.vectorstores import FAISS


# ---------------------------------------------------------------
# Pattern 1: Basic chain — prompt -> LLM -> output parser
# Use this for simple question-answering without retrieval
# ---------------------------------------------------------------
llm = ChatOpenAI(model='gpt-4o', temperature=0)  # temperature=0 for deterministic output

prompt = ChatPromptTemplate.from_template(
    "You are a helpful assistant.\n\n"
    "Question: {question}\n\n"
    "Answer:"
)

basic_chain = prompt | llm | StrOutputParser()
result = basic_chain.invoke({"question": "What is retrieval-augmented generation?"})
print(result)


# ---------------------------------------------------------------
# Pattern 2: RAG chain — retrieve context, then generate
# Use this for any question-answering over your documents
# This is the pattern you will use 80% of the time
# ---------------------------------------------------------------
rag_prompt = ChatPromptTemplate.from_template(
    """Answer the question using ONLY the following context.
If the context does not contain the answer, respond with:
'I don't have that information in my knowledge base.'
Do not use your general knowledge.

Context:
{context}

Question: {question}

Answer:"""
)

# Assume a vector store has been built and loaded
# retriever returns the top-4 most similar chunks for each query
retriever = vector_store.as_retriever(
    search_type='similarity',
    search_kwargs={'k': 4}
)

def format_docs(docs):
    """Join retrieved chunks into a single context string."""
    return "\n\n".join(
        f"[Source {i+1}]: {doc.page_content}"
        for i, doc in enumerate(docs)
    )

rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | rag_prompt
    | llm
    | StrOutputParser()
)

result = rag_chain.invoke("What is the return window for electronics?")
print(result)


# ---------------------------------------------------------------
# Pattern 3: Conversational RAG with memory
# Use when you need multi-turn chat over your documents
# ---------------------------------------------------------------
from langchain_core.messages import HumanMessage, AIMessage
from langchain_core.chat_history import InMemoryChatMessageHistory

chat_history = InMemoryChatMessageHistory()

conversational_prompt = ChatPromptTemplate.from_messages([
    ("system",
     "You are a helpful assistant. Answer using ONLY the provided context."),
    ("human", "Context: {context}\n\nQuestion: {question}")
])

# Track conversation history for follow-up questions
def answer_with_history(question: str, history: list) -> str:
    context_docs = retriever.invoke(question)
    context = format_docs(context_docs)
    response = (conversational_prompt | llm | StrOutputParser()).invoke({
        "context": context,
        "question": question
    })
    return response

# Prompt engineering is structured, not magical.
# A production prompt has four parts: role, context, task, and format.
# Treat prompt design the same way you treat feature design — systematic,
# version-controlled, and evaluated against your test set.


# ---------------------------------------------------------------
# THE FOUR-PART PROMPT STRUCTURE
# ---------------------------------------------------------------

BASE_SYSTEM_PROMPT = """
ROLE:
You are a customer support specialist for Acme Electronics.
You have access to our product documentation, return policies,
and warranty terms.

CONSTRAINTS:
- Answer ONLY using the provided context documents.
- If the context does not contain the answer, say:
  'I don't have that information. Let me connect you with a specialist.'
- Do not speculate, estimate, or use your general knowledge.
- Do not fabricate product specifications, prices, or policy terms.

TASK:
Answer the customer's question accurately, concisely, and helpfully.
If the question requires a policy decision that exceeds your authority,
say so and offer to escalate.

FORMAT:
Respond in 2-4 sentences maximum.
If listing steps, use numbered format.
End with: 'Is there anything else I can help you with?'
"""


# ---------------------------------------------------------------
# FEW-SHOT EXAMPLES: Dramatically improve consistency
# ---------------------------------------------------------------
# Few-shot examples in prompts are the equivalent of providing
# labeled training examples in classical ML. They show the model
# exactly what format, tone, and reasoning pattern you expect.

FEW_SHOT_EXAMPLES = """
Example 1:
Customer: Can I return a laptop I bought 45 days ago?
Agent: Our standard return window for laptops is 30 days for unopened
items and 14 days for opened items. A 45-day return would fall outside
our standard policy. I can escalate this to our returns team for a
case-by-case review if you'd like. Is there anything else I can help
you with?

Example 2:
Customer: What's the warranty on your 4K monitors?
Agent: Our 4K monitors carry a 3-year limited warranty covering
manufacturing defects. This does not cover physical damage or
accidents. You can register your product at acme.com/warranty to
activate coverage. Is there anything else I can help you with?
"""


# ---------------------------------------------------------------
# PROMPT VERSIONING: Version prompts like code
# ---------------------------------------------------------------
# Prompts are production artifacts. A changed prompt changes model
# behavior across ALL queries — not just the ones you tested.
# Version them, test them in CI/CD, and never deploy blind.

prompt_config = {
    "version": "2.3.1",
    "description": "Added explicit non-speculation constraint after hallucination audit",
    "system": BASE_SYSTEM_PROMPT,
    "few_shot": FEW_SHOT_EXAMPLES,
    "temperature": 0,
    "max_tokens": 300,
    "eval_score": {
        "faithfulness": 0.91,
        "correctness": 0.87,
        "hallucination_rate": 0.03
    },
    "deployed": False,
    "tested_against_eval_set": True
}


def load_prompt(version: str, config_path: str = "prompts/") -> dict:
    """Load a versioned prompt from config files.

    Never hardcode prompts in application code.
    Store in YAML or JSON config files that can be version-controlled,
    diffed, and rolled back independently of the application code.
    """
    import json
    with open(f"{config_path}/prompt_v{version}.json") as f:
        return json.load(f)


# ---------------------------------------------------------------
# PROMPT TESTING: Evaluate before deploying
# ---------------------------------------------------------------

def test_prompt_regression(
    new_prompt: dict,
    eval_dataset: list,
    evaluator,
    threshold: float = 0.85
) -> bool:
    """Test a new prompt version against the evaluation dataset.

    Returns True if the new prompt meets all metric thresholds.
    This runs in CI/CD before any prompt change is merged.
    """
    results = evaluator.evaluate(new_prompt, eval_dataset)

    passed = (
        results['faithfulness'] >= threshold and
        results['hallucination_rate'] <= 0.05 and
        results['correctness'] >= threshold
    )

    print(f"Prompt v{new_prompt['version']} evaluation:")
    print(f"  Faithfulness:      {results['faithfulness']:.2f} "
          f"({'PASS' if results['faithfulness'] >= threshold else 'FAIL'})")
    print(f"  Correctness:       {results['correctness']:.2f} "
          f"({'PASS' if results['correctness'] >= threshold else 'FAIL'})")
    print(f"  Hallucination Rate:{results['hallucination_rate']:.2f} "
          f"({'PASS' if results['hallucination_rate'] <= 0.05 else 'FAIL'})")
    print(f"  Overall: {'PASS — safe to deploy' if passed else 'FAIL — do not deploy'}")

    return passed

from typing import List, Dict, Any
from dataclasses import dataclass


@dataclass
class EvalExample:
    """One example in your evaluation dataset.

    The ground_truth is the canonical correct answer, verified by
    a domain expert. This is your labeled test set — the same concept
    as y_test in classical ML.
    """
    question: str
    ground_truth: str       # Verified correct answer
    source_documents: List[str]  # The documents that contain the answer


class LLMEvaluator:
    """Systematic evaluation of LLM pipeline outputs.

    Measures three core metrics:
    - Faithfulness: Is the answer grounded in the retrieved context?
      (High faithfulness = low hallucination risk)
    - Relevance: Did retrieval surface useful context?
      (Low relevance = retrieval problem, not generation problem)
    - Correctness: Is the answer factually accurate vs. ground truth?
      (The only metric that directly measures real-world quality)
    """

    def __init__(self, judge_llm, metrics: List[str] = None):
        self.judge_llm = judge_llm  # LLM used to score outputs
        self.metrics = metrics or ['faithfulness', 'relevance', 'correctness']

    def evaluate(
        self,
        pipeline,
        eval_dataset: List[EvalExample]
    ) -> Dict[str, float]:
        """Evaluate pipeline on all examples. Returns mean scores."""
        all_results = []

        for example in eval_dataset:
            # Run the pipeline
            pipeline_output = pipeline.answer(example.question)

            # Score this example
            example_scores = self._score_example(
                question=example.question,
                answer=pipeline_output['answer'],
                context=pipeline_output['sources'],
                ground_truth=example.ground_truth
            )
            all_results.append(example_scores)

        # Aggregate scores
        aggregated = {}
        for metric in self.metrics:
            scores = [r[metric] for r in all_results]
            aggregated[metric] = sum(scores) / len(scores)
            aggregated[f'{metric}_min'] = min(scores)  # Worst case matters too

        aggregated['hallucination_rate'] = sum(
            1 for r in all_results if r.get('hallucination', False)
        ) / len(all_results)

        return aggregated

    def _score_example(
        self,
        question: str,
        answer: str,
        context: List[str],
        ground_truth: str
    ) -> Dict[str, float]:
        """Score one example using the judge LLM.

        The judge LLM scores each metric from 0 to 1.
        Validate a sample of these scores against human labels
        to catch judge model bias.
        """
        context_str = '\n'.join(context)

        faithfulness_prompt = f"""
Score whether this answer is fully supported by the provided context.
Answer: {answer}
Context: {context_str}
Score: Return a number from 0.0 (completely unsupported) to 1.0 (fully supported).
Just the number, nothing else."""

        correctness_prompt = f"""
Score whether this answer is factually correct given the ground truth.
Answer: {answer}
Ground Truth: {ground_truth}
Score: Return a number from 0.0 (completely wrong) to 1.0 (fully correct).
Just the number, nothing else."""

        faithfulness = float(self.judge_llm.complete(faithfulness_prompt).strip())
        correctness = float(self.judge_llm.complete(correctness_prompt).strip())

        return {
            'faithfulness': min(max(faithfulness, 0.0), 1.0),
            'correctness': min(max(correctness, 0.0), 1.0),
            'hallucination': faithfulness < 0.5  # Flag low-faithfulness answers
        }


def compare_pipelines(
    baseline,
    candidate,
    eval_dataset: List[EvalExample],
    evaluator: LLMEvaluator
) -> Dict[str, Any]:
    """A/B test two pipeline versions against the same eval set.

    Same principle as comparing two model versions in classical ML:
    hold the evaluation data constant, vary the pipeline.
    """
    print("Evaluating baseline pipeline...")
    baseline_scores = evaluator.evaluate(baseline, eval_dataset)

    print("Evaluating candidate pipeline...")
    candidate_scores = evaluator.evaluate(candidate, eval_dataset)

    improvements = {
        metric: candidate_scores[metric] - baseline_scores[metric]
        for metric in ['faithfulness', 'relevance', 'correctness']
    }

    winner = 'candidate' if sum(improvements.values()) > 0 else 'baseline'

    print(f"\nResults ({len(eval_dataset)} examples):")
    for metric in ['faithfulness', 'correctness', 'hallucination_rate']:
        delta = candidate_scores.get(metric, 0) - baseline_scores.get(metric, 0)
        direction = '+' if delta > 0 else ''
        print(f"  {metric:25}: "
              f"baseline={baseline_scores.get(metric, 0):.3f} "
              f"candidate={candidate_scores.get(metric, 0):.3f} "
              f"delta={direction}{delta:.3f}")

    print(f"\nWinner: {winner}")
    return {'winner': winner, 'baseline': baseline_scores,
            'candidate': candidate_scores, 'improvements': improvements}

# Recommended learning path from classical ML to production LLM pipelines
# Time estimates assume 1-2 hours of focused work per day.
# Each step includes a concrete project to ship, not just concepts to read.

LEARNING_PATH = [
    {
        "step": 1,
        "topic": "LLM API Basics",
        "description": "Call OpenAI or Anthropic APIs directly. Understand tokens, "
                      "context windows, temperature, top-p, and system prompts. "
                      "See how small changes in these parameters change output.",
        "time_estimate": "1 week",
        "prerequisite": "Python fluency and basic HTTP/API concepts",
        "ship_this": "A CLI tool that takes a user question and returns an LLM answer. "
                    "Log token usage and cost per call."
    },
    {
        "step": 2,
        "topic": "Prompt Engineering",
        "description": "Design structured prompts with role, context, task, and format. "
                      "Test few-shot examples. Observe how explicit output format "
                      "constraints reduce variance. Learn why temperature=0 matters.",
        "time_estimate": "2 weeks",
        "prerequisite": "Step 1",
        "ship_this": "A structured prompt for a specific task (summarization, classification, "
                    "extraction) tested on 20 diverse examples. Document failure cases."
    },
    {
        "step": 3,
        "topic": "Embeddings and Vector Search",
        "description": "Convert text to dense vectors using an embedding model. "
                      "Build similarity search with FAISS or ChromaDB. "
                      "Understand semantic similarity vs. keyword matching.",
        "time_estimate": "2 weeks",
        "prerequisite": "Step 1 + classical ML basics (distance metrics, nearest neighbors)",
        "ship_this": "A semantic search engine over a small document set. "
                    "Compare results to keyword search on the same queries."
    },
    {
        "step": 4,
        "topic": "RAG Pipelines",
        "description": "Combine retrieval with generation. Chunk documents, embed them, "
                      "store in a vector database, retrieve on query, and generate "
                      "grounded answers. Tune chunk size and top-k on real queries.",
        "time_estimate": "3 weeks",
        "prerequisite": "Steps 2 and 3",
        "ship_this": "An end-to-end Q&A system over a set of real documents you care about. "
                    "It should decline to answer when the context is insufficient."
    },
    {
        "step": 5,
        "topic": "LangChain Orchestration",
        "description": "Rebuild your Step 4 RAG pipeline using LangChain. "
                      "Add memory for multi-turn conversation. Understand when "
                      "LangChain abstractions help vs. when they add unnecessary complexity.",
        "time_estimate": "2 weeks",
        "prerequisite": "Step 4",
        "ship_this": "A multi-turn chatbot over your document corpus that remembers "
                    "conversation context and cites sources in every answer."
    },
    {
        "step": 6,
        "topic": "LLM Evaluation",
        "description": "Build an evaluation dataset of 100+ real queries with verified "
                      "ground truth. Implement automated scoring for faithfulness, "
                      "relevance, and correctness. Run A/B tests between pipeline versions.",
        "time_estimate": "2 weeks",
        "prerequisite": "Steps 4 and 5",
        "ship_this": "An evaluation pipeline that scores your Step 4 RAG system and "
                    "produces a report showing which query types fail and why."
    },
    {
        "step": 7,
        "topic": "Fine-tuning (When RAG Fails)",
        "description": "Fine-tune a smaller model (Llama 3, Mistral) using LoRA on a "
                      "specific task where RAG has provably failed. Evaluate the "
                      "fine-tuned model against your Step 6 eval dataset. "
                      "Compare cost and quality vs. RAG.",
        "time_estimate": "3 weeks",
        "prerequisite": "Step 6 — you must have eval results showing RAG is insufficient",
        "ship_this": "A fine-tuned model with before/after eval scores that justify "
                    "the fine-tuning investment. If scores do not improve significantly, "
                    "the fine-tuning was premature."
    }
]

for item in LEARNING_PATH:
    print(f"Step {item['step']}: {item['topic']} ({item['time_estimate']})")
    print(f"  What: {item['description'][:80]}...")
    print(f"  Ship: {item['ship_this'][:80]}...")
    print()