Una implementación de codificación para establecer flujos de trabajo rigurosos de control de versiones y pruebas de regresión para modelos de lenguaje grandes utilizando MLflow

En este tutorial, mostramos cómo tratamos las indicaciones como artefactos versionados de primera clase y aplicamos pruebas de regresión rigurosas al comportamiento de modelos de lenguaje grandes usando MLflow. Diseñamos un proceso de evaluación que registra versiones de solicitudes, diferencias de solicitudes, resultados de modelos y múltiples métricas de calidad de una manera totalmente reproducible. Al combinar métricas de texto clásicas con similitud semántica y indicadores de regresión automatizados, demostramos cómo podemos detectar sistemáticamente la desviación del rendimiento causada por cambios rápidos aparentemente pequeños. A lo largo del tutorial, nos centramos en crear un flujo de trabajo que refleje las prácticas reales de ingeniería de software, pero aplicado a una evaluación rápida de ingeniería y LLM. Consulta los CÓDIGOS COMPLETOS aquí.

!pip -q install -U “openai>=1.0.0” mlflow rouge-score nltk transformadores de oraciones scikit-learn pandas import os, json, time, difflib, re desde escribir import List, Dict, Any, Tuple import mlflow import pandas as pd import numpy as np from openai import OpenAI from rouge_score import rouge_scorer import nltk from nltk.translate.bleu_score importar sentencia_bleu, SmoothingFunction de sentencia_transformers importar SentenceTransformer de sklearn.metrics.pairwise importar coseno_similaridad nltk.download(“punkt”, quiet=True) nltk.download(“punkt_tab”, quiet=True) si no os.getenv(“OPENAI_API_KEY”): intente: de google.colab importar datos de usuario # tipo: ignorar k = userdata.get(“OPENAI_API_KEY”) si k: os.environ[“OPENAI_API_KEY”] = k excepto Excepción: pasar si no os.getenv(“OPENAI_API_KEY”): importar getpass os.environ[“OPENAI_API_KEY”] = getpass.getpass(“Ingrese OPENAI_API_KEY (entrada oculta): “).strip() afirmar os.getenv(“OPENAI_API_KEY”), “Se requiere OPENAI_API_KEY”.

Configuramos el entorno de ejecución instalando todas las dependencias necesarias e importando las bibliotecas principales utilizadas a lo largo del tutorial. Cargamos de forma segura la clave API de OpenAI en tiempo de ejecución, asegurando que las credenciales nunca estén codificadas en el cuaderno. También inicializamos recursos esenciales de PNL para garantizar que el proceso de evaluación se ejecute de manera confiable en diferentes entornos.

MODELO = “gpt-4o-mini” TEMPERATURA = 0.2 MAX_OUTPUT_TOKENS = 250 ABS_SEM_SIM_MIN = 0.78 DELTA_SEM_SIM_MAX_DROP = 0.05 DELTA_ROUGE_L_MAX_DROP = 0.08 DELTA_BLEU_MAX_DROP = 0.10 mlflow.set_tracking_uri(“file:/content/mlruns”) mlflow.set_experiment(“prompt_versioning_llm_regression”) cliente = OpenAI() incrustador = SentenceTransformer(“all-MiniLM-L6-v2”) EVAL_SET = [
{
“id”: “q1”,
“input”: “Summarize in one sentence: MLflow tracks experiments, runs, parameters, metrics, and artifacts.”,
“reference”: “MLflow helps track machine learning experiments by logging runs with parameters, metrics, and artifacts.”
},
{
“id”: “q2”,
“input”: “Rewrite professionally: ‘this model is kinda slow but it works ok.'”,
“reference”: “The model is somewhat slow, but it performs reliably.”
},
{
“id”: “q3”,
“input”: “Extract key fields as JSON: ‘Order 5531 by Alice costs $42.50 and ships to Toronto.'”,
“reference”: ‘{“order_id”:”5531″,”customer”:”Alice”,”amount_usd”:42.50,”city”:”Toronto”}’
},
{
“id”: “q4”,
“input”: “Answer briefly: What is prompt regression testing?”,
“reference”: “Prompt regression testing checks whether prompt changes degrade model outputs compared to a baseline.”
},
]

