Gemma 4 Fine-Tuning: Production Recipes for 2026

1{
2    "messages": [
3        {
4            "role": "system",
5            "content": "You are a technical support specialist for a PostgreSQL database product."
6        },
7        {
8            "role": "user", 
9            "content": "I'm getting 'connection refused' errors. What should I check?"
10        },
11        {
12            "role": "assistant",
13            "content": "Let's troubleshoot systematically:\n\n1. **Verify PostgreSQL is running**..."
14        }
15    ]
16}

1import pandas as pd
2from transformers import AutoTokenizer
3
4class DataValidator:
5    def __init__(self, model_name="google/gemma-4-9b"):
6        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
7        self.max_length = 8192
8    
9    def validate_dataset(self, dataset_path: str):
10        df = pd.read_json(dataset_path, lines=True)
11        issues = []
12        
13        for idx, row in df.iterrows():
14            messages = row["messages"]
15            
16            valid_roles = {"system", "user", "assistant", "tool"}
17            for msg in messages:
18                if msg["role"] not in valid_roles:
19                    issues.append(f"Row {idx}: Invalid role '{msg['role']}'")
20            
21            for i in range(1, len(messages)):
22                if messages[i]["role"] == messages[i-1]["role"]:
23                    issues.append(f"Row {idx}: Duplicate roles at positions {i-1},{i}")
24            
25            full_text = "\n".join(m["content"] for m in messages)
26            tokens = self.tokenizer.encode(full_text)
27            if len(tokens) > self.max_length:
28                issues.append(f"Row {idx}: {len(tokens)} tokens > {self.max_length} limit")
29            
30            assistant_msgs = [m for m in messages if m["role"] == "assistant"]
31            if not all(len(m["content"]) > 20 for m in assistant_msgs):
32                issues.append(f"Row {idx}: Short assistant response detected")
33        
34        return issues

1class DataAugmenter:
2    def paraphrase_user_queries(self, dataset, num_variations=3):
3        """Generate semantic variations of user prompts"""
4        
5        paraphrase_prompt = """Generate {n} different ways a user might ask this question.
6        Keep the intent identical but vary phrasing, formality, and length:
7        
8        Original: {original}
9        
10        Variations:"""
11        
12        augmented = []
13        for example in dataset:
14            user_msg = next(m for m in example["messages"] if m["role"] == "user")
15            
16            variations = self.model.generate(
17                paraphrase_prompt.format(n=num_variations, original=user_msg["content"])
18            )
19            
20            for variation in variations:
21                new_example = deepcopy(example)
22                user_msg_idx = next(i for i, m in enumerate(new_example["messages"]) if m["role"] == "user")
23                new_example["messages"][user_msg_idx]["content"] = variation
24                augmented.append(new_example)
25        
26        return dataset + augmented

1from unsloth import FastLanguageModel
2from trl import SFTTrainer
3from transformers import TrainingArguments
4from datasets import load_dataset
5
6model, tokenizer = FastLanguageModel.from_pretrained(
7    model_name="google/gemma-4-9b",
8    max_seq_length=8192,
9    load_in_4bit=True,
10    fast_inference=True,
11)
12
13model = FastLanguageModel.get_peft_model(
14    model,
15    r=64,              # LoRA rank (higher = more capacity)
16    lora_alpha=128,    # Scaling factor (typically 2x rank)
17    target_modules=[
18        "q_proj", "k_proj", "v_proj", "o_proj",
19        "gate_proj", "up_proj", "down_proj",
20    ],
21    lora_dropout=0.05,
22    bias="none",
23    use_gradient_checkpointing="unsloth",
24    random_state=42,
25)
26
27dataset = load_dataset("json", data_files="training_data.jsonl", split="train")
28
29training_args = TrainingArguments(
30    output_dir="./gemma-4-9b-finetuned",
31    num_train_epochs=3,
32    per_device_train_batch_size=2,
33    gradient_accumulation_steps=4,
34    warmup_steps=100,
35    learning_rate=2e-4,
36    optim="adamw_8bit",
37    weight_decay=0.01,
38    lr_scheduler_type="cosine",
39    seed=42,
40    report_to="wandb",
41)
42
43trainer = SFTTrainer(
44    model=model,
45    tokenizer=tokenizer,
46    train_dataset=dataset,
47    dataset_text_field="messages",
48    max_seq_length=8192,
49    args=training_args,
50)
51
52trainer.train()
53
54model.save_pretrained("./gemma-4-9b-lora")
55tokenizer.save_pretrained("./gemma-4-9b-lora")

