How to Build Type-Safe, Schema-Constrained, and Function-Driven LLM Pipelines Using Lists and Pydantic
In this tutorial, we create a workflow using The outline generating systematic and type-safe results from language models. We work with typed constraints like Literal, int, and bool, and design information templates using outline.Template, and enforce strict schema validation with Pydantic models. We also use robust JSON recovery and a callback style that generates validated arguments and executes Python functions safely. Throughout the course, we focus on reliability, enforceability, and systematic production of the production grade.
import os, sys, subprocess, json, textwrap, re
subprocess.check_call([sys.executable, "-m", "pip", "install", "-q",
"outlines", "transformers", "accelerate", "sentencepiece", "pydantic"])
import torch
import outlines
from transformers import AutoTokenizer, AutoModelForCausalLM
from typing import Literal, List, Union, Annotated
from pydantic import BaseModel, Field
from enum import Enum
print("Torch:", torch.__version__)
print("CUDA available:", torch.cuda.is_available())
print("Outlines:", getattr(outlines, "__version__", "unknown"))
device = "cuda" if torch.cuda.is_available() else "cpu"
print("Using device:", device)
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 device == "cuda" else torch.float32,
device_map="auto" if device == "cuda" else None,
)
if device == "cpu":
hf_model = hf_model.to(device)
model = outlines.from_transformers(hf_model, tokenizer)
def build_chat(user_text: str, system_text: str = "You are a precise assistant. Follow instructions exactly.") -> str:
try:
msgs = [{"role": "system", "content": system_text}, {"role": "user", "content": user_text}]
return tokenizer.apply_chat_template(msgs, tokenize=False, add_generation_prompt=True)
except Exception:
return f"{system_text}nnUser: {user_text}nAssistant:"
def banner(title: str):
print("n" + "=" * 90)
print(title)
print("=" * 90)We install all necessary dependencies and implement the Outlines pipeline with a lightweight tutorial model. We configure device management so that the system automatically switches between CPU and GPU based on availability. We’ve also created reusable helper functions for dialog formatting and clean category banners to streamline workflows.
def extract_json_object(s: str) -> str:
s = s.strip()
start = s.find("{")
if start == -1:
return s
depth = 0
in_str = False
esc = False
for i in range(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 == "{":
depth += 1
elif ch == "}":
depth -= 1
if depth == 0:
return s[start:i + 1]
return s[start:]
def json_repair_minimal(bad: str) -> str:
bad = bad.strip()
last = bad.rfind("}")
if last != -1:
return bad[:last + 1]
return bad
def safe_validate(model_cls, raw_text: str):
raw = extract_json_object(raw_text)
try:
return model_cls.model_validate_json(raw)
except Exception:
raw2 = json_repair_minimal(raw)
return model_cls.model_validate_json(raw2)
banner("2) Typed outputs (Literal / int / bool)")
sentiment = model(
build_chat("Analyze the sentiment: 'This product completely changed my life!'. Return one label only."),
Literal["Positive", "Negative", "Neutral"],
max_new_tokens=8,
)
print("Sentiment:", sentiment)
bp = model(build_chat("What's the boiling point of water in Celsius? Return integer only."), int, max_new_tokens=8)
print("Boiling point (int):", bp)
prime = model(build_chat("Is 29 a prime number? Return true or false only."), bool, max_new_tokens=6)
print("Is prime (bool):", prime)We use a robust JSON framework and minimal maintenance resources to safely return structured results from incomplete generations. We then demonstrate strongly typed generation using Literal, int, and bool, ensuring that the model returns strongly bound values. We validate how Outlines implements type-safe output directly at production time.
banner("3) Prompt templating (outlines.Template)")
tmpl = outlines.Template.from_string(textwrap.dedent("""
<|system|>
You are a strict classifier. Return ONLY one label.
<|user|>
Classify sentiment of this text:
{{ text }}
Labels: Positive, Negative, Neutral
<|assistant|>
""").strip())
templated = model(tmpl(text="The food was cold but the staff were kind."), Literal["Positive","Negative","Neutral"], max_new_tokens=8)
print("Template sentiment:", templated)We use outlines.Template to create structured information templates with strong output controls. We inject user input into the model while preserving paragraph formatting and partition parameters. We show how templating improves reusability and ensures consistent, constrained responses.
banner("4) Pydantic structured output (advanced constraints)")
class TicketPriority(str, Enum):
low = "low"
medium = "medium"
high = "high"
urgent = "urgent"
IPv4 = Annotated[str, Field(pattern=r"^((25[0-5]|2[0-4]d|[01]?dd?).){3}(25[0-5]|2[0-4]d|[01]?dd?)$")]
ISODate = Annotated[str, Field(pattern=r"^d{4}-d{2}-d{2}$")]
class ServiceTicket(BaseModel):
priority: TicketPriority
category: Literal["billing", "login", "bug", "feature_request", "other"]
requires_manager: bool
summary: str = Field(min_length=10, max_length=220)
action_items: List[str] = Field(min_length=1, max_length=6)
class NetworkIncident(BaseModel):
affected_service: Literal["dns", "vpn", "api", "website", "database"]
severity: Literal["sev1", "sev2", "sev3"]
public_ip: IPv4
start_date: ISODate
mitigation: List[str] = Field(min_length=2, max_length=6)
email = """
Subject: URGENT - Cannot access my account after payment
I paid for the premium plan 3 hours ago and still can't access any features.
I have a client presentation in an hour and need the analytics dashboard.
Please fix this immediately or refund my payment.
""".strip()
ticket_text = model(
build_chat(
"Extract a ServiceTicket from this message.n"
"Return JSON ONLY matching the ServiceTicket schema.n"
"Action items must be distinct.nnMESSAGE:n" + email
),
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))We describe advanced Pydantic schemes with enums, regex constraints, field constraints, and ordered lists. We extract the complex ServiceTicket object from the raw email text and validate it using schema-driven encoding. We also use secure authentication logic to handle edge cases and ensure robustness at production scale.
banner("5) Function-calling style (schema -> args -> call)")
class 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("Return JSON ONLY with two integers a and b. Make a odd and b even."),
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("Tip: For best speed and fewer truncations, switch Colab Runtime → GPU.")We implement a callback style workflow by generating structured arguments that match the defined schema. We validate the generated arguments, then safely execute a Python function with that validated input. We show how the first generation of the schema allows controlled tool persuasion and reliable computation driven by LLM.
In conclusion, we implemented a fully structured production pipeline using Outline with strong typing, schema validation, and controlled code generation. We’ve shown how to move from simple typed outputs to advanced Pydantic-based outputs and task-style usage patterns. We also built robustness by using JSON salvage and validation methods, making the system robust against incomplete model outputs. Overall, we have created a functional and production-oriented framework for deterministic, secure, and schema-driven LLM applications.
Check it out Full Codes here. Also, feel free to follow us Twitter and don’t forget to join our 120k+ ML SubReddit and Subscribe to Our newspaper. Wait! are you on telegram? now you can join us on telegram too.
