Cómo crear canalizaciones de LLM con tipos seguros, con restricciones de esquema y basadas en funciones utilizando Outlines y Pydantic

En este tutorial, creamos un flujo de trabajo utilizando Outlines para generar resultados estructurados y con seguridad de tipos a partir de modelos de lenguaje. Trabajamos con restricciones escritas como Literal, int y bool, y diseñamos plantillas de mensajes usando contornos.Template, y aplicamos una validación estricta de esquemas con modelos Pydantic. También implementamos una recuperación JSON sólida y un estilo de llamada de funciones que genera argumentos validados y ejecuta funciones de Python de forma segura. A lo largo del tutorial, nos centramos en la confiabilidad, el cumplimiento de restricciones y la generación estructurada de nivel de producción.

importar sistema operativo, sistema, subproceso, json, ajuste de texto, re subproceso.check_call ([sys.executable, “-m”, “pip”, “install”, “-q”,
“outlines”, “transformers”, “accelerate”, “sentencepiece”, “pydantic”]) importar antorcha importar esquemas desde transformadores importar AutoTokenizer, AutoModelForCausalLM desde escribir importar Literal, Lista, Unión, Anotado desde pydantic importar BaseModel, Campo desde enum importar Enum print(“Torch:”, torch.__version__) print(“CUDA disponible:”, torch.cuda.is_available()) print(“Outlines:”, getattr(outlines, “__version__”, “unknown”)) dispositivo = “cuda” if torch.cuda.is_available() else “cpu” print(“Usando dispositivo:”, dispositivo) MODEL_NAME = “HuggingFaceTB/SmolLM2-135M-Instruct” tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME, use_fast=True) hf_model = AutoModelForCausalLM.from_pretrained( MODEL_NAME, torch_dtype=torch.float16 if dispositivo == “cuda” else torch.float32, device_map=”auto” if dispositivo == “cuda” else Ninguno, ) if dispositivo == “cpu”: hf_model = hf_model.to(device) model = contornos.from_transformers(hf_model, tokenizer) def build_chat(user_text: str, system_text: str = “Eres un asistente preciso. Sigue las instrucciones exactamente.”) -> str: try: msgs = [{“role”: “system”, “content”: system_text}, {“role”: “user”, “content”: user_text}]
return tokenizer.apply_chat_template(msgs, tokenize=False, add_spawn_prompt=True) excepto Excepción: return f”{system_text}\n\nUsuario: {user_text}\nAsistente:” def banner(title: str): print(“\n” + “=” * 90) print(title) print(“=” * 90)

Instalamos todas las dependencias necesarias e inicializamos la canalización de Outlines con un modelo de instrucciones ligero. Configuramos el manejo del dispositivo para que el sistema cambie automáticamente entre CPU y GPU según la disponibilidad. También creamos funciones auxiliares reutilizables para formatear el chat y limpiar banners de secciones para estructurar el flujo de trabajo.

def extract_json_object(s: str) -> str: s = s.strip() start = s.find(“{“) if start == -1: return s profundidad = 0 in_str = False esc = False para i en rango(start, len(s)): ch = s[i]
if in_str: if esc: esc = False elif ch == “\\”: esc = True elif ch == ‘”‘: in_str = False else: if ch == ‘”‘: in_str = True elif ch == “{“: profundidad += 1 elif ch == “}”: profundidad -= 1 if profundidad == 0: return s[start:i + 1]
regresar[start:]

def json_repair_minimal(malo: str) -> str: bad = bad.strip() last = bad.rfind(“}”) si es el último! = -1: devuelve bad[:last + 1]
return bad def safe_validate(model_cls, raw_text: str): raw = extract_json_object(raw_text) intente: return model_cls.model_validate_json(raw) excepto Excepción: raw2 = json_repair_minimal(raw) return model_cls.model_validate_json(raw2) banner(“2) Salidas escritas (Literal / int / bool)”) sentimiento = model( build_chat(“Analice el sentimiento: ‘¡Este producto cambió completamente mi vida!’. Devuelve solo una etiqueta.”), Literal[“Positive”, “Negative”, “Neutral”]max_new_tokens=8, ) print(“Sentimiento:”, sentimiento) bp = model(build_chat(“¿Cuál es el punto de ebullición del agua en grados Celsius? Devuelve solo números enteros.”), int, max_new_tokens=8) print(“Punto de ebullición (int):”, pb) prime = model(build_chat(“¿29 es un número primo? Devuelve solo verdadero o falso.”), bool, max_new_tokens=6) print(“Es primo (bool):”, primo)

Implementamos una extracción JSON sólida y utilidades de reparación mínimas para recuperar de forma segura resultados estructurados de generaciones imperfectas. Luego demostramos la generación fuertemente tipada usando Literal, int y bool, asegurando que el modelo devuelva valores estrictamente restringidos. Validamos cómo Outlines aplica salidas deterministas con seguridad de tipos directamente en el momento de la generación.

