Saltar a contenido

Traducción automática

Este artículo se tradujo automáticamente a partir de la versión original en inglés.

Razonamiento Guiado por Esquemas en vLLM: salidas estructuradas con xgrammar y Pydantic

Un bucle de reintentos es una confesión. Dice: esperaba que el modelo devolviera JSON válido, no lo hizo, así que volveré a tirar los dados. La mayor parte del código de agentes con LLM con el que he trabajado se apoya muchísimo en los reintentos, y aun así el JSON sigue rompiéndose con la suficiente frecuencia como para que alguien termine configurando una alerta.

El Razonamiento Guiado por Esquemas (SGR) se salta el juego de los reintentos imponiendo el esquema a nivel de token. Defines la topología de razonamiento como un esquema de Pydantic, y el motor de inferencia enmascara cualquier token que lo incumpliría antes del muestreo. La salida es válida por construcción, no por reintento.

TL;DR. SGR usa decodificación restringida para fijar la salida de un LLM a un esquema de Pydantic que controlas tú. Si lo combinas con el backend xgrammar de vLLM, obtienes JSON válido siempre, con una sobrecarga de latencia despreciable.

¿Qué es Schema-Guided Reasoning?

Schema-Guided Reasoning es una técnica que Rinat Abdullin describió en 2024. En lugar de dejar que el modelo complete texto libremente (algo que puede ser inconsistente o ambiguo), le das una plantilla estricta que define:

  • qué pasos debe seguir, para que no pueda saltarse el análisis
  • el orden de esos pasos, para que el razonamiento vaya de los datos a la decisión
  • dónde debe centrar la atención, para que la profundidad se aplique donde importa

Piensa en ello como una lista de comprobación cognitiva que el modelo tiene que seguir.

Visión general de SGR

Por qué importa SGR

Cuando defines un esquema con campos como churn_analysis, margin_math y max_discount_percent, el modelo tiene que rellenarlos en orden. No puede saltar directamente a la decisión sobre el descuento sin antes escribir el análisis.

Eso te da:

  • razonamientos reproducibles en ejecuciones repetidas
  • salidas auditables donde cada paso se puede inspeccionar
  • campos intermedios que puedes evaluar contra un dataset de prueba
  • modelos más pequeños que pasan a ser viables, ya que el esquema impone lo que de otro modo el modelo tendría que aprender
  • el aumento del 5-10% de precisión habitualmente reportado en cargas de trabajo de producción

SGR frente a Chain of Thought frente a prompt engineering

Los tres enfoques se diferencian sobre todo en el grado en que restringen el modelo.

Comparativa de SGR

Característica Prompt Engineering Chain of Thought Schema-Guided Reasoning
Estructura de salida Texto variable Prosa de formato libre JSON/Pydantic rígido
Mecanismo de control Persuasión semántica ("Please output JSON") Prompting heurístico ("Let's think step by step") Decodificación restringida (basada en gramática)
Flujo de razonamiento Lo determina el modelo Lo determina el modelo Lo determina el desarrollador (topología del esquema)
Auditabilidad Baja (requiere parsing) Baja (requiere leer prosa) Alta (inspección a nivel de campo)
Integración Difícil (parsing con regex) Difícil (formato variable) Trivial (deserialización nativa a objetos)
Tasa de error Alta (variabilidad de formato) Moderada (alucinación del formato) Casi nula (la sintaxis la impone el motor)
Requisito del modelo Buen seguimiento de instrucciones Buena capacidad de razonamiento Funciona también con modelos más pequeños

Prompt engineering: persuasión semántica

Please analyze the customer data and output your response as valid JSON
with the following structure: {"discount": <number>, "reason": <string>}
Be careful with the formatting!

Estás esperando que la comprensión del modelo de "output JSON" pese más que su tendencia a ser conversacional. Una actualización del modelo, un cambio de temperatura o un ejemplo few-shot distinto pueden romper tu parser.

Chain of Thought: mejor razonamiento, mismo problema de estructura

Let's think step by step:
1. First, I'll analyze the customer's churn risk...
2. Then I'll calculate the margin...
3. Therefore, I recommend a 15% discount.