INDICACIONES = [
{
“version”: “v1_baseline”,
“prompt”: (
“You are a precise assistant.\n”
“Follow the user request carefully.\n”
“If asked for JSON, output valid JSON only.\n”
“User: {user_input}”
)
},
{
“version”: “v2_formatting”,
“prompt”: (
“You are a helpful, structured assistant.\n”
“Respond clearly and concisely.\n”
“Prefer clean formatting.\n”
“User request: {user_input}”
)
},
{
“version”: “v3_guardrailed”,
“prompt”: (
“You are a rigorous assistant.\n”
“Rules:\n”
“1) If user asks for JSON, output ONLY valid minified JSON.\n”
“2) Otherwise, keep the answer short and factual.\n”
“User: {user_input}”
)
},
]

Definimos todas las configuraciones experimentales, incluidos los parámetros del modelo, los umbrales de regresión y la configuración de seguimiento de MLflow. Construimos el conjunto de datos de evaluación y declaramos explícitamente múltiples versiones de indicaciones para comparar y probar regresiones. Al centralizar estas definiciones, garantizamos que los cambios rápidos y la lógica de evaluación permanezcan controlados y reproducibles.

def call_llm(formatted_prompt: str) -> str: resp = client.responses.create( model=MODEL, input=formatted_prompt, temperatura=TEMPERATURA, max_output_tokens=MAX_OUTPUT_TOKENS, ) out = getattr(resp, “output_text”, Ninguno) si sale: devuelve out.strip() prueba: textos = []
para elemento en resp.output: if getattr(item, “type”, “”) == “message”: para c en item.content: if getattr(c, “type”, “”) in (“output_text”, “text”): texts.append(getattr(c, “text”, “”)) return “\n”.join(texts).strip() excepto Excepción: return “” smooth = SmoothingFunction().method3 rouge = rouge_scorer.RougeScorer([“rougeL”]use_stemmer=True) def safe_tokenize(s: str) -> Lista[str]: s = (s o “”).strip().lower() si no s: devolver []
intente: devolver nltk.word_tokenize(s) excepto LookupError: devolver re.findall(r”\b\w+\b”, s) def bleu_score(ref: str, hyp: str) -> float: r = safe_tokenize(ref) h = safe_tokenize(hyp) si len(h) == 0 o len(r) == 0: devolver 0.0 devolver flotador (oración_bleu ([r]h, smoothing_function=smooth)) def rougeL_f1(ref: str, hyp: str) -> float: puntuaciones = rouge.score(ref o “”, hyp o “”) return float(puntuaciones[“rougeL”].fmeasure) def semantic_sim(ref: str, hyp: str) -> float: embs = embedder.encode([ref or “”, hyp or “”]normalize_embeddings=True) return float(cosine_similarity([embs[0]], [embs[1]])[0][0])

Implementamos las métricas principales de invocación y evaluación de LLM que se utilizan para evaluar la calidad inmediata. Calculamos BLEU, ROUGE-L y puntuaciones de similitud semántica para capturar diferencias semánticas y de nivel superficial en los resultados del modelo. Nos permite evaluar cambios rápidos desde múltiples perspectivas complementarias en lugar de depender de una única métrica.

def evaluar_prompt(prompt_template: str) -> Tupla[pd.DataFrame, Dict[str, float]cadena]: filas = []
por ejemplo en EVAL_SET: p = Prompt_template.format(user_input=ex[“input”]) y = call_llm(p) ref = ex[“reference”]

filas.append({ “id”: ex[“id”]”entrada”: ej.[“input”]”referencia”: ref, “salida”: y, “bleu”: bleu_score(ref, y), “rougeL_f1”: rougeL_f1(ref, y), “semantic_sim”: semantic_sim(ref, y), }) df = pd.DataFrame(rows) agg = { “bleu_mean”: float(df[“bleu”].mean()), “rougeL_f1_mean”: flotador(df[“rougeL_f1”].mean()), “semantic_sim_mean”: flotante(df[“semantic_sim”].mean()), } outputs_jsonl = “\n”.join(json.dumps(r, sure_ascii=False) para r en filas) return df, agg, outputs_jsonl def log_text_artifact(text: str, artefacto_path: str): mlflow.log_text(text, artefacto_path) def Prompt_diff(antiguo: str, nuevo: str) -> str: a = old.splitlines(keepends=True) b = new.splitlines(keepends=True) return “”.join(difflib.unified_diff(a, b, fromfile=”previous_prompt”, tofile=”current_prompt”)) def compute_regression_flags(baseline: Dict[str, float]actual: Dictado[str, float]) -> Dictar[str, Any]: d_sem = línea base[“semantic_sim_mean”] – actual[“semantic_sim_mean”]
d_rouge = línea de base[“rougeL_f1_mean”] – actual[“rougeL_f1_mean”]
d_bleu = línea de base[“bleu_mean”] – actual[“bleu_mean”]

