scripts/dual_task_sft_pipeline.py

"""
Dual-task SFT pipeline: train model on both question generation and solution tasks.

This pipeline trains a single model that can:
1. Generate math questions when prompted with "### Task: Generate Question"
2. Solve math problems when prompted with "### Task: Solve Problem"

Examples
--------
  # Train dual-task model
  python scripts/dual_task_sft_pipeline.py train \\
      --data data/sft/dual_task_train.jsonl \\
      --output-dir checkpoints/dual_task_v1 \\
      --epochs 2

  # Infer - Question Generation
  python scripts/dual_task_sft_pipeline.py infer \\
      --adapter checkpoints/dual_task_v1 \\
      --task generate \\
      --prompt "Create a word problem about fractions and money requiring 3 steps."

  # Infer - Solution Generation
  python scripts/dual_task_sft_pipeline.py infer \\
      --adapter checkpoints/dual_task_v1 \\
      --task solve \\
      --problem "Janet has 16 eggs. She eats 3. How many are left?"

Dependencies: torch, transformers, peft, datasets, accelerate, bitsandbytes, trl
"""

from __future__ import annotations

import os

if "HF_HUB_DISABLE_XET" not in os.environ:
    os.environ["HF_HUB_DISABLE_XET"] = "1"

import argparse
import json
import math
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))

from src.config.prompts import (
    SOLVE_TASK_PREFIX,
    GENERATE_TASK_PREFIX,
    SOLVER_SYSTEM_PROMPT,
    GENERATOR_SYSTEM_PROMPT,
)