1{
2    "fp16": {
3        "enabled": true,
4        "loss_scale": 0,
5        "loss_scale_window": 1000,
6        "initial_scale_power": 16,
7        "hysteresis": 2,
8        "min_loss_scale": 1
9    },
10    "zero_optimization": {
11        "stage": 2,
12        "offload_optimizer": {
13            "device": "cpu",
14            "pin_memory": true
15        },
16        "allgather_partitions": true,
17        "allgather_bucket_size": 2e8,
18        "overlap_comm": true,
19        "reduce_scatter": true,
20        "reduce_bucket_size": 2e8,
21        "contiguous_gradients": true
22    },
23    "train_batch_size": "auto",
24    "train_micro_batch_size_per_gpu": "auto",
25    "gradient_accumulation_steps": "auto"
26}

1from accelerate import Accelerator
2from transformers import AutoModelForCausalLM, TrainingArguments
3from trl import SFTTrainer
4
5accelerator = Accelerator(deepspeed_plugin=deepspeed_plugin)
6
7model = AutoModelForCausalLM.from_pretrained(
8    "google/gemma-4-27b",
9    torch_dtype=torch.bfloat16,
10)
11
12training_args = TrainingArguments(
13    output_dir="./gemma-4-27b-full",
14    deepspeed="./deepspeed_config.json",
15    num_train_epochs=2,
16    per_device_train_batch_size=1,
17    gradient_accumulation_steps=8,
18    learning_rate=1e-5,  # Lower LR for full fine-tuning
19    warmup_ratio=0.03,
20    lr_scheduler_type="cosine",
21    save_strategy="epoch",
22    logging_steps=10,
23)
24
25trainer = SFTTrainer(
26    model=model,
27    train_dataset=dataset,
28    args=training_args,
29)
30
31trainer.train()

1{
2    "messages": [
3        {"role": "user", "content": "Book me a flight to Tokyo next Tuesday"},
4        {"role": "assistant", "content": "", "tool_calls": [
5            {"name": "search_flights", "arguments": {"destination": "Tokyo", "date": "2026-05-12"}}
6        ]},
7        {"role": "tool", "name": "search_flights", "content": "[...flight data...]"},
8        {"role": "assistant", "content": "I found 3 flights... Which would you prefer?"}
9    ]
10}
11
12tool_schema = json.dumps([{
13    "type": "function",
14    "function": {
15        "name": "search_flights",
16        "parameters": {...}
17    }
18}])
19
20system_msg = f"You have access to these tools: {tool_schema}\nUse them when appropriate."

1from torch.optim import AdamW
2import matplotlib.pyplot as plt
3
4class LRFinder:
5    def find(self, model, dataset, min_lr=1e-6, max_lr=1e-3, num_steps=100):
6        optimizer = AdamW(model.parameters(), lr=min_lr)
7        
8        lr_mult = (max_lr / min_lr) ** (1 / num_steps)
9        lr = min_lr
10        
11        losses = []
12        lrs = []
13        
14        for step, batch in enumerate(dataset):
15            if step >= num_steps:
16                break
17                
18            optimizer.param_groups[0]["lr"] = lr
19            
20            loss = model(**batch).loss
21            loss.backward()
22            optimizer.step()
23            optimizer.zero_grad()
24            
25            losses.append(loss.item())
26            lrs.append(lr)
27            
28            lr *= lr_mult
29        
30        plt.plot(lrs, losses)
31        plt.xscale("log")
32        plt.xlabel("Learning Rate")
33        plt.ylabel("Loss")
34        
35        optimal_idx = losses.index(min(losses))
36        optimal_lr = lrs[optimal_idx] / 10
37        
38        return optimal_lr