banderas = { “abs_semantic_fail”: actual[“semantic_sim_mean”] < ABS_SEM_SIM_MIN, "drop_semantic_fail": d_sem > DELTA_SEM_SIM_MAX_DROP, “drop_rouge_fail”: d_rouge > DELTA_ROUGE_L_MAX_DROP, “drop_bleu_fail”: d_bleu > DELTA_BLEU_MAX_DROP, “delta_semantic”: float(d_sem), “delta_rougeL”: flotador(d_rouge), “delta_bleu”: flotador(d_bleu), } banderas[“regression”] = cualquiera([flags[“abs_semantic_fail”]banderas[“drop_semantic_fail”]banderas[“drop_rouge_fail”]banderas[“drop_bleu_fail”]]) devolver banderas

Construimos la lógica de evaluación y regresión que ejecuta cada mensaje con el conjunto de evaluación y agrega los resultados. Registramos artefactos de avisos, diferencias de avisos y resultados de evaluación en MLflow, asegurando que cada experimento siga siendo auditable. También calculamos indicadores de regresión que identifican automáticamente si una versión solicitada degrada el rendimiento en relación con la línea de base. Consulta los CÓDIGOS COMPLETOS aquí.

print(“Ejecutando control de versiones + pruebas de regresión con MLflow…”) print(f”URI de seguimiento: {mlflow.get_tracking_uri()}”) print(f”Experimento: {mlflow.get_experiment_by_name(‘prompt_versioning_llm_regression’).name}”) run_summary = []
baseline_metrics = Ninguno baseline_prompt = Ninguno baseline_df = Ninguno baseline_metrics_name = Ninguno con mlflow.start_run(run_name=f”prompt_regression_suite_{int(time.time())}”) como parent_run: mlflow.set_tag(“task”, “prompt_versioning_regression_testing”) mlflow.log_param(“model”, MODEL) mlflow.log_param(“temperatura”, TEMPERATURA) mlflow.log_param(“max_output_tokens”, MAX_OUTPUT_TOKENS) mlflow.log_param(“eval_set_size”, len(EVAL_SET)) para pv en PROMPTS: ver = pv[“version”]
mensaje_t = pv[“prompt”]

con mlflow.start_run(run_name=ver, nested=True) como child_run: mlflow.log_param(“prompt_version”, ver) log_text_artifact(prompt_t, f”prompts/{ver}.txt”) si baseline_prompt no es Ninguno y baseline_metrics_name no es Ninguno: diff = Prompt_diff(baseline_prompt, Prompt_t) log_text_artifact(diff, f”prompt_diffs/{baseline_metrics_name}_to_{ver}.diff”) else: log_text_artifact(“BASELINE_PROMPT (sin diff)”, f”prompt_diffs/{ver}.diff”) df, agg, outputs_jsonl = evalua_prompt(prompt_t) mlflow.log_dict(agg, f”metrics/{ver}_agg.json”) log_text_artifact(outputs_jsonl, f”outputs/{ver}_outputs.jsonl”) mlflow.log_metric(“bleu_mean”, agg[“bleu_mean”]) mlflow.log_metric(“rougeL_f1_mean”, agg[“rougeL_f1_mean”]) mlflow.log_metric(“semantic_sim_mean”, agg[“semantic_sim_mean”]) si baseline_metrics es Ninguno: baseline_metrics = agg baseline_prompt = Prompt_t baseline_df = df baseline_metrics_name = ver flags = {“regression”: False, “delta_bleu”: 0.0, “delta_rougeL”: 0.0, “delta_semantic”: 0.0} mlflow.set_tag(“regression”, “false”) else: flags = Compute_regression_flags(baseline_metrics, agg) mlflow.log_metric(“delta_bleu”, banderas[“delta_bleu”]) mlflow.log_metric(“delta_rougeL”, banderas[“delta_rougeL”]) mlflow.log_metric(“delta_semantic”, banderas[“delta_semantic”]) mlflow.set_tag(“regresión”, str(banderas[“regression”]).lower()) para k en [“abs_semantic_fail”,”drop_semantic_fail”,”drop_rouge_fail”,”drop_bleu_fail”]: mlflow.set_tag(k, str(banderas[k]).lower()) run_summary.append({ “prompt_version”: ver, “bleu_mean”: agg[“bleu_mean”]”rougeL_f1_mean”: agg[“rougeL_f1_mean”]”semantic_sim_mean”: agg[“semantic_sim_mean”]”delta_bleu_vs_baseline”: float(flags.get(“delta_bleu”, 0.0)), “delta_rougeL_vs_baseline”: float(flags.get(“delta_rougeL”, 0.0)), “delta_semantic_vs_baseline”: float(flags.get(“delta_semantic”, 0.0)), “regression_flag”: bool(banderas[“regression”]), “mlflow_run_id”: child_run.info.run_id, }) resumen_df = pd.DataFrame(run_summary).sort_values(“prompt_version”) print(“\n=== Resultados agregados (cuanto más alto, mejor) ===”) display(summary_df) regresed = resumen_df[summary_df[“regression_flag”] == Verdadero]si len(regresión) > 0: print(“\n🚩 Regresiones detectadas:”) display(regresión[[“prompt_version”,”delta_bleu_vs_baseline”,”delta_rougeL_vs_baseline”,”delta_semantic_vs_baseline”,”mlflow_run_id”]]) else: print(“\n✅ No se detectaron regresiones bajo los umbrales actuales.”) si len(regresed) > 0 y baseline_df no es Ninguno: peor_ver = regressed.sort_values(“delta_semantic_vs_baseline”, ascending=False).iloc[0][“prompt_version”]

