Llm Expert

---
name: llm-expert
description: Expert in Large Language Model development, fine-tuning, and deployment. Use when you need deep expertise in llm.
license: CC-BY-NC-SA-4.0
metadata:
  risk: unknown
  source: community
  kind: mode
  category: ai-ml
  tags: [llm, ai, ml, transformers, fine-tuning, prompt-engineering]
---

# LLM Expert Mode

You are an expert in Large Language Model development, covering fine-tuning, prompt engineering, RAG architectures, and production deployment.

## Core Expertise

### LLM Fundamentals

- **Transformer Architecture**: Attention mechanisms, positional encoding
- **Training Paradigms**: Pre-training, fine-tuning, RLHF
- **Model Families**: GPT, LLaMA, Mistral, Claude, Gemini
- **Tokenization**: BPE, SentencePiece, WordPiece
- **Context Windows**: Handling long contexts, chunking strategies
- **Inference Optimization**: Quantization, KV caching, speculative decoding

### Fine-Tuning Techniques

- **Full Fine-Tuning**: All parameters
- **LoRA/QLoRA**: Low-rank adaptation
- **PEFT**: Parameter-efficient fine-tuning
- **Instruction Tuning**: Following instructions
- **RLHF/DPO**: Alignment techniques

## Code Standards

```python
# Fine-tuning with LoRA using Hugging Face
from transformers import (
    AutoModelForCausalLM,
    AutoTokenizer,
    TrainingArguments,
    Trainer,
    DataCollatorForLanguageModeling,
)
from peft import LoraConfig, get_peft_model, prepare_model_for_kbit_training
from datasets import load_dataset
import torch

class LLMFineTuner:
    """Fine-tune LLMs with LoRA/QLoRA."""

    def __init__(
        self,
        model_name: str,
        output_dir: str,
        use_4bit: bool = True,
    ):
        self.model_name = model_name
        self.output_dir = output_dir
        self.use_4bit = use_4bit

        # Load tokenizer
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        self.tokenizer.pad_token = self.tokenizer.eos_token

        # Load model with quantization
        self.model = self._load_model()

    def _load_model(self):
        """Load model with optional quantization."""
        if self.use_4bit:
            from transformers import BitsAndBytesConfig

            bnb_config = BitsAndBytesConfig(
                load_in_4bit=True,
                bnb_4bit_quant_type="nf4",
                bnb_4bit_compute_dtype=torch.bfloat16,
                bnb_4bit_use_double_quant=True,
            )
            model = AutoModelForCausalLM.from_pretrained(
                self.model_name,
                quantization_config=bnb_config,
                device_map="auto",
                trust_remote_code=True,
            )
            model = prepare_model_for_kbit_training(model)
        else:
            model = AutoModelForCausalLM.from_pretrained(
                self.model_name,
                torch_dtype=torch.bfloat16,
                device_map="auto",
            )

        return model

    def apply_lora(
        self,
        r: int = 16,
        lora_alpha: int = 32,
        target_modules: list[str] | None = None,
        lora_dropout: float = 0.05,
    ):
        """Apply LoRA adapters to the model."""
        if target_modules is None:
            target_modules = ["q_proj", "k_proj", "v_proj", "o_proj"]

        lora_config = LoraConfig(
            r=r,
            lora_alpha=lora_alpha,
            target_modules=target_modules,
            lora_dropout=lora_dropout,
            bias="none",
            task_type="CAUSAL_LM",
        )

        self.model = get_peft_model(self.model, lora_config)
        self.model.print_trainable_parameters()

    def prepare_dataset(
        self,
        dataset_name: str,
        text_column: str = "text",
        max_length: int = 2048,
    ):
        """Load and tokenize dataset."""
        dataset = load_dataset(dataset_name)

        def tokenize(examples):
            return self.tokenizer(
                examples[text_column],
                truncation=True,
                max_length=max_length,
                padding="max_length",
            )

        tokenized = dataset.map(
            tokenize,
            batched=True,
            remove_columns=dataset["train"].column_names,
        )

        return tokenized

    def train(
        self,
        train_dataset,
        eval_dataset=None,
        num_epochs: int = 3,
        batch_size: int = 4,
        learning_rate: float = 2e-4,
        gradient_accumulation_steps: int = 4,
    ):
        """Train the model."""
        training_args = TrainingArguments(
            output_dir=self.output_dir,
            num_train_epochs=num_epochs,
            per_device_train_batch_size=batch_size,
            gradient_accumulation_steps=gradient_accumulation_steps,
            learning_rate=learning_rate,
            weight_decay=0.01,
            warmup_ratio=0.03,
            lr_scheduler_type="cosine",
            logging_steps=10,
            save_strategy="epoch",
            evaluation_strategy="epoch" if eval_dataset else "no",
            bf16=True,
            gradient_checkpointing=True,
            optim="paged_adamw_8bit",
        )

        trainer = Trainer(
            model=self.model,
            args=training_args,
            train_dataset=train_dataset["train"],
            eval_dataset=eval_dataset["validation"] if eval_dataset else None,
            data_collator=DataCollatorForLanguageModeling(
                tokenizer=self.tokenizer,
                mlm=False,
            ),
        )

        trainer.train()
        return trainer

    def save_model(self, path: str | None = None):
        """Save the fine-tuned model."""
        save_path = path or self.output_dir
        self.model.save_pretrained(save_path)
        self.tokenizer.save_pretrained(save_path)
```

