# SkillPatch skill: finetuning This skill guides agents through the full fine-tuning lifecycle on Azure AI Foundry, supporting Supervised Fine-Tuning (SFT), Direct Preference Optimization (DPO), and Reinforcement Fine-Tuning (RFT) with graders. It covers dataset preparation and validation, training job submission and monitoring, grader calibration, model deployment, and evaluation. Helper scripts and reference guides are provided for every stage of the pipeline. You (the agent) were given this URL and asked to use this skill. This is a **public** skill — no SkillPatch account, API key, or prior setup is required. Two ways to use it: **1. Use it right now** — the skill's complete file tree (SKILL.md + all reference files) is inlined below; read `SKILL.md` first, then follow it, consulting the other files as it directs. **2. Install the exact package onto disk** (recommended if you can run a shell — this reproduces the skill byte-for-byte, including any binary assets that can't be inlined): ```bash mkdir -p .claude/skills/finetuning curl -sSL https://skillpatch.dev/install_skill/finetuning | tar -xz -C .claude/skills/ ``` (`.claude/skills/` is Claude Code's convention; use whatever directory your agent loads skills from.) --- ## Skill files (34) - `SKILL.md` - `references/agentic-rft.md` - `references/dataset-formats.md` - `references/deployment.md` - `references/evaluation.md` - `references/grader-design.md` - `references/hyperparameters.md` - `references/large-file-uploads.md` - `references/platform-gotchas.md` - `references/reward-hacking.md` - `references/training-curves.md` - `references/training-types.md` - `references/vision-fine-tuning.md` - `scripts/calibrate_grader.py` - `scripts/check_training.py` - `scripts/cleanup.py` - `scripts/common.py` - `scripts/convert_dataset.py` - `scripts/deploy_model.py` - `scripts/evaluate_model.py` - `scripts/generate_distillation_data.py` - `scripts/monitor_training.py` - `scripts/score_dataset.py` - `scripts/submit_training.py` - `scripts/validate/__init__.py` - `scripts/validate/data_stats.py` - `scripts/validate/validate_dpo.py` - `scripts/validate/validate_rft.py` - `scripts/validate/validate_sft.py` - `workflows/dataset-creation.md` - `workflows/diagnose-poor-results.md` - `workflows/full-pipeline.md` - `workflows/iterative-training.md` - `workflows/quickstart.md` ### `SKILL.md` ```markdown --- name: finetuning description: "Fine-tune models on Azure AI Foundry using SFT (supervised), DPO (preference), or RFT (reinforcement with graders). Covers dataset preparation, training job submission, deployment, and evaluation. USE FOR: fine-tune, SFT, DPO, RFT, training data, grader, distillation, fine-tuned model, training job, large file upload, calibrate grader, deploy fine-tuned model, evaluate fine-tuned model. DO NOT USE FOR: general model deployment without fine-tuning (use deploy-model), agent creation (use agents), prompt optimization without training (use prompt-optimizer)." license: MIT metadata: author: Microsoft version: "0.0.0-placeholder" --- # Fine-Tuning on Azure AI Foundry Fine-tune models using SFT (supervised), DPO (preference), or RFT (reinforcement with graders). Covers dataset prep, training, deployment, and evaluation. ## When to Use Use this sub-skill when the user asks about: - Fine-tuning a model (SFT, DPO, or RFT) - Preparing, validating, or formatting training data - Submitting, monitoring, or diagnosing training jobs - Calibrating graders or pass thresholds for RFT - Deploying or evaluating a fine-tuned model - Choosing between training types (SFT vs DPO vs RFT) - Distillation, synthetic data generation, or dataset quality scoring - Large file uploads for training data - Cleaning up fine-tuning resources (files, deployments) **Do NOT use for:** General model deployment without fine-tuning (use deploy-model), agent creation (use agents), prompt optimization without training (use prompt-optimizer). ## Workflows | Stage | Guide | |-------|-------| | **Quick start** | [workflows/quickstart.md](workflows/quickstart.md) | | **Full pipeline** | [workflows/full-pipeline.md](workflows/full-pipeline.md) | | **Create data** | [workflows/dataset-creation.md](workflows/dataset-creation.md) | | **Iterate** | [workflows/iterative-training.md](workflows/iterative-training.md) | | **Diagnose** | [workflows/diagnose-poor-results.md](workflows/diagnose-poor-results.md) | ## References | Topic | File | |-------|------| | SFT vs DPO vs RFT | [references/training-types.md](references/training-types.md) | | Hyperparameters | [references/hyperparameters.md](references/hyperparameters.md) | | Data formats | [references/dataset-formats.md](references/dataset-formats.md) | | Grader design (RFT) | [references/grader-design.md](references/grader-design.md) | | Reward hacking | [references/reward-hacking.md](references/reward-hacking.md) | | Agentic RFT (tools) | [references/agentic-rft.md](references/agentic-rft.md) | | Deployment | [references/deployment.md](references/deployment.md) | | Training curves | [references/training-curves.md](references/training-curves.md) | | Evaluation | [references/evaluation.md](references/evaluation.md) | | Vision fine-tuning | [references/vision-fine-tuning.md](references/vision-fine-tuning.md) | | Large file uploads | [references/large-file-uploads.md](references/large-file-uploads.md) | | Platform gotchas | [references/platform-gotchas.md](references/platform-gotchas.md) | ## Scripts | Script | Purpose | |--------|---------| | `scripts/submit_training.py` | Submit SFT/DPO/RFT jobs | | `scripts/monitor_training.py` | Poll job until completion | | `scripts/calibrate_grader.py` | Find optimal RFT pass_threshold | | `scripts/check_training.py` | Analyze curves, list checkpoints | | `scripts/deploy_model.py` | Deploy via ARM REST API | | `scripts/evaluate_model.py` | LLM judge evaluation | | `scripts/convert_dataset.py` | Convert between SFT/DPO/RFT formats | | `scripts/generate_distillation_data.py` | Generate synthetic training data | | `scripts/score_dataset.py` | Quality scoring on training data | | `scripts/cleanup.py` | Delete old files and deployments | | `scripts/validate/` | Data validators (SFT, DPO, RFT) + stats | ## Rules 1. **Always baseline first** — evaluate the base model before fine-tuning 2. **Validate data** before submitting — run `scripts/validate/validate_sft.py` 3. **Calibrate RFT graders** — target 25-50% failure rate on the base model 4. **Evaluate checkpoints** — don't blindly deploy the final one 5. **Measure token cost** alongside accuracy when comparing models ## Quick Reference | Task | Command | |------|---------| | Validate SFT data | `python scripts/validate/validate_sft.py data.jsonl` | | Submit SFT job | `python scripts/submit_training.py --model gpt-4.1-mini --training-file train.jsonl --validation-file val.jsonl --type sft` | | Monitor job | `python scripts/monitor_training.py --job-id ftjob-xxx` | | Analyze curves | `python scripts/check_training.py --job-id ftjob-xxx` | | Deploy model | `python scripts/deploy_model.py --model-id ft:gpt-4.1-mini:... --name my-eval` | | Evaluate model | `python scripts/evaluate_model.py --deployment-name my-eval --test-file test.jsonl` | ## Error Handling | Error | Cause | Fix | |-------|-------|-----| | "API version not supported" | Older `openai` SDK on `/v1/` endpoint | Upgrade to `openai>=1.0` | | "does not support fine-tuning with Standard TrainingType" | OSS model needs `globalStandard` | Use `--use-rest` flag or script auto-falls back | | Job stuck in post-training eval | Under-provisioned tool endpoint (RFT) | Scale to S2+, enable Always On | | "DeploymentNotReady" after ARM succeeds | ARM/data-plane race condition | Delete and recreate deployment, wait 5 min | | Content safety block at deployment | PII-dense training data | Remove problematic document types | ``` ### `references/agentic-rft.md` ````markdown # Agentic RFT — Tool Calling Train reasoning models (o4-mini) for agentic scenarios where the model invokes external tools during chain-of-thought reasoning. > ⚠️ **Access required**: Agentic RFT with tool calling and GPT-5 RFT are behind feature flags. You must request access through the Azure AI Foundry portal or your Microsoft account team. o4-mini RFT without tools is generally available. ## Tool Definition Format ```python tools = [ { "name": "search", "server_url": "https://your-function-app.azurewebsites.net/api/tools", "headers": { "Authorization": "Bearer " } }, { "name": "get_by_id", "server_url": "https://your-function-app.azurewebsites.net/api/tools", "headers": { "Authorization": "Bearer " } } ] ``` ## Submitting an Agentic RFT Job ```python job = client.fine_tuning.jobs.create( model="o4-mini-2025-04-16", training_file=train.id, validation_file=valid.id, method={ "type": "reinforcement", "reinforcement": { "grader": grader, "tools": tools, "max_episode_steps": 10, "hyperparameters": { "eval_interval": 5, "eval_samples": 10, "compute_multiplier": 1.5, "reasoning_effort": "medium" } } } ) ``` ## Tool Response Format Your tool endpoint must return: ```json { "type": "function_call_output", "call_id": "call_12345xyz", "output": "The result of the tool call...", "id": "fc_12345xyz" } ``` ## Tool Endpoint Requirements | Constraint | Limit | |-----------|-------| | Recommended throughput | 50 QPS | | Max input payload | 1 MB | | Max return payload | 1 MB (413 error if exceeded) | | Timeout | 10 minutes | | Parallel calls | Supported — handle race conditions | | Retry on 5xx | 3 attempts, then rollout discarded | | On 4xx | Error serialized and shown to model | **Infrastructure**: Use Always On, sufficient compute (S2+), multiple instances. Under-provisioned endpoints can cause jobs to hang during post-training eval. ## RFT Hyperparameters | Parameter | Description | Recommended Start | |-----------|-------------|-------------------| | `reasoning_effort` | `"low"`, `"medium"`, `"high"` | `"medium"` | | `compute_multiplier` | Scales rollouts per step | `1.5` | | `learning_rate_multiplier` | Scales the learning rate | `1.0` | | `n_epochs` | Data passes | `2–3` | | `eval_interval` | Eval every N steps | `5` | | `eval_samples` | Validation examples per eval | `10` | | `max_episode_steps` | Max tool calls + reasoning steps per rollout | `5–10` | **Notes:** Higher LR increases output verbosity without improving accuracy. Compute multiplier 1.5 balances rollout quality and training time. Platform may early-stop before all epochs. ## When to Use Agentic RFT - Model needs to **decide when to call tools** (not just follow instructions) - Task involves **multi-step reasoning** with external data lookups - Model needs to learn **tool selection** — choosing the right tool for the job - Standard RFT (without tools) can't capture the agentic behavior ```` ### `references/dataset-formats.md` ````markdown # Dataset Formats ## SFT Format (Supervised Fine-Tuning) Standard chat-completion JSONL. Each line: JSON object with `messages` array. ```jsonl {"messages": [{"role": "system", "content": "You are a helpful assistant."}, {"role": "user", "content": "What is 2+2?"}, {"role": "assistant", "content": "4"}]} ``` **Rules:** - Each line must be valid JSON - `messages` must contain at least one `user` and one `assistant` message - `system` message is optional but recommended - Multi-turn supported: alternate `user`/`assistant` - Last message must be `assistant` (that's what the model learns) **Validation checklist:** `.jsonl` extension, valid JSON per line, every example has `messages`, every message has `role` and `content`, no empty `content`. ## DPO Format (Direct Preference Optimization) Three top-level fields: `input`, `preferred_output`, `non_preferred_output`. ```jsonl {"input": {"messages": [{"role": "system", "content": "You are a helpful assistant."}, {"role": "user", "content": "Explain gravity."}]}, "preferred_output": [{"role": "assistant", "content": "Gravity is a fundamental force that attracts objects with mass toward each other."}], "non_preferred_output": [{"role": "assistant", "content": "Gravity is when stuff falls down."}]} ``` **Rules:** - `input`: Object with `messages` array (system + user turns). May include `tools` and `parallel_tool_calls`. - `preferred_output` / `non_preferred_output`: Array of messages (`assistant` or `tool` role only) - Both must contain at least one `assistant` message - Exactly two completions compared per example **DPO REST API example:** ```json { "model": "gpt-4.1-mini-2025-04-14", "training_file": "file-abc123", "method": { "type": "dpo", "dpo": { "beta": 0.1, "l2_multiplier": 0.1 } } } ``` ## RFT Format (Reinforcement Fine-Tuning) Chat-completion format with key differences from SFT: ```jsonl {"messages": [{"role": "user", "content": "Write a Python function to reverse a string."}], "reference_code": "def reverse_string(s):\n return s[::-1]", "expected_output": "olleh"} ``` **Rules:** - Last message **MUST** be `user` role (model generates its own response) - Extra fields alongside `messages` are accessible to grader via `item.*` - Both training and validation datasets are **required** - ⚠️ Do NOT put `assistant` as last message — unlike SFT, RFT generates its own outputs **API version**: Python graders require `api-version=2025-04-01-preview` or later. **Grader types:** `string_check` (exact match), `text_similarity` (fuzzy/BLEU/ROUGE), `python` (custom function), `score_model` (LLM judge), `multi` (weighted combination). **Python grader template:** ```python def grade(sample, item): """ sample: dict with 'output_text' (model's generation) item: dict with extra fields from JSONL Returns: float 0.0–1.0 """ output = sample.get("output_text", "") reference = item.get("reference_code", "") return score ``` **Python grader constraints:** 256KB code max, no network, 2GB memory, 1GB disk, 2min timeout. **Grader field access:** - `sample.output_text` → model's generation - `sample.output_json` → structured output (if using response_format) - `item.*` → extra JSONL fields - Template variables: `{{item.field_name}}` — no spaces inside braces, no array indexing ## Converting Between Formats - **SFT → RFT**: Strip assistant messages (RFT last message must be `user`), add grader reference fields. Use `scripts/convert_dataset.py --format rft`. - **SFT → DPO**: Generate rejected responses (run base model on same prompts, intentionally degrade good outputs, or use human ranking). - **DPO → SFT**: Extract chosen responses from the preferred output. ```` ### `references/deployment.md` ````markdown # Deployment Formats ## Model Format and SKU Mapping | Base model family | `model.format` | `sku.name` | Endpoint type | |-------------------|---------------|------------|---------------| | gpt-4.1-mini | `"OpenAI"` | `"Standard"` | Project | | gpt-4.1-nano | `"OpenAI"` | `"Standard"` | Project | | o4-mini (RFT) | `"OpenAI"` | `"Standard"` | Project | | gpt-oss-20b | `"Microsoft"` | `"GlobalStandard"` | Cognitive Services | | Ministral-3B | `"Mistral AI"` | `"GlobalStandard"` | Cognitive Services | | Llama-3.3-70B | `"Meta"` | `"GlobalStandard"` | Cognitive Services | | Qwen-3-32B | `"Alibaba"` | `"GlobalStandard"` | Cognitive Services | **Format strings are case-sensitive.** `"Mistral AI"` works; `"mistral"` does not. ## Two Endpoint Types **Project Endpoint** (OpenAI models): `https://.services.ai.azure.com/api/projects//openai/v1/` - Use `openai.OpenAI(base_url=..., api_key=...)` — NOT `AzureOpenAI` **Cognitive Services Endpoint** (OSS models): `https://.cognitiveservices.azure.com/openai/deployments//chat/completions?api-version=2025-04-01-preview` - Use `openai.AzureOpenAI(azure_endpoint=..., api_key=..., api_version=...)` ## CLI Deployment (`az cognitiveservices`) The CLI uses **different** format strings than the ARM REST API for OSS models: ```bash az cognitiveservices account deployment create \ --name \ --resource-group \ --deployment-name \ --model-name \ --model-version "1" \ --model-format "OpenAI-OSS" \ --sku-capacity 100 \ --sku-name "GlobalStandard" ``` | Base model family | ARM REST `model.format` | CLI `--model-format` | |-------------------|------------------------|----------------------| | gpt-4.1-mini/nano | `"OpenAI"` | `"OpenAI"` | | gpt-oss-20b | `"Microsoft"` | `"OpenAI-OSS"` | | Ministral-3B | `"Mistral AI"` | `"OpenAI-OSS"` | | Llama-3.3-70B | `"Meta"` | `"OpenAI-OSS"` | | Qwen-3-32B | `"Alibaba"` | `"OpenAI-OSS"` | > ⚠️ Using `"OpenAI-OSS"` in ARM REST or `"Microsoft"` in CLI will fail with HTTP 500. ## ARM REST API Deployment ``` PUT https://management.azure.com/subscriptions/{sub_id}/resourceGroups/{rg}/providers/Microsoft.CognitiveServices/accounts/{account}/deployments/{deploy_name}?api-version=2024-10-01 ``` ```json { "sku": { "name": "GlobalStandard", "capacity": 100 }, "properties": { "model": { "format": "Microsoft", "name": "gpt-oss-20b.ft-{jobid}-suffix", "version": "1" } } } ``` **ARM token:** `az account get-access-token --query accessToken -o tsv` (expires ~60min). ## Capacity Notes - Capacity = tokens-per-minute in thousands. `100` = 100K TPM. - Set capacity ≥ 100 for eval workloads. At capacity=1, OSS FT models hit "Failed to load LoRA" errors. - Quota is per-resource. After deleting a deployment, wait 15–20s before creating a new one. - Deployment names: max 64 chars, alphanumeric + hyphens, unique within resource. ## Common Deployment Errors | Error | Cause | Fix | |-------|-------|-----| | HTTP 500, no message | Wrong `model.format` | Check format table above | | HTTP 409, deployment exists | Name collision | Use unique deployment name | | HTTP 403 | ARM token expired | Refresh token | | HTTP 400, "api-version not allowed" | `AzureOpenAI` client on `/v1/` endpoint | Switch to `openai.OpenAI` | | HTTP 429, quota exceeded | Too many deployments | Delete unused, wait 20s | | ProvisioningState: Failed | Model not available in region | Try different region | ```` ### `references/evaluation.md` ````markdown # Evaluation Methodology ## Principles 1. **Always establish a baseline**: Evaluate the base (un-tuned) model first. Without a baseline, you can't measure improvement. 2. **Use a held-out test set**: Never evaluate on training or validation data. The model has seen those. 3. **Use the same test set for every model**: This is the only way to compare results fairly. 4. **Use task-specific graders**: Built-in generic evaluators (Coherence, Fluency) measure general quality and won't detect fine-tuning improvements. Use custom graders (Python, score model, string check) for task-specific evaluation. 5. **Measure cost alongside accuracy**: Report completion tokens per response when comparing models or checkpoints. A model that achieves the same accuracy with fewer tokens is strictly better — cheaper inference and lower latency. ## Two-Layer Evaluation Strategy Use the **Azure AI Evaluation SDK** (`azure-ai-evaluation`) for all evaluation. | Layer | Purpose | Grader Type | When | |-------|---------|-------------|------| | **Task-specific** (primary) | Measure FT improvement | `AzureOpenAIScoreModelGrader`, `AzureOpenAIPythonGrader`, `AzureOpenAIStringCheckGrader` | Every eval | | **General quality** (guardrail) | Verify model didn't degrade | `CoherenceEvaluator`, `FluencyEvaluator` | Spot-check only | Generic built-in evaluators (Coherence, Fluency, TaskAdherence) are guardrails, not metrics — they often show no difference between base and fine-tuned models even when domain-specific evaluation reveals clear improvement. ## Custom Graders (Primary FT Evaluation) ### 1. Score Model Grader (LLM judge with task-specific rubric) Best for: subjective tasks (summarization, alignment, style). ```python from azure.ai.evaluation import AzureOpenAIScoreModelGrader summarization_grader = AzureOpenAIScoreModelGrader( model_config=model_config, name="summarization_quality", prompt="""Rate this news summary on a scale of 1-5. Article: {{item.article}} Summary: {{sample.output_text}} Criteria: - Captures ALL key facts (who, what, when, where) - No hallucinated information not in the article - Concise (under 3 sentences) Score 1: Missing key facts or hallucinations Score 3: Captures main point but misses details Score 5: Perfect summary — all facts, no extras, concise Return ONLY a number 1-5.""", output_type="numeric", pass_threshold=3, ) ``` ### 2. Python Grader (programmatic/exact-match evaluation) Best for: code generation, math, entity extraction, structured output. ```python from azure.ai.evaluation import AzureOpenAIPythonGrader entity_grader = AzureOpenAIPythonGrader( name="entity_extraction_accuracy", source=""" import json def grade(item, sample): try: extracted = json.loads(sample["output_text"]) reference = json.loads(item["ground_truth"]) except (json.JSONDecodeError, KeyError): return {"score": 0, "reason": "Invalid JSON output"} required_keys = ["people", "organizations", "locations", "dates"] missing = [k for k in required_keys if k not in extracted] if missing: return {"score": 0.5, "reason": f"Missing keys: {missing}"} total, matched = 0, 0 for key in required_keys: ref_set = set(str(v).lower() for v in reference.get(key, [])) ext_set = set(str(v).lower() for v in extracted.get(key, [])) total += len(ref_set) matched += len(ref_set & ext_set) score = matched / total if total > 0 else 1.0 return {"score": score, "reason": f"{matched}/{total} entities matched"} """, pass_threshold=0.7, ) ``` ### 3. String Check Grader (pattern matching) Best for: classification, format compliance, tool calling format. ```python from azure.ai.evaluation import AzureOpenAIStringCheckGrader tool_format_grader = AzureOpenAIStringCheckGrader( name="tool_call_format", input="{{sample.output_text}}", operation="like", # or "eq", "starts_with", "contains" reference="function_call", pass_threshold=1, ) classification_grader = AzureOpenAIStringCheckGrader( name="classification_accuracy", input="{{sample.output_text}}", operation="eq", reference="{{item.expected_label}}", pass_threshold=1, ) ``` ## Running an Evaluation The `evaluate()` function runs multiple graders over an entire dataset: ```python from azure.ai.evaluation import evaluate, F1ScoreEvaluator result = evaluate( data="eval_data.jsonl", evaluators={ "task_grader": my_custom_score_grader, # primary "f1": F1ScoreEvaluator(), # token overlap }, output_path="./eval_results.json", ) for metric, value in result["metrics"].items(): print(f"{metric}: {value}") ``` ## Test Set Design - **Size**: 30–100 examples is sufficient. - **Diversity**: Cover easy/medium/hard, edge cases, and different sub-categories. - **Quality**: Reference answers must be gold-standard correct. A wrong reference penalizes correct outputs. ## Interpreting Results | Score Type | Range | Meaning | |-----------|-------|---------| | AI quality (1–5) | 1–2 Poor, 3 Adequate, 4 Good, 5 Excellent | | | NLP (0–1) | <0.3 Wrong, 0.3–0.6 Partial, 0.6–0.8 Good, >0.8 Strong | | With 50+ eval examples, a difference of ~0.3 points (on 1–5 scale) is usually meaningful. ## Evaluating RFT Models 1. **Evaluate with a DIFFERENT rubric than the training grader** — otherwise you measure overfitting to the grader. 2. Use `F1ScoreEvaluator` for exact-match accuracy. 3. Use `SimilarityEvaluator` to catch semantically correct but differently formatted answers. 4. **Compare against the base model**, not just other fine-tunes. ## Reference - [Azure AI Evaluation SDK docs](https://learn.microsoft.com/en-us/python/api/overview/azure/ai-evaluation-readme) - [Evaluation samples](https://github.com/Azure-Samples/azureai-samples/tree/main/scenarios/evaluate) ```` ### `references/grader-design.md` ````markdown # RFT Grader Design Guide ## Grader Type Selection | Grader Type | Best For | Tradeoffs | |------------|---------|-----------| | **Python grader** (default) | Most tasks incl. tool-calling. Accesses `output_text` and `output_tools`. | Can't call external APIs or execute code. | | **Multi grader** | Combining multiple scoring dimensions. | `score_model` component adds LLM cost per rollout. | | **Endpoint grader** | Tasks requiring external API calls (test suites, DB queries). | HTTP latency, scaling risk. Under-provisioned endpoints can hang jobs. | | **String check** | Exact-match tasks (classification, yes/no, numeric). | Binary 0/1 only — no partial credit. | Start with Python grader unless you need external API calls. Python graders are fast, deterministic, reliable, and tool-aware (`sample.output_tools` provides tool call metadata). ## Partial Credit Pattern Binary pass/fail gives sparse reward. Decompose into 2–4 scored dimensions: ```python def grade(sample, item): output_text = sample.get("output_text", "") or "" expected = item.get("expected_answer", "") score = 0.0 # Core correctness (highest weight) if correct_action(output_text, expected): score += 0.4 # Precision (exact amounts, specific values) score += 0.3 * precision_score(output_text, expected) # Reasoning quality (cited correct rules/facts) score += 0.2 * reasoning_score(output_text, expected) # Process quality (used the right tools) if used_correct_tools(sample.get("output_tools", [])): score += 0.1 return round(min(score, 1.0), 3) ``` ### Weight Guidelines | Dimension | Typical Weight | Examples | |-----------|---------------|----------| | Core correctness | 0.3–0.5 | Right action/answer/classification | | Precision | 0.2–0.3 | Exact amounts, correct format | | Reasoning | 0.1–0.2 | Cited correct rules, justified decision | | Process quality | 0.05–0.1 | Used right tools, followed steps | ## Threshold Calibration Workflow The `pass_threshold` determines what score counts as pass vs fail — the most important RFT hyperparameter. 1. Run the **base model** on your training/validation set 2. Score every output with your grader 3. Compute pass rates at multiple thresholds: ```python for threshold in [0.5, 0.6, 0.7, 0.8, 0.85, 0.9, 0.95]: pass_rate = sum(1 for s in scores if s >= threshold) / len(scores) print(f" @{threshold}: pass={pass_rate:.0%}, fail={1 - pass_rate:.0%}") ``` 4. Choose where **25–50% of base model rollouts fail**: | Failure Rate | Signal Quality | |-------------|----------------| | < 10% | ❌ Too easy — no learning signal | | 10–25% | ⚠️ Weak signal | | **25–50%** | ✅ Good — enough failures to learn from | | 50–70% | ⚠️ Harsh — mostly negative reward | | > 70% | ❌ Too hard — training may diverge | **Always re-run calibration when you change your dataset.** ## Consistency Rules When using multiple graders (Python for training, endpoint for debugging, local script for eval): 1. **Identical scoring logic** — same weights, keywords, dimension breakdown 2. **Identical default scores** — same behavior when no action found, no amounts expected 3. **Test with same examples** — run 10 samples through all graders and verify scores match Mismatched scoring causes the model to learn different behavior than what your evaluation measures. ```` ### `references/hyperparameters.md` ````markdown # Hyperparameter Guide ## SFT / DPO Core Parameters | Parameter | What it controls | Default | Typical range | |-----------|-----------------|---------|---------------| | **Epochs** | Passes through data | 2 | 1–5 | | **Learning rate multiplier** | Weight change aggressiveness | 1.0 | 0.1–2.0 | | **Batch size** | Examples per gradient step | Model-dependent | 4–32 | ### Dataset Size vs Epochs | Dataset size | Recommended epochs | |-------------|-------------------| | < 100 examples | 3–5 | | 100–500 examples | 2–3 | | 500–2,000 examples | 1–2 | | > 2,000 examples | 1 | ### Learning Rate Guidelines - **Higher LR** (1.5–2.0): Large/diverse datasets, task very different from pre-training - **Lower LR** (0.1–0.5): Small datasets (<200), refining not overwriting base behavior - For 1,000+ examples, LR 0.2–0.5 often beats default 1.0 ### DPO-Specific Parameters - `beta` (default 0.1): Alignment strength. Lower = more conservative. - `l2_multiplier` (default 0.1): Regularization to prevent drift from base model. ## HP Sweep Strategy | Run | Epochs | LR | Why | |-----|--------|----|-----| | 1 | 2 | 1.0 | Baseline | | 2 | 2 | 0.5 | Conservative | | 3 | 2 | 1.5 | Aggressive | | 4 | 3 | 1.0 | More training | | 5 | 1 | 1.0 | Minimal intervention | ## Checkpoint Trick When overfitting (val loss rises after epoch 2): deploy the epoch-2 checkpoint directly instead of retraining. Azure saves checkpoints at each epoch boundary. ```python checkpoints = client.fine_tuning.jobs.checkpoints.list(job_id) for cp in checkpoints.data: print(f"Step {cp.step_number}: val_loss={cp.metrics.valid_loss}") ``` ## Model-Specific Recommendations | Model | Recommended Start | Notes | |-------|------------------|-------| | gpt-4.1-mini | 2ep, lr=0.5–1.0 | Very capable base; small nudges work | | gpt-4.1-nano | 2–3ep, lr=1.0–1.5 | Smaller capacity, needs more epochs | | gpt-oss-20b | 2ep, lr=0.2–0.5 | Lower LR critical; deployment may need capacity=100 | | o4-mini (RFT) | Grader quality > HPs | Focus on grader, not HP sweep | ## OSS Model Parameters All OSS models require `trainingType: "globalStandard"` in the API request. | Model | Recommended Start | Best Found | Notes | |-------|------------------|------------|-------| | Ministral-3B | 5ep, lr=1.0 | 10ep, lr=0.5 | Small model, slow convergence | | gpt-oss-20b | 2ep, lr=0.3 | 2ep, lr=0.3 | lr=1.0 overfits quickly | | Llama-3.3-70B | 3ep, lr=0.3 | 5ep, lr=0.5 | lr=2.0 causes catastrophic degradation | | Qwen-3-32B | 3ep, lr=0.3 | 3ep, lr=0.3 | Most fragile — more data can hurt | **Key patterns**: OSS models need 2–5× more epochs than nano. Lower LR (0.3–0.5) is safer. More data doesn't always help. ## RFT Hyperparameters | Parameter | Description | Recommended Start | |-----------|-------------|-------------------| | `reasoning_effort` | `"low"`, `"medium"`, `"high"` | `"medium"` | | `compute_multiplier` | Scales rollouts per step | `1.5` | | `learning_rate_multiplier` | Scales LR | `1.0` | | `n_epochs` | Data passes | `2–3` | | `eval_interval` | Eval every N steps | `5` | | `eval_samples` | Validation examples per eval | `10` | | `max_episode_steps` | Max tool calls + reasoning steps | `5–10` | ```` ### `references/large-file-uploads.md` ````markdown # Large File Uploads The standard `client.files.create()` silently fails on JSONL files >~150MB (Azure returns 500 during job execution). Use the chunked Uploads API: ```python upload = client.uploads.create(filename="data.jsonl", purpose="fine-tune", bytes=file_size, mime_type="application/jsonl") part_ids = [] with open(filepath, "rb") as f: while chunk := f.read(64 * 1024 * 1024): # 64MB chunks part = client.uploads.parts.create(upload_id=upload.id, data=chunk) part_ids.append(part.id) completed = client.uploads.complete(upload_id=upload.id, part_ids=part_ids) file_id = completed.file.id ``` **Important:** Requires `openai.AzureOpenAI()` client, NOT `openai.OpenAI()` with `/v1/` URL. The project endpoint returns 404 for upload operations. | File Size | Method | |-----------|--------| | < 100MB | Standard `files.create()` | | 100MB–5GB | Chunked Uploads API | | > 5GB | Split dataset | ```` ### `references/platform-gotchas.md` ```markdown # Platform Gotchas — Top 10 1. **OSS models require `"trainingType": "globalStandard"`** in the request body — undocumented, and all OSS FT jobs fail without it. 2. **Model catalog `fine_tune` flag is wrong for OSS models** — API returns `fine_tune = false` for all OSS models despite being FT-supported. Hardcode the supported list. 3. **Older SDK versions may fail on `/v1/` project endpoints** — `client.files.create()` throws "API version not supported" with older `openai` package versions. Upgrade to `openai>=1.0` and use the `/v1/` project endpoint (preferred). If you must use an older SDK, fall back to REST API with the non-project `/openai/` endpoint. 4. **ARM "Succeeded" doesn't mean deployment is ready** — `provisioningState: Succeeded` but data plane returns `DeploymentNotReady` indefinitely. Delete and recreate the deployment, then wait ~5 minutes. 5. **OSS FT deployments may fail with InternalServerError** — use the correct provider-specific `model.format` (e.g., `"Mistral AI"` not `"OpenAI"`) and try `capacity=100`. 6. **OSS FT inference hits "Failed to load LoRA" intermittently** — deploy with capacity ≥ 100, use 8+ retries with exponential backoff, and wait 2+ minutes after deployment before first call. 7. **ARM REST and `az cognitiveservices` use different format strings for OSS models** — ARM uses provider names (`"Microsoft"`, `"Meta"`), CLI uses `"OpenAI-OSS"` for all OSS. Mixing them produces HTTP 500. 8. **Content safety false positives on entity extraction data** — PII-dense data (medical records, legal docs, resumes) can trigger "Hate/Fairness" blocks at deployment time. Remove problematic document types. 9. **FT deployments at capacity=1 are severely rate-limited (~1 RPM)** — evaluating 10 samples takes ~10 minutes. Use capacity ≥ 100 for eval workloads and exponential backoff. 10. **Wrong resource endpoint is a silent killer** — jobs submitted to the wrong Foundry resource succeed via API but don't appear in the portal. Always verify the endpoint matches your Foundry project. ``` ### `references/reward-hacking.md` ````markdown # Reward Hacking Prevention in RFT ## What Is Reward Hacking? The model optimizes for the grader's scoring function rather than the actual task. The training grader becomes a proxy reward that diverges from true quality — the model games the proxy instead of improving. **Core rule: Your training grader MUST produce the same ranking as your evaluation methodology.** | If you evaluate with… | Then train with… | NOT with… | |------------------------|------------------|-----------| | LLM judge (semantic) | LLM judge | AST / regex / structural matching | | Exact match | Exact match | Fuzzy or partial matching | | Unit tests | Unit tests | Static analysis alone | Misaligned graders are the #1 cause of reward hacking. ## Train-Val Gap Thresholds | Train-Val Gap | Status | Action | |---------------|--------|--------| | ≤ 0.05 | ✅ Healthy | Continue training | | 0.05–0.10 | ⚠️ Warning | Monitor closely, check outputs qualitatively | | > 0.10 | 🛑 Stop | Stop training — reward hacking is likely | ## Pre-Training Checklist 1. **Baseline the grader**: Run training grader on base model outputs. Record scores as your floor. 2. **Cross-validate graders**: If training grader ≠ eval grader, generate 50 outputs, score with both, compute Spearman ρ. Proceed only if ρ ≥ 0.8. If ρ < 0.6, fix alignment first. 3. **Test hackability**: Generate 5 intentionally bad outputs that might score well. If grader scores any > 5/10, redesign it. 4. **Set gap threshold**: Monitor train-val gap every eval_interval. Stop if > 0.10. ## Grader Iteration Loop When reward hacking is detected: ``` 1. STOP the training run ↓ 2. COLLECT "hacked" outputs (high train score, low eval score) ↓ 3. ANALYZE what pattern the model exploited (structural mimicry? verbosity? keyword stuffing?) ↓ 4. UPDATE the grader to penalize that pattern ↓ 5. RE-BASELINE the updated grader on base model outputs ↓ 6. RESTART training with the improved grader ``` ## Red Flags Checklist Investigate immediately if **any** are true: - [ ] Train-val gap > 0.10 - [ ] Training reward increasing but eval quality stable or declining - [ ] Model outputs are longer/more verbose than base model - [ ] Outputs structurally match references but are semantically wrong - [ ] Different LLM judges disagree on quality - [ ] Conciseness/style scores dropping while correctness climbs - [ ] Model produces "template" responses ## Key Principles | Principle | Action | |-----------|--------| | Align graders | Training grader must rank outputs same as eval | | Cross-validate first | Spearman ρ ≥ 0.8 between training and eval graders | | Monitor train-val gap | ≤ 0.05 healthy, > 0.10 stop | | Test hackability | Bad outputs should score < 5/10 | | Prefer SFT when possible | Use RFT only for verifiable-answer tasks | | Iterate graders, not models | Fix grader before restarting training | ```` ### `references/training-curves.md` ````markdown # Training Curve Analysis ## SFT Metrics | Column | What it means | |--------|---------------| | `train_loss` | Loss on training batch (should decrease) | | `train_mean_token_accuracy` | Token-level accuracy on training data | | `valid_loss` | Loss on validation set (**primary metric**) | | `valid_mean_token_accuracy` | Token-level accuracy on validation data | | `full_valid_loss` | Full-pass validation loss (more accurate, less frequent) | | `full_valid_mean_token_accuracy` | Full-pass token accuracy | ## Overfitting Detection **Overfitting ratio** at each checkpoint: `valid_loss / train_loss` | Ratio | Interpretation | |-------|---------------| | < 1.2 | Healthy — generalizes well | | 1.2–1.5 | Mild overfitting — acceptable for small datasets | | 1.5–2.0 | Moderate — consider reducing epochs | | > 2.0 | Severe — deploy an earlier checkpoint | ```python val_losses = [cp.metrics.valid_loss for cp in checkpoints if cp.metrics.valid_loss] best_val = min(val_losses) final_val = val_losses[-1] if final_val > best_val * 1.2: print(f"⚠️ OVERFIT: Best={best_val:.4f}, final={final_val:.4f}") ``` ## Best Checkpoint Selection (SFT) ```python checkpoints = client.fine_tuning.jobs.checkpoints.list(job_id) best_cp = min(checkpoints.data, key=lambda cp: cp.metrics.valid_loss or float('inf')) print(f"Best: step {best_cp.step_number}, valid_loss={best_cp.metrics.valid_loss:.4f}, " f"model={best_cp.fine_tuned_model_checkpoint}") ``` ## Diagnosis Table | Observation | Diagnosis | Action | |-------------|-----------|--------| | Train loss barely decreases | LR too low or noisy data | Increase LR or clean data | | Train loss crashes to ~0 | LR too high or easy data | Decrease LR or add harder examples | | Valid loss rises after epoch 2 | Overfitting | Deploy epoch-2 checkpoint | | Valid loss plateaus after epoch 1 | Learned quickly | Try epoch=1 or lower LR | | Valid loss oscillates | Small batch or inconsistent data | Increase batch size or audit data | | Both losses stay high | Task too hard | Larger model or simplify task | | Large train-valid gap from start | Insufficient/mismatched data | Add diverse training data | ## RFT Metrics | Column | What it means | |--------|---------------| | `train_mean_reward` | Average reward across rollouts (**primary** — should increase) | | `full_valid_mean_reward` | Validation reward (overfitting check) | | `completion_tokens_mean` | Average response length per rollout | | `reasoning_tokens_mean` | Average reasoning tokens (o-series models) | | `mean_unresponsive_rewards` | Rollouts with no scoreable output | | `train_sample_parse_error_count` | Grader couldn't parse output | | `train_other_error_count` | Grader logic bugs — should be 0 | ## RFT Reward Curve Patterns - **Reward flat at ~0**: Grader broken or threshold too strict - **Reward always negative**: pass_threshold too high - **Reward immediately high + flat**: Threshold too lenient - **Train-valid reward gap > 0.10**: Possible reward hacking ### Token Growth - **Moderate** (tokens double): Normal — model becoming more thorough - **Excessive** (3x+): Grader may incentivize verbosity — check scoring dimensions - When comparing checkpoints, equal accuracy at fewer tokens is strictly better ### Parse Errors vs Logic Errors - `sample_parse_error_count`: Often high in agentic RFT (mid-reasoning captures). Training still works if reward is climbing. - `other_error_count`: Bugs in grader logic. Fix before continuing. ## RFT Checkpoint Selection ```python checkpoints = client.fine_tuning.jobs.checkpoints.list(job_id) for cp in checkpoints: m = cp.metrics tr = f"{m.train_mean_reward:.3f}" if m.train_mean_reward is not None else "n/a" vr = f"{m.full_valid_mean_reward:.3f}" if m.full_valid_mean_reward is not None else "n/a" ct = f"{m.completion_tokens_mean:.0f}" if m.completion_tokens_mean is not None else "n/a" print(f"Step {cp.step_number}: train_reward={tr}, valid_reward={vr}, tokens={ct}") ``` Don't rely solely on `valid_reward` for RFT — deploy 2–3 candidates (peak reward, final, mid-training) and evaluate with your real task harness including tool execution. ```` ### `references/training-types.md` ````markdown # Training Types: SFT vs DPO vs RFT ## Decision Matrix | Factor | SFT | DPO | RFT | |--------|-----|-----|-----| | **Best for** | Teaching a new skill or format | Aligning preferences/style | Improving reasoning chains | | **Data needed** | Input–output pairs | Chosen/rejected pairs | Prompts + grading function | | **Data volume** | 50–5,000 examples | 500–5,000 pairs | 200–2,000 prompts | | **Effort to prepare data** | Low | High (need contrasting pairs) | Medium (need grader, not outputs) | | **Risk of regression** | Low | Medium | High (sensitive to grader quality) | | **Typical improvement** | 5–30% on task metrics | Subtle style/safety shifts | 0–15% on reasoning tasks | | **Supported models** | Most models | Select models | o4-mini | ## When to Use Each ### SFT (Supervised Fine-Tuning) - You have high-quality input–output pairs - Task is well-defined (code generation, classification, extraction, summarization) - You want reliable, repeatable outputs in a specific format or style - **Key insight**: 300–500 high-quality examples often outperforms 1,500+ lower-quality ones ### DPO (Direct Preference Optimization) - You want to adjust tone, verbosity, safety, or style - You have examples of "good" and "bad" outputs for the same input - SFT already works but outputs need refinement - DPO-specific params: `beta` (default 0.1), `l2_multiplier` (default 0.1) ### RFT (Reinforcement Fine-Tuning) - Task has objectively verifiable answers (code execution, math, logic) - You can write a programmatic or LLM-based grader - You want to improve the model's reasoning, not just its outputs - **Critical**: RFT is extremely sensitive to grader quality. Train–val gap should be ≤ 0.05. ## Choosing a Path ``` ├─ Do you have labeled input–output pairs? │ ├─ Yes → SFT │ └─ No │ ├─ Can you write a grading function? → RFT │ └─ Can you rank "good" vs "bad" outputs? → DPO │ After SFT: ├─ Results good enough? → Ship it ├─ Need style refinement? → DPO on top of SFT model └─ Reasoning needs improvement? → RFT (if model supports it) ``` ## Model Compatibility (Azure AI Foundry) | Model | SFT | DPO | RFT | Vision FT | |-------|-----|-----|-----|-----------| | gpt-4.1 | ✅ | ✅ | ❌ | ✅ | | gpt-4.1-mini | ✅ | ❌ | ❌ | ❌ | | gpt-4.1-nano | ✅ | ❌ | ❌ | ❌ | | gpt-4o (2024-08-06) | ✅ | ✅ | ❌ | ✅ | | gpt-4o-mini | ✅ | ❌ | ❌ | ❌ | | o4-mini | ❌ | ❌ | ✅ | ❌ | | gpt-5 | ❌ | ❌ | ✅ ⚠️ | ❌ | | gpt-oss-20b | ✅ | ❌ | ❌ | ❌ | | Ministral-3B | ✅ | ❌ | ❌ | ❌ | | Llama-3.3-70B | ✅ | ❌ | ❌ | ❌ | | Qwen-3-32B | ✅ | ❌ | ❌ | ❌ | DPO can be applied on top of an already SFT-fine-tuned model. Vision fine-tuning follows the same SFT workflow but with image data in messages. > ⚠️ **Feature flags**: GPT-5 RFT and agentic RFT with tool calling require access requests. Contact your Microsoft account team or request access through the Azure AI Foundry portal. o4-mini RFT without tools is generally available. *Check Azure AI Foundry docs for the latest model availability.* ```` ### `references/vision-fine-tuning.md` ````markdown # Vision Fine-Tuning Fine-tune models with image data to customize visual understanding. Uses the same chat-completions JSONL format as text SFT, but with image content blocks in user messages. ## Supported Models | Model | Version | |-------|---------| | gpt-4o | 2024-08-06 | | gpt-4.1 | 2025-04-14 | ## Image Requirements | Constraint | Limit | |-----------|-------| | Max examples with images per training file | 50,000 | | Max images per example | 64 | | Max image file size | 10 MB | | Supported formats | JPEG, PNG, WEBP | | Color mode | RGB or RGBA | | Min examples | 10 | **Important**: Images can only appear in `user` messages, never in `assistant` responses. ## Data Format Each training example follows the standard SFT `messages` format. Images are included as `image_url` content blocks within user messages. ```jsonl {"messages": [{"role": "system", "content": "You are a helpful AI assistant that describes images."}, {"role": "user", "content": [{"type": "text", "text": "Describe this image."}, {"type": "image_url", "image_url": {"url": "https://example.com/photo.png", "detail": "high"}}]}, {"role": "assistant", "content": "The image shows a cityscape with tall buildings against a blue sky."}]} ``` ### Image Sources Images can be provided in two ways: **1. Public URL:** ```json {"type": "image_url", "image_url": {"url": "https://example.com/image.png"}} ``` **2. Base64 data URI:** ```json {"type": "image_url", "image_url": {"url": "data:image/png;base64,iVBORw0KGgo..."}} ``` ### Detail Control The `detail` parameter controls image processing fidelity and cost: | Value | Behavior | Cost | |-------|----------|------| | `low` | Downscales to 512×512 pixels | Lower | | `high` | Full resolution processing | Higher | | `auto` | Model decides based on image size | Default | ```json {"type": "image_url", "image_url": {"url": "https://example.com/image.png", "detail": "low"}} ``` Use `low` for tasks where fine visual detail doesn't matter (classification, general description). Use `high` for tasks needing precise detail (OCR, diagram reading, defect detection). ## Content Moderation Images are screened before training. The following are **automatically excluded**: - Images containing **people or faces** (face detection only — no identification) - **CAPTCHAs** - Content violating Azure usage policies This screening may add latency to file upload validation. ## Best Practices - **Diverse examples**: Vary image content, angles, lighting, and resolution - **Consistent annotations**: Keep assistant response style and detail level uniform - **Start with `detail: low`**: Cheaper and faster — upgrade to `high` only if results need it - **Check for excluded images**: After upload, verify the training count matches expectations — some images may be silently skipped due to content moderation - **Mixed text+image**: You can include both text-only and image examples in the same training file ## Training Workflow Vision fine-tuning follows the exact same workflow as text SFT: 1. Prepare JSONL with image content blocks 2. Upload training file (validation may take longer due to image screening) 3. Create fine-tuning job with a supported vision model 4. Monitor and evaluate as usual ```python # Upload (image validation may take longer) train_file = client.files.create(purpose="fine-tune", file=open("vision_train.jsonl", "rb")) client.files.wait_for_processing(train_file.id) # Submit — same as text SFT job = client.fine_tuning.jobs.create( model="gpt-4.1-2025-04-14", training_file=train_file.id, validation_file=val_file.id, method={"type": "supervised"} ) ``` ## Troubleshooting | Issue | Resolution | |-------|-----------| | Images skipped silently | Check for people/faces, oversized files, unsupported formats | | URL not accessible | Ensure URLs are publicly accessible, or use base64 data URIs | | Exceeds 10 MB | Resize or compress the image | | Wrong color mode | Convert to RGB or RGBA | | Low quality results | Try `detail: high`, add more diverse examples, increase dataset size | ## Reference - [Official docs](https://learn.microsoft.com/en-us/azure/foundry/openai/how-to/fine-tuning-vision) ```` ### `scripts/calibrate_grader.py` ``` # /// script # dependencies = [ # "openai>=1.0", # "azure-identity", # "azure-ai-projects", # ] # /// """ calibrate_grader.py — Calibrate RFT grader pass_threshold before submitting a job. Runs the base model on your training/validation data, scores each output with your Python grader, and recommends the optimal pass_threshold. Usage: python calibrate_grader.py --base-url --api-key KEY \ --model o4-mini --data train.jsonl --grader grader.py --n 30 python calibrate_grader.py --model gpt-4.1-mini --data val.jsonl \ --grader grader.py --n 20 --tools '[{"name": "search", "server_url": "https://..."}]' """ import argparse import json import os import random import sys try: sys.stdout.reconfigure(encoding="utf-8") sys.stderr.reconfigure(encoding="utf-8") except (AttributeError, OSError): pass # Stream not reconfigurable (older Python or non-tty); default encoding is fine import time sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from common import HelpOnErrorParser, get_clients def load_grader(grader_path): """Load and compile a Python grader file. Returns the grade() function. SECURITY: This executes the grader file as Python code. Only load grader files that you wrote or reviewed — never load untrusted files from the internet or unknown sources. The grader runs with the same permissions as this script. """ grader_path = os.path.abspath(grader_path) if not os.path.isfile(grader_path): print(f"❌ Grader file not found: {grader_path}") sys.exit(1) with open(grader_path, encoding="utf-8") as f: source = f.read() namespace = {} exec(compile(source, grader_path, "exec"), namespace) if "grade" not in namespace: print(f"❌ Grader file must define a grade(sample, item) function") sys.exit(1) return namespace["grade"] def run_model(client, model, messages, tools_schema=None, max_retries=3): """Run the model and return (output_text, output_tools).""" kwargs = {"model": model, "messages": messages, "max_completion_tokens": 4096} if tools_schema: kwargs["tools"] = tools_schema for attempt in range(max_retries): try: resp = client.chat.completions.create(**kwargs) msg = resp.choices[0].message output_text = msg.content or "" output_tools = [] if msg.tool_calls: output_tools = [ {"type": "function", "function": {"name": tc.function.name, "arguments": tc.function.arguments}} for tc in msg.tool_calls ] return output_text, output_tools except Exception as e: if "429" in str(e) and attempt < max_retries - 1: time.sleep(5 * (attempt + 1)) else: return f"ERROR: {e}", [] return "ERROR: max retries", [] def calibrate(client, model, data, grade_fn, tools_schema=None, n=30): """Run base model on data, score with grader, output threshold analysis.""" if not data: print("No examples to evaluate. Check your data file.") return # Sample if dataset is larger than n if len(data) > n: data = random.sample(data, n) print(f"Running {model} on {len(data)} examples...\n") scores = [] for i, ex in enumerate(data): messages = ex["messages"] user_msg = messages[-1]["content"] if messages else "" output_text, output_tools = run_model(client, model, messages, tools_schema) if output_text.startswith("ERROR:"): print(f" [{i+1:3d}] ❌ {output_text[:60]}") scores.append(0.0) continue # Build sample dict matching what the grader expects sample = {"output_text": output_text, "output_tools": output_tools} # Build item dict from all fields in the training example item = {k: v for k, v in ex.items() if k != "messages"} try: score = grade_fn(sample, item) except Exception as e: print(f" [{i+1:3d}] ❌ Grader error: {e}") scores.append(0.0) continue status = "✅" if score >= 0.9 else ("⚠️" if score >= 0.5 else "❌") print(f" [{i+1:3d}] {score:.3f} {status} {user_msg[:55]}") scores.append(score) time.sleep(0.5) # Rate limiting # Analysis scored = [s for s in scores if s is not None] if not scored: print("\n❌ No examples were scored successfully. Check model access and data format.") return avg = sum(scored) / len(scored) print(f"\n{'='*60}") print(f" BASE MODEL GRADER CALIBRATION ({len(scores)} examples)") print(f" Average score: {avg:.1%}") print(f"{'='*60}") print(f"\n {'Threshold':>10} {'Pass Rate':>10} {'Fail Rate':>10} {'Signal':>20}") print(f" {'-'*10} {'-'*10} {'-'*10} {'-'*20}") best_threshold = None best_distance = float("inf") for threshold in [0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.85, 0.9, 0.95, 1.0]: pass_rate = sum(1 for s in scored if s >= threshold) / len(scored) fail_rate = 1 - pass_rate if 0.25 <= fail_rate <= 0.50: signal = "✅ Good (25-50%)" distance = abs(fail_rate - 0.35) # Ideal is ~35% if distance < best_distance: best_distance = distance best_threshold = threshold elif fail_rate < 0.10: signal = "❌ Too easy" elif fail_rate < 0.25: signal = "⚠️ Weak signal" elif fail_rate <= 0.70: signal = "⚠️ Harsh" else: signal = "❌ Too hard" print(f" {threshold:>10.2f} {pass_rate:>9.0%} {fail_rate:>9.0%} {signal:>20}") if best_threshold: print(f"\n ✅ Recommended pass_threshold: {best_threshold}") print(f" (~{sum(1 for s in scores if s < best_threshold)/len(scores):.0%} failure rate)") else: print(f"\n ⚠️ No threshold in the ideal 25-50% failure range.") print(f" Consider adjusting your grader scoring dimensions.") # Score distribution print(f"\n Score distribution:") buckets = {"0.0-0.2": 0, "0.2-0.4": 0, "0.4-0.6": 0, "0.6-0.8": 0, "0.8-0.9": 0, "0.9-1.0": 0} for s in scores: if s < 0.2: buckets["0.0-0.2"] += 1 elif s < 0.4: buckets["0.2-0.4"] += 1 elif s < 0.6: buckets["0.4-0.6"] += 1 elif s < 0.8: buckets["0.6-0.8"] += 1 elif s < 0.9: buckets["0.8-0.9"] += 1 else: buckets["0.9-1.0"] += 1 for bucket, count in buckets.items(): bar = "█" * count print(f" {bucket}: {count:3d} {bar}") def build_parser(): parser = HelpOnErrorParser( description="Calibrate RFT grader pass_threshold on base model outputs", epilog=( "Example:\n" " python calibrate_grader.py --model o4-mini --data train.jsonl --grader grader.py\n" " python calibrate_grader.py --model o4-mini --data val.jsonl --grader grader.py --n 20" ), formatter_class=argparse.RawTextHelpFormatter, ) parser.add_argument("--base-url", default=os.environ.get("OPENAI_BASE_URL"), help="Project /v1/ endpoint URL") parser.add_argument("--endpoint", default=os.environ.get("AZURE_OPENAI_ENDPOINT"), help="Azure OpenAI endpoint (fallback)") parser.add_argument("--api-key", default=os.environ.get("AZURE_OPENAI_API_KEY"), help="API key") parser.add_argument("--project-endpoint", default=os.environ.get("AZURE_AI_PROJECT_ENDPOINT"), help="Azure AI project endpoint") parser.add_argument("--model", required=True, help="Base model deployment name to calibrate against") parser.add_argument("--data", required=True, help="Path to training or validation JSONL file") parser.add_argument("--grader", required=True, help="Path to Python grader file (must define grade(sample, item))") parser.add_argument("--n", type=int, default=30, help="Number of examples to evaluate (default: 30)") parser.add_argument("--tools", default=None, help="Tool schemas as JSON array (for tool-calling models). Pass as a JSON string.") parser.add_argument("--seed", type=int, default=42, help="Random seed for sampling (default: 42)") return parser if __name__ == "__main__": parser = build_parser() if len(sys.argv) == 1: parser.print_help() sys.exit(0) args = parser.parse_args() random.seed(args.seed) client, method = get_clients(base_url=args.base_url, azure_endpoint=args.endpoint, project_endpoint=args.project_endpoint, api_key=args.api_key) # Load data with open(args.data, encoding="utf-8") as f: data = [] for ln, line in enumerate(f, 1): if not line.strip(): continue try: data.append(json.loads(line)) except json.JSONDecodeError as e: print(f"⚠️ Skipping malformed JSON on line {ln}: {e}") print(f"Loaded {len(data)} examples from {args.data}") # Load grader grade_fn = load_grader(args.grader) print(f"Loaded grader from {args.grader}") # Parse tools if provided tools_schema = None if args.tools: tools_schema = json.loads(args.tools) calibrate(client, args.model, data, grade_fn, tools_schema, args.n) ``` ### `scripts/check_training.py` ``` # /// script # dependencies = [ # "openai>=1.0", # "azure-identity", # "azure-ai-projects", # ] # /// """ check_training.py — Analyze training curves, detect overfitting, list checkpoints. Usage: python check_training.py --job-id ftjob-abc123 python check_training.py --job-id ftjob-abc123 --download-csv results.csv python check_training.py --base-url https://.services.ai.azure.com/api/projects//openai/v1/ --api-key KEY --job-id ftjob-abc123 """ import csv import io import os import sys try: sys.stdout.reconfigure(encoding="utf-8") sys.stderr.reconfigure(encoding="utf-8") except (AttributeError, OSError): pass # Stream not reconfigurable (older Python or non-tty); default encoding is fine sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from common import HelpOnErrorParser, get_clients def analyze_job(client, job_id, download_csv=None): """Pull training results, analyze curves, detect overfitting.""" job = client.fine_tuning.jobs.retrieve(job_id) print(f"Job: {job.id}") print(f" Model: {job.model}") print(f" Status: {job.status}") print(f" Fine-tuned model: {job.fine_tuned_model}") if job.hyperparameters: hp = job.hyperparameters print(f" Epochs: {getattr(hp, 'n_epochs', 'N/A')}") print(f" LR multiplier: {getattr(hp, 'learning_rate_multiplier', 'N/A')}") print(f" Batch size: {getattr(hp, 'batch_size', 'N/A')}") # Allow analysis while still running if result files exist if job.status not in ("succeeded", "running"): print(f"\n Job status is '{job.status}'. Cannot analyze curves.") return if not job.result_files: if job.status == "running": print("\n Job is still running and no result files available yet. Check back later.") else: print("\n No result files available.") return # Download results CSV content = client.files.content(job.result_files[0]) csv_data = content.read() if download_csv: with open(download_csv, "wb") as f: f.write(csv_data) print(f"\n Results CSV saved to {download_csv}") # Parse CSV reader = csv.DictReader(io.StringIO(csv_data.decode("utf-8"))) rows = list(reader) if job.status == "running": print(f"\n ⚡ Job still running — showing partial results ({len(rows)} steps so far)") # Extract validation checkpoints val_points = [] for row in rows: step = int(row.get("step", 0)) train_loss = float(row["train_loss"]) if row.get("train_loss", "").strip() else None val_loss = None for col in ["valid_loss", "full_valid_loss", "eval_loss"]: if row.get(col, "").strip(): val_loss = float(row[col]) break if val_loss is not None: val_points.append((step, val_loss, train_loss)) if not val_points: print("\n No validation loss data found in results CSV.") return # Find best validation checkpoint best_step, best_val, best_train = min(val_points, key=lambda x: x[1]) final_step, final_val, final_train = val_points[-1] print(f"\n Training Curve Analysis:") print(f" {'Step':>6} {'Val Loss':>10} {'Train Loss':>12} {'Ratio':>8}") print(f" {'─'*6} {'─'*10} {'─'*12} {'─'*8}") for step, val, train in val_points: ratio = val / train if train and train > 0 else 0 marker = " ← best" if step == best_step else "" train_str = f"{train:12.4f}" if train is not None else " N/A" print(f" {step:>6} {val:>10.4f} {train_str} {ratio:>8.2f}{marker}") print(f"\n Best val_loss: {best_val:.4f} at step {best_step}") print(f" Final val_loss: {final_val:.4f} at step {final_step}") # Overfitting detection if best_val > 0 and final_val > best_val * 1.2: pct = (final_val - best_val) / best_val * 100 print(f"\n ⚠️ OVERFITTING DETECTED: Final val_loss is {pct:.0f}% above best.") elif best_val == 0 and final_val > 0: print(f"\n ⚠️ Best val_loss was 0.0; final val_loss is {final_val:.4f} — possible overfitting from a near-perfect early checkpoint.") elif final_train and final_val / final_train > 1.5: ratio = final_val / final_train print(f"\n ⚠️ MODERATE OVERFITTING: val/train ratio = {ratio:.2f}") else: print(f"\n ✅ Training looks healthy. No significant overfitting detected.") # List checkpoints and recommend best deployable one print(f"\n Checkpoints:") available_checkpoints = [] try: cps = client.fine_tuning.jobs.checkpoints.list(job_id) if cps.data: for cp in sorted(cps.data, key=lambda c: c.step_number): vl = cp.metrics.valid_loss if cp.metrics and cp.metrics.valid_loss is not None else None model_id = cp.fine_tuned_model_checkpoint or "N/A" vl_str = f"{vl:.4f}" if vl is not None else "N/A" available_checkpoints.append((cp.step_number, vl, model_id)) print(f" Step {cp.step_number}: val_loss={vl_str}, model={model_id}") else: print(" No checkpoints available.") except Exception as e: print(f" Could not retrieve checkpoints: {e}") # Recommend the best deployable checkpoint if available_checkpoints and best_val > 0 and final_val > best_val * 1.2: # Find the checkpoint with the lowest val_loss, or nearest to best_step best_cp = None if any(vl is not None for _, vl, _ in available_checkpoints): # Use checkpoint with lowest val_loss scored_cps = [(s, vl, m) for s, vl, m in available_checkpoints if vl is not None] if scored_cps: best_cp = min(scored_cps, key=lambda x: x[1]) else: # No val_loss on checkpoints — pick the one nearest to (but not exceeding) best_step earlier_cps = [(s, vl, m) for s, vl, m in available_checkpoints if s <= best_step] if earlier_cps: best_cp = max(earlier_cps, key=lambda x: x[0]) elif available_checkpoints: best_cp = available_checkpoints[0] if best_cp: cp_step, cp_vl, cp_model = best_cp vl_info = f" (val_loss={cp_vl:.4f})" if cp_vl is not None else "" print(f"\n 🎯 Recommended checkpoint: step {cp_step}{vl_info}") print(f" Model ID: {cp_model}") print(f" (Best val_loss was at step {best_step}, nearest deployable checkpoint is step {cp_step})") print(f" Alternatively, retrain with fewer epochs to avoid overfitting.") else: print(f"\n Recommendation: Retrain with fewer epochs (best val_loss was at step {best_step}).") def main(): parser = HelpOnErrorParser(description="Analyze fine-tuning training curves") parser.add_argument("--base-url", default=os.environ.get("OPENAI_BASE_URL"), help="Project /v1/ URL (preferred)") parser.add_argument("--endpoint", default=os.environ.get("AZURE_OPENAI_ENDPOINT"), help="Azure OpenAI endpoint (fallback)") parser.add_argument("--project-endpoint", default=os.environ.get("AZURE_AI_PROJECT_ENDPOINT"), help="Azure AI project endpoint (Foundry SDK)") parser.add_argument("--api-key", default=os.environ.get("AZURE_OPENAI_API_KEY")) parser.add_argument("--job-id", required=True, help="Fine-tuning job ID") parser.add_argument("--download-csv", help="Save results CSV to this path") args = parser.parse_args() client, method = get_clients( base_url=args.base_url, azure_endpoint=args.endpoint, project_endpoint=args.project_endpoint, api_key=args.api_key ) analyze_job(client, args.job_id, args.download_csv) if __name__ == "__main__": main() ``` ### `scripts/cleanup.py` ``` # /// script # dependencies = [ # "openai>=1.0", # "azure-identity", # "azure-ai-projects", # ] # /// """ cleanup.py — Clean up fine-tuning resources to avoid quota exhaustion. Lists and optionally deletes uploaded files and cancels pending jobs. Useful after experimentation to reclaim quota (max 100 files per resource, deployment slots are limited). Usage: python cleanup.py --list # List all resources python cleanup.py --list --type files # List only files python cleanup.py --delete-files --older-than 7 # Delete files older than 7 days python cleanup.py --delete-files --dry-run # Preview what would be deleted python cleanup.py --cancel-pending # Cancel queued jobs """ import argparse import os import sys try: sys.stdout.reconfigure(encoding="utf-8") sys.stderr.reconfigure(encoding="utf-8") except (AttributeError, OSError): pass # Stream not reconfigurable (older Python or non-tty); default encoding is fine from datetime import datetime, timezone sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from common import HelpOnErrorParser, get_clients def list_deployments(client): """List fine-tuned model deployments. Returns deployment info from jobs.""" deployments = [] for job in _iter_all_jobs(client): if job.fine_tuned_model and job.status == "succeeded": deployments.append({ "job_id": job.id, "model": job.fine_tuned_model, "base_model": job.model, "created": datetime.fromtimestamp(job.created_at, tz=timezone.utc), "tokens": job.trained_tokens, }) return deployments def _iter_all_jobs(client, page_size=100): """Yield every fine-tuning job, paginating through the API. The OpenAI/Azure SDK's `jobs.list(limit=N)` returns at most N jobs with no auto-paging. Users with >100 jobs would otherwise miss older jobs entirely. """ after = None while True: kwargs = {"limit": page_size} if after: kwargs["after"] = after page = client.fine_tuning.jobs.list(**kwargs) items = list(page) if not items: break for job in items: yield job if len(items) < page_size: break # Cursor-based paging: use last job's id as `after` after = items[-1].id def list_files(client): """List uploaded files.""" files = client.files.list() result = [] for f in files: result.append({ "id": f.id, "filename": f.filename, "purpose": f.purpose, "bytes": f.bytes, "created": datetime.fromtimestamp(f.created_at, tz=timezone.utc), "status": f.status, }) return result def list_jobs(client): """List fine-tuning jobs.""" result = [] for job in _iter_all_jobs(client): result.append({ "id": job.id, "status": job.status, "model": job.model, "fine_tuned_model": job.fine_tuned_model or "—", "created": datetime.fromtimestamp(job.created_at, tz=timezone.utc), }) return result def format_age(dt): """Format a datetime as a human-readable age.""" delta = datetime.now(timezone.utc) - dt if delta.days > 0: return f"{delta.days}d ago" hours = delta.seconds // 3600 return f"{hours}h ago" def format_bytes(b): """Format bytes as human-readable size.""" if not b: return "—" if b > 1_000_000: return f"{b/1_000_000:.1f} MB" if b > 1_000: return f"{b/1_000:.0f} KB" return f"{b} B" def show_list(client, resource_type="all"): """Display current resources.""" if resource_type in ("all", "jobs"): jobs = list_jobs(client) print(f"\n📋 Fine-tuning jobs ({len(jobs)}):") if jobs: print(f" {'ID':<25} {'Status':<12} {'Model':<20} {'Age':<10}") print(f" {'-'*25} {'-'*12} {'-'*20} {'-'*10}") for j in jobs: print(f" {j['id'][:24]:<25} {j['status']:<12} {j['model']:<20} {format_age(j['created'])}") else: print(" (none)") if resource_type in ("all", "deployments"): deps = list_deployments(client) print(f"\n🚀 Fine-tuned models ({len(deps)}):") if deps: print(f" {'Model':<60} {'Age':<10}") print(f" {'-'*60} {'-'*10}") for d in deps: name = d['model'][:59] print(f" {name:<60} {format_age(d['created'])}") else: print(" (none)") if resource_type in ("all", "files"): files = list_files(client) print(f"\n📁 Uploaded files ({len(files)}):") if files: print(f" {'ID':<40} {'Size':>8} {'Purpose':<12} {'Age':<10} {'Status':<10}") print(f" {'-'*40} {'-'*8} {'-'*12} {'-'*10} {'-'*10}") for f in files: print(f" {f['id']:<40} {format_bytes(f['bytes']):>8} {f['purpose']:<12} {format_age(f['created']):<10} {f['status']}") else: print(" (none)") # Quota warning if len(files) >= 80: print(f"\n ⚠️ {len(files)}/100 file slots used — approaching quota limit!") def delete_files(client, older_than_days=None, dry_run=False): """Delete uploaded files, optionally filtering by age.""" files = list_files(client) now = datetime.now(timezone.utc) to_delete = [] for f in files: if older_than_days: age_days = (now - f["created"]).days if age_days < older_than_days: continue to_delete.append(f) if not to_delete: print("No files to delete.") return label = f"older than {older_than_days} days" if older_than_days else "all" print(f"\n{'[DRY RUN] ' if dry_run else ''}Deleting {len(to_delete)} files ({label}):") deleted = 0 for f in to_delete: print(f" {'Would delete' if dry_run else 'Deleting'}: {f['id']} ({f['filename']}, {format_age(f['created'])})") if not dry_run: try: client.files.delete(f["id"]) deleted += 1 except Exception as e: print(f" ❌ Failed: {e}") if not dry_run: print(f"\n✅ Deleted {deleted}/{len(to_delete)} files") def cancel_pending_jobs(client, dry_run=False): """Cancel any pending or queued jobs.""" jobs = list_jobs(client) pending = [j for j in jobs if j["status"] in ("pending", "queued", "validating_files")] if not pending: print("No pending jobs to cancel.") return print(f"\n{'[DRY RUN] ' if dry_run else ''}Cancelling {len(pending)} pending jobs:") for j in pending: print(f" {'Would cancel' if dry_run else 'Cancelling'}: {j['id']} ({j['status']})") if not dry_run: try: client.fine_tuning.jobs.cancel(j["id"]) except Exception as e: print(f" ❌ Failed: {e}") def build_parser(): parser = HelpOnErrorParser( description="Clean up fine-tuning resources (deployments, files, jobs)", epilog=( "Examples:\n" " python cleanup.py --list # Show all resources\n" " python cleanup.py --list --type files # Show files only\n" " python cleanup.py --delete-files --older-than 7 # Delete files older than 7 days\n" " python cleanup.py --delete-files --dry-run # Preview what would be deleted\n" " python cleanup.py --cancel-pending # Cancel queued jobs" ), formatter_class=argparse.RawTextHelpFormatter, ) parser.add_argument("--base-url", default=os.environ.get("OPENAI_BASE_URL"), help="Project /v1/ endpoint URL") parser.add_argument("--endpoint", default=os.environ.get("AZURE_OPENAI_ENDPOINT"), help="Azure OpenAI endpoint (fallback)") parser.add_argument("--api-key", default=os.environ.get("AZURE_OPENAI_API_KEY"), help="API key") parser.add_argument("--project-endpoint", default=os.environ.get("AZURE_AI_PROJECT_ENDPOINT"), help="Azure AI project endpoint") parser.add_argument("--list", action="store_true", help="List resources") parser.add_argument("--type", choices=["all", "jobs", "deployments", "files"], default="all", help="Resource type to list (default: all)") parser.add_argument("--delete-files", action="store_true", help="Delete uploaded files") parser.add_argument("--older-than", type=int, default=None, help="Only delete files older than N days (use with --delete-files)") parser.add_argument("--cancel-pending", action="store_true", help="Cancel pending/queued jobs") parser.add_argument("--dry-run", action="store_true", help="Preview changes without executing") return parser if __name__ == "__main__": parser = build_parser() if len(sys.argv) == 1: parser.print_help() sys.exit(0) args = parser.parse_args() client, method = get_clients(base_url=args.base_url, azure_endpoint=args.endpoint, project_endpoint=args.project_endpoint, api_key=args.api_key) if args.list: show_list(client, args.type) if args.delete_files: delete_files(client, older_than_days=args.older_than, dry_run=args.dry_run) if args.cancel_pending: cancel_pending_jobs(client, dry_run=args.dry_run) ``` ### `scripts/common.py` ``` """ common.py — Shared Azure AI Foundry authentication and client setup. Supports three connection methods in order of preference: 1. /v1/ project endpoint (simplest, preferred) 2. Foundry SDK with DefaultAzureCredential (no API key needed, cloud-native) 3. Azure OpenAI endpoint (classic) AAD tokens are auto-refreshed via azure.identity for long-running scripts (monitor_training.py, generate_distillation_data.py, etc.). Usage: from common import get_clients, upload_file # Method 1: Project /v1/ endpoint (preferred) clients = get_clients(base_url="https://.services.ai.azure.com/api/projects//openai/v1/", api_key="KEY") # Method 2: Foundry SDK (DefaultAzureCredential — no API key needed) clients = get_clients(project_endpoint="https://.services.ai.azure.com/api/projects/") # Method 3: Azure OpenAI endpoint clients = get_clients(azure_endpoint="https://.openai.azure.com", api_key="KEY") """ import argparse import os import sys try: sys.stdout.reconfigure(encoding="utf-8") sys.stderr.reconfigure(encoding="utf-8") except (AttributeError, OSError): pass # Stream not reconfigurable (older Python or non-tty); default encoding is fine _AZURE_COGSERVICES_SCOPE = "https://cognitiveservices.azure.com/.default" def _clamp_score(v, default=0): """Clamp a judge score to [1, 10]. Returns `default` for missing/non-numeric values. LLM judges occasionally return out-of-range integers (e.g., 15) or non-numeric strings ("high"). Without clamping, these distort aggregate scores or crash `int()`. We use 0 as a sentinel for "missing/failed" so callers can filter via `score > 0`. """ if v is None: return default try: return max(1, min(10, int(v))) except (ValueError, TypeError): return default class HelpOnErrorParser(argparse.ArgumentParser): """ArgumentParser that prints full help when arguments are invalid. Standard ArgumentParser only prints a one-line usage summary on error, which isn't helpful for first-time users. This prints the full --help. """ def error(self, message): self.print_help(sys.stderr) self.exit(2, f"\nerror: {message}\n") def _make_token_provider(): """Create an auto-refreshing AAD token provider for long-running scripts. Returns a callable that the OpenAI SDK calls before each request to get a fresh token. Tokens are cached and refreshed ~5 min before expiry. """ from azure.identity import DefaultAzureCredential credential = DefaultAzureCredential() def get_token(): try: token = credential.get_token(_AZURE_COGSERVICES_SCOPE) return token.token except Exception as e: raise RuntimeError( f"Azure AD authentication failed: {e}\n" "Ensure you're logged in (az login) or have valid " "AZURE_CLIENT_ID/AZURE_TENANT_ID/AZURE_CLIENT_SECRET set." ) from e return get_token def get_clients(base_url=None, azure_endpoint=None, project_endpoint=None, api_key=None): """Initialize and return OpenAI-compatible client. Tries in order: 1. Project /v1/ endpoint with openai.OpenAI() (simplest, preferred) 2. Foundry SDK with AIProjectClient.get_openai_client() (no API key needed) 3. Azure OpenAI endpoint with openai.AzureOpenAI() (classic) When using DefaultAzureCredential (no API key), tokens are auto-refreshed so long-running scripts won't fail with 401 after ~60 min. Returns: (openai_client, method_name) """ # Method 1: /v1/ project endpoint base_url = base_url or os.environ.get("OPENAI_BASE_URL") api_key = api_key or os.environ.get("AZURE_OPENAI_API_KEY") if base_url: import openai if not api_key: try: token_provider = _make_token_provider() token_provider() # verify it works # Use a custom httpx auth class that refreshes the token on each request import httpx class _AzureADAuth(httpx.Auth): def __init__(self, provider): self._provider = provider def auth_flow(self, request): request.headers["Authorization"] = f"Bearer {self._provider()}" yield request client = openai.OpenAI( base_url=base_url, api_key="aad", # required by SDK but overridden by auth http_client=httpx.Client(auth=_AzureADAuth(token_provider)), ) print(f"✅ Connected via /v1/ project endpoint (DefaultAzureCredential, auto-refresh)") return client, "project-v1-aad" except Exception as e: print(f"⚠️ No API key and DefaultAzureCredential failed: {e}") else: client = openai.OpenAI(base_url=base_url, api_key=api_key) print(f"✅ Connected via /v1/ project endpoint") return client, "project-v1" # Method 2: Foundry SDK project_endpoint = project_endpoint or os.environ.get("AZURE_AI_PROJECT_ENDPOINT") if project_endpoint: try: from azure.ai.projects import AIProjectClient from azure.identity import DefaultAzureCredential credential = DefaultAzureCredential() project_client = AIProjectClient(endpoint=project_endpoint, credential=credential) openai_client = project_client.get_openai_client() print(f"✅ Connected via Foundry SDK") return openai_client, "foundry-sdk" except Exception as e: print(f"⚠️ Foundry SDK failed: {e}") # Method 3: Azure OpenAI endpoint azure_endpoint = azure_endpoint or os.environ.get("AZURE_OPENAI_ENDPOINT") if azure_endpoint: import openai if api_key: client = openai.AzureOpenAI( azure_endpoint=azure_endpoint, api_key=api_key, api_version="2025-04-01-preview", ) print(f"✅ Connected via Azure OpenAI endpoint") return client, "azure-openai" else: # No API key — use DefaultAzureCredential with auto-refresh try: token_provider = _make_token_provider() token_provider() # verify it works client = openai.AzureOpenAI( azure_endpoint=azure_endpoint, azure_ad_token_provider=token_provider, api_version="2025-04-01-preview", ) print(f"✅ Connected via Azure OpenAI endpoint (DefaultAzureCredential, auto-refresh)") return client, "azure-openai-aad" except Exception as e: print(f"⚠️ DefaultAzureCredential failed for Azure endpoint: {e}") print("❌ No valid connection method. Set one of:") print(" OPENAI_BASE_URL (preferred)") print(" AZURE_AI_PROJECT_ENDPOINT (Foundry SDK)") print(" AZURE_OPENAI_ENDPOINT + AZURE_OPENAI_API_KEY") raise SystemExit(1) def upload_file(openai_client, filepath: str, purpose: str = "fine-tune") -> str: """Upload a file to Azure AI Foundry and wait for processing.""" print(f"📤 Uploading {filepath}...") with open(filepath, "rb") as f: file_obj = openai_client.files.create(file=f, purpose=purpose) print(f" File ID: {file_obj.id}") print(f" Waiting for processing...") openai_client.files.wait_for_processing(file_obj.id) print(f" ✅ File ready") return file_obj.id ``` ### `scripts/convert_dataset.py` ``` # /// script # dependencies = [ # "openai>=1.0", # ] # /// """ convert_dataset.py — Convert between SFT, DPO, and RFT dataset formats. Usage: # Parquet/CSV to SFT JSONL python convert_dataset.py --input data.parquet --output train.jsonl --format sft \ --user-column prompt --assistant-column response --system-prompt "You are helpful." # SFT JSONL to DPO (generates rejected via base model) python convert_dataset.py --input train.jsonl --output dpo.jsonl --format dpo \ --base-model gpt-4.1-mini --endpoint $ENDPOINT --api-key $KEY # SFT JSONL to RFT JSONL (passthrough — same format, different intent) python convert_dataset.py --input train.jsonl --output rft.jsonl --format rft # DPO JSONL to SFT (extract chosen responses) python convert_dataset.py --input dpo.jsonl --output sft.jsonl --format sft-from-dpo """ import json import os import sys try: sys.stdout.reconfigure(encoding="utf-8") sys.stderr.reconfigure(encoding="utf-8") except (AttributeError, OSError): pass # Stream not reconfigurable (older Python or non-tty); default encoding is fine import time sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from common import HelpOnErrorParser, get_clients def parquet_to_sft(input_path, output_path, user_col, assistant_col, system_prompt=None): """Convert a parquet or CSV file to SFT JSONL.""" try: import pandas as pd except ImportError: print("Error: pandas required. Install with: pip install pandas pyarrow") sys.exit(1) if input_path.endswith(".parquet"): df = pd.read_parquet(input_path) elif input_path.endswith(".csv"): df = pd.read_csv(input_path) elif input_path.endswith(".json"): df = pd.read_json(input_path) else: print(f"Unsupported format: {input_path}. Use .parquet, .csv, or .json") sys.exit(1) if user_col not in df.columns or assistant_col not in df.columns: print(f"Error: Columns '{user_col}' and/or '{assistant_col}' not found.") print(f"Available columns: {list(df.columns)}") sys.exit(1) count = 0 with open(output_path, "w", encoding="utf-8") as f: for _, row in df.iterrows(): user_content = str(row[user_col]).strip() asst_content = str(row[assistant_col]).strip() if not user_content or not asst_content: continue messages = [] if system_prompt: messages.append({"role": "system", "content": system_prompt}) messages.append({"role": "user", "content": user_content}) messages.append({"role": "assistant", "content": asst_content}) f.write(json.dumps({"messages": messages}, ensure_ascii=False) + "\n") count += 1 print(f"Converted {count} examples to SFT JSONL → {output_path}") def sft_to_dpo(input_path, output_path, client, base_model): """Convert SFT to DPO by generating non-preferred responses from a base model. DPO format uses: input (system+user messages), preferred_output, non_preferred_output. """ with open(input_path, encoding="utf-8") as inf: examples = [] for ln, raw in enumerate(inf, 1): if not raw.strip(): continue try: examples.append(json.loads(raw)) except json.JSONDecodeError as e: print(f" ⚠️ Skipping malformed JSON on line {ln}: {e}") count = 0 with open(output_path, "w", encoding="utf-8") as f: for i, ex in enumerate(examples): msgs = ex["messages"] system_msgs = [m for m in msgs if m["role"] == "system"] user_msg = next((m for m in msgs if m["role"] == "user"), None) asst_msg = next((m for m in msgs if m["role"] == "assistant"), None) if not user_msg or not asst_msg: continue # Generate a non-preferred response from the base model try: gen_msgs = system_msgs + [user_msg] resp = client.chat.completions.create( model=base_model, messages=gen_msgs, temperature=1.0, # High temp for diversity max_completion_tokens=2048, ) rejected_content = resp.choices[0].message.content except Exception as e: print(f" Skipping example {i}: {e}") continue if not rejected_content: # None or empty — content filter, finish=length with no text, etc. # Skip rather than emit a DPO entry with null content (trainer rejects). print(f" Skipping example {i}: base model returned no content") continue # Build DPO entry with correct format input_messages = system_msgs + [user_msg] dpo_entry = { "input": {"messages": input_messages}, "preferred_output": [asst_msg], "non_preferred_output": [{"role": "assistant", "content": rejected_content}], } f.write(json.dumps(dpo_entry, ensure_ascii=False) + "\n") count += 1 if (i + 1) % 50 == 0: print(f" Processed {i+1}/{len(examples)}") time.sleep(1) print(f"Converted {count} examples to DPO JSONL → {output_path}") def sft_to_rft(input_path, output_path): """Convert SFT to RFT format. Strips assistant messages (RFT last message must be user) and adds a placeholder grader field. The user must populate grader reference fields (e.g., expected_answer) before training. """ count = 0 skipped = 0 with open(output_path, "w", encoding="utf-8") as out: with open(input_path, encoding="utf-8") as inf: for ln, line in enumerate(inf, 1): if not line.strip(): continue try: ex = json.loads(line) except json.JSONDecodeError as e: print(f" ⚠️ Skipping malformed JSON on line {ln}: {e}") skipped += 1 continue msgs = ex.get("messages", []) # Keep only system + user messages; RFT last message must be user rft_msgs = [m for m in msgs if m["role"] in ("system", "user")] if not rft_msgs or rft_msgs[-1]["role"] != "user": skipped += 1 continue # Extract assistant content as a reference answer placeholder asst_msgs = [m for m in msgs if m["role"] == "assistant"] expected = asst_msgs[-1]["content"] if asst_msgs else "" rft_entry = {"messages": rft_msgs, "expected_answer": expected} out.write(json.dumps(rft_entry, ensure_ascii=False) + "\n") count += 1 print(f"Converted {count} examples to RFT JSONL → {output_path}") if skipped: print(f" Skipped {skipped} examples (no user message)") print("Note: Review 'expected_answer' fields and update your grader to use item.expected_answer.") def dpo_to_sft(input_path, output_path, system_prompt=None): """Extract chosen responses from DPO format to SFT format.""" count = 0 with open(output_path, "w", encoding="utf-8") as f: with open(input_path, encoding="utf-8") as inf: for ln, line in enumerate(inf, 1): if not line.strip(): continue try: ex = json.loads(line) except json.JSONDecodeError as e: print(f" ⚠️ Skipping malformed JSON on line {ln}: {e}") continue input_messages = ex["input"]["messages"] chosen_messages = ex["preferred_output"] messages = [] if system_prompt: messages.append({"role": "system", "content": system_prompt}) messages.extend(m for m in input_messages if m["role"] != "system") else: messages.extend(input_messages) messages.extend(chosen_messages) f.write(json.dumps({"messages": messages}, ensure_ascii=False) + "\n") count += 1 print(f"Extracted {count} chosen examples to SFT JSONL → {output_path}") def main(): parser = HelpOnErrorParser(description="Convert between fine-tuning dataset formats") parser.add_argument("--input", required=True, help="Input file path") parser.add_argument("--output", required=True, help="Output file path") parser.add_argument("--format", required=True, choices=["sft", "dpo", "rft", "sft-from-dpo"], help="Target format") # SFT from raw data parser.add_argument("--user-column", default="prompt", help="Column name for user input") parser.add_argument("--assistant-column", default="response", help="Column name for assistant output") parser.add_argument("--system-prompt", default=None, help="System prompt to prepend") # DPO generation (needs API connection) parser.add_argument("--base-url", default=os.environ.get("OPENAI_BASE_URL"), help="Project /v1/ URL (preferred)") parser.add_argument("--endpoint", default=os.environ.get("AZURE_OPENAI_ENDPOINT"), help="Azure OpenAI endpoint (fallback)") parser.add_argument("--project-endpoint", default=os.environ.get("AZURE_AI_PROJECT_ENDPOINT"), help="Azure AI project endpoint (Foundry SDK)") parser.add_argument("--api-key", default=os.environ.get("AZURE_OPENAI_API_KEY")) parser.add_argument("--base-model", default="gpt-4.1-mini", help="Base model for generating rejections") args = parser.parse_args() if args.format == "sft": if args.input.endswith(".jsonl"): print("Input is already JSONL — assuming SFT format. Nothing to convert.") if args.input != args.output: import shutil shutil.copy2(args.input, args.output) else: parquet_to_sft(args.input, args.output, args.user_column, args.assistant_column, args.system_prompt) elif args.format == "dpo": client, method = get_clients( base_url=args.base_url, azure_endpoint=args.endpoint, project_endpoint=args.project_endpoint, api_key=args.api_key ) sft_to_dpo(args.input, args.output, client, args.base_model) elif args.format == "rft": sft_to_rft(args.input, args.output) elif args.format == "sft-from-dpo": dpo_to_sft(args.input, args.output, args.system_prompt) if __name__ == "__main__": main() ``` ### `scripts/deploy_model.py` ``` # /// script # dependencies = [ # "openai>=1.0", # "requests", # "azure-identity", # ] # /// """ deploy_model.py — Deploy fine-tuned models on Azure AI Foundry via ARM REST API. Supports all model families with correct format/SKU mapping. Usage: python deploy_model.py --model-id "ft:gpt-4.1-mini-2025-04-14:..." --name "my-ft-eval" --capacity 100 python deploy_model.py --model-id "ft:gpt-oss-20b:..." --name "oss-eval" --format Microsoft --sku GlobalStandard python deploy_model.py --delete --name "my-ft-eval" python deploy_model.py --list """ import os import subprocess import sys try: sys.stdout.reconfigure(encoding="utf-8") sys.stderr.reconfigure(encoding="utf-8") except (AttributeError, OSError): pass # Stream not reconfigurable (older Python or non-tty); default encoding is fine import time sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from common import HelpOnErrorParser import requests def _safe_error_msg(resp): """Extract error message from response, handling non-JSON bodies (HTML 502/503).""" try: return resp.json().get("error", {}).get("message", resp.text[:200]) except (ValueError, KeyError): return resp.text[:200] if resp.text else "Unknown error" # Default Azure resource coordinates — override with env vars or args DEFAULT_SUB = os.environ.get("AZURE_SUBSCRIPTION_ID", "") DEFAULT_RG = os.environ.get("AZURE_RESOURCE_GROUP", "") DEFAULT_ACCOUNT = os.environ.get("AZURE_COGSERVICES_ACCOUNT", "") AZ_CLI = os.environ.get("AZ_CLI_PATH") if not AZ_CLI: import shutil AZ_CLI = shutil.which("az") if not AZ_CLI: # Common Windows paths for candidate in [ r"C:\Program Files (x86)\Microsoft SDKs\Azure\CLI2\wbin\az.cmd", r"C:\Program Files\Microsoft SDKs\Azure\CLI2\wbin\az.cmd", ]: if os.path.exists(candidate): AZ_CLI = candidate break if not AZ_CLI: AZ_CLI = "az" # last resort, hope it's on PATH # Model format auto-detection rules FORMAT_RULES = [ (lambda m: "oss-20b" in m.lower() or "oss20b" in m.lower(), "Microsoft", "GlobalStandard"), (lambda m: "ministral" in m.lower() or "mistral" in m.lower(), "Mistral AI", "GlobalStandard"), (lambda m: "llama" in m.lower() or "meta" in m.lower(), "Meta", "GlobalStandard"), (lambda m: "qwen" in m.lower() or "alibaba" in m.lower(), "Alibaba", "GlobalStandard"), (lambda m: True, "OpenAI", "Standard"), # Default fallback ] def get_arm_token(): """Get a fresh ARM token from Azure CLI.""" result = subprocess.run( [AZ_CLI, "account", "get-access-token", "--query", "accessToken", "-o", "tsv"], capture_output=True, text=True, ) token = result.stdout.strip() if not token: raise RuntimeError(f"Failed to get ARM token: {result.stderr}") return token def arm_url(sub, rg, account, deploy_name=None): """Build the ARM REST API URL.""" base = (f"https://management.azure.com/subscriptions/{sub}" f"/resourceGroups/{rg}" f"/providers/Microsoft.CognitiveServices/accounts/{account}" f"/deployments") if deploy_name: base += f"/{deploy_name}" return base + "?api-version=2024-10-01" def detect_format(model_id): """Auto-detect model format and SKU from model ID.""" for check, fmt, sku in FORMAT_RULES: if check(model_id): return fmt, sku return "OpenAI", "Standard" def create_deployment(sub, rg, account, name, model_id, model_format, sku, capacity): """Create a deployment via ARM REST API.""" token = get_arm_token() url = arm_url(sub, rg, account, name) body = { "sku": {"name": sku, "capacity": capacity}, "properties": { "model": { "format": model_format, "name": model_id, "version": "1", } }, } resp = requests.put(url, headers={ "Authorization": f"Bearer {token}", "Content-Type": "application/json", }, json=body, timeout=(10, 120)) if resp.status_code in (200, 201): print(f"✅ Deployment '{name}' created (format={model_format}, sku={sku}, capacity={capacity})") return True else: print(f"❌ Deployment failed ({resp.status_code}): {_safe_error_msg(resp)}") return False def wait_for_deployment(sub, rg, account, name, timeout=600, poll_interval=15): """Wait for deployment to reach 'Succeeded' state.""" url = arm_url(sub, rg, account, name) start = time.time() while time.time() - start < timeout: token = get_arm_token() try: resp = requests.get(url, headers={"Authorization": f"Bearer {token}"}, timeout=(10, 60)) except requests.exceptions.RequestException as e: print(f" ⚠️ Polling error: {e} — retrying in {poll_interval}s") time.sleep(poll_interval) continue if resp.status_code == 200: try: state = resp.json().get("properties", {}).get("provisioningState", "Unknown") except (ValueError, KeyError): state = "Unknown" print(f" Status: {state}") if state == "Succeeded": return True if state in ("Failed", "Canceled"): print(f" Deployment {state}.") return False time.sleep(poll_interval) print(f" Timed out after {timeout}s") return False def delete_deployment(sub, rg, account, name): """Delete a deployment.""" token = get_arm_token() url = arm_url(sub, rg, account, name) resp = requests.delete(url, headers={"Authorization": f"Bearer {token}"}, timeout=(10, 60)) if resp.status_code in (200, 202, 204): print(f"✅ Deployment '{name}' deleted.") else: print(f"❌ Delete failed ({resp.status_code}): {_safe_error_msg(resp)}") def list_deployments(sub, rg, account): """List all deployments.""" token = get_arm_token() url = arm_url(sub, rg, account) resp = requests.get(url, headers={"Authorization": f"Bearer {token}"}, timeout=(10, 60)) if resp.status_code != 200: print(f"❌ Failed to list deployments ({resp.status_code}): {_safe_error_msg(resp)}") return try: deployments = resp.json().get("value", []) except (ValueError, KeyError): print(f"❌ Failed to parse deployment list: {resp.text[:200]}") return if not deployments: print("No deployments found.") return print(f"{'Name':<40} {'Model':<40} {'SKU':<15} {'State':<15}") print("─" * 110) for d in deployments: name = d.get("name", "?") model = d.get("properties", {}).get("model", {}).get("name", "?") sku = d.get("sku", {}).get("name", "?") state = d.get("properties", {}).get("provisioningState", "?") print(f"{name:<40} {model:<40} {sku:<15} {state:<15}") def main(): parser = HelpOnErrorParser(description="Deploy fine-tuned models on Azure AI Foundry") parser.add_argument("--sub", default=DEFAULT_SUB, help="Azure subscription ID") parser.add_argument("--rg", default=DEFAULT_RG, help="Resource group") parser.add_argument("--account", default=DEFAULT_ACCOUNT, help="Cognitive Services account") # Actions parser.add_argument("--list", action="store_true", help="List all deployments") parser.add_argument("--delete", action="store_true", help="Delete a deployment") parser.add_argument("--wait", action="store_true", help="Wait for deployment to succeed") # Deployment config parser.add_argument("--name", help="Deployment name (max 64 chars, alphanumeric + hyphens)") parser.add_argument("--model-id", help="Fine-tuned model ID (e.g., ft:gpt-4.1-mini:...)") parser.add_argument("--format", help="Model format (auto-detected if not specified)") parser.add_argument("--sku", help="SKU name (auto-detected if not specified)") parser.add_argument("--capacity", type=int, default=100, help="TPM capacity in thousands") args = parser.parse_args() if not all([args.sub, args.rg, args.account]): print("Error: Set --sub/--rg/--account or AZURE_SUBSCRIPTION_ID/AZURE_RESOURCE_GROUP/AZURE_COGSERVICES_ACCOUNT") sys.exit(1) if args.list: list_deployments(args.sub, args.rg, args.account) return if not args.name: print("Error: --name required for create/delete/wait") sys.exit(1) if args.delete: delete_deployment(args.sub, args.rg, args.account, args.name) return if args.wait and not args.model_id: # Wait-only mode: poll an existing deployment success = wait_for_deployment(args.sub, args.rg, args.account, args.name) sys.exit(0 if success else 1) if not args.model_id: print("Error: --model-id required for create") sys.exit(1) # Auto-detect format/SKU if not specified model_format = args.format sku = args.sku if not model_format or not sku: auto_fmt, auto_sku = detect_format(args.model_id) model_format = model_format or auto_fmt sku = sku or auto_sku print(f"Auto-detected: format={model_format}, sku={sku}") created = create_deployment(args.sub, args.rg, args.account, args.name, args.model_id, model_format, sku, args.capacity) if args.wait and created: wait_for_deployment(args.sub, args.rg, args.account, args.name) if __name__ == "__main__": main() ``` ### `scripts/evaluate_model.py` ``` # /// script # dependencies = [ # "openai>=1.0", # "azure-identity", # ] # /// """ evaluate_model.py — Custom 2-dimension LLM judge evaluator for fine-tuned models. This is a lightweight evaluation script using the OpenAI API directly. For production evaluation, prefer the Azure AI Evaluation SDK which provides built-in graders, batch evaluation, and guardrail metrics. See references/evaluation.md for SDK patterns. Uses the OpenAI API directly to: 1. Generate responses from a deployed fine-tuned model 2. Grade each response on correctness and conciseness using an LLM judge 3. Produce aggregate quality scores (weighted 70% correctness, 30% conciseness) By default, system prompts from each test example's messages array are used during generation. The --system-prompt flag overrides this for all examples. Usage: python evaluate_model.py \ --deployment-name my-ft-eval \ --test-file test.jsonl \ --judge-model gpt-4o \ --output results.json python evaluate_model.py \ --base-url "$BASE_URL" --api-key "$API_KEY" \ --deployment-name my-ft-eval \ --test-file test.jsonl \ --concurrency 4 """ import json import os import re import sys try: sys.stdout.reconfigure(encoding="utf-8") sys.stderr.reconfigure(encoding="utf-8") except (AttributeError, OSError): pass # Stream not reconfigurable (older Python or non-tty); default encoding is fine import time from concurrent.futures import ThreadPoolExecutor, as_completed sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from common import HelpOnErrorParser, get_clients, _clamp_score JUDGE_PROMPT = """You are evaluating the quality of a model's output for a given task. ## Task prompt {prompt} ## Reference answer {reference} ## Model output {output} ## Scoring Rate the output on two dimensions, each on a scale of 1-10: **Correctness** (1-10): Does the output correctly accomplish the task? - 1-3: Fundamentally wrong or broken - 4-6: Partially correct with significant issues - 7-8: Mostly correct with minor issues - 9-10: Fully correct **Conciseness** (1-10): Is the output appropriately concise? - 1-3: Extremely verbose or padded - 4-6: Contains unnecessary content - 7-8: Mostly concise with minor excess - 9-10: Clean and focused Return ONLY a JSON object: {{"correctness": , "conciseness": }}""" def load_test_data(filepath): """Load held-out test set. Expects JSONL with 'messages' array. Extracts the system prompt (if present), user prompt, and assistant reference from each example so per-example system prompts are preserved. """ data = [] with open(filepath, encoding="utf-8") as f: for i, line in enumerate(f): if not line.strip(): continue try: ex = json.loads(line) except json.JSONDecodeError as e: print(f"⚠️ Skipping malformed JSON on line {i+1}: {e}") continue msgs = ex.get("messages") if not isinstance(msgs, list): print(f"⚠️ Skipping example {i}: missing or invalid 'messages' list") continue prompt = next((m["content"] for m in msgs if m["role"] == "user"), None) reference = next((m["content"] for m in msgs if m["role"] == "assistant"), None) if not prompt: print(f"⚠️ Skipping example {i}: missing 'user' message") continue if not reference: print(f"⚠️ Skipping example {i}: missing 'assistant' message") continue system_msgs = [m["content"] for m in msgs if m["role"] == "system"] system_prompt = system_msgs[0] if system_msgs else None data.append({"prompt": prompt, "reference": reference, "system_prompt": system_prompt}) return data def generate_response(client, deployment, prompt, system_prompt=None, max_retries=3): """Generate a single response from the deployed model.""" messages = [] if system_prompt: messages.append({"role": "system", "content": system_prompt}) messages.append({"role": "user", "content": prompt}) for attempt in range(max_retries): try: resp = client.chat.completions.create( model=deployment, messages=messages, temperature=0.0, max_completion_tokens=2048, ) content = resp.choices[0].message.content if content is None: # Content filter or empty completion — surface as an error sentinel # so the aggregate filter at line ~`.startswith("ERROR:")` skips it. finish = getattr(resp.choices[0], "finish_reason", "unknown") return f"ERROR: empty content (finish_reason={finish})" return content except Exception as e: if attempt >= max_retries - 1: return f"ERROR: {e}" time.sleep(3 * (attempt + 1)) return "ERROR: max retries exceeded" def grade_response(judge_client, judge_model, prompt, reference, output, max_retries=3): """Grade a response using the LLM judge.""" judge_input = JUDGE_PROMPT.format(prompt=prompt, reference=reference, output=output) for attempt in range(max_retries): try: resp = judge_client.chat.completions.create( model=judge_model, messages=[{"role": "user", "content": judge_input}], temperature=0.0, max_completion_tokens=200, ) text = (resp.choices[0].message.content or "").strip() # Extract JSON from response match = re.search(r'\{[^}]+\}', text) if match: scores = json.loads(match.group()) return { "correctness": _clamp_score(scores.get("correctness")), "conciseness": _clamp_score(scores.get("conciseness")), } except Exception as e: if attempt < max_retries - 1: time.sleep(2) else: return {"correctness": 0, "conciseness": 0, "error": str(e)} return {"correctness": 0, "conciseness": 0, "error": "All retries failed"} def main(): parser = HelpOnErrorParser(description="Evaluate a fine-tuned model with LLM judge") parser.add_argument("--base-url", default=os.environ.get("OPENAI_BASE_URL"), help="Project /v1/ URL (preferred)") parser.add_argument("--endpoint", default=os.environ.get("AZURE_OPENAI_ENDPOINT"), help="Azure OpenAI endpoint (fallback)") parser.add_argument("--project-endpoint", default=os.environ.get("AZURE_AI_PROJECT_ENDPOINT"), help="Azure AI project endpoint (Foundry SDK)") parser.add_argument("--api-key", default=os.environ.get("AZURE_OPENAI_API_KEY")) parser.add_argument("--deployment-name", required=True, help="Deployed model name") parser.add_argument("--test-file", required=True, help="Held-out test set (JSONL)") parser.add_argument("--system-prompt", default=None, help="Override system prompt for all examples (default: use per-example system prompt from test data)") # Judge config parser.add_argument("--judge-model", default="gpt-4o", help="Model for LLM judge") parser.add_argument("--judge-endpoint", help="Endpoint for judge (default: same as model)") parser.add_argument("--judge-api-key", help="API key for judge (default: same as model)") # Output parser.add_argument("--output", default="eval_results.json", help="Output file") parser.add_argument("--concurrency", type=int, default=1, help="Parallel grading workers (generation is always sequential)") args = parser.parse_args() # Set up model client via shared auth (supports /v1/, Foundry SDK, AzureOpenAI) model_client, method = get_clients( base_url=args.base_url, azure_endpoint=args.endpoint, project_endpoint=args.project_endpoint, api_key=args.api_key ) # Set up judge client (defaults to same connection as model) judge_key = args.judge_api_key or args.api_key if args.judge_endpoint: judge_client, _ = get_clients(azure_endpoint=args.judge_endpoint, api_key=judge_key) elif args.judge_api_key: # Different API key but same endpoint — create a new client with the judge key judge_client, _ = get_clients( base_url=args.base_url, azure_endpoint=args.endpoint, project_endpoint=args.project_endpoint, api_key=judge_key ) else: judge_client = model_client # Load data test_data = load_test_data(args.test_file) print(f"Loaded {len(test_data)} test examples from {args.test_file}") # Phase 1: Generate responses (sequential to avoid rate limits) print(f"\nGenerating responses from {args.deployment_name}...") for i, ex in enumerate(test_data): # Use CLI override if provided, otherwise use per-example system prompt effective_system_prompt = args.system_prompt if args.system_prompt is not None else ex.get("system_prompt") ex["output"] = generate_response( model_client, args.deployment_name, ex["prompt"], effective_system_prompt ) if (i + 1) % 10 == 0: print(f" Generated {i+1}/{len(test_data)}") errors = sum(1 for ex in test_data if ex["output"].startswith("ERROR:")) print(f" Done. {errors} errors out of {len(test_data)}.") # Phase 2: Grade responses (parallel) print(f"\nGrading with {args.judge_model} (concurrency={args.concurrency})...") def grade_one(ex): return grade_response(judge_client, args.judge_model, ex["prompt"], ex["reference"], ex["output"]) with ThreadPoolExecutor(max_workers=args.concurrency) as pool: futures = {pool.submit(grade_one, ex): i for i, ex in enumerate(test_data)} for future in as_completed(futures): idx = futures[future] test_data[idx]["scores"] = future.result() # Aggregate valid_scores = [ex["scores"] for ex in test_data if ex["scores"]["correctness"] > 0] if not valid_scores: print("No valid scores — all grading failed.") sys.exit(1) avg_corr = sum(s["correctness"] for s in valid_scores) / len(valid_scores) avg_conc = sum(s["conciseness"] for s in valid_scores) / len(valid_scores) combined = 0.7 * avg_corr + 0.3 * avg_conc print(f"\n{'='*50}") print(f"Results for {args.deployment_name}") print(f" Correctness: {avg_corr:.2f}") print(f" Conciseness: {avg_conc:.2f}") print(f" Combined: {combined:.2f}") print(f" (N={len(valid_scores)} scored, {len(test_data)-len(valid_scores)} failed)") print(f"{'='*50}") # Save results = { "deployment": args.deployment_name, "judge_model": args.judge_model, "n_examples": len(test_data), "n_scored": len(valid_scores), "correctness": round(avg_corr, 2), "conciseness": round(avg_conc, 2), "combined": round(combined, 2), "details": [ { "prompt": ex["prompt"][:200], "scores": ex.get("scores", {}), } for ex in test_data ], } with open(args.output, "w", encoding="utf-8") as f: json.dump(results, f, indent=2) print(f"\nDetailed results saved to {args.output}") if __name__ == "__main__": main() ``` ### `scripts/generate_distillation_data.py` ``` # /// script # dependencies = [ # "openai>=1.0", # "azure-identity", # ] # /// """ generate_distillation_data.py — Generate training data from a teacher model for distillation. Creates a synthetic SFT dataset by: 1. Generating diverse prompts from combinatorial axes (topics × formats × contexts) 2. Having the teacher model produce responses 3. Quality-grading each response with an LLM judge 4. Filtering low-quality examples 5. Splitting into train/val/test JSONL files Usage: python generate_distillation_data.py \ --teacher gpt-4.1-mini \ --system-prompt "You are a formal business writer." \ --topics "earnings,risk,compliance" \ --num-prompts 300 \ --min-score 7.0 \ --output-dir ./my_dataset # Or with a prompts file (one prompt per line): python generate_distillation_data.py \ --teacher gpt-4.1-mini \ --prompts-file my_prompts.txt \ --output-dir ./my_dataset """ import json import os import random import re import sys try: sys.stdout.reconfigure(encoding="utf-8") sys.stderr.reconfigure(encoding="utf-8") except (AttributeError, OSError): pass # Stream not reconfigurable (older Python or non-tty); default encoding is fine import time sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from common import HelpOnErrorParser, get_clients, _clamp_score import openai def verify_deployment(client, model): """Verify a model deployment exists by sending a trivial request.""" try: client.chat.completions.create( model=model, messages=[{"role": "user", "content": "Hi"}], max_completion_tokens=1, ) return True except openai.NotFoundError: return False except Exception: return True # other errors (rate limit, etc.) mean the deployment exists def generate_combinatorial_prompts(topics, formats, contexts, n): """Generate diverse prompts from combinatorial axes.""" prompts = [] for _ in range(n): t = random.choice(topics) f = random.choice(formats) c = random.choice(contexts) prompts.append(f"Context: {c}\n\nWrite {f} about: {t}.") return prompts def teacher_generate(client, model, system_prompt, prompt, retries=3): """Generate a single response from the teacher.""" for attempt in range(retries): try: resp = client.chat.completions.create( model=model, messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": prompt}, ], temperature=0.7, max_completion_tokens=1024, ) return resp.choices[0].message.content except Exception as e: if attempt >= retries - 1: print(f" Failed after {retries} attempts: {e}") return None time.sleep(2 * (attempt + 1)) return None QUALITY_PROMPT = """Rate this AI-generated text on quality dimensions (1-10 each). ## Text to evaluate {output} ## Dimensions **Accuracy** (1-10): Is the content factually sound and coherent? **Quality** (1-10): Is it well-written, clear, and professional? **Task-fit** (1-10): Does it match the requested format and purpose? Return ONLY JSON: {{"accuracy": , "quality": , "task_fit": }}""" def grade_output(client, judge_model, output, retries=3): for attempt in range(retries): try: resp = client.chat.completions.create( model=judge_model, messages=[{"role": "user", "content": QUALITY_PROMPT.format(output=output)}], temperature=0.0, max_completion_tokens=100, ) text = (resp.choices[0].message.content or "").strip() match = re.search(r'\{[^}]+\}', text) if match: scores = json.loads(match.group()) return {k: _clamp_score(v) for k, v in scores.items()} except Exception: if attempt < retries - 1: time.sleep(2) return None def main(): parser = HelpOnErrorParser(description="Generate distillation training data from a teacher model") parser.add_argument("--base-url", default=os.environ.get("OPENAI_BASE_URL"), help="Project /v1/ URL (preferred)") parser.add_argument("--endpoint", default=os.environ.get("AZURE_OPENAI_ENDPOINT"), help="Azure OpenAI endpoint (fallback)") parser.add_argument("--project-endpoint", default=os.environ.get("AZURE_AI_PROJECT_ENDPOINT"), help="Azure AI project endpoint (Foundry SDK)") parser.add_argument("--api-key", default=os.environ.get("AZURE_OPENAI_API_KEY")) parser.add_argument("--teacher", required=True, help="Teacher model deployment name") parser.add_argument("--judge", default=None, help="Judge model (default: same as teacher)") parser.add_argument("--system-prompt", default="You are a helpful assistant.", help="System prompt for teacher") # Prompt generation (either combinatorial or from file) parser.add_argument("--prompts-file", help="File with one prompt per line (skips combinatorial generation)") parser.add_argument("--topics", help="Comma-separated topics for combinatorial prompts") parser.add_argument("--formats", default="a concise response,a brief summary,a detailed explanation", help="Comma-separated output formats") parser.add_argument("--contexts", default="", help="Comma-separated context sentences") parser.add_argument("--num-prompts", type=int, default=300, help="Number of prompts to generate") # Quality parser.add_argument("--min-score", type=float, default=7.0, help="Minimum average quality score to keep") parser.add_argument("--skip-grading", action="store_true", help="Skip quality grading (keep all)") # Output parser.add_argument("--output-dir", default="./distillation_data", help="Output directory") parser.add_argument("--train-split", type=float, default=0.8) parser.add_argument("--val-split", type=float, default=0.1) args = parser.parse_args() client, method = get_clients( base_url=args.base_url, azure_endpoint=args.endpoint, project_endpoint=args.project_endpoint, api_key=args.api_key ) judge = args.judge or args.teacher # Step 0: Verify deployments exist print(f"Verifying deployment '{args.teacher}'...") if not verify_deployment(client, args.teacher): print(f" ERROR: Deployment '{args.teacher}' not found. Available deployments can be listed in Azure Portal.") sys.exit(1) print(f" ✅ Teacher deployment verified.") if judge != args.teacher: print(f"Verifying judge deployment '{judge}'...") if not verify_deployment(client, judge): print(f" ERROR: Judge deployment '{judge}' not found.") sys.exit(1) print(f" ✅ Judge deployment verified.") # Step 1: Generate or load prompts if args.prompts_file: with open(args.prompts_file, encoding="utf-8") as pf: prompts = [line.strip() for line in pf if line.strip()] print(f"Loaded {len(prompts)} prompts from {args.prompts_file}") else: topics = [t.strip() for t in (args.topics or "general knowledge").split(",")] formats = [f.strip() for f in args.formats.split(",")] contexts = [c.strip() for c in args.contexts.split(",") if c.strip()] or [""] prompts = generate_combinatorial_prompts(topics, formats, contexts, args.num_prompts) print(f"Generated {len(prompts)} prompts ({len(topics)} topics × {len(formats)} formats × {len(contexts)} contexts)") # Step 2: Teacher generates responses print(f"\nTeacher ({args.teacher}) generating responses...") examples = [] for i, prompt in enumerate(prompts): response = teacher_generate(client, args.teacher, args.system_prompt, prompt) if response: examples.append({"prompt": prompt, "response": response}) if (i + 1) % 25 == 0: print(f" {i+1}/{len(prompts)} ({len(examples)} successful)") print(f" Teacher produced {len(examples)}/{len(prompts)} responses") # Step 3: Quality grade and filter if not args.skip_grading: print(f"\nGrading with {judge}...") for i, ex in enumerate(examples): scores = grade_output(client, judge, ex["response"]) if scores: ex["scores"] = scores ex["avg_score"] = sum(scores.values()) / len(scores) else: ex["avg_score"] = 0 if (i + 1) % 25 == 0: print(f" Graded {i+1}/{len(examples)}") filtered = [ex for ex in examples if ex["avg_score"] >= args.min_score] avgs = [ex["avg_score"] for ex in examples if ex["avg_score"] > 0] print(f" Passed filter (>= {args.min_score}): {len(filtered)}/{len(examples)}") if avgs: print(f" Scores: min={min(avgs):.1f}, max={max(avgs):.1f}, mean={sum(avgs)/len(avgs):.1f}") else: filtered = examples print(f"Skipping grading — keeping all {len(filtered)} examples") # Step 4: Convert to SFT format and split sft_data = [{"messages": [ {"role": "system", "content": args.system_prompt}, {"role": "user", "content": ex["prompt"]}, {"role": "assistant", "content": ex["response"]}, ]} for ex in filtered] random.shuffle(sft_data) n = len(sft_data) t_end = int(n * args.train_split) v_end = int(n * (args.train_split + args.val_split)) splits = {"train": sft_data[:t_end], "validation": sft_data[t_end:v_end], "test": sft_data[v_end:]} os.makedirs(args.output_dir, exist_ok=True) for name, data in splits.items(): path = os.path.join(args.output_dir, f"{name}.jsonl") with open(path, "w", encoding="utf-8") as f: for ex in data: f.write(json.dumps(ex, ensure_ascii=False) + "\n") print(f" {name}: {len(data)} examples → {path}") print(f"\n✅ Done! Dataset ready in {args.output_dir}/") if __name__ == "__main__": main() ``` ### `scripts/monitor_training.py` ``` # /// script # dependencies = [ # "openai>=1.0", # "azure-identity", # "azure-ai-projects", # ] # /// """ monitor_training.py — Monitor a fine-tuning job until completion. Polls the job status and streams training events (reward, loss, errors) in real time. Exits when the job reaches a terminal state. Usage: python monitor_training.py --job-id ftjob-abc123 python monitor_training.py --base-url https://.services.ai.azure.com/api/projects//openai/v1/ --api-key KEY --job-id ftjob-abc123 python monitor_training.py --job-id ftjob-abc123 --poll-interval 30 """ import argparse import os import sys try: sys.stdout.reconfigure(encoding="utf-8") sys.stderr.reconfigure(encoding="utf-8") except (AttributeError, OSError): pass # Stream not reconfigurable (older Python or non-tty); default encoding is fine import time sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from common import HelpOnErrorParser, get_clients TERMINAL_STATUSES = {"succeeded", "failed", "cancelled"} def monitor_job(client, job_id, poll_interval=15): """Poll a fine-tuning job until it reaches a terminal state.""" # Cap memory for long-running jobs (RFT can run hours/days, accumulating thousands of events) seen_events = set() MAX_SEEN_EVENTS = 5000 print(f"Monitoring job: {job_id}") print(f"Polling every {poll_interval}s. Ctrl+C to stop.\n") while True: try: job = client.fine_tuning.jobs.retrieve(job_id) except Exception as e: print(f"⚠️ Error retrieving job: {e}") time.sleep(poll_interval) continue status = (job.status or "").lower() # Fetch and display new events try: events = list(client.fine_tuning.jobs.list_events(job_id, limit=20)) for event in reversed(events): event_key = (event.created_at, event.message) if event_key not in seen_events: if len(seen_events) >= MAX_SEEN_EVENTS: # Keep only the most recent half — a fully-flushed dedup window # would risk re-printing old events on transient API hiccups, but # without trimming this set grows unbounded for long RFT runs. seen_events = set(list(seen_events)[-(MAX_SEEN_EVENTS // 2):]) seen_events.add(event_key) ts = time.strftime("%H:%M:%S", time.localtime(event.created_at)) level = event.level or "info" # Highlight step events if "Step" in event.message and "reward" in event.message: print(f" 📈 [{ts}] {event.message}") elif "Step" in event.message and "loss" in event.message: print(f" 📉 [{ts}] {event.message}") elif "error" in event.message.lower() or level == "error": print(f" ❌ [{ts}] {event.message}") elif "started" in event.message.lower() or "completed" in event.message.lower(): print(f" 🔔 [{ts}] {event.message}") else: print(f" ℹ️ [{ts}] {event.message}") except Exception: pass # Events API may not be available for all job states # Check terminal state if status in TERMINAL_STATUSES: print(f"\n{'='*50}") if status == "succeeded": model = job.fine_tuned_model or "unknown" print(f" ✅ Job succeeded!") print(f" Fine-tuned model: {model}") if job.trained_tokens: print(f" Trained tokens: {job.trained_tokens:,}") elif status == "failed": print(f" ❌ Job failed.") if hasattr(job, "error") and job.error: print(f" Error: {job.error}") elif status == "cancelled": print(f" ⚠️ Job was cancelled.") print(f"{'='*50}") return status time.sleep(poll_interval) def build_parser(): parser = HelpOnErrorParser( description="Monitor a fine-tuning job until completion", epilog=( "Example:\n" " python monitor_training.py --job-id ftjob-abc123\n" " python monitor_training.py --base-url https://.services.ai.azure.com/api/projects//openai/v1/ --api-key KEY --job-id ftjob-abc123" ), formatter_class=argparse.RawTextHelpFormatter, ) parser.add_argument("--base-url", default=os.environ.get("OPENAI_BASE_URL"), help="Project /v1/ endpoint URL") parser.add_argument("--endpoint", default=os.environ.get("AZURE_OPENAI_ENDPOINT"), help="Azure OpenAI endpoint (fallback)") parser.add_argument("--api-key", default=os.environ.get("AZURE_OPENAI_API_KEY"), help="API key") parser.add_argument("--project-endpoint", default=os.environ.get("AZURE_AI_PROJECT_ENDPOINT"), help="Azure AI project endpoint (alternative to --base-url)") parser.add_argument("--job-id", required=True, help="Fine-tuning job ID (e.g., ftjob-abc123)") parser.add_argument("--poll-interval", type=int, default=15, help="Seconds between status checks (default: 15)") return parser if __name__ == "__main__": parser = build_parser() if len(sys.argv) == 1: parser.print_help() sys.exit(0) args = parser.parse_args() client, method = get_clients(base_url=args.base_url, azure_endpoint=args.endpoint, project_endpoint=args.project_endpoint, api_key=args.api_key) status = monitor_job(client, args.job_id, args.poll_interval) sys.exit(0 if status == "succeeded" else 1) ``` ### `scripts/score_dataset.py` ``` # /// script # dependencies = [ # "openai>=1.0", # "azure-identity", # ] # /// """ score_dataset.py — Assess training data quality using an LLM judge. Scores each example on correctness and relevance, optionally filters out low-quality examples. Usage: # Score all examples python score_dataset.py --input training.jsonl --output scored.jsonl # Score and filter (keep only score >= 7) python score_dataset.py --input training.jsonl --output filtered.jsonl --min-score 7 # Custom scoring dimensions python score_dataset.py --input training.jsonl --output scored.jsonl \ --dimensions "correctness,clarity,completeness" """ import json import os import re import sys try: sys.stdout.reconfigure(encoding="utf-8") sys.stderr.reconfigure(encoding="utf-8") except (AttributeError, OSError): pass # Stream not reconfigurable (older Python or non-tty); default encoding is fine import time from concurrent.futures import ThreadPoolExecutor, as_completed sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from common import HelpOnErrorParser, get_clients, _clamp_score QUALITY_PROMPT = """You are a data quality assessor for machine learning training data. ## Task Evaluate this training example for quality. ## User input (what the model receives) {user_content} ## Assistant output (what the model should learn to produce) {assistant_content} ## Scoring dimensions {dimensions_text} Rate each dimension on a scale of 1-10. Return ONLY a JSON object with dimension names as keys and integer scores as values. Example: {example_json}""" DEFAULT_DIMENSIONS = { "correctness": "Is the assistant's output factually/functionally correct?", "relevance": "Does the output directly address the user's request?", "quality": "Is the output well-written, well-formatted, and professional?", } def score_example(client, model, user_content, assistant_content, dimensions): """Score a single training example.""" dims_text = "\n".join(f"**{k}** (1-10): {v}" for k, v in dimensions.items()) example = {k: 8 for k in dimensions} prompt = QUALITY_PROMPT.format( user_content=user_content[:2000], assistant_content=assistant_content[:2000], dimensions_text=dims_text, example_json=json.dumps(example), ) for attempt in range(3): try: resp = client.chat.completions.create( model=model, messages=[{"role": "user", "content": prompt}], temperature=0.0, max_completion_tokens=200, ) text = (resp.choices[0].message.content or "").strip() match = re.search(r'\{[^}]+\}', text) if match: scores = json.loads(match.group()) return {k: _clamp_score(scores.get(k)) for k in dimensions} except Exception: if attempt < 2: time.sleep(2) return {k: 0 for k in dimensions} def main(): parser = HelpOnErrorParser(description="Score training data quality with LLM judge") parser.add_argument("--base-url", default=os.environ.get("OPENAI_BASE_URL"), help="Project /v1/ URL (preferred)") parser.add_argument("--endpoint", default=os.environ.get("AZURE_OPENAI_ENDPOINT"), help="Azure OpenAI endpoint (fallback)") parser.add_argument("--project-endpoint", default=os.environ.get("AZURE_AI_PROJECT_ENDPOINT"), help="Azure AI project endpoint (Foundry SDK)") parser.add_argument("--api-key", default=os.environ.get("AZURE_OPENAI_API_KEY")) parser.add_argument("--model", default="gpt-4o", help="Judge model") parser.add_argument("--input", required=True, help="Input JSONL file") parser.add_argument("--output", required=True, help="Output JSONL file (with scores)") parser.add_argument("--min-score", type=float, default=None, help="Minimum average score to keep (filters below this)") parser.add_argument("--dimensions", default=None, help="Comma-separated dimension names (default: correctness,relevance,quality)") parser.add_argument("--concurrency", type=int, default=4, help="Parallel scoring workers") parser.add_argument("--strip-metadata", action="store_true", help="Remove _quality_scores and _avg_quality from output (safe for training input)") args = parser.parse_args() client, method = get_clients( base_url=args.base_url, azure_endpoint=args.endpoint, project_endpoint=args.project_endpoint, api_key=args.api_key ) # Parse dimensions if args.dimensions: dim_names = [d.strip() for d in args.dimensions.split(",")] dimensions = {d: f"Rate the {d} of the output" for d in dim_names} else: dimensions = DEFAULT_DIMENSIONS # Load data examples = [] with open(args.input, encoding="utf-8") as f: for i, line in enumerate(f): if not line.strip(): continue try: ex = json.loads(line) except json.JSONDecodeError as e: print(f"⚠️ Skipping malformed JSON on line {i+1}: {e}") continue msgs = ex.get("messages", []) user = next((m["content"] for m in msgs if m["role"] == "user"), "") asst = next((m["content"] for m in msgs if m["role"] == "assistant"), "") examples.append({"data": ex, "user": user, "assistant": asst}) print(f"Loaded {len(examples)} examples. Scoring with {args.model}...") # Score in parallel def score_one(idx): ex = examples[idx] scores = score_example(client, args.model, ex["user"], ex["assistant"], dimensions) return idx, scores with ThreadPoolExecutor(max_workers=args.concurrency) as pool: futures = {pool.submit(score_one, i): i for i in range(len(examples))} done = 0 for future in as_completed(futures): idx, scores = future.result() examples[idx]["scores"] = scores done += 1 if done % 25 == 0: print(f" Scored {done}/{len(examples)}") # Calculate stats all_avgs = [] for ex in examples: scores = ex.get("scores", {}) if scores and any(v > 0 for v in scores.values()): avg = sum(scores.values()) / len(scores) ex["avg_score"] = avg all_avgs.append(avg) if all_avgs: print(f"\nQuality Distribution:") print(f" Mean: {sum(all_avgs)/len(all_avgs):.1f}") print(f" Min: {min(all_avgs):.1f}") print(f" Max: {max(all_avgs):.1f}") sorted_avgs = sorted(all_avgs) n_avgs = len(sorted_avgs) if n_avgs % 2 == 1: median = sorted_avgs[n_avgs // 2] else: median = (sorted_avgs[n_avgs // 2 - 1] + sorted_avgs[n_avgs // 2]) / 2 print(f" Median: {median:.1f}") # Filter and write kept = 0 filtered = 0 with open(args.output, "w", encoding="utf-8") as f: for ex in examples: if not args.strip_metadata: ex["data"]["_quality_scores"] = ex.get("scores", {}) ex["data"]["_avg_quality"] = ex.get("avg_score", 0) if args.min_score and ex.get("avg_score", 0) < args.min_score: filtered += 1 continue f.write(json.dumps(ex["data"], ensure_ascii=False) + "\n") kept += 1 print(f"\nKept: {kept}, Filtered: {filtered}") if args.min_score: print(f"(min_score threshold: {args.min_score})") if args.strip_metadata: print("(metadata stripped — output is safe for training input)") print(f"Output: {args.output}") if __name__ == "__main__": main() ``` ### `scripts/submit_training.py` ``` # /// script # dependencies = [ # "openai>=1.0", # "requests", # "azure-identity", # "azure-ai-projects", # ] # /// """ submit_training.py — Submit SFT, DPO, or RFT training jobs on Azure AI Foundry. Handles both SDK and REST API submission (REST fallback for OSS models). Supports /v1/ project endpoint (preferred) and Azure endpoint (fallback). Usage: python submit_training.py --base-url https://.services.ai.azure.com/api/projects//openai/v1/ \ --api-key KEY --training-file training.jsonl --validation-file validation.jsonl \ --model gpt-4.1-mini --type sft --epochs 2 --lr 1.0 python submit_training.py --endpoint https://.openai.azure.com --api-key KEY \ --training-file-id file-abc123 --validation-file-id file-def456 \ --model gpt-oss-20b --type sft --epochs 2 --lr 0.5 --use-rest python submit_training.py --base-url --api-key KEY \ --training-file-id file-abc123 --validation-file-id file-def456 \ --model o4-mini-2025-04-16 --type rft --grader-file grader.py """ import json import os import sys try: sys.stdout.reconfigure(encoding="utf-8") sys.stderr.reconfigure(encoding="utf-8") except (AttributeError, OSError): pass # Stream not reconfigurable (older Python or non-tty); default encoding is fine sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from common import HelpOnErrorParser, get_clients, upload_file import requests def submit_sft_sdk(client, model, train_id, val_id, epochs=2, lr=1.0, batch_size=None, suffix=None, training_type="globalStandard"): """Submit SFT job using the Python SDK.""" hp = {"n_epochs": epochs, "learning_rate_multiplier": lr} if batch_size: hp["batch_size"] = batch_size kwargs = dict( model=model, training_file=train_id, validation_file=val_id, method={"type": "supervised"}, hyperparameters=hp, # Azure-specific: passed via extra_body since the OpenAI SDK has no # top-level trainingType kwarg. extra_body={"trainingType": training_type}, ) if suffix: kwargs["suffix"] = suffix job = client.fine_tuning.jobs.create(**kwargs) return {"id": job.id, "status": job.status, "model": model, "method": "sdk"} def submit_sft_rest(endpoint, api_key, model, train_id, val_id, epochs=2, lr=1.0, batch_size=None, suffix=None, training_type="globalStandard"): """Submit SFT job via REST API (fallback for models like gpt-oss-20b).""" url = f"{endpoint}/openai/fine_tuning/jobs?api-version=2025-04-01-preview" body = { "model": model, "training_file": train_id, "validation_file": val_id, "method": {"type": "supervised"}, "hyperparameters": {"n_epochs": epochs, "learning_rate_multiplier": lr}, "trainingType": training_type, } if batch_size: body["hyperparameters"]["batch_size"] = batch_size if suffix: body["suffix"] = suffix resp = requests.post(url, headers={ "Content-Type": "application/json", "api-key": api_key, }, json=body, timeout=(10, 60)) if resp.status_code in (200, 201): try: data = resp.json() except ValueError: raise RuntimeError( f"REST submission returned {resp.status_code} but body was not JSON: {resp.text[:200]}" ) if "id" not in data or "status" not in data: raise RuntimeError(f"REST response missing 'id' or 'status' fields: {data}") return {"id": data["id"], "status": data["status"], "model": model, "method": "rest"} else: try: err_msg = resp.json().get('error', {}).get('message', 'Unknown error') except (ValueError, KeyError): err_msg = resp.text[:200] if resp.text else "Unknown error" raise RuntimeError( f"REST submission failed ({resp.status_code}): {err_msg}" ) def submit_rft(client, model, train_id, val_id, grader_source): """Submit RFT job.""" job = client.fine_tuning.jobs.create( model=model, training_file=train_id, validation_file=val_id, method={ "type": "reinforcement", "reinforcement": { "grader": { "type": "python", "name": "custom_grader", "source": grader_source, }, }, }, ) return {"id": job.id, "status": job.status, "model": model, "method": "sdk-rft"} def submit_dpo(client, model, train_id, val_id, epochs=2, lr=1.0, beta=0.1, suffix=None): """Submit DPO job.""" job = client.fine_tuning.jobs.create( model=model, training_file=train_id, validation_file=val_id, suffix=suffix or None, method={ "type": "dpo", "dpo": { "hyperparameters": { "n_epochs": epochs, "beta": beta, "learning_rate_multiplier": lr, }, }, }, ) return {"id": job.id, "status": job.status, "model": model, "method": "sdk-dpo"} def main(): parser = HelpOnErrorParser(description="Submit fine-tuning jobs on Azure AI Foundry") parser.add_argument("--base-url", default=os.environ.get("OPENAI_BASE_URL"), help="Project /v1/ URL (preferred)") parser.add_argument("--endpoint", default=os.environ.get("AZURE_OPENAI_ENDPOINT"), help="Azure OpenAI endpoint (fallback)") parser.add_argument("--project-endpoint", default=os.environ.get("AZURE_AI_PROJECT_ENDPOINT"), help="Azure AI project endpoint (Foundry SDK)") parser.add_argument("--api-key", default=os.environ.get("AZURE_OPENAI_API_KEY"), help="API key") parser.add_argument("--model", required=True, help="Base model name (e.g., gpt-4.1-mini)") parser.add_argument("--type", choices=["sft", "dpo", "rft"], default="sft", help="Training type: sft, dpo, or rft") # Data files — either paths (will upload) or IDs (already uploaded) parser.add_argument("--training-file", help="Path to training JSONL file (will upload)") parser.add_argument("--validation-file", help="Path to validation JSONL file (will upload)") parser.add_argument("--training-file-id", help="Already-uploaded training file ID") parser.add_argument("--validation-file-id", help="Already-uploaded validation file ID") # Hyperparameters parser.add_argument("--epochs", type=int, default=2) parser.add_argument("--lr", type=float, default=1.0, help="Learning rate multiplier") parser.add_argument("--batch-size", type=int, default=None) parser.add_argument("--suffix", help="Model suffix for identification") # DPO-specific parser.add_argument("--beta", type=float, default=0.1, help="DPO beta (alignment strength)") # RFT-specific parser.add_argument("--grader-file", help="Path to Python grader file (for RFT)") # REST fallback parser.add_argument("--use-rest", action="store_true", help="Force REST API (needed for gpt-oss-20b and other OSS models)") parser.add_argument("--training-type", choices=["globalStandard", "developerTier", "standard"], default="globalStandard", help="Azure training tier (default: globalStandard). developerTier is ~50%% off " "globalStandard with lower quotas. OSS models (gpt-oss-20b, Ministral, " "Llama, Qwen) only support globalStandard.") args = parser.parse_args() client, method = get_clients( base_url=args.base_url, azure_endpoint=args.endpoint, project_endpoint=args.project_endpoint, api_key=args.api_key ) # Resolve file IDs train_id = args.training_file_id val_id = args.validation_file_id if args.training_file: train_id = upload_file(client, args.training_file) if args.validation_file: val_id = upload_file(client, args.validation_file) if not train_id or not val_id: print("Error: Provide training and validation file paths or IDs") sys.exit(1) # Submit if args.type == "rft": if not args.grader_file: print("Error: --grader-file required for RFT") sys.exit(1) with open(args.grader_file, encoding="utf-8") as f: grader_source = f.read() result = submit_rft(client, args.model, train_id, val_id, grader_source) elif args.type == "dpo": result = submit_dpo(client, args.model, train_id, val_id, args.epochs, args.lr, args.beta, args.suffix) elif args.use_rest: if not args.endpoint or not args.api_key: print("Error: --use-rest requires --endpoint and --api-key (REST does not support DefaultAzureCredential)") sys.exit(1) result = submit_sft_rest(args.endpoint, args.api_key, args.model, train_id, val_id, args.epochs, args.lr, args.batch_size, args.suffix, args.training_type) else: # SFT via SDK with REST fallback for OSS models try: result = submit_sft_sdk(client, args.model, train_id, val_id, args.epochs, args.lr, args.batch_size, args.suffix, args.training_type) except Exception as e: err_str = str(e).lower() # Match a wider set of "use REST instead" signals than the original # exact-string comparison: Azure changes error text periodically. if ("trainingtype" in err_str or "globalstandard" in err_str or "global_standard" in err_str or "does not support fine-tuning" in err_str): if not args.endpoint or not args.api_key: print(f"SDK failed for {args.model}. REST fallback requires --endpoint and --api-key.") sys.exit(1) print(f"SDK failed for {args.model}, falling back to REST API...") result = submit_sft_rest(args.endpoint, args.api_key, args.model, train_id, val_id, args.epochs, args.lr, args.batch_size, args.suffix, args.training_type) else: raise print(f"\nJob submitted successfully:") print(json.dumps(result, indent=2)) # Save job info outfile = f"ft_job_{result['id']}.json" with open(outfile, "w", encoding="utf-8") as f: json.dump({**result, "epochs": args.epochs, "lr": args.lr, "batch_size": args.batch_size, "train_file": train_id, "val_file": val_id}, f, indent=2) print(f"Job info saved to {outfile}") if __name__ == "__main__": main() ``` ### `scripts/validate/__init__.py` ``` # Validation scripts for SFT, DPO, and RFT JSONL files. # Usage: # python -m scripts.validate.validate_sft # python -m scripts.validate.validate_dpo # python -m scripts.validate.validate_rft # python -m scripts.validate.data_stats ``` ### `scripts/validate/data_stats.py` ``` #!/usr/bin/env python3 """Compute dataset statistics for any fine-tuning JSONL file. Adapted from foundry-ft agent. Auto-detects SFT/DPO/RFT format and reports token estimates, role distribution, and rough cost estimates. """ import json import sys try: sys.stdout.reconfigure(encoding="utf-8") sys.stderr.reconfigure(encoding="utf-8") except (AttributeError, OSError): pass # Stream not reconfigurable (older Python or non-tty); default encoding is fine from collections import Counter def estimate_tokens(text: str) -> int: """Rough token estimate: ~4 chars per token for English text.""" return max(1, len(text) // 4) def extract_text(record: dict) -> str: """Extract all text content from a record regardless of format.""" texts = [] if "messages" in record: for msg in record["messages"]: if "content" in msg and msg["content"]: texts.append(str(msg["content"])) if "input" in record and "messages" in record["input"]: for msg in record["input"]["messages"]: if "content" in msg and msg["content"]: texts.append(str(msg["content"])) for field in ["preferred_output", "non_preferred_output"]: if field in record: for msg in record[field]: if "content" in msg and msg["content"]: texts.append(str(msg["content"])) # Include any extra fields beyond messages/input/preferred_output/non_preferred_output known_structural = {"messages", "input", "preferred_output", "non_preferred_output"} for field in record: if field not in known_structural and isinstance(record[field], (str, int, float)): texts.append(str(record[field])) return " ".join(texts) def data_stats(filepath: str) -> None: records = [] format_type = "unknown" parse_errors = 0 with open(filepath, "r", encoding="utf-8") as f: for line in f: line = line.strip() if not line: continue try: records.append(json.loads(line)) except json.JSONDecodeError: parse_errors += 1 if not records: print(f"No valid records found in {filepath}") sys.exit(1) # Detect format first = records[0] if "input" in first and "preferred_output" in first: format_type = "DPO" elif "messages" in first: msgs = first["messages"] extra_fields = set(first.keys()) - {"messages"} last_role = msgs[-1].get("role") if isinstance(msgs, list) and msgs else None if extra_fields and last_role == "user": format_type = "RFT" else: format_type = "SFT" # Compute stats token_counts = [estimate_tokens(extract_text(r)) for r in records] total_tokens = sum(token_counts) avg_tokens = total_tokens / len(records) min_tokens = min(token_counts) max_tokens = max(token_counts) print(f"\n{'='*60}") print(f"Dataset Statistics: {filepath}") print(f"{'='*60}") print(f"Format: {format_type}") print(f"Total records: {len(records)}") print(f"Parse errors: {parse_errors}") print(f"") print(f"Token Estimates (approx):") print(f" Total: {total_tokens:,}") print(f" Average/record: {avg_tokens:,.0f}") print(f" Min: {min_tokens:,}") print(f" Max: {max_tokens:,}") if format_type == "SFT": role_counts = Counter() for r in records: for msg in r.get("messages", []): role_counts[msg.get("role", "unknown")] += 1 print(f"\nRole Distribution:") for role, count in role_counts.most_common(): print(f" {role}: {count}") has_system = sum(1 for r in records if any(m.get("role") == "system" for m in r.get("messages", []))) print(f"\nRecords with system message: {has_system}/{len(records)}") elif format_type == "DPO": pref_lens = [] non_pref_lens = [] for r in records: pref_text = " ".join(m.get("content", "") for m in r.get("preferred_output", [])) non_pref_text = " ".join(m.get("content", "") for m in r.get("non_preferred_output", [])) pref_lens.append(estimate_tokens(pref_text)) non_pref_lens.append(estimate_tokens(non_pref_text)) print(f"\nPreferred output avg tokens: {sum(pref_lens)/len(pref_lens):,.0f}") print(f"Non-preferred output avg tokens: {sum(non_pref_lens)/len(non_pref_lens):,.0f}") elif format_type == "RFT": grader_field_counts = Counter() grader_values = [] for r in records: extra = set(r.keys()) - {"messages"} grader_field_counts.update(extra) for field in sorted(extra): grader_values.append(str(r[field])) unique = len(set(grader_values)) avg_val_len = sum(len(v) for v in grader_values) / len(grader_values) if grader_values else 0 print(f"\nGrader fields found:") for field, count in grader_field_counts.most_common(): print(f" • '{field}' — in {count}/{len(records)} records") print(f"Unique grader values: {unique}/{len(grader_values)}") print(f"Avg grader value length: {avg_val_len:.0f} chars") # Dataset size guidance print(f"\n📊 Dataset size guidance:") if len(records) < 50: print(f" ⚠️ Very small dataset ({len(records)} records). May only learn format, not domain knowledge.") elif len(records) < 200: print(f" ⚠️ Small dataset. Good for initial experiments — evaluate results and add more data if needed.") elif len(records) <= 500: print(f" ✅ Sweet spot for getting started (200-500). Evaluate results to decide if you need more.") elif len(records) <= 2000: print(f" ✅ Good dataset size. Watch for diminishing returns — check if quality beats quantity.") else: print(f" ⚠️ Large dataset ({len(records):,}). Larger isn't always better — especially for OSS models where 335-500 examples outperformed 4K.") if __name__ == "__main__": if len(sys.argv) != 2: print("Usage: python data_stats.py ") sys.exit(1) data_stats(sys.argv[1]) ``` ### `scripts/validate/validate_dpo.py` ``` #!/usr/bin/env python3 """Validate DPO (Direct Preference Optimization) JSONL files for Azure AI Foundry. Adapted from foundry-ft agent with additional checks: - Identical preferred/non_preferred detection - DPO overtraining risk (small dataset warning) """ import json import sys try: sys.stdout.reconfigure(encoding="utf-8") sys.stderr.reconfigure(encoding="utf-8") except (AttributeError, OSError): pass # Stream not reconfigurable (older Python or non-tty); default encoding is fine def validate_dpo(filepath: str) -> None: errors = [] warnings = [] total = 0 with open(filepath, "r", encoding="utf-8") as f: for line_num, line in enumerate(f, 1): line = line.strip() if not line: continue total += 1 try: record = json.loads(line) except json.JSONDecodeError as e: errors.append(f"Line {line_num}: Invalid JSON — {e}") continue for field in ["input", "preferred_output", "non_preferred_output"]: if field not in record: errors.append(f"Line {line_num}: Missing '{field}' field") if "input" not in record: continue inp = record["input"] if "messages" not in inp: errors.append(f"Line {line_num}: 'input' missing 'messages' field") else: msgs = inp["messages"] if not any(m.get("role") == "user" for m in msgs): errors.append(f"Line {line_num}: 'input.messages' has no 'user' message") for output_field in ["preferred_output", "non_preferred_output"]: if output_field in record: out = record[output_field] if not isinstance(out, list) or len(out) == 0: errors.append(f"Line {line_num}: '{output_field}' must be a non-empty array") elif not any(m.get("role") == "assistant" for m in out): errors.append(f"Line {line_num}: '{output_field}' has no 'assistant' message") if "preferred_output" in record and "non_preferred_output" in record: pref = json.dumps(record["preferred_output"], sort_keys=True) non_pref = json.dumps(record["non_preferred_output"], sort_keys=True) if pref == non_pref: warnings.append(f"Line {line_num}: preferred and non_preferred outputs are identical") print(f"\n{'='*60}") print(f"DPO Validation Report: {filepath}") print(f"{'='*60}") print(f"Total records: {total}") print(f"Errors: {len(errors)}") print(f"Warnings: {len(warnings)}") # DPO-specific guidance from our experiments if total < 500 and total > 0: print(f"\n⚠️ DPO tip: With {total} pairs, use n_epochs=1-2 max (Azure defaults to 3, which causes overtraining on small datasets).") if total > 0: print(f"\n💡 DPO tip: If your base model already scores >9/10 on this task, DPO may hurt more than help.") if errors: print(f"\n❌ ERRORS (must fix):") for e in errors[:20]: print(f" • {e}") if len(errors) > 20: print(f" ... and {len(errors) - 20} more errors") if warnings: print(f"\n⚠️ WARNINGS:") for w in warnings[:10]: print(f" • {w}") if not errors: print(f"\n✅ Data is valid for DPO fine-tuning!") else: print(f"\n❌ Fix {len(errors)} error(s) before submitting.") sys.exit(1) if __name__ == "__main__": if len(sys.argv) != 2: print("Usage: python validate_dpo.py ") sys.exit(1) validate_dpo(sys.argv[1]) ``` ### `scripts/validate/validate_rft.py` ``` #!/usr/bin/env python3 """Validate RFT (Reinforcement Fine-Tuning) JSONL files for Azure AI Foundry. Adapted from foundry-ft agent with critical additions from our platform gotchas: - Grader escaping warnings for newlines (\\n must be \\\\n in JSON strings) - Content moderation risk detection ("chain of thought" triggers RAI filter) - Reference answer diversity check """ import argparse import json import sys try: sys.stdout.reconfigure(encoding="utf-8") sys.stderr.reconfigure(encoding="utf-8") except (AttributeError, OSError): pass # Stream not reconfigurable (older Python or non-tty); default encoding is fine from collections import Counter RISKY_PHRASES = [ "chain of thought", "step by step reasoning", "let me think", "think carefully", "reason through", ] def validate_rft(filepath, expected_field=None): errors = [] warnings = [] total = 0 extra_fields_per_line: list[set[str]] = [] all_extra_field_counts: Counter = Counter() grader_values: list[str] = [] with open(filepath, "r", encoding="utf-8") as f: for line_num, line in enumerate(f, 1): raw_line = line line = line.strip() if not line: continue total += 1 try: record = json.loads(line) except json.JSONDecodeError as e: errors.append(f"Line {line_num}: Invalid JSON — {e}") continue if "messages" not in record: errors.append(f"Line {line_num}: Missing 'messages' field") else: msgs = record["messages"] if not isinstance(msgs, list) or len(msgs) == 0: errors.append(f"Line {line_num}: 'messages' must be a non-empty array") elif not any(m.get("role") == "user" for m in msgs): errors.append(f"Line {line_num}: 'messages' has no 'user' message") elif msgs[-1].get("role") != "user": errors.append( f"Line {line_num}: Last message must be 'user' role for RFT " f"(found '{msgs[-1].get('role')}') — unlike SFT, the model generates its own response" ) # Detect extra fields (grader fields) beyond 'messages' extra_fields = set(record.keys()) - {"messages"} extra_fields_per_line.append(extra_fields) all_extra_field_counts.update(extra_fields) if expected_field: if expected_field not in record: errors.append(f"Line {line_num}: Missing expected field '{expected_field}'") else: val = str(record[expected_field]).strip() if not val: errors.append(f"Line {line_num}: '{expected_field}' is empty") else: grader_values.append(val) else: if not extra_fields: errors.append( f"Line {line_num}: No grader fields found — RFT requires at least " "one field beyond 'messages' (e.g. 'answer', 'reference_code')" ) else: # Collect values from extra fields for diversity check for field in sorted(extra_fields): val = str(record[field]).strip() if val: grader_values.append(val) # Check for unescaped newlines in extra fields (CRITICAL platform gotcha) # Instead of regex-parsing the raw JSON line (which risks catastrophic # backtracking), we compare the parsed value against the raw line to # detect single-escaped \n that should be double-escaped \\n. for field in extra_fields: parsed_val = str(record.get(field, "")) if "\n" in parsed_val: # The parsed value contains actual newlines — check if the raw # JSON has them properly double-escaped field_needle = f'"{field}"' if field_needle in raw_line: field_start = raw_line.index(field_needle) field_region = raw_line[field_start:field_start + 500] # Single-escaped \n in raw JSON (not \\n) means the source # code newlines aren't properly escaped for the platform if "\\n" in field_region and "\\\\n" not in field_region: warnings.append( f"Line {line_num}: '{field}' contains \\n sequences — " "if this is grader source code embedded in JSON, " "ensure newlines are escaped as \\\\n." ) # Content moderation risk all_text = json.dumps(record).lower() for phrase in RISKY_PHRASES: if phrase in all_text: warnings.append( f"Line {line_num}: Contains '{phrase}' — may trigger Azure content moderation filter." ) break # Check for inconsistent extra-field schemas across examples field_sets = [fs for fs in extra_fields_per_line if fs] if len(field_sets) > 1: first_schema = field_sets[0] inconsistent_lines = [ i + 1 for i, fs in enumerate(extra_fields_per_line) if fs and fs != first_schema ] if inconsistent_lines: warnings.append( f"Inconsistent grader fields across examples — " f"line 1 has {sorted(first_schema)}, but {len(inconsistent_lines)} " f"line(s) differ (e.g. line {inconsistent_lines[0]}). " "Ensure your grader handles all field variants." ) # Diversity check if grader_values: unique_values = set(grader_values) if len(unique_values) == 1: warnings.append( f"All grader field values are identical ('{list(unique_values)[0][:50]}...') — " "grader may not learn effectively" ) avg_len = sum(len(v) for v in grader_values) / len(grader_values) if avg_len > 500: warnings.append( f"Average grader field value length is {avg_len:.0f} chars — " "consider using a model_grader instead of string_check" ) print(f"\n{'='*60}") print(f"RFT Validation Report: {filepath}") print(f"{'='*60}") print(f"Total records: {total}") print(f"Errors: {len(errors)}") print(f"Warnings: {len(warnings)}") if all_extra_field_counts: print(f"\nGrader fields found:") for field, count in all_extra_field_counts.most_common(): print(f" • '{field}' — in {count}/{total} records") if errors: print(f"\n❌ ERRORS (must fix):") for e in errors[:20]: print(f" • {e}") if len(errors) > 20: print(f" ... and {len(errors) - 20} more errors") if warnings: print(f"\n⚠️ WARNINGS:") for w in warnings[:10]: print(f" • {w}") if len(warnings) > 10: print(f" ... and {len(warnings) - 10} more warnings") # RFT-specific guidance if total > 0: print(f"\n💡 RFT tips:") print(f" • Ensure your training grader matches your eval grader (alignment gotcha)") print(f" • Start with reasoning_effort='medium', pass_threshold=0.5") print(f" • RFT is primarily for o-series models (o4-mini). Check Azure docs for the latest supported model list.") if not errors: print(f"\n✅ Data is valid for RFT fine-tuning!") else: print(f"\n❌ Fix {len(errors)} error(s) before submitting.") sys.exit(1) if __name__ == "__main__": parser = argparse.ArgumentParser( description="Validate RFT (Reinforcement Fine-Tuning) JSONL files for Azure AI Foundry." ) parser.add_argument("filepath", help="Path to the JSONL file to validate") parser.add_argument( "--expected-field", default=None, help="Specific grader field name to require (e.g. 'answer'). " "If omitted, any extra field beyond 'messages' is accepted.", ) args = parser.parse_args() validate_rft(args.filepath, expected_field=args.expected_field) ``` ### `scripts/validate/validate_sft.py` ``` #!/usr/bin/env python3 """Validate SFT (Supervised Fine-Tuning) JSONL files for Azure AI Foundry. Adapted from foundry-ft agent with additional checks from our platform gotchas: - Token length warnings (4096 limit varies by model) - System prompt consistency check """ import json import sys try: sys.stdout.reconfigure(encoding="utf-8") sys.stderr.reconfigure(encoding="utf-8") except (AttributeError, OSError): pass # Stream not reconfigurable (older Python or non-tty); default encoding is fine VALID_ROLES = {"system", "user", "assistant", "tool"} def estimate_tokens(text: str) -> int: """Rough token estimate: ~4 chars per token for English text.""" return max(1, len(text) // 4) def validate_sft(filepath: str) -> None: errors = [] warnings = [] total = 0 token_counts = [] system_prompts = set() with open(filepath, "r", encoding="utf-8") as f: for line_num, line in enumerate(f, 1): line = line.strip() if not line: continue total += 1 try: record = json.loads(line) except json.JSONDecodeError as e: errors.append(f"Line {line_num}: Invalid JSON — {e}") continue if "messages" not in record: errors.append(f"Line {line_num}: Missing 'messages' field") continue messages = record["messages"] if not isinstance(messages, list) or len(messages) == 0: errors.append(f"Line {line_num}: 'messages' must be a non-empty array") continue roles_found = set() total_text = "" for i, msg in enumerate(messages): if "role" not in msg: errors.append(f"Line {line_num}, message {i}: Missing 'role'") elif msg["role"] not in VALID_ROLES: errors.append(f"Line {line_num}, message {i}: Invalid role '{msg['role']}' (expected: {VALID_ROLES})") else: roles_found.add(msg["role"]) if "content" not in msg and "tool_calls" not in msg: errors.append(f"Line {line_num}, message {i}: Missing 'content' (and no 'tool_calls')") elif "content" in msg and msg["content"] is not None: content = str(msg["content"]) if not content.strip(): warnings.append(f"Line {line_num}, message {i}: Empty content string") total_text += content if msg.get("role") == "system": system_prompts.add(content.strip()[:100]) if "user" not in roles_found: errors.append(f"Line {line_num}: No 'user' message found") if "assistant" not in roles_found: errors.append(f"Line {line_num}: No 'assistant' message found") tokens = estimate_tokens(total_text) token_counts.append(tokens) if tokens > 4096: warnings.append(f"Line {line_num}: ~{tokens} tokens (exceeds 4096 limit for most models)") # Report print(f"\n{'='*60}") print(f"SFT Validation Report: {filepath}") print(f"{'='*60}") print(f"Total records: {total}") print(f"Errors: {len(errors)}") print(f"Warnings: {len(warnings)}") if token_counts: avg_tok = sum(token_counts) / len(token_counts) print(f"\nToken stats (approx):") print(f" Avg: {avg_tok:.0f} Min: {min(token_counts)} Max: {max(token_counts)}") print(f" Total: {sum(token_counts):,}") if len(system_prompts) > 1: warnings.append(f"Found {len(system_prompts)} different system prompts — ensure this is intentional") if system_prompts: print(f"\nSystem prompts: {len(system_prompts)} unique") if errors: print(f"\n❌ ERRORS (must fix):") for e in errors[:20]: print(f" • {e}") if len(errors) > 20: print(f" ... and {len(errors) - 20} more errors") if warnings: print(f"\n⚠️ WARNINGS:") for w in warnings[:10]: print(f" • {w}") if not errors: print(f"\n✅ Data is valid for SFT fine-tuning!") else: print(f"\n❌ Fix {len(errors)} error(s) before submitting.") sys.exit(1) if __name__ == "__main__": if len(sys.argv) != 2: print("Usage: python validate_sft.py ") sys.exit(1) validate_sft(sys.argv[1]) ``` ### `workflows/dataset-creation.md` ````markdown # Dataset Creation Workflow Three paths to training data (these combine well: curate seeds → augment → generate at scale): > If you already have data, skip to validation: `python scripts/validate/validate_sft.py your_data.jsonl` ## Approach 1: Manual Curation Write examples by hand, collect from production logs, or adapt existing datasets. **When to use:** - You have real-world examples (production logs, support tickets, labeled data) - Your task requires domain expertise an LLM can't reliably generate - You need a gold-standard evaluation set (always curate manually) **Tips:** - Start with 10-20 examples to establish quality standards and format consistency - These seed examples also serve as the foundation of your evaluation test set - For RFT, you only need prompts + expected answers — no model responses needed ## Approach 2: LLM Augmentation Expand a small curated dataset through **rephrasing** — generating diverse variations while keeping the same expected answer. Especially useful for RFT. **When to use:** - Well-defined task with clear correct answers - You can write quality examples but need more volume - Diversity of phrasing matters more than diversity of scenarios **Workflow:** 1. Write base examples with correct expected answers 2. For each, use an LLM to generate rephrasings varying tone, detail, and wording 3. Each rephrasing gets the same expected answer — only the phrasing changes 4. Validate the augmented dataset **Rephrasing prompt:** ``` Generate N different phrasings of this request. Each should: - Use different wording, tone, or level of detail - Include the same key identifiers (order IDs, item names) - Vary between formal, casual, frustrated, brief, and detailed styles Return a JSON array of N strings. Original: [your example] ``` A cheap model (gpt-4.1-mini) works well — no new ground truth needed, just phrasing diversity. ## Approach 3: Synthetic Generation Generate training data from scratch using LLM prompts. 1. Define topic/scenario categories for diversity 2. Generate prompts from an LLM 3. Generate responses (or preferred/non-preferred pairs for DPO) 4. Grade quality with an LLM judge 5. Filter to a quality threshold 6. Split into train/validation/test sets 7. Write JSONL in the correct format (see `references/dataset-formats.md`) ## Quality Checklist Before training, verify: - [ ] **No duplicates**: Exact or near-duplicate examples waste budget - [ ] **Balanced distribution**: Topics, difficulty, output lengths well-distributed - [ ] **Consistent formatting**: All examples follow the same structure - [ ] **Correct outputs**: Spot-check 20 random examples manually - [ ] **Reasonable lengths**: No extremely short or extremely long outputs - [ ] **Clean text**: No encoding errors, garbled text, or template artifacts ## Dataset Size vs. Quality From experiments: - **335 high-quality examples** (carefully curated) → best combined eval score (9.15) - **1,576 examples** (broader but noisier) → higher correctness but lower conciseness (8.53) **Takeaway**: A small, pristine dataset usually beats a large, noisy one. Quality filter aggressively. ```` ### `workflows/diagnose-poor-results.md` ```markdown # Diagnosing Poor Results When your fine-tuned model performs worse than expected, work through this checklist top-down (most common causes first). ## Diagnostic Table | # | Symptom | Likely Cause | Fix | |---|---------|-------------|-----| | 1 | Training loss → 0, validation loss rises | Overfitting | 1) Deploy earlier checkpoint. 2) Reduce epochs. 3) Lower LR. 4) Add more diverse data. Overfitting ratio > 1.5 is concerning. | | 2 | High correctness, low conciseness (or reverse) | Dataset style mismatch | **Verbose**: Add concise examples, use "Be concise" system prompt, filter to shortest correct examples. **Terse**: Add detailed examples, increase dataset with quality-filtered data. | | 3 | Model seems good on spot-check but auto-eval is low | Evaluation rubric issue | Manually grade 10 examples vs. LLM judge. Check: Is judge model strong enough? Is rubric clear? Do reference answers match desired output? | | 4 | Garbage, empty outputs, or errors | Deployment/client bug | Check: wrong model format (→ HTTP 500), `AzureOpenAI` on project endpoint (→ "api-version not allowed"), low capacity (→ timeouts), wrong deployment name. Test with curl. | | 5 | RFT model scores below base model | RFT-specific issue | See RFT section below. | ## RFT-Specific Diagnosis | Signal | Meaning | Fix | |--------|---------|-----| | Train-val grader gap > 0.2 | Model gaming the grader | Use stricter/more deterministic grader (Python execution > LLM judge) | | Grader too easy | High grader scores but bad outputs | Add multi-criteria grading (syntax + semantic) | | Grader too noisy | Random signal, no learning | Use deterministic grader or increase val set size | | All of the above fail | RFT may not suit this task | Switch back to SFT | ## Escalation Path If nothing above helps: 1. **Try a different base model** — some fine-tune better for certain tasks 2. **Increase dataset 2x-5x** with synthetic data 3. **Simplify the task** — fine-tune for a narrower sub-task first 4. **Try prompt engineering instead** — sometimes a well-crafted system prompt beats fine-tuning 5. **Combine approaches** — prompt engineering + fine-tuning together ## Red Flags: Don't Fine-Tune - Base model already scores > 9.0 (minimal headroom) - Task changes frequently (constant retraining needed) - < 50 examples and can't generate synthetic data - "Correct" output is highly subjective ``` ### `workflows/full-pipeline.md` ````markdown # Full Pipeline Workflow End-to-end fine-tuning on Azure AI Foundry in 9 phases. ## Prerequisites - Azure AI Foundry resource with fine-tuning enabled - Python 3.10+ with `openai` and `requests` - Azure CLI (`az`) authenticated - A clear task definition: what should the model do differently after fine-tuning? ## Phase 1: Define the Task Answer before touching data or models: 1. **What task?** (e.g., "translate natural language to Python code") 2. **What does good output look like?** Write 5 examples by hand. 3. **What does bad output look like?** Write 3 anti-examples. 4. **How will you measure success?** Define evaluation dimensions (see `references/grader-design.md`). 5. **Which base model?** Pick 1-3 candidates from the supported model list. ## Phase 2: Prepare the Dataset ### Option A: You Have Data 1. Convert to SFT JSONL format (see `references/dataset-formats.md`) 2. Split: 80% train, 10% validation, 10% held-out test 3. Remove or fix low-quality examples ### Option B: Synthetic Data 1. Generate using LLM prompts (see `workflows/dataset-creation.md`) 2. Convert to SFT JSONL with `scripts/convert_dataset.py` ### Option C: Hybrid (Seed + Synthetic) 1. Use existing data as seed, generate synthetic variations 2. Merge, deduplicate, and quality-filter **Checkpoint**: You should have `training.jsonl`, `validation.jsonl`, and `test.jsonl` (never used for training). ## Phase 3: Establish Baselines 1. Deploy base model (or use existing deployment) 2. Record scores — this is your "zero" that every fine-tune must beat ## Phase 4: Choose Training Type See `references/training-types.md` for the full decision framework. | Condition | Training Type | |-----------|--------------| | Have input-output pairs | SFT | | Can write a grading function | RFT (reasoning models only) | | Need style alignment | DPO | Most projects start with SFT. Move to RFT/DPO only if SFT isn't sufficient. ## Phase 5: Upload and Submit Training Use `scripts/submit_training.py` or the API directly. See `references/hyperparameters.md` for starting HP values. **Foundry CLI** alternative (no Python): ```bash azd ai finetuning jobs submit -f ./fine-tune-job.yaml ``` ## Phase 6: Monitor and Analyze 1. Wait for completion or use `scripts/monitor_training.py` 2. Analyze training curves with `scripts/check_training.py` 3. Read `references/training-curves.md` to interpret results 4. Check for overfitting — consider deploying an earlier checkpoint if detected ## Phase 7: Evaluate Fine-Tuned Model 1. Deploy fine-tuned model (see `references/deployment.md` for format/SKU) 2. Compare against baseline and previous experiments 3. Delete deployment after evaluation ## Phase 8: Iterate Follow `workflows/iterative-training.md`: - Adjust hyperparameters based on training curves - Try different data subsets or augmentations - Test different base models - Track everything in your leaderboard ## Phase 9: Ship When the model convincingly beats baseline: 1. Deploy with production-appropriate capacity 2. Monitor with Application Insights 3. Periodically re-evaluate against test set for regression 4. Retrain as new data becomes available ```` ### `workflows/iterative-training.md` ````markdown # Iterative Training Workflow Systematically improve a fine-tuned model through successive experiments. ## The Core Loop ``` 1. Train with current config 2. Analyze training curves 3. Evaluate on held-out set 4. Diagnose what to change 5. Plan next experiment → Better than baseline? → Good enough? → Ship it (or loop back to 4) ``` **Rule**: Change ONE variable per experiment. ## Experiment Tracking | Run | Base model | Dataset | Epochs | LR | Batch | Best val_loss | Combined eval | |-----|-----------|---------|--------|-----|-------|--------------|---------------| | R1 | gpt-4.1-mini | v1 (335 ex) | 2 | 1.0 | default | 0.320 | 8.05 | | R2 | gpt-4.1-mini | v1 (335 ex) | 2 | 0.5 | default | 0.310 | 9.15 | | ... | ... | ... | ... | ... | ... | ... | ... | ## What to Try (Priority Order) ### Priority 1: Data Quality (highest leverage) - **Fix inconsistencies**: Contradicting examples confuse the model - **Add diversity**: Add examples for input types the model fails on - **Reduce noise**: Remove "correct but not ideal" outputs ### Priority 2: Hyperparameters See `references/hyperparameters.md` for full guide. **Quick sweep strategy:** 1. Baseline: epochs=2, lr=1.0 2. Overfitting → lr=0.5 or epochs=1 3. Underfitting → lr=1.5 or epochs=3 4. Good LR found → try batch_size=16 or 32 ### Priority 3: Base Model | Model | Best for | |-------|----------| | gpt-4.1-mini | Best quality-per-dollar, most tasks | | gpt-4.1-nano | Fastest inference, simple tasks | | gpt-oss-20b | Large datasets, lowest absolute loss | | Ministral-3B | Lightweight, fast inference | | Qwen-3-32B, Llama-3.3-70B | Multilingual or specialized tasks | ### Priority 4: Training Type - SFT plateaued + need better reasoning → RFT (if model supports it) - Need style alignment → DPO - See `references/training-types.md` before switching ## Diagnostic Decision Tree ``` Training curves healthy (no overfitting)? ├─ Yes │ ├─ Eval improved? → Refine further │ └─ Eval same/worse? → Data quality issue — filter or augment └─ No (overfitting) ├─ Earlier checkpoint evals well? → Deploy that checkpoint ├─ Not severe → Reduce epochs or lower LR └─ Severe (ratio > 2.0) ├─ Dataset too small → Add more data └─ Dataset large → Lower LR dramatically (0.1-0.3) ``` ## When to Stop 1. Beaten baseline by meaningful margin (>5%) and last 3 experiments didn't improve 2. Diminishing returns: each experiment improves < 0.1 points 3. Model is "good enough" for production 4. Budget exhausted (time or money) ## Multi-Model Strategy Run the same dataset through 2-3 base models: 1. **gpt-4.1-mini** — primary candidate 2. **gpt-oss-20b** — large-dataset specialist (500+ examples) 3. **gpt-4.1-nano** — fast inference option ## Common Mistakes 1. Not establishing a baseline first 2. Changing multiple variables at once 3. Overfitting to the eval set (keep a separate final test set) 4. Ignoring training curves (they tell you what to change next) 5. More data without quality check (lower-quality data often makes things worse) 6. Not cleaning up old deployments (wastes quota and money) ```` ### `workflows/quickstart.md` ````markdown # Quickstart: Fine-Tune Your First Model 6 steps from zero to a fine-tuned model using SFT with synthetic data. > **Time**: ~20 min active + 1-3 hours training. ## Prerequisites - Azure AI Foundry project with a deployed model (e.g., `gpt-4.1-mini`) - Python 3.10+ with `openai` installed - Project endpoint URL and API key (Foundry portal → Project Settings) ## Step 1: Connect to Your Project ```bash export OPENAI_BASE_URL="https://.services.ai.azure.com/api/projects//openai/v1/" export AZURE_OPENAI_API_KEY="" ``` ```python from openai import OpenAI import os client = OpenAI(base_url=os.environ["OPENAI_BASE_URL"], api_key=os.environ["AZURE_OPENAI_API_KEY"]) resp = client.chat.completions.create(model="gpt-4.1-mini", messages=[{"role": "user", "content": "Hello"}], max_tokens=10) print(resp.choices[0].message.content) ``` ## Step 2: Generate Training Data ```python import json, re SYSTEM_PROMPT = "You are a concise technical support agent. Answer in 1-2 sentences." generation_prompt = """Generate 50 diverse technical support conversations. Each should have a customer question and an ideal agent response (1-2 sentences). Cover: password resets, billing, product setup, account changes, shipping, troubleshooting. Return a JSON array where each element has "question" and "answer" fields.""" resp = client.chat.completions.create( model="gpt-4.1-mini", messages=[{"role": "user", "content": generation_prompt}], max_tokens=8000, temperature=1.0, ) content = resp.choices[0].message.content match = re.search(r'```(?:json)?\s*\n(.*?)\n```', content, re.DOTALL) json_str = match.group(1) if match else content.strip().strip("`").replace("json\n", "") examples = json.loads(json_str) for split, name, rng in [("train", "train.jsonl", examples[:40]), ("val", "val.jsonl", examples[40:])]: with open(name, "w") as f: for ex in rng: f.write(json.dumps({"messages": [ {"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": ex["question"]}, {"role": "assistant", "content": ex["answer"]}, ]}) + "\n") ``` Validate: `python scripts/validate/validate_sft.py train.jsonl` ## Step 3: Baseline the Base Model ```python with open("val.jsonl") as f: test_examples = [json.loads(line) for line in f][:5] for ex in test_examples: resp = client.chat.completions.create( model="gpt-4.1-mini", messages=ex["messages"][:2], max_tokens=200) print(f"Q: {ex['messages'][1]['content']}") print(f"Expected: {ex['messages'][2]['content']}") print(f"Base model: {resp.choices[0].message.content}\n") ``` ## Step 4: Upload Data and Submit Job ```python import time with open("train.jsonl", "rb") as f: train = client.files.create(file=f, purpose="fine-tune") with open("val.jsonl", "rb") as f: val = client.files.create(file=f, purpose="fine-tune") for _ in range(30): if client.files.retrieve(train.id).status == "processed" and client.files.retrieve(val.id).status == "processed": break time.sleep(10) job = client.fine_tuning.jobs.create( model="gpt-4.1-mini", training_file=train.id, validation_file=val.id, suffix="my-first-ft", method={"type": "supervised"}, hyperparameters={"n_epochs": 2, "learning_rate_multiplier": 1.0}, ) print(f"Job submitted: {job.id}") ``` Or via script: ```bash python scripts/submit_training.py --model gpt-4.1-mini --training-file train.jsonl --validation-file val.jsonl --type sft --suffix my-first-ft --epochs 2 ``` ## Step 5: Monitor ```bash python scripts/monitor_training.py --job-id ``` Or check [Azure AI Foundry portal](https://ai.azure.com) → Fine-tuning → Jobs. ## Step 6: Deploy, Test, and Compare ```bash python scripts/deploy_model.py --model-id --name my-ft-deployment --capacity 50 ``` ```python for ex in test_examples: base = client.chat.completions.create(model="gpt-4.1-mini", messages=ex["messages"][:2], max_tokens=200) ft = client.chat.completions.create(model="my-ft-deployment", messages=ex["messages"][:2], max_tokens=200) print(f"Q: {ex['messages'][1]['content']}") print(f"Base: {base.choices[0].message.content}") print(f"Fine-tuned: {ft.choices[0].message.content}\n") ``` ## What's Next - **Scale data**: 200-500 examples → `workflows/dataset-creation.md` - **Try RFT**: For verifiable answers → `references/training-types.md` - **Debug**: `workflows/diagnose-poor-results.md` - **Full guide**: `workflows/full-pipeline.md` ````