def _warmup_steps_from_ratio(
    num_examples: int,
    per_device_train_batch_size: int,
    gradient_accumulation_steps: int,
    num_train_epochs: float,
    warmup_ratio: float,
) -> int:
    """Calculate warmup steps from ratio."""
    if warmup_ratio <= 0:
        return 0
    num_batches = max(
        1,
        (num_examples + per_device_train_batch_size - 1) // per_device_train_batch_size,
    )
    num_update_steps_per_epoch = max(1, num_batches // gradient_accumulation_steps)
    total_optimizer_steps = max(1, math.ceil(num_train_epochs * num_update_steps_per_epoch))
    return min(total_optimizer_steps, int(total_optimizer_steps * warmup_ratio))


def cmd_train(args: argparse.Namespace) -> None:
    try:
        import torch
        from datasets import load_dataset
        from peft import LoraConfig, get_peft_model, prepare_model_for_kbit_training
        from transformers import AutoModelForCausalLM, AutoTokenizer, BitsAndBytesConfig
        from trl import SFTConfig, SFTTrainer
    except ImportError as e:
        raise SystemExit(
            "Missing dependency for training. Install:\n"
            "  pip install torch transformers peft datasets accelerate bitsandbytes trl\n"
            f"Original error: {e}"
        ) from e

    data_path = Path(args.data)
    if not data_path.is_file():
        raise SystemExit(f"Data file not found: {data_path}")

    out_dir = Path(args.output_dir)
    out_dir.mkdir(parents=True, exist_ok=True)

    compute_dtype = getattr(torch, args.bnb_compute_dtype)
    bnb_config = BitsAndBytesConfig(
        load_in_4bit=True,
        bnb_4bit_compute_dtype=compute_dtype,
        bnb_4bit_quant_type="nf4",
        bnb_4bit_use_double_quant=True,
    )

    tokenizer = AutoTokenizer.from_pretrained(args.model, trust_remote_code=True)
    if tokenizer.pad_token is None:
        tokenizer.pad_token = tokenizer.eos_token
    tokenizer.padding_side = "right"

    print(f"Loading model {args.model} …")
    model = AutoModelForCausalLM.from_pretrained(
        args.model,
        quantization_config=bnb_config,
        device_map="auto",
        trust_remote_code=True,
        dtype=compute_dtype,
    )
    model = prepare_model_for_kbit_training(model)
    
    peft = LoraConfig(
        r=args.lora_rank,
        lora_alpha=args.lora_alpha,
        lora_dropout=args.lora_dropout,
        bias="none",
        task_type="CAUSAL_LM",
        target_modules=list(args.target_modules.split(",")),
    )
    model = get_peft_model(model, peft)
    model.config.use_cache = False
    model.print_trainable_parameters()

    print(f"Loading dual-task dataset from {data_path} …")
    ds = load_dataset("json", data_files=str(data_path), split="train")
    if args.max_samples and args.max_samples > 0:
        ds = ds.select(range(min(args.max_samples, len(ds))))
    
    task_counts = {"solve": 0, "generate": 0, "unknown": 0}
    for example in ds:
        task_type = example.get("task_type", "unknown")
        task_counts[task_type] = task_counts.get(task_type, 0) + 1
    
    print(f"Dataset composition:")
    print(f"  Total examples: {len(ds)}")
    print(f"  Solve tasks: {task_counts['solve']} ({task_counts['solve']/len(ds):.1%})")
    print(f"  Generate tasks: {task_counts['generate']} ({task_counts['generate']/len(ds):.1%})")
    if task_counts['unknown'] > 0:
        print(f"  Unknown tasks: {task_counts['unknown']}")

    def formatting_func(example):
        return tokenizer.apply_chat_template(
            example["messages"],
            tokenize=False,
            add_generation_prompt=False,
        )

    if args.warmup_steps is not None:
        warmup_steps = max(0, args.warmup_steps)
    else:
        warmup_steps = _warmup_steps_from_ratio(
            len(ds),
            args.batch_size,
            args.grad_accum,
            args.epochs,
            args.warmup_ratio,
        )

    sft_args = SFTConfig(
        output_dir=str(out_dir),
        num_train_epochs=args.epochs,
        per_device_train_batch_size=args.batch_size,
        gradient_accumulation_steps=args.grad_accum,
        learning_rate=args.learning_rate,
        logging_steps=args.logging_steps,
        save_steps=args.save_steps,
        save_total_limit=3,
        bf16=args.bf16 and torch.cuda.is_available(),
        fp16=args.fp16 and torch.cuda.is_available() and not args.bf16,
        max_length=args.max_seq_length,
        warmup_steps=warmup_steps,
        lr_scheduler_type="cosine",
        report_to="none",
        gradient_checkpointing=True,
    )

    print("\nStarting dual-task training...")
    trainer = SFTTrainer(
        model=model,
        args=sft_args,
        train_dataset=ds,
        processing_class=tokenizer,
        formatting_func=formatting_func,
    )

    trainer.train()
    trainer.save_model(str(out_dir))
    tokenizer.save_pretrained(str(out_dir))

    with (out_dir / "pipeline_meta.json").open("w", encoding="utf-8") as f:
        json.dump(
            {
                "pipeline_type": "dual_task",
                "base_model": args.model,
                "data": str(data_path),
                "lora_rank": args.lora_rank,
                "epochs": args.epochs,
                "task_distribution": task_counts,
            },
            f,
            indent=2,
        )
    print(f"\nSaved dual-task adapter and tokenizer to {out_dir}")


def cmd_infer(args: argparse.Namespace) -> None:
    import torch
    from peft import PeftModel
    from transformers import AutoModelForCausalLM, AutoTokenizer, BitsAndBytesConfig

    adapter = Path(args.adapter)
    meta_path = adapter / "pipeline_meta.json"
    base_model = args.base_model
    
    if meta_path.is_file():
        meta = json.loads(meta_path.read_text(encoding="utf-8"))
        base_model = meta.get("base_model", base_model)
        pipeline_type = meta.get("pipeline_type", "unknown")
        if pipeline_type != "dual_task":
            print(f"Warning: Adapter trained with pipeline_type='{pipeline_type}', expected 'dual_task'")

    compute_dtype = getattr(torch, args.bnb_compute_dtype)
    bnb_config = BitsAndBytesConfig(
        load_in_4bit=True,
        bnb_4bit_compute_dtype=compute_dtype,
        bnb_4bit_quant_type="nf4",
        bnb_4bit_use_double_quant=True,
    )

    tokenizer = AutoTokenizer.from_pretrained(adapter, trust_remote_code=True)
    if tokenizer.pad_token is None:
        tokenizer.pad_token = tokenizer.eos_token

    print(f"Loading base {base_model} + adapter {adapter} …")
    base = AutoModelForCausalLM.from_pretrained(
        base_model,
        quantization_config=bnb_config,
        device_map="auto",
        trust_remote_code=True,
    )
    model = PeftModel.from_pretrained(base, str(adapter))
    model.eval()

    if args.task == "solve":
        system_prompt = SOLVER_SYSTEM_PROMPT
        user_content = (
            f"{SOLVE_TASK_PREFIX}"
            "Solve the following problem. Show your reasoning as numbered steps, "
            "then give the final numeric answer on the last line.\n\n"
            f"Problem:\n{args.problem.strip()}"
        )
    elif args.task == "generate":
        system_prompt = GENERATOR_SYSTEM_PROMPT
        user_content = f"{GENERATE_TASK_PREFIX}{args.prompt.strip()}"
    else:
        raise ValueError(f"Unknown task: {args.task}. Must be 'solve' or 'generate'")

    messages = [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": user_content},
    ]
    
    prompt = tokenizer.apply_chat_template(
        messages, tokenize=False, add_generation_prompt=True
    )
    inputs = tokenizer(prompt, return_tensors="pt").to(model.device)

    print(f"\nTask: {args.task}")
    print(f"Prompt length: {inputs['input_ids'].shape[1]} tokens")
    print("\nGenerating...")

    with torch.no_grad():
        out = model.generate(
            **inputs,
            max_new_tokens=args.max_new_tokens,
            temperature=args.temperature,
            top_p=args.top_p,
            do_sample=not args.greedy,
            pad_token_id=tokenizer.pad_token_id,
        )

    gen_ids = out[0, inputs["input_ids"].shape[1] :]
    text = tokenizer.decode(gen_ids, skip_special_tokens=True).strip()
    
    print("\n" + "=" * 60)
    print("Generated Output")
    print("=" * 60)
    print(text)
    print("=" * 60)

    if args.task == "solve":
        print("\n--- Format Validation ---")
        from src.sft.solution_format import validate_sympy_solution_format
        r = validate_sympy_solution_format(text)
        print(json.dumps(r.__dict__, indent=2))


def build_parser() -> argparse.ArgumentParser:
    p = argparse.ArgumentParser(description="Dual-task SFT pipeline (train / infer)")
    sub = p.add_subparsers(dest="command", required=True)

    tr = sub.add_parser("train", help="Train dual-task model on mixed dataset")
    tr.add_argument("--data", type=str, required=True, help="Dual-task training JSONL")
    tr.add_argument("--output-dir", type=str, required=True, help="Output directory for adapter")
    tr.add_argument("--model", type=str, default="Qwen/Qwen2.5-Math-1.5B-Instruct", help="Base model")
    tr.add_argument("--epochs", type=float, default=2.0, help="Training epochs (default: 2.0 for dual-task)")
    tr.add_argument("--batch-size", type=int, default=1)
    tr.add_argument("--grad-accum", type=int, default=8)
    tr.add_argument("--learning-rate", type=float, default=2e-4)
    tr.add_argument("--max-samples", type=int, default=0, help="0 = use full dataset")
    tr.add_argument("--lora-rank", type=int, default=16)
    tr.add_argument("--lora-alpha", type=int, default=32)
    tr.add_argument("--lora-dropout", type=float, default=0.05)
    tr.add_argument(
        "--target-modules",
        type=str,
        default="q_proj,v_proj,o_proj,gate_proj",
    )
    tr.add_argument("--max-seq-length", type=int, default=2048)
    tr.add_argument("--save-steps", type=int, default=200)
    tr.add_argument("--logging-steps", type=int, default=10)
    tr.add_argument("--warmup-ratio", type=float, default=0.03)
    tr.add_argument("--warmup-steps", type=int, default=None)
    tr.add_argument("--bf16", action="store_true", default=True)
    tr.add_argument("--no-bf16", dest="bf16", action="store_false")
    tr.add_argument("--fp16", action="store_true")
    tr.add_argument("--bnb-compute-dtype", type=str, default="bfloat16")
    tr.set_defaults(func=cmd_train)

    inf = sub.add_parser("infer", help="Generate with dual-task model")
    inf.add_argument("--adapter", type=str, required=True, help="Adapter directory")
    inf.add_argument(
        "--base-model",
        type=str,
        default="Qwen/Qwen2.5-Math-1.5B-Instruct",
        help="Base model (auto-detected from pipeline_meta.json if present)",
    )
    inf.add_argument(
        "--task",
        type=str,
        required=True,
        choices=["solve", "generate"],
        help="Task type: 'solve' for problem solving, 'generate' for question generation",
    )
    inf.add_argument(
        "--problem",
        type=str,
        default="",
        help="Math problem to solve (required if --task solve)",
    )
    inf.add_argument(
        "--prompt",
        type=str,
        default="",
        help="Question generation prompt (required if --task generate)",
    )
    inf.add_argument("--max-new-tokens", type=int, default=1024)
    inf.add_argument("--temperature", type=float, default=0.7)
    inf.add_argument("--top-p", type=float, default=0.95)
    inf.add_argument("--greedy", action="store_true", help="Use greedy decoding")
    inf.add_argument("--bnb-compute-dtype", type=str, default="bfloat16")
    inf.set_defaults(func=cmd_infer)

    return p


def main() -> None:
    parser = build_parser()
    args = parser.parse_args()
    
    if args.command == "infer":
        if args.task == "solve" and not args.problem:
            raise SystemExit("Error: --problem is required when --task solve")
        if args.task == "generate" and not args.prompt:
            raise SystemExit("Error: --prompt is required when --task generate")
    
    if str(ROOT) not in sys.path:
        sys.path.insert(0, str(ROOT))
    
    args.func(args)


if __name__ == "__main__":
    main()