1from deepeval import evaluate
2from deepeval.metrics import GEval, FaithfulnessMetric
3from deepeval.test_case import LLMTestCase
4
5class ProductionEvaluator:
6    def __init__(self, base_model, finetuned_model):
7        self.base = base_model
8        self.finetuned = finetuned_model
9        self.metrics = [
10            GEval(
11                name="Correctness",
12                criteria="Determine if the response is factually correct",
13                evaluation_params=[LLMTestCaseParams.ACTUAL_OUTPUT]
14            ),
15            FaithfulnessMetric(threshold=0.7),
16        ]
17    
18    def evaluate_on_test_set(self, test_cases: list):
19        """Compare base vs fine-tuned on identical inputs"""
20        
21        results = []
22        for case in test_cases:
23            base_output = self.base.generate(case["input"])
24            finetuned_output = self.finetuned.generate(case["input"])
25            
26            test_case = LLMTestCase(
27                input=case["input"],
28                actual_output=finetuned_output,
29                expected_output=case["expected"],
30                retrieval_context=case.get("context", [])
31            )
32            
33            scores = evaluate([test_case], self.metrics)
34            
35            results.append({
36                "input": case["input"][:100],
37                "base_score": self.score_output(base_output, case["expected"]),
38                "finetuned_score": scores[0].score,
39                "improvement": scores[0].score - self.score_output(base_output, case["expected"]),
40                "regression": scores[0].score < 0.5
41            })
42        
43        return results

1class RegressionTester:
2    def test_for_catastrophic_forgetting(self, base_model, finetuned_model):
3        """Ensure model didn't forget general capabilities"""
4        
5        general_tasks = [
6            "Explain quantum computing in simple terms",
7            "Write a Python function to reverse a string",
8            "What is the capital of Australia?",
9            "Summarize the theory of relativity"
10        ]
11        
12        regressions = []
13        for task in general_tasks:
14            base_output = base_model.generate(task)
15            finetuned_output = finetuned_model.generate(task)
16            
17            similarity = self.semantic_similarity(base_output, finetuned_output)
18            
19            if similarity < 0.85:  # Significant divergence
20                regressions.append({
21                    "task": task,
22                    "base": base_output[:200],
23                    "finetuned": finetuned_output[:200],
24                    "similarity": similarity
25                })
26        
27        return regressions

1from peft import AutoPeftModelForCausalLM
2
3model = AutoPeftModelForCausalLM.from_pretrained(
4    "./gemma-4-9b-finetuned",
5    device_map="auto"
6)
7
8merged_model = model.merge_and_unload()
9merged_model.save_pretrained("./gemma-4-9b-merged")
10

1python -m vllm.entrypoints.openai.api_server \
2    --model ./gemma-4-9b-merged \
3    --served-model-name gemma-4-customer-support \
4    --max-model-len 16384 \
5    --quantization fp8 \
6    --gpu-memory-utilization 0.92
7
8curl http://localhost:8000/v1/chat/completions \
9  -H "Content-Type: application/json" \
10  -d '{
11    "model": "gemma-4-customer-support",
12    "messages": [{"role": "user", "content": "My database is slow"}]
13  }'

1from prometheus_client import Counter, Histogram, Gauge
2import time
3
4class ProductionMonitor:
5    def __init__(self):
6        self.inference_latency = Histogram(
7            "gemma4_inference_latency_seconds",
8            "End-to-end inference time",
9            ["model_version", "quantization"]
10        )
11        self.token_throughput = Gauge(
12            "gemma4_tokens_per_second",
13            "Generation speed",
14            ["model_version"]
15        )
16        self.drift_detector = DriftDetector(window_size=1000)
17    
18    def monitor_inference(self, model_version, quantization, prompt, output):
19        start = time.time()
20        result = self.model.generate(prompt)
21        latency = time.time() - start
22        
23        self.inference_latency.labels(
24            model_version=model_version,
25            quantization=quantization
26        ).observe(latency)
27        
28        self.drift_detector.add_sample(output)
29        if self.drift_detector.detect_drift():
30            self.alert_engineer("Output distribution drift detected!")
31        
32        if self.contains_unexpected_patterns(output):
33            self.log_suspicious_output(prompt, output)
34    
35    def contains_unexpected_patterns(self, text: str) -> bool:
36        """Detect potential fine-tuning degradation"""
37        indicators = [
38            r"I'm sorry, but I",  # Generic refusal patterns returning
39            r"As an AI language model",  # Base model leakage
40            r"\[object Object\]",  # JSON formatting errors
41            r"undefined|null",  # Code generation regression
42        ]
43        return any(re.search(pattern, text) for pattern in indicators)