banner(“3) Plantilla de solicitud (outlines.Template)”) tmpl = contornos.Template.from_string(textwrap.dedent(“”” <|system|> Eres un clasificador estricto. Devuelve SÓLO una etiqueta. <|user|> Clasifica el sentimiento de este texto: {{ text }} Etiquetas: Positivo, Negativo, Neutral <|assistant|> “””).strip()) templated = model(tmpl(text=”La comida estaba fría pero el personal fueron amables.”), Literal[“Positive”,”Negative”,”Neutral”]max_new_tokens=8) print(“Sentimiento de plantilla:”, con plantilla)

Usamos contornos.Template para crear plantillas de mensajes estructurados con un estricto control de salida. Inyectamos dinámicamente la entrada del usuario en la plantilla mientras preservamos el formato de rol y las restricciones de clasificación. Demostramos cómo las plantillas mejoran la reutilización y garantizan respuestas consistentes y limitadas.

banner(“4) Salida estructurada de Pydantic (restricciones avanzadas)”) clase TicketPriority(str, Enum): bajo = “bajo” medio = “medio” alto = “alto” urgente = “urgente” IPv4 = Anotado[str, Field(pattern=r”^((25[0-5]|2[0-4]\d|[01]?\d\d?)\.){3}(25[0-5]|2[0-4]\d|[01]?\d\d?)$”)]ISODate = Anotado[str, Field(pattern=r”^\d{4}-\d{2}-\d{2}$”)]

clase ServiceTicket(BaseModel): prioridad: TicketPriority categoría: Literal[“billing”, “login”, “bug”, “feature_request”, “other”]
require_manager: resumen bool: str = Campo(min_length=10, max_length=220) action_items: Lista[str] = Campo(min_length=1, max_length=6) clase NetworkIncident(BaseModel): servicio_afectado: Literal[“dns”, “vpn”, “api”, “website”, “database”]
gravedad: Literal[“sev1”, “sev2”, “sev3”]
public_ip: IPv4 start_date: ISODate mitigación: Lista[str] = Field(min_length=2, max_length=6) email = “”” Asunto: URGENTE – No puedo acceder a mi cuenta después del pago. Pagué el plan premium hace 3 horas y todavía no puedo acceder a ninguna función. Tengo una presentación de cliente en una hora y necesito el panel de análisis. Por favor solucione esto inmediatamente o reembolse mi pago. “””.strip() ticket_text = model( build_chat( “Extraiga un ServiceTicket de este mensaje.\n” “Devuelva JSON SÓLO que coincida con el ServiceTicket esquema.\n” “Los elementos de acción deben ser distintos.\n\nMENSAJE:\n” + correo electrónico ), ServiceTicket, max_new_tokens=240, ) ticket = safe_validate(ServiceTicket, ticket_text) if isinstance(ticket_text, str) else ticket_text print(“ServiceTicket JSON:\n”, ticket.model_dump_json(indent=2))

Definimos esquemas Pydantic avanzados con enumeraciones, restricciones de expresiones regulares, límites de campo y listas estructuradas. Extraemos un objeto ServiceTicket complejo del texto sin formato del correo electrónico y lo validamos mediante decodificación basada en esquemas. También aplicamos una lógica de validación segura para manejar casos extremos y garantizar la solidez a escala de producción.

banner(“5) Estilo de llamada de función (esquema -> args -> llamada)”) clase AddArgs(BaseModel): a: int = Field(ge=-1000, le=1000) b: int = Field(ge=-1000, le=1000) def add(a: int, b: int) -> int: return a + b args_text = model( build_chat(“Devolver SÓLO JSON con dos números enteros a y b. Haz un impar y b par.”), AddArgs, max_new_tokens=80, ) args = safe_validate(AddArgs, args_text) if isinstance(args_text, str) else args_text print(“Args:”, args.model_dump()) print(“add(a,b) =”, add(args.a, args.b)) print (“Consejo: para obtener la mejor velocidad y menos truncamientos, cambie Colab Runtime → GPU”).

Implementamos un flujo de trabajo de estilo de llamada de funciones generando argumentos estructurados que se ajustan a un esquema definido. Validamos los argumentos generados y luego ejecutamos de forma segura una función de Python con esas entradas validadas. Demostramos cómo la primera generación de esquemas permite la invocación controlada de herramientas y un cálculo confiable basado en LLM.

En conclusión, implementamos un canal de generación completamente estructurado utilizando Outlines con escritura sólida, validación de esquemas y decodificación controlada. Demostramos cómo pasar de salidas escritas simples a extracción avanzada basada en Pydantic y patrones de ejecución de estilo de función. También construimos resiliencia a través de mecanismos de recuperación y validación de JSON, lo que hizo que el sistema fuera robusto frente a resultados imperfectos del modelo. En general, creamos un marco práctico y orientado a la producción para aplicaciones LLM deterministas, seguras y basadas en esquemas.

Consulte los códigos completos aquí. Además, no dude en seguirnos en Twitter y no olvide unirse a nuestro SubReddit de más de 120.000 ML y suscribirse a nuestro boletín. ¡Esperar! estas en telegrama? Ahora también puedes unirte a nosotros en Telegram.