peor_prompt = siguiente (p[“prompt”] para p en PROMPTS si p[“version”] == peor_ver) peor_df, _, _ = evaluar_prompt(peor_prompt) fusionado = baseline_df[[“id”,”output”,”bleu”,”rougeL_f1″,”semantic_sim”]].merge( peor_df[[“id”,”output”,”bleu”,”rougeL_f1″,”semantic_sim”]], on=”id”, sufijos=(“_baseline”, f”_{worst_ver}”) ) fusionados[“delta_semantic”] = fusionado[“semantic_sim_baseline”] – fusionado[f”semantic_sim_{worst_ver}”]
fusionado[“delta_rougeL”] = fusionado[“rougeL_f1_baseline”] – fusionado[f”rougeL_f1_{worst_ver}”]
fusionado[“delta_bleu”] = fusionado[“bleu_baseline”] – fusionado[f”bleu_{worst_ver}”]
print(f”\n=== Deltas por ejemplo: línea base vs {peor_ver} (delta positivo = peor) ===”) display( fusionado[[“id”,”delta_semantic”,”delta_rougeL”,”delta_bleu”,”output_baseline”,f”output_{worst_ver}”]].sort_values(“delta_semantic”, ascending=False) ) print(“\nAbra la interfaz de usuario de MLflow (opcional) ejecutando:”) print(“!mlflow ui –backend-store-uri file:/content/mlruns –host 0.0.0.0 –port 5000”)

Orquestamos el flujo de trabajo completo de pruebas de regresión mediante ejecuciones de MLflow anidadas. Comparamos cada versión del mensaje con la línea de base, registramos deltas métricos y registramos los resultados de la regresión en una tabla de resumen estructurada. Esto completa un proceso repetible y de nivel de ingeniería para pruebas de regresión y versiones rápidas que podemos extender a conjuntos de datos más grandes y aplicaciones del mundo real.

En conclusión, establecimos un marco práctico y orientado a la investigación para el control rápido de versiones y pruebas de regresión que nos permite evaluar el comportamiento de LLM con disciplina y transparencia. Mostramos cómo MLflow nos permite realizar un seguimiento de la evolución rápida, comparar resultados entre versiones y marcar automáticamente regresiones en función de umbrales bien definidos. Este enfoque nos ayuda a alejarnos del ajuste rápido ad hoc y acercarnos a una experimentación mensurable y repetible. Al adoptar este flujo de trabajo, nos aseguramos de que las actualizaciones rápidas mejoren el comportamiento del modelo intencionalmente en lugar de introducir regresiones de rendimiento ocultas.

Consulta los CÓDIGOS COMPLETOS aquí. Además, no dude en seguirnos en Twitter y no olvide unirse a nuestro SubReddit de más de 100.000 ML y suscribirse a nuestro boletín. ¡Esperar! estas en telegrama? Ahora también puedes unirte a nosotros en Telegram.