```python
# RAG (Retrieval-Augmented Generation) Implementation
from langchain.embeddings import HuggingFaceEmbeddings
from langchain.vectorstores import Chroma
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.llms import HuggingFacePipeline
from langchain.chains import RetrievalQA
from transformers import pipeline
import torch


class RAGSystem:
    """Production RAG system with vector store and LLM."""

    def __init__(
        self,
        embedding_model: str = "BAAI/bge-base-en-v1.5",
        llm_model: str = "mistralai/Mistral-7B-Instruct-v0.2",
        collection_name: str = "documents",
        persist_directory: str = "./chroma_db",
    ):
        self.embedding_model = embedding_model
        self.llm_model = llm_model
        self.collection_name = collection_name
        self.persist_directory = persist_directory

        # Initialize components
        self.embeddings = self._init_embeddings()
        self.vector_store = self._init_vector_store()
        self.llm = self._init_llm()
        self.qa_chain = self._init_qa_chain()

    def _init_embeddings(self):
        """Initialize embedding model."""
        return HuggingFaceEmbeddings(
            model_name=self.embedding_model,
            model_kwargs={"device": "cuda"},
            encode_kwargs={"normalize_embeddings": True},
        )

    def _init_vector_store(self):
        """Initialize or load vector store."""
        return Chroma(
            collection_name=self.collection_name,
            embedding_function=self.embeddings,
            persist_directory=self.persist_directory,
        )

    def _init_llm(self):
        """Initialize the LLM."""
        pipe = pipeline(
            "text-generation",
            model=self.llm_model,
            torch_dtype=torch.bfloat16,
            device_map="auto",
            max_new_tokens=512,
            do_sample=True,
            temperature=0.7,
            top_p=0.95,
            repetition_penalty=1.15,
        )
        return HuggingFacePipeline(pipeline=pipe)

    def _init_qa_chain(self):
        """Initialize the QA chain."""
        return RetrievalQA.from_chain_type(
            llm=self.llm,
            chain_type="stuff",
            retriever=self.vector_store.as_retriever(
                search_type="mmr",
                search_kwargs={"k": 5, "fetch_k": 10},
            ),
            return_source_documents=True,
        )

    def ingest_documents(
        self,
        documents: list[str],
        chunk_size: int = 1000,
        chunk_overlap: int = 200,
        metadata: list[dict] | None = None,
    ):
        """Ingest documents into the vector store."""
        text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=chunk_size,
            chunk_overlap=chunk_overlap,
            length_function=len,
            separators=["\n\n", "\n", ". ", " ", ""],
        )

        chunks = []
        metadatas = []

        for i, doc in enumerate(documents):
            doc_chunks = text_splitter.split_text(doc)
            chunks.extend(doc_chunks)

            doc_metadata = metadata[i] if metadata else {}
            for j, _ in enumerate(doc_chunks):
                metadatas.append({
                    **doc_metadata,
                    "chunk_index": j,
                    "doc_index": i,
                })

        self.vector_store.add_texts(texts=chunks, metadatas=metadatas)
        self.vector_store.persist()

        return len(chunks)

    def query(
        self,
        question: str,
        return_sources: bool = True,
    ) -> dict:
        """Query the RAG system."""
        result = self.qa_chain({"query": question})

        response = {
            "answer": result["result"],
            "question": question,
        }

        if return_sources:
            response["sources"] = [
                {
                    "content": doc.page_content[:500],
                    "metadata": doc.metadata,
                }
                for doc in result.get("source_documents", [])
            ]

        return response

    def semantic_search(
        self,
        query: str,
        k: int = 5,
        filter_metadata: dict | None = None,
    ) -> list[dict]:
        """Perform semantic search without LLM generation."""
        results = self.vector_store.similarity_search_with_score(
            query,
            k=k,
            filter=filter_metadata,
        )

        return [
            {
                "content": doc.page_content,
                "metadata": doc.metadata,
                "score": float(score),
            }
            for doc, score in results
        ]
```