CoT mejora la precisión del razonamiento, pero empeora la estructura. La salida es prosa impredecible y es casi imposible parsearla de forma fiable. Normalmente acabas haciendo una segunda llamada al LLM solo para extraer los datos estructurados.

SGR: chain of thought estructurado

SGR mantiene la intuición de CoT de que el razonamiento intermedio mejora la precisión. Simplemente formaliza los pasos:

class PricingLogic(BaseModel):
    # 1. Data Analysis (must complete before decision)
    churn_analysis: str = Field(..., description="Analyze churn_probability")
    financial_analysis: str = Field(..., description="Analyze cart_value and margin")

    # 2. Math Enforcement (explicit calculation)
    margin_math: str = Field(..., description="Calculate: 'Cart $X * Y% = $Z'")

    # 3. Decision Constraint (bounded by prior analysis)
    max_discount_percent: float = Field(..., description="Max allowed discount")

    # 4. Final Output
    offer_code: str
    customer_message: str

El modelo no puede emitir max_discount_percent hasta que churn_analysis, financial_analysis y margin_math estén rellenados. El esquema impone el orden del razonamiento.

Patrones de SGR

SGR tiene tres patrones básicos que se componen en flujos de trabajo mayores.

Patrones de SGR

1. Cascade: pasos de razonamiento secuenciales

Cascade impone un orden de razonamiento. Cada campo tiene que completarse antes que el siguiente.

from pydantic import BaseModel
from typing import Literal, Annotated
from annotated_types import Ge, Le

class CandidateEvaluation(BaseModel):
    """Evaluate a job candidate with enforced reasoning order."""

    # Step 1: Summarize (forces context awareness)
    brief_candidate_summary: str

    # Step 2: Rate (bounded integer)
    rate_skill_match: Annotated[int, Ge(1), Le(10)]

    # Step 3: Decide (constrained choices)
    final_recommendation: Literal["hire", "reject", "hold"]

Buenos encajes: evaluación de candidatos, clasificación de documentos, análisis de cumplimiento, diagnóstico médico.

El modelo tiene que escribir brief_candidate_summary antes de poder puntuar, y puntuar antes de poder recomendar. No hay atajo posible.

2. Routing: un switch semántico

Routing hace que el modelo se comprometa con una ruta de entre varias opciones, implementada con tipos Union.

from pydantic import BaseModel
from typing import Literal, Union

class FeatureLookup(BaseModel):
    """Route to database lookup."""
    rationale: str
    tool_name: Literal["fetch_user_features"] = "fetch_user_features"
    user_id: str

class GeneralResponse(BaseModel):
    """Standard response for non-pricing queries."""
    tool_name: Literal["respond"] = "respond"
    content: str

class RouterSchema(BaseModel):
    """The model must pick exactly ONE branch."""
    action: Union[FeatureLookup, GeneralResponse]

Buenos encajes: clasificación de intenciones, selección de herramientas, triaje de soporte, despacho multiagente.

El discriminador Literal (tool_name) hace que el modelo elija una sola rama y rellene únicamente los campos que necesita esa rama.

3. Cycle: razonamiento repetido con listas

Cycle obliga al modelo a producir varios elementos, con límites sobre cuántos.

from pydantic import BaseModel
from typing import List, Literal, Annotated
from annotated_types import MinLen, MaxLen

class RiskFactor(BaseModel):
    explanation: str
    severity: Literal["low", "medium", "high"]

class RiskAssessment(BaseModel):
    """Generate 2-4 risk factors."""
    factors: Annotated[List[RiskFactor], MinLen(2), MaxLen(4)]

Buenos encajes: evaluación de riesgos, extracción de incidencias, llamadas paralelas a herramientas, planificación multi-paso.

Los límites MinLen y MaxLen fuerzan al menos 2 elementos y como máximo 4. Combinado con Routing, así es como despachas un lote de llamadas a herramientas de anchura fija.

Hacer que SGR funcione: decodificación restringida

Los patrones anteriores son solo esquemas de Pydantic. Lo que hace que sean vinculantes es la decodificación restringida (también llamada Structured Output).

