Le pipeline analyse des tickets de support en français et produit un objet JSON contenant l’intention, la file de routage, la priorité, le besoin de revue humaine, une réponse suggérée et la langue. Le format est volontairement contraint afin de rendre l’évaluation reproductible.
Bitext Customer Support (anglais, licence à vérifier)
│
▼
Traduction EN → FR avec Qwen + contrôles qualité
│
▼
Données acceptées / rejetées + métadonnées de traçabilité
│
▼
Transformation en tâches de support avec sortie JSON stricte
│
▼
Splits déterministes : entraînement / validation / test
│
▼
SFT QLoRA avec Unsloth et Qwen2.5
│
├── comparaison modèle de base vs adaptateur LoRA
▼
Métriques automatiques + revue humaine + API FastAPI locale
Le corpus Bitext en anglais est traduit vers le français. Les scripts conservent les métadonnées d’origine, l’identifiant source, le modèle de traduction et la version de prompt afin de documenter la provenance. Les contrôles cherchent les traductions vides, les longueurs anormales, les résidus anglais et la perte d’éléments à préserver : URL, e-mails, variables et identifiants.
La licence du corpus source doit être vérifiée avant toute publication d’un corpus dérivé ou d’un adaptateur. Les traductions automatiques ne sont pas présentées comme des annotations humaines.
{
"instruction": "Analyse cette demande de support client et réponds uniquement avec un objet JSON valide respectant le schéma attendu.",
"input": "J'ai été facturé deux fois pour mon abonnement.",
"output": {
"intent": "payment_issue", "queue": "billing", "priority": "high",
"needs_human_review": true, "suggested_reply": "...", "language": "fr"
}
}
Le projet utilise QLoRA : le modèle de base est chargé en 4 bits et seuls des adaptateurs LoRA sont entraînés. Pour un GPU NVIDIA de 8 Go, le profil de départ est Qwen2.5-1.5B-Instruct en 4 bits, batch size 1, gradient accumulation de 16 et contexte de 1 024 tokens. Un profil Qwen2.5-3B expérimental est disponible seulement après un entraînement 1.5B stable ; en cas de saturation mémoire, le contexte doit être réduit à 768 puis 512 tokens.
Le modèle de base et l’adaptateur LoRA doivent être comparés sur un même test figé, avec le même prompt et la même génération déterministe. Les métriques : validité JSON, clés exactes, intention, file, priorité et escalade. Un CSV permet une revue humaine de la réponse suggérée. Un second test manuel de 50 à 100 tickets inédits en français doit couvrir les fautes, demandes multiples, cas sensibles et hors périmètre.
├── README.md ├── configs/support_training_config.yaml ├── configs/training_config.yaml ├── docs/colab.md ├── docs/description_cv.md ├── docs/end_to_end.md ├── docs/translation_pipeline.md ├── requirements-api.txt ├── requirements-translation.txt ├── requirements.txt ├── src/evaluate.py ├── src/evaluate_support.py ├── src/infer.py ├── src/prepare_support_dataset.py ├── src/serve_api.py ├── src/train.py ├── src/translate_bitext_to_french.py ├── src/validate_dataset.py
# 1. Environnement officiel Unsloth : Python 3.13 et uv # Windows : installer Python 3.13 et uv, puis : uv venv .venv --python 3.13 # Windows : .venv\Scripts\Activate.ps1 # Linux : source .venv/bin/activate # 2. Installer Unsloth avec le build PyTorch/CUDA adapté uv pip install unsloth --torch-backend=auto uv pip install -r requirements.txt python scripts_verify_environment.py # 3. Traduction pilote et revue manuelle uv pip install -r requirements-translation.txt python src/translate_bitext_to_french.py --limit 50 --output-dir data/bitext_fr_sample # 4. Traduction complète, après validation python src/translate_bitext_to_french.py --output-dir data/bitext_fr --batch-size 4 # 5. Dataset structuré + validation python src/prepare_support_dataset.py --input-dir data/bitext_fr --output-dir data/support_fr_structured python src/validate_dataset.py --data-dir data/support_fr_structured # 6. Fine-tuning CUDA — profil Qwen2.5-1.5B, 8 Go de VRAM python src/train.py --config configs/support_training_config.yaml # Profil 3B expérimental seulement après succès du 1.5B # python src/train.py --config configs/support_training_3b_8gb_experimental.yaml # 7. Évaluation et démonstration python src/evaluate_support.py --adapter outputs/qwen25-3b-support-fr-lora pip install -r requirements-api.txt python src/serve_api.py --adapter outputs/qwen25-3b-support-fr-lora
src/translate_bitext_to_french.py#!/usr/bin/env python3
"""Create a traceable French customer-support dataset from Bitext with Qwen.
The script translates only source text, preserves original fields, performs
basic quality checks and writes train/validation/test JSONL files. It does not
publish derivative data: review the Bitext CDLA-Sharing-1.0 license first.
"""
import argparse
import hashlib
import json
import random
import re
from pathlib import Path
from datasets import load_dataset
from transformers import AutoModelForCausalLM, AutoTokenizer, pipeline
import torch
SYSTEM = """You are a professional English-to-French translator for customer support.
Translate into natural, polite French using 'vous'. Preserve exactly URLs, email
addresses, values in {{double braces}}, values in {single braces}, order IDs,
currency amounts, product names and line breaks. Do not add, remove, summarize
or explain anything. Return only the French translation."""
PLACEHOLDER = re.compile(r"\{\{[^{}]+\}\}|\{[^{}]+\}")
URL = re.compile(r"https?://[^\s]+|www\.[^\s]+|[\w.+-]+@[\w.-]+\.[A-Za-z]{2,}")
ENGLISH_WORDS = re.compile(r"\b(the|and|with|your|please|order|refund|account|help|thank)\b", re.I)
def value(row, candidates):
for key in candidates:
if key in row and row[key] is not None:
return str(row[key]).strip()
return ""
def protected_tokens(text):
return set(PLACEHOLDER.findall(text)) | set(URL.findall(text))
def quality_reason(source, translated):
if not translated or len(translated) < max(3, len(source) * 0.18):
return "empty_or_too_short"
if len(translated) > max(5000, len(source) * 3.0):
return "too_long"
missing = protected_tokens(source) - protected_tokens(translated)
if missing:
return "missing_protected_tokens: " + ", ".join(sorted(missing))
# A small number of English words can be valid proper names; this is a review flag.
if len(ENGLISH_WORDS.findall(translated)) >= 4:
return "possible_english_residue"
return ""
def make_messages(text):
return [
{"role": "system", "content": SYSTEM},
{"role": "user", "content": text},
]
def translate(batch, generator, tokenizer, max_new_tokens):
prompts = [
tokenizer.apply_chat_template(make_messages(text), tokenize=False, add_generation_prompt=True)
for text in batch
]
results = generator(
prompts,
max_new_tokens=max_new_tokens,
do_sample=False,
temperature=None,
top_p=None,
return_full_text=False,
batch_size=len(prompts),
)
return [item[0]["generated_text"].strip() for item in results]
def split_name(identifier, validation_ratio, test_ratio):
bucket = int(hashlib.sha256(identifier.encode("utf-8")).hexdigest()[:8], 16) % 10000
if bucket < int(test_ratio * 10000):
return "test"
if bucket < int((test_ratio + validation_ratio) * 10000):
return "validation"
return "train"
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--model", default="Qwen/Qwen2.5-7B-Instruct")
parser.add_argument("--dataset", default="bitext/Bitext-customer-support-llm-chatbot-training-dataset")
parser.add_argument("--dataset-config", default=None)
parser.add_argument("--split", default="train")
parser.add_argument("--output-dir", default="data/bitext_fr")
parser.add_argument("--limit", type=int, default=0, help="0 = all examples")
parser.add_argument("--batch-size", type=int, default=4)
parser.add_argument("--max-new-tokens", type=int, default=384)
parser.add_argument("--validation-ratio", type=float, default=0.1)
parser.add_argument("--test-ratio", type=float, default=0.1)
parser.add_argument("--device-map", default="auto")
args = parser.parse_args()
if not 0 < args.validation_ratio < 1 or not 0 < args.test_ratio < 1 or args.validation_ratio + args.test_ratio >= 1:
raise ValueError("Les ratios validation/test doivent être positifs et leur somme inférieure à 1.")
output_dir = Path(args.output_dir)
output_dir.mkdir(parents=True, exist_ok=True)
rejected_path = output_dir / "rejected.jsonl"
files = {name: (output_dir / f"{name}.jsonl").open("w", encoding="utf-8") for name in ("train", "validation", "test")}
rejected = rejected_path.open("w", encoding="utf-8")
print(f"Chargement du dataset : {args.dataset}")
ds = load_dataset(args.dataset, args.dataset_config, split=args.split)
if args.limit:
ds = ds.select(range(min(args.limit, len(ds))))
print(f"Colonnes détectées : {ds.column_names}; exemples à traiter : {len(ds)}")
tokenizer = AutoTokenizer.from_pretrained(args.model)
model = AutoModelForCausalLM.from_pretrained(
args.model,
torch_dtype="auto",
device_map=args.device_map,
)
generator = pipeline("text-generation", model=model, tokenizer=tokenizer)
rows = []
for index, row in enumerate(ds):
instruction = value(row, ["instruction", "instruction_en", "question", "user", "prompt"])
response = value(row, ["response", "response_en", "answer", "assistant", "completion"])
category = value(row, ["category", "intent", "topic"])
if not instruction or not response:
rejected.write(json.dumps({"source_index": index, "reason": "missing_instruction_or_response", "raw": row}, ensure_ascii=False) + "\n")
continue
rows.append({"source_index": index, "instruction_en": instruction, "response_en": response, "category": category})
accepted, rejected_count = 0, 0
for start in range(0, len(rows), args.batch_size):
chunk = rows[start : start + args.batch_size]
source_texts = []
for item in chunk:
source_texts.extend([item["instruction_en"], item["response_en"]])
translations = translate(source_texts, generator, tokenizer, args.max_new_tokens)
for pos, item in enumerate(chunk):
instruction_fr, response_fr = translations[2 * pos], translations[2 * pos + 1]
issue = quality_reason(item["instruction_en"], instruction_fr) or quality_reason(item["response_en"], response_fr)
record_id = f"bitext_{item['source_index']:07d}"
record = {
"id": record_id,
"instruction": instruction_fr,
"input": "",
"output": response_fr,
"source": "bitext/Bitext-customer-support-llm-chatbot-training-dataset",
"source_index": item["source_index"],
"instruction_en": item["instruction_en"],
"response_en": item["response_en"],
"category": item["category"],
"translation_model": args.model,
"translation_prompt_version": "v1",
"quality_status": "accepted" if not issue else "rejected",
"quality_reason": issue,
}
if issue:
rejected.write(json.dumps(record, ensure_ascii=False) + "\n")
rejected_count += 1
else:
target = split_name(record_id, args.validation_ratio, args.test_ratio)
files[target].write(json.dumps(record, ensure_ascii=False) + "\n")
accepted += 1
print(f"{min(start + len(chunk), len(rows))}/{len(rows)} traités | acceptés={accepted} rejetés={rejected_count}")
for handle in files.values():
handle.close()
rejected.close()
manifest = {
"source_dataset": args.dataset,
"source_split": args.split,
"source_license": "CDLA-Sharing-1.0 (verify obligations before distribution)",
"translation_model": args.model,
"translation_prompt_version": "v1",
"accepted": accepted,
"rejected": rejected_count,
"validation_ratio": args.validation_ratio,
"test_ratio": args.test_ratio,
"notes": "Review a random sample manually before fine-tuning. Do not publish derivative data without checking source license obligations.",
}
(output_dir / "manifest.json").write_text(json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8")
print(f"Terminé. Fichiers créés dans : {output_dir.resolve()}")
if __name__ == "__main__":
main()
src/prepare_support_dataset.py#!/usr/bin/env python3
"""Convert accepted translated Bitext rows into structured support-agent SFT data.
Input must be the output of translate_bitext_to_french.py. The script derives a
stable intent/category mapping, assigns conservative routing labels, creates a
strict JSON response target, removes duplicates and writes reproducible splits.
"""
import argparse
import hashlib
import json
import re
from collections import Counter
from pathlib import Path
PLACEHOLDER = re.compile(r"\{\{[^{}]+\}\}|\{[^{}]+\}")
INTENT_ROUTING = {
"cancel_order": ("order_management", "medium", False),
"change_order": ("order_management", "medium", False),
"track_order": ("order_tracking", "low", False),
"edit_personal_details": ("account", "low", False),
"get_invoice": ("billing", "low", False),
"payment_issue": ("billing", "high", True),
"refund": ("refunds", "medium", False),
"complaint": ("customer_care", "high", True),
"contact_customer_service": ("customer_care", "medium", False),
}
def normalise_label(value):
value = (value or "unknown").strip().lower()
value = re.sub(r"[^a-z0-9]+", "_", value).strip("_")
return value or "unknown"
def stable_split(record_id, validation_ratio, test_ratio):
value = int(hashlib.sha256(record_id.encode()).hexdigest()[:8], 16) % 10000
if value < int(test_ratio * 10000):
return "test"
if value < int((test_ratio + validation_ratio) * 10000):
return "validation"
return "train"
def has_sensitive_marker(text):
lowered = text.lower()
return any(word in lowered for word in ("fraude", "fraud", "piraté", "hacked", "vol", "stolen", "urgent"))
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--input-dir", default="data/bitext_fr")
parser.add_argument("--output-dir", default="data/support_fr_structured")
parser.add_argument("--validation-ratio", type=float, default=0.1)
parser.add_argument("--test-ratio", type=float, default=0.1)
args = parser.parse_args()
if args.validation_ratio + args.test_ratio >= 1:
raise ValueError("La somme validation + test doit être inférieure à 1.")
input_dir, output_dir = Path(args.input_dir), Path(args.output_dir)
output_dir.mkdir(parents=True, exist_ok=True)
handles = {s: (output_dir / f"{s}.jsonl").open("w", encoding="utf-8") for s in ("train", "validation", "test")}
seen, counts, intents = set(), Counter(), Counter()
for source_file in (input_dir / "train.jsonl", input_dir / "validation.jsonl", input_dir / "test.jsonl"):
if not source_file.exists():
raise FileNotFoundError(f"Fichier introuvable : {source_file}")
for line in source_file.read_text(encoding="utf-8").splitlines():
if not line.strip():
continue
item = json.loads(line)
customer = item.get("instruction", "").strip()
reply = item.get("output", "").strip()
if not customer or not reply or item.get("quality_status") != "accepted":
continue
key = " ".join(customer.lower().split())
if key in seen:
continue
seen.add(key)
intent = normalise_label(item.get("category"))
queue, priority, review = INTENT_ROUTING.get(intent, ("general_support", "medium", False))
if has_sensitive_marker(customer):
priority, review = "high", True
record_id = item.get("id") or hashlib.sha256(key.encode()).hexdigest()[:16]
target = {
"intent": intent,
"queue": queue,
"priority": priority,
"needs_human_review": review,
"suggested_reply": reply,
"language": "fr",
}
example = {
"id": record_id,
"instruction": "Analyse cette demande de support client et réponds uniquement avec un objet JSON valide respectant le schéma attendu.",
"input": customer,
"output": json.dumps(target, ensure_ascii=False, separators=(",", ":")),
"source": item.get("source"),
"source_index": item.get("source_index"),
"category_original": item.get("category"),
"translation_model": item.get("translation_model"),
"translation_prompt_version": item.get("translation_prompt_version"),
}
split = stable_split(record_id, args.validation_ratio, args.test_ratio)
handles[split].write(json.dumps(example, ensure_ascii=False) + "\n")
counts[split] += 1
intents[intent] += 1
for handle in handles.values():
handle.close()
report = {"counts": dict(counts), "intent_counts": dict(intents), "total": sum(counts.values())}
(output_dir / "dataset_report.json").write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")
print(json.dumps(report, ensure_ascii=False, indent=2))
if __name__ == "__main__":
main()
src/train.py#!/usr/bin/env python3
"""Supervised fine-tuning with Unsloth + QLoRA."""
import argparse
import os
from pathlib import Path
import yaml
from datasets import load_dataset
from unsloth import FastLanguageModel
from trl import SFTTrainer
from transformers import TrainingArguments
SYSTEM_PROMPT = (
"Tu es un assistant technique pédagogique spécialisé en RAG et fine-tuning de LLM. "
"Réponds en français, de manière factuelle, concise et structurée. "
"N'invente pas de faits lorsque l'information manque."
)
def load_config(path):
with open(path, encoding="utf-8") as handle:
return yaml.safe_load(handle)
def format_example(example, tokenizer):
messages = [{"role": "system", "content": SYSTEM_PROMPT}]
user = example["instruction"].strip()
if example.get("input", "").strip():
user += "\n\nContexte : " + example["input"].strip()
messages += [
{"role": "user", "content": user},
{"role": "assistant", "content": example["output"].strip()},
]
return {"text": tokenizer.apply_chat_template(messages, tokenize=False, add_generation_prompt=False)}
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--config", default="configs/training_config.yaml")
args = parser.parse_args()
cfg = load_config(args.config)
max_seq_length = int(cfg["max_seq_length"])
model, tokenizer = FastLanguageModel.from_pretrained(
model_name=cfg["model_name"],
max_seq_length=max_seq_length,
load_in_4bit=bool(cfg.get("load_in_4bit", True)),
)
model = FastLanguageModel.get_peft_model(
model,
r=int(cfg["lora_r"]),
target_modules=cfg["lora_target_modules"],
lora_alpha=int(cfg["lora_alpha"]),
lora_dropout=float(cfg["lora_dropout"]),
bias="none",
use_gradient_checkpointing="unsloth",
random_state=int(cfg["random_seed"]),
)
dataset = load_dataset(
"json",
data_files={"train": "data/train.jsonl", "validation": "data/validation.jsonl"},
)
dataset = dataset.map(lambda x: format_example(x, tokenizer), remove_columns=dataset["train"].column_names)
output_dir = cfg["output_dir"]
training_args = TrainingArguments(
output_dir=output_dir,
num_train_epochs=float(cfg["num_train_epochs"]),
learning_rate=float(cfg["learning_rate"]),
per_device_train_batch_size=int(cfg["per_device_train_batch_size"]),
per_device_eval_batch_size=int(cfg["per_device_eval_batch_size"]),
gradient_accumulation_steps=int(cfg["gradient_accumulation_steps"]),
warmup_ratio=float(cfg["warmup_ratio"]),
logging_steps=int(cfg["logging_steps"]),
eval_strategy="steps",
eval_steps=int(cfg["eval_steps"]),
save_strategy="steps",
save_steps=int(cfg["save_steps"]),
save_total_limit=2,
fp16=not os.environ.get("BF16", ""),
bf16=bool(os.environ.get("BF16", "")),
seed=int(cfg["random_seed"]),
report_to="none",
optim="adamw_8bit",
)
trainer = SFTTrainer(
model=model,
tokenizer=tokenizer,
train_dataset=dataset["train"],
eval_dataset=dataset["validation"],
dataset_text_field="text",
max_seq_length=max_seq_length,
args=training_args,
)
trainer.train()
trainer.save_model(output_dir)
tokenizer.save_pretrained(output_dir)
print(f"Adaptateur et tokenizer sauvegardés dans : {Path(output_dir).resolve()}")
if __name__ == "__main__":
main()
src/evaluate_support.py#!/usr/bin/env python3
"""Evaluate structured customer-support outputs from a LoRA adapter.
Produces a reviewable CSV and automatic metrics: JSON validity, exact intent,
queue, priority and human-review accuracy. This does not claim factual quality
of suggested replies; review those manually.
"""
import argparse
import csv
import json
from pathlib import Path
from unsloth import FastLanguageModel
SYSTEM = """Tu es un copilote de support client. Tu réponds uniquement avec un objet JSON valide.
Le JSON doit contenir exactement les clés : intent, queue, priority,
needs_human_review, suggested_reply, language. Utilise le français et n'invente
pas de données de compte ou de commande."""
EXPECTED = {"intent", "queue", "priority", "needs_human_review", "suggested_reply", "language"}
def generate(model, tokenizer, customer, max_new_tokens):
messages = [{"role": "system", "content": SYSTEM}, {"role": "user", "content": customer}]
inputs = tokenizer.apply_chat_template(messages, tokenize=True, add_generation_prompt=True, return_tensors="pt").to("cuda")
output = model.generate(input_ids=inputs, max_new_tokens=max_new_tokens, do_sample=False, use_cache=True)
return tokenizer.decode(output[0][inputs.shape[-1]:], skip_special_tokens=True).strip()
def parse_json(text):
start, end = text.find("{"), text.rfind("}")
if start < 0 or end < start:
return None
try:
parsed = json.loads(text[start:end + 1])
return parsed if isinstance(parsed, dict) else None
except json.JSONDecodeError:
return None
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--adapter", required=True)
parser.add_argument("--test-file", default="data/support_fr_structured/test.jsonl")
parser.add_argument("--output", default="outputs/support_evaluation.csv")
parser.add_argument("--max-new-tokens", type=int, default=320)
args = parser.parse_args()
rows = [json.loads(x) for x in Path(args.test_file).read_text(encoding="utf-8").splitlines() if x.strip()]
model, tokenizer = FastLanguageModel.from_pretrained(model_name=args.adapter, max_seq_length=1024, load_in_4bit=True)
FastLanguageModel.for_inference(model)
output = Path(args.output); output.parent.mkdir(parents=True, exist_ok=True)
metrics = {"total": len(rows), "json_valid": 0, "exact_schema": 0, "intent_correct": 0, "queue_correct": 0, "priority_correct": 0, "review_correct": 0}
fields = ["id", "customer_message", "expected", "prediction_raw", "prediction_json", "json_valid", "intent_correct", "queue_correct", "priority_correct", "review_correct", "reply_quality_1_5", "notes"]
with output.open("w", newline="", encoding="utf-8") as fh:
writer = csv.DictWriter(fh, fieldnames=fields); writer.writeheader()
for i, item in enumerate(rows, 1):
expected = json.loads(item["output"])
raw = generate(model, tokenizer, item["input"], args.max_new_tokens)
predicted = parse_json(raw)
valid = predicted is not None
if valid:
metrics["json_valid"] += 1
if set(predicted) == EXPECTED: metrics["exact_schema"] += 1
for metric, key in (("intent_correct", "intent"), ("queue_correct", "queue"), ("priority_correct", "priority"), ("review_correct", "needs_human_review")):
if predicted.get(key) == expected.get(key): metrics[metric] += 1
writer.writerow({
"id": item.get("id"), "customer_message": item["input"], "expected": json.dumps(expected, ensure_ascii=False),
"prediction_raw": raw, "prediction_json": json.dumps(predicted, ensure_ascii=False) if predicted else "",
"json_valid": valid, "intent_correct": valid and predicted.get("intent") == expected.get("intent"),
"queue_correct": valid and predicted.get("queue") == expected.get("queue"),
"priority_correct": valid and predicted.get("priority") == expected.get("priority"),
"review_correct": valid and predicted.get("needs_human_review") == expected.get("needs_human_review"),
"reply_quality_1_5": "", "notes": "",
})
print(f"[{i}/{len(rows)}]")
summary = {k: (v / len(rows) if k != "total" and rows else v) for k, v in metrics.items()}
summary_path = output.with_suffix(".metrics.json")
summary_path.write_text(json.dumps({"counts": metrics, "rates": summary}, indent=2), encoding="utf-8")
print(json.dumps({"counts": metrics, "rates": summary}, indent=2))
if __name__ == "__main__":
main()
src/serve_api.py#!/usr/bin/env python3
"""Minimal FastAPI demo for the trained support-routing adapter."""
import argparse
import json
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from unsloth import FastLanguageModel
import uvicorn
SYSTEM = """Tu es un copilote de support client. Réponds uniquement par un objet JSON valide avec :
intent, queue, priority, needs_human_review, suggested_reply, language.
N'invente pas de données de compte, de commande ou de remboursement."""
app = FastAPI(title="Support Copilot", version="1.0")
model = tokenizer = None
class Ticket(BaseModel):
message: str
@app.post('/analyze')
def analyze(ticket: Ticket):
if not ticket.message.strip():
raise HTTPException(400, "Le message ne peut pas être vide.")
messages = [{"role": "system", "content": SYSTEM}, {"role": "user", "content": ticket.message}]
inputs = tokenizer.apply_chat_template(messages, tokenize=True, add_generation_prompt=True, return_tensors="pt").to("cuda")
output = model.generate(input_ids=inputs, max_new_tokens=320, do_sample=False, use_cache=True)
raw = tokenizer.decode(output[0][inputs.shape[-1]:], skip_special_tokens=True).strip()
try:
start, end = raw.find('{'), raw.rfind('}')
return json.loads(raw[start:end + 1])
except Exception:
return {"needs_human_review": True, "raw_model_output": raw}
if __name__ == '__main__':
parser = argparse.ArgumentParser(); parser.add_argument('--adapter', required=True); parser.add_argument('--port', type=int, default=8000)
args = parser.parse_args()
model, tokenizer = FastLanguageModel.from_pretrained(model_name=args.adapter, max_seq_length=1024, load_in_4bit=True)
FastLanguageModel.for_inference(model)
uvicorn.run(app, host='127.0.0.1', port=args.port)
src/validate_dataset.py#!/usr/bin/env python3
"""Validate JSONL instruction datasets and detect obvious data leakage."""
import argparse
import json
from pathlib import Path
REQUIRED = {"instruction", "input", "output"}
def read_jsonl(path: Path):
rows = []
with path.open(encoding="utf-8") as handle:
for number, line in enumerate(handle, start=1):
line = line.strip()
if not line:
continue
try:
item = json.loads(line)
except json.JSONDecodeError as exc:
raise ValueError(f"{path}:{number}: JSON invalide: {exc}") from exc
missing = REQUIRED - item.keys()
if missing:
raise ValueError(f"{path}:{number}: champs manquants: {sorted(missing)}")
if not all(isinstance(item[key], str) for key in REQUIRED):
raise ValueError(f"{path}:{number}: instruction/input/output doivent être des chaînes")
if not item["instruction"].strip() or not item["output"].strip():
raise ValueError(f"{path}:{number}: instruction et output ne peuvent pas être vides")
rows.append(item)
return rows
def fingerprint(item):
return "\n".join(item[key].strip().lower() for key in ("instruction", "input"))
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--data-dir", default="data", help="Dossier contenant les fichiers JSONL")
args = parser.parse_args()
data_dir = Path(args.data_dir)
splits = {}
for split in ("train", "validation", "test"):
path = data_dir / f"{split}.jsonl"
if not path.exists():
raise FileNotFoundError(f"Fichier introuvable : {path}")
splits[split] = read_jsonl(path)
print(f"{split:10s}: {len(splits[split])} exemples")
all_fingerprints = {}
for split, rows in splits.items():
seen = set()
for item in rows:
key = fingerprint(item)
if key in seen:
raise ValueError(f"Doublon interne détecté dans {split}: {item['instruction']}")
seen.add(key)
all_fingerprints.setdefault(key, []).append(split)
leakage = {key: where for key, where in all_fingerprints.items() if len(where) > 1}
if leakage:
example, where = next(iter(leakage.items()))
raise ValueError(f"Fuite entre splits {where}: {example}")
print("Validation réussie : format correct, pas de doublons ni fuite détectés.")
if __name__ == "__main__":
main()
configs/support_training_config.yamlmodel_name: unsloth/Qwen2.5-3B-Instruct-bnb-4bit
max_seq_length: 1024
load_in_4bit: true
random_seed: 3407
output_dir: outputs/qwen25-3b-support-fr-lora
num_train_epochs: 2
learning_rate: 0.00015
per_device_train_batch_size: 2
per_device_eval_batch_size: 2
gradient_accumulation_steps: 8
warmup_ratio: 0.03
logging_steps: 10
eval_steps: 100
save_steps: 100
lora_r: 16
lora_alpha: 32
lora_dropout: 0.05
lora_target_modules: [q_proj, k_proj, v_proj, o_proj, gate_proj, up_proj, down_proj]
requirements.txtunsloth
transformers>=4.45.0
trl>=0.12.0
peft>=0.13.0
datasets>=3.0.0
accelerate>=1.0.0
bitsandbytes>=0.44.0
pyyaml>=6.0
sentencepiece>=0.2.0
requirements-translation.txttorch
transformers>=4.45.0
accelerate>=1.0.0
datasets>=3.0.0
sentencepiece>=0.2.0
requirements-api.txtfastapi>=0.115.0
uvicorn[standard]>=0.30.0
pydantic>=2.0.0