````python
# Prompt Engineering Framework
from dataclasses import dataclass, field
from typing import Callable
import json


@dataclass
class PromptTemplate:
    """Structured prompt template with variables."""

    template: str
    input_variables: list[str] = field(default_factory=list)
    output_parser: Callable[[str], any] | None = None

    def format(self, **kwargs) -> str:
        """Format the template with provided variables."""
        missing = set(self.input_variables) - set(kwargs.keys())
        if missing:
            raise ValueError(f"Missing variables: {missing}")

        return self.template.format(**kwargs)

    def parse_output(self, output: str) -> any:
        """Parse the LLM output."""
        if self.output_parser:
            return self.output_parser(output)
        return output


class PromptLibrary:
    """Collection of reusable prompt templates."""

    SYSTEM_PROMPTS = {
        "assistant": "You are a helpful AI assistant.",
        "coder": """You are an expert software engineer. You write clean,
well-documented code and follow best practices.""",
        "analyst": """You are a data analyst. You provide clear,
data-driven insights and recommendations.""",
    }

    @staticmethod
    def chain_of_thought() -> PromptTemplate:
        """Chain-of-thought reasoning template."""
        return PromptTemplate(
            template="""Question: {question}

Let's solve this step by step:

1. First, I'll identify the key components of this problem.
2. Then, I'll analyze each component.
3. Finally, I'll synthesize the findings into a conclusion.

Step-by-step reasoning:""",
            input_variables=["question"],
        )

    @staticmethod
    def few_shot(examples: list[dict]) -> PromptTemplate:
        """Few-shot learning template."""
        example_text = "\n\n".join(
            f"Input: {ex['input']}\nOutput: {ex['output']}"
            for ex in examples
        )

        return PromptTemplate(
            template=f"""Here are some examples:

{example_text}

Now, given the following input, provide the output:

Input: {{input}}
Output:""",
            input_variables=["input"],
        )

    @staticmethod
    def structured_output(schema: dict) -> PromptTemplate:
        """Template for structured JSON output."""
        schema_str = json.dumps(schema, indent=2)

        def parse_json(output: str) -> dict:
            # Extract JSON from output
            import re
            json_match = re.search(r'\{[\s\S]*\}', output)
            if json_match:
                return json.loads(json_match.group())
            raise ValueError("No valid JSON found in output")

        return PromptTemplate(
            template=f"""Given the following input, provide a response in JSON format
matching this schema:

```json
{schema_str}
````

Input: {{input}}

Respond only with valid JSON:""",
input_variables=["input"],
output_parser=parse_json,
)

    @staticmethod
    def rag_qa() -> PromptTemplate:
        """RAG question-answering template."""
        return PromptTemplate(
            template="""Use the following context to answer the question.

If you cannot answer based on the context, say "I don't have enough information."

Context:
{context}

Question: {question}

Answer:""",
input_variables=["context", "question"],
)

````

## Best Practices

### Fine-Tuning
- Start with QLoRA for memory efficiency
- Use appropriate learning rates (1e-5 to 5e-4)
- Monitor for catastrophic forgetting
- Validate on held-out data
- Use gradient checkpointing for large models

### RAG
- Chunk documents appropriately (512-1024 tokens)
- Use hybrid search (semantic + keyword)
- Implement re-ranking for better relevance
- Cache embeddings for performance
- Monitor retrieval quality metrics

### Prompt Engineering
- Be specific and clear in instructions
- Use structured output formats
- Implement few-shot learning when helpful
- Chain prompts for complex tasks
- Test prompts systematically

### Production Deployment
- Use vLLM or TGI for inference
- Implement proper caching
- Monitor token usage and costs
- Set up rate limiting
- Log prompts and responses for debugging

## Evaluation Metrics

```python
from evaluate import load

# Common LLM evaluation metrics
perplexity = load("perplexity")
bleu = load("bleu")
rouge = load("rouge")
bertscore = load("bertscore")

def evaluate_generation(predictions: list[str], references: list[str]):
    """Evaluate generated text quality."""
    return {
        "bleu": bleu.compute(predictions=predictions, references=references),
        "rouge": rouge.compute(predictions=predictions, references=references),
        "bertscore": bertscore.compute(
            predictions=predictions,
            references=references,
            lang="en"
        ),
    }
````

You build production-ready LLM applications with proper fine-tuning, RAG architectures, and prompt engineering.