La decodificación restringida modifica el paso de generación de tokens. En lugar de dejar que el modelo muestree libremente de su vocabulario, el motor aplica una máscara gramatical que bloquea los tokens que violarían el esquema. Ocurre en el motor de inferencia, no en el código de tu aplicación.

[!TIP] SGR no requiere "reasoning models" como o1 o DeepSeek-R1. Funciona bien con modelos ajustados por instrucciones, y especialmente bien con modelos destilados a partir de modelos de razonamiento.

Proveedores cloud que lo soportan

La mayoría de los proveedores modernos de LLM ofrecen salidas estructuradas mediante decodificación restringida:

Proveedor Soporte
OpenAI Structured Outputs (incluyendo Azure). GPT-5 usa JSON Schema mediante llguidance
Google/Gemini Soporte de JSON Schema desde nov 2025 (Pydantic y Zod)
Mistral Custom Structured Output
Grok Structured Outputs para varios modelos
Fireworks AI JSON Schema
Cerebras Structured Outputs
OpenRouter Depende del proveedor subyacente, se mapea a JSON Schema

Motores de inferencia que lo soportan

Para modelos self-hosted, todos los motores principales tienen un backend de decodificación restringida:

Motor Backend
vLLM xgrammar o guidance
SGLang Outlines, XGrammar o llguidance
TensorRT-LLM GuidedDecoding
Ollama Structured Outputs

Por qué este artículo se centra en vLLM y xgrammar

Hay varios motivos:

  • vLLM es el motor open source de inferencia de LLM más desplegado, así que lo que construyas aquí se porta fácilmente.
  • xgrammar está implementado en C++ y añade una latencia despreciable.
  • La API de vLLM es compatible con OpenAI, lo que hace barata la migración desde proveedores cloud.
  • xgrammar maneja esquemas anidados complejos, uniones y estructuras recursivas.

La siguiente sección recorre cómo xgrammar impone realmente un esquema a nivel de token.

Cómo xgrammar impone esquemas

Vale la pena entender esta parte con precisión, porque cambia cómo depuras y ajustas los flujos SGR.

Aplicación de xgrammar

Dónde ocurre el enmascarado

xgrammar modifica los logits de salida después del forward pass del modelo y antes del muestreo. No cambia el modelo en sí; filtra qué tokens pueden seleccionarse.

Un bucle de inferencia estándar tiene este aspecto:

1. Input tokens → GPU Forward Pass → Logits (probability scores for all ~128K tokens)
2. Logits → Sampling (temperature, top-p, etc.) → Next Token
3. Repeat until done

xgrammar se cuela entre los pasos 1 y 2:

1. Input tokens → GPU Forward Pass → Raw Logits
2. Raw Logits → xgrammar Logits Processor → Masked Logits
3. Masked Logits → Sampling → Next Token (guaranteed valid)
4. Repeat until done

El modelo sigue calculando su distribución de probabilidad completa en la GPU. Después xgrammar se ejecuta en la CPU y aplica una máscara de bits a esos logits antes del muestreo. A los tokens inválidos se les pone el logit a -∞, lo que hace que su probabilidad sea exactamente 0 tras el softmax.

Dos fases

xgrammar divide el trabajo entre tiempo de compilación y tiempo de ejecución, que es lo que lo hace rápido.

Fase 1: compilación de la gramática, una vez por esquema

# This happens once per schema
tokenizer_info = xgr.TokenizerInfo.from_huggingface(tokenizer)
grammar_compiler = xgr.GrammarCompiler(tokenizer_info)
compiled_grammar = grammar_compiler.compile_json_schema(schema_json)

Durante la compilación, xgrammar:

  1. Convierte el JSON Schema en una Context-Free Grammar.
  2. Construye un Pushdown Automaton (PDA), que es una máquina de estados con pila para poder manejar estructuras anidadas como {"a": {"b": {"c": ...}}}.
  3. Precalcula qué tokens son válidos en cada posición de la gramática. El resultado es la "adaptive token mask cache".
  4. Clasifica los tokens como "context-independent" (cacheables) o "context-dependent" (deben comprobarse en tiempo de ejecución contra el estado de la pila).

[!NOTE] Aproximadamente el 99% de los tokens resultan ser context-independent y acaban en caché (artículo de XGrammar). La mayoría de las comprobaciones de validez en tiempo de ejecución son solo accesos a caché, y por eso xgrammar es rápido.

Fase 2: generación de máscara en tiempo de ejecución, en cada token

En cada paso de generación:

  1. El GrammarMatcher rastrea la posición actual en la gramática.
  2. Busca la máscara precalculada para los tokens context-independent.
  3. Ejecuta el PDA para comprobar los tokens context-dependent restantes.
  4. Los combina en una máscara de bits final y la aplica a los logits.

¿Por qué pushdown automata y no regex?

Por el anidamiento. Una expresión regular (una máquina de estados finitos) no puede hacer matching de forma fiable con estructuras como:

{ "user": { "profile": { "settings": { "theme": "dark" } } } }

La parte difícil son las llaves de cierre }}}: necesitas recordar cuántas has abierto. Un Pushdown Automaton tiene una pila que lleva esa cuenta, así que puede manejar una profundidad de anidamiento arbitraria. Esa es también la razón por la que xgrammar puede imponer tipos Union, objetos anidados y esquemas recursivos, donde los enfoques basados en regex se quedan cortos.

Un ejemplo concreto: generar un campo float

Cuando el modelo está generando "max_discount_percent":, xgrammar sabe por el esquema que a continuación va un float. La máscara:

  • permite (probabilidad sin cambios): 0, 1, 2, ..., 9, ., -
  • bloquea (probabilidad puesta a 0): ", {, [, true, false, null y el resto del vocabulario de más de 128K

El forward pass podría haber asignado una probabilidad alta a la palabra "fifteen". Tras la máscara de xgrammar, ese token tiene probabilidad 0. El modelo tiene que emitir dígitos.

Por qué la sobrecarga es "casi nula"

Por tres motivos:

  1. Ejecución en paralelo. El cálculo de la máscara en CPU se solapa con el siguiente forward pass en GPU. Mientras la GPU calcula los logits del token N+1, la CPU calcula la máscara del token N.
  2. Caché. La mayor parte del trabajo de validación se hace en tiempo de compilación. En ejecución casi todo son accesos a caché.
  3. Implementación en C++. El hot path está en C++, no en Python, y la máscara se aplica a los logits in place.

En benchmarks, xgrammar muestra una sobrecarga despreciable, y la generación estructurada puede ocasionalmente ser más rápida que la generación no restringida porque el vocabulario restringido abarata el muestreo.

Implementación práctica con vLLM

La referencia es el proyecto sgr-discount-manager, una pequeña demo que usa SGR para pricing dinámico.

Flujo de trabajo del agente

Estructura del proyecto

sgr/
├── agent.py            # Main orchestration
├── models/
│   └── schemas.py      # Pydantic SGR schemas
├── prompts/
│   ├── routing.py      # Phase 1 prompts
│   └── pricing.py      # Phase 3 prompts
├── store/
│   └── hybrid_store.py # Hot/Cold data retrieval
└── utils/
    └── llm_client.py   # LLM client wrapper with xgrammar

Paso 1: definir los esquemas

# sgr/models/schemas.py
from pydantic import BaseModel, Field
from typing import Literal, Union


# --- Phase 1: Routing (Union for branching) ---
class FeatureLookup(BaseModel):
    """Route to DB lookup if pricing context is needed."""
    rationale: str
    tool_name: Literal["fetch_user_features"] = "fetch_user_features"
    user_id: str


class GeneralResponse(BaseModel):
    """Standard response for non-pricing queries."""
    tool_name: Literal["respond"] = "respond"
    content: str


class RouterSchema(BaseModel):
    action: Union[FeatureLookup, GeneralResponse]


# --- Phase 2: Pricing Logic (Cascade for sequential reasoning) ---
class PricingLogic(BaseModel):
    """
    Strict reasoning topology for dynamic pricing.
    Fields are ordered to enforce the analysis→decision flow.
    """
    # 1. Data Analysis (Reflection)
    churn_analysis: str = Field(...,
        description="Analyze churn_probability (High > 0.7).")
    financial_analysis: str = Field(...,
        description="Analyze cart_value and profit_margin.")

    # 2. Hard Math Enforcement
    margin_math: str = Field(...,
        description="Calculate absolute profit: 'Cart $200 * 0.20 Margin = $40'.")

    # 3. The Decision Constraint
    max_discount_percent: float = Field(...,
        description="Max allowed discount %. NEVER exceed margin.")

    # 4. Final Output
    offer_code: str = Field(..., description="Generated code (e.g. SAVE20).")
    customer_message: str = Field(..., description="The final polite offer text.")

Paso 2: un cliente LLM que activa xgrammar

# sgr/utils/llm_client.py
from openai import OpenAI
from pydantic import BaseModel
from typing import TypeVar
import json

T = TypeVar("T", bound=BaseModel)


class LLMClient:
    """Wrapper for vLLM with xgrammar-enforced structured generation."""

    def __init__(self, base_url: str = "http://localhost:8000/v1"):
        self.client = OpenAI(base_url=base_url, api_key="EMPTY")
        self.model = self._get_available_model()

    def _get_available_model(self) -> str:
        """Auto-detect the model running on vLLM server."""
        try:
            models = self.client.models.list()
            if models.data:
                return models.data[0].id
        except Exception:
            pass
        return "Qwen/Qwen2.5-7B-Instruct"

    def run_sgr(self, messages: list[dict], schema_class: type[T]) -> T:
        """Run inference with Schema-Guided Response constraints.

        Uses vLLM's guided_json with xgrammar backend to enforce
        strict schema constraints at the token generation level.
        """
        schema_dict = schema_class.model_json_schema()

        # Enhance system message with schema for model guidance
        enhanced_messages = messages.copy()
        if enhanced_messages and enhanced_messages[0]["role"] == "system":
            schema_json = json.dumps(schema_dict, indent=2)
            enhanced_messages[0] = {
                "role": "system",
                "content": (
                    enhanced_messages[0]["content"]
                    + f"\n\nRespond with JSON matching this schema:\n{schema_json}"
                ),
            }

        # vLLM's guided_json with the xgrammar backend
        completion = self.client.chat.completions.create(
            model=self.model,
            messages=enhanced_messages,
            temperature=0.1,  # Low temp for deterministic reasoning
            extra_body={
                "guided_json": schema_dict,  # Pydantic schema as dict
                "guided_decoding_backend": "xgrammar",  # Hardware-enforced
            },
        )

        raw_response = completion.choices[0].message.content
        return schema_class.model_validate_json(raw_response)

[!NOTE] > guided_json acepta un dict de JSON Schema. Con guided_decoding_backend: "xgrammar", el LLM solo puede generar tokens que formen JSON válido y que encaje con tu esquema.

Paso 3: orquestar el agente

# sgr/agent.py
from .models.schemas import PricingLogic, RouterSchema
from .prompts.routing import build_routing_prompt
from .prompts.pricing import build_pricing_context_prompt, ASSISTANT_FETCH_MESSAGE
from .store.hybrid_store import HybridFeatureStore
from .utils.llm_client import LLMClient


def pricing_agent(user_query: str, user_id: str) -> str:
    """Process a pricing query with three-phase SGR workflow."""

    llm = LLMClient()
    feature_store = HybridFeatureStore()

    # Build conversation history
    history = [
        {"role": "system", "content": build_routing_prompt(user_id)},
        {"role": "user", "content": user_query},
    ]

    # --- Phase 1: Routing (Uses RouterSchema) ---
    print(f"🤖 Processing: '{user_query}' for {user_id}")
    decision = llm.run_sgr(history, RouterSchema)
    print(f"📍 Routing decision: {decision.action.tool_name}")

    if decision.action.tool_name == "respond":
        return decision.action.content

    # --- Phase 2: Context Retrieval ---
    if decision.action.tool_name == "fetch_user_features":
        print(f"🔍 Fetching features for {user_id}...")
        context = feature_store.get_user_context(user_id)

        if not context:
            return "Error: User profile not found."

        print(f"   [Data] LTV: ${context.get('user_ltv')} | "
              f"Margin: {context.get('cart_profit_margin', 0) * 100}%")

        # Inject context into conversation
        history.append({"role": "assistant", "content": ASSISTANT_FETCH_MESSAGE})
        history.append({
            "role": "user",
            "content": build_pricing_context_prompt(
                churn_prob=context.get("churn_probability", 0.5),
                cart_val=context.get("current_cart_value", 100),
                margin=context.get("cart_profit_margin", 0.2),
                user_ltv=context.get("user_ltv", 0),
            ),
        })

        # --- Phase 3: SGR Logic Execution (Uses PricingLogic) ---
        print("🧠 Calculating Offer (Schema Enforced)...")
        offer = llm.run_sgr(history, PricingLogic)

        # Audit log — the SGR benefit: explicit reasoning traces
        print(f"   [Audit] Math: {offer.margin_math}")
        print(f"   [Audit] Max Allowed: {offer.max_discount_percent}%")

        return offer.customer_message

    return "I'm sorry, I couldn't process your request."


if __name__ == "__main__":
    response = pricing_agent("I want a discount or I'm leaving!", "user_102")
    print(f"\n💬 Final Reply: {response}")

Paso 4: ejecutar vLLM con xgrammar

# Start vLLM server with xgrammar backend (default in recent versions)
python -m vllm.entrypoints.openai.api_server \
    --model Qwen/Qwen2.5-7B-Instruct \
    --port 8000

# Run the agent
uv run python -m sgr.agent

Ejemplo de salida

🤖 Processing: 'I want a discount or I'm leaving!' for user_102
📍 Routing decision: fetch_user_features
🔍 Fetching features for user_102...
   [Data] LTV: $1,500 | Margin: 20%
🧠 Calculating Offer (Schema Enforced)...
   [Audit] Math: Cart $200 * 0.20 Margin = $40
   [Audit] Max Allowed: 15.0%

💬 Final Reply: We value your loyalty! Here's a special 15% discount
   with code SAVE15. This reflects our appreciation for your continued
   business with us.

El log de auditoría muestra el trabajo real del modelo: calculó el margen ($40 en un carrito de $200 al 20%) y acotó el descuento para que la oferta se mantenga dentro de la restricción de beneficio.

Buenas prácticas

Diseño de esquemas

  1. Ordena los campos según el flujo de razonamiento. Los campos de análisis van antes que los de decisión.
  2. Escribe descripciones Field descriptivas. Guían la atención del modelo tanto como el propio nombre del campo.
  3. Restringe con Literal y Annotated. Usa Literal["a", "b"] para enums y Annotated[int, Ge(1), Le(10)] para límites.
  4. Mantén los esquemas enfocados. Un esquema por fase de razonamiento, y luego compón con varias llamadas.

Configuración de vLLM

  1. Usa una temperatura baja (0.1-0.3) para un razonamiento determinista.
  2. Deja que xgrammar maneje la estructura. No luches contra ello con instrucciones de formato en el prompt.
  3. Vigila el uso de tokens. SGR suele usar menos tokens que CoT porque no hay prosa verbosa.

Consideraciones de producción

  1. Versiona tus esquemas igual que versionas APIs.
  2. Incluso con SGR, los errores de red y de servidor siguen necesitando un manejo elegante.
  3. Registra las salidas SGR en bruto para cumplimiento y depuración.
  4. Prueba con casos límite para que el esquema se mantenga en los bordes.

Conclusión

SGR es lo que te lleva de "funciona en una demo" a "funciona en producción". Defines la topología de razonamiento en Pydantic, dejas que xgrammar la imponga en tiempo de decodificación, y la salida es:

  • válida siempre, sin bucles de reintentos ni fallos de parsing
  • auditable a nivel de campo
  • utilizable con modelos más pequeños, porque ya no tienen que clavar el formato por sí solos
  • más barata de ejecutar, ya que usas menos tokens, menos reintentos y modelos más pequeños

La demo sgr-discount-manager conecta cada ejemplo de código de este post contra un servidor vLLM real. Clónala y empieza a adaptar los esquemas a tu propio flujo de trabajo.

Referencias

Framework SGR

xgrammar

vLLM

Proyecto demo