[!NOTE] Tradução automática Este artigo foi traduzido automaticamente a partir da versão original em inglês.

Raciocínio Guiado por Esquema em vLLM: Structured Outputs com xgrammar e Pydantic

Retentar uma chamada ao LLM não garante a obtenção de um JSON válido. O próximo exemplo pode falhar da mesma forma, e chamadas repetidas aumentam a latência e os custos associados.

O Reasoning Orientado por Esquema (SGR) aplica um esquema específico à medida que o modelo gera cada token. Definem-se os campos obrigatórios utilizando Pydantic, e o motor de inferência bloqueia os tokens que violam essa estrutura. Como resultado, a saída obtida é sintaticamente válida desde a sua génese, e não apenas após tentativas repetidas.

TL;DR. SGR utiliza constrained decoding para fixar a saída de um LLM a um esquema Pydantic que se encontra sob o nosso controlo. Em combinação com o xgrammar backend de vLLM, obtém-se sempre JSON válidos, sem quase nenhum custo adicional em termos de latência.


O que é o Raciocínio Guiado por Esquema?

O Reasoning Orientado por Esquema é uma técnica que Rinat Abdullin escrito em 2024. Em vez de permitir que o modelo complete o texto livremente (o que pode resultar em inconsistências ou ambiguidades), é-lhe fornecido um template rigoroso que define:

Pense nisso como uma lista de verificação cognitiva que o modelo deve seguir.

SGR Visão geral

O que o esquema controla

Campos como churn_analysis, margin_math, e max_discount_percent Deve-se tornar explícitos os resultados intermediários esperados. O modelo tem de preencher a estrutura obrigatória antes de poder retornar a decisão final relativa ao desconto.

Isso fornece a você:


SGR vs Cadeia de Pensamento vs prompt engineering

As três abordagens diferem principalmente no grau de restrição que impõem ao modelo.

Comparação de SGR

FuncionalidadePrompt EngineeringCadeia de PensamentoRaciocínio Guiado por Esquema
Estrutura de SaídaTexto variávelProsa de formato livreRígido JSON/Pydantic
Mecanismo de ControloPersuasão semântica (“Por favor, exiba JSON”)Estímulo heurístico (“Vamos pensar passo a passo”)Constrained decoding (baseado em gramática)
Fluxo de RaciocínioO modelo determinaO modelo determinaO desenvolvedor determina a topologia do esquema.
AuditoriaBaixo (requer análise de sintaxe)Baixo (requer leitura de texto)Inspeção de alto nível (a nível de campo)
IntegraçãoTrivial (deserialização de objetos nativos)
Taxa de ErroAlto (variabilidade de formato)Moderado (alucinação de formato)
Requisitos do ModeloForte capacidade de seguimento de instruçõesCapacidade avançada de raciocínioFunciona igualmente com modelos mais pequenos.

Prompt engineering: persuasão 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!

Espera-se que a compreensão do modelo em relação a “output JSON” seja mais forte do que a sua tendência para adotar um tom conversacional. Uma atualização do modelo, uma alteração na temperatura ou um exemplo de few-shot diferente podem comprometer o seu analisador de texto.

Cadeia de Pensamento: raciocínio mais eficaz, mesmo problema de estrutura

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 melhora a precisão do raciocínio, mas piora a estruturação dos dados. O resultado é um texto imprevisível que é quase impossível de ser analisado de forma confiável. Geralmente, é necessário realizar uma segunda chamada a LLM apenas para extrair os dados estruturados.

SGR: cadeia estruturada de raciocínio

SGR mantém a intuição de CoT de que o raciocínio intermédio melhora a precisão. Ele apenas formaliza os passos necessários:

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

O modelo não consegue gerar saída. max_discount_percent até que churn_analysis, financial_analysis, e margin_math São preenchidos. O esquema impõe a ordem de raciocínio.


Padrões SGR

SGR possui três padrões fundamentais que se combinam para formar fluxos de trabalho mais complexos.

SGR Padrões

1. Cascata: passos sequenciais de raciocínio

A cascata impõe uma ordem de raciocínio: cada campo tem de ser preenchido antes do seguinte.

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"]

Casos de uso adequados: avaliação de candidatos, classificação de documentos, análise de conformidade, diagnóstico médico.

O modelo tem de escrever brief_candidate_summary Antes de poder avaliar, é preciso primeiro fazer a avaliação; e antes de conseguir recomendar, é necessário ter realizado a avaliação. Não existe atalho.


2. Roteamento: uma instrução de comutação semântica

O encaminhamento faz com que o modelo escolha um caminho específico entre um conjunto de opções, sendo implementado com Union tipos.

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]

Casos de uso adequados: classificação de intenções, seleção de ferramentas, triagem de suporte e envio de multi-agent.

O Literal discriminador (tool_name) faz com que o modelo selecione um único ramo e preencha apenas os campos necessários para esse ramo.


3. Ciclo: raciocínio repetido com listas

O ciclo obriga o modelo a gerar vários itens, estabelecendo limites para o seu número.

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)]

Casos de uso adequados: avaliação de risco, extração de problemas, processamento paralelo tool calls, planeamento em várias etapas.

O MinLen e MaxLen A restrição de limites exige pelo menos 2 e no máximo 4 elementos. Em conjunto com o mecanismo de Roteamento, é assim que se envia um lote de largura fixa de tool calls.


Fazer com que SGR funcione: constrained decoding

Os padrões acima são simplesmente esquemas Pydantic. O que os torna vinculativos é constrained decoding (também conhecido como Structured Output).

Constrained decoding modifica a etapa de geração de tokens. Em vez de permitir que o modelo selecione livremente a partir do seu vocabulário, o motor aplica uma máscara gramatical que bloqueia os tokens que poderiam violar o esquema. Este processo ocorre no motor de inferência, e não no código da aplicação.

[!DICA] SGR não requer “modelos de raciocínio” como o1 ou DeepSeek-R1. Funciona perfeitamente com modelos ajustados por instruções e, especialmente, com modelos derivados de aqueles baseados em raciocínio.

Fornecedores de nuvem que o suportam

A maioria dos fornecedores modernos de LLM oferece structured outputs através de constrained decoding:

FornecedorSuporte
OpenAIStructured Outputs (incluindo o Azure). O GPT-5 utiliza JSON Schema por meio de llguidance
Google/GeminiJSON Schema Suporte disponível a partir de novembro de 2025 (Pydantic e Zod)
MistralPersonalizado Structured Output
Structured Outputs para vários modelos
Fogos de artifício AIJSON Schema
CerebrasStructured Outputs
OpenRouterDepende do fornecedor a montante, o que corresponde a JSON Schema

Mecanismos de inferência que o suportam

Nos modelos auto-hospedados, todos os principais motores dispõem de um constrained decoding backend:

MecanismoBackend
vLLMxgrammar ou orientação
Esboços, XGrammar, ou orientação LL
TensorRT-LLMDecodificação Guiada
OllamaStructured Outputs

Por que este artigo foca em vLLM e xgrammar

Algumas razões:

A secção seguinte explica de que forma o xgrammar aplica efetivamente um esquema ao nível dos tokens.


Como o xgrammar aplica os esquemas

Esta parte merece ser compreendida com precisão, uma vez que altera a forma como se efetua a depuração e o ajuste dos fluxos de trabalho SGR.

Aplicação de regras XGrammar

Onde ocorre o mascaramento

O xgrammar modifica os logits de saída após a passagem forward do modelo e antes da amostragem. Ele não altera o próprio modelo; em vez disso, filtra quais tokens podem ser selecionados.

Um ciclo de inferência padrão tem a seguinte aparência:

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

O xgrammar falha entre os passos 1 e 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

O modelo ainda calcula a sua distribuição de probabilidade completa sobre o GPU. Em seguida, o xgrammar é executado no CPU e aplica uma máscara de bits a esses logits antes da amostragem. Os tokens inválidos têm os seus logits definidos como -∞o que faz com que a sua probabilidade seja exatamente 0 após a aplicação da função softmax.

Duas fases

A xgrammar divide o processo em fase de compilação e runtime, e é precisamente isso que a torna rápida.

Fase 1: compilação da gramática, uma 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 a compilação, o xgrammar:

  1. Converte o JSON Schema numa Gramática Livre de Contexto.
  2. Constrói um Autómato de Empilhamento (PDA), que é uma máquina de estados com uma pilha para poder tratar estruturas aninhadas como {"a": {"b": {"c": ...}}}.
  3. Pré-computa quais tokens são válidos em cada posição gramatical. O resultado obtido é o “cache de máscara de token adaptativa”.
  4. Classifica os tokens como “independentes do contexto” (passíveis de armazenamento em cache) ou “dependentes do contexto” (que devem ser verificados em runtime em relação ao estado da pilha).

[!NOTA] Cerca de 99% dos tokens revelam-se independentes do contexto e acabam por ser armazenados em cache.Artigo XGrammar). A maioria das verificações de validade em runtime consiste apenas em consultas ao cache, e é por isso que o xgrammar é tão rápido.

Fase 2: geração da máscara runtime, para cada token

Em cada passo de geração:

  1. O GrammarMatcher Acompanha a posição atual na gramática.
  2. Consulta a máscara pré-calculada para os tokens independentes de contexto.
  3. Executa o PDA para verificar os tokens dependentes de contexto restantes.
  4. Combina-os numa máscara de bits final e aplica-a aos logits.

Porque automatos de deslocamento e não expressões regulares?

Devido ao encadeamento. Uma expressão regular (uma máquina de estados finitos) não consegue corresponder de forma fiável a estruturas como:

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

A parte mais difícil são os chaves de fechamento. }}}: é necessário lembrar-se de quantos foram abertos. Um Autómato de Empilhamento dispõe de uma pilha que regista este valor, permitindo assim lidar com profundidades de aninhamento arbitrárias. É também por isso que o xgrammar consegue impor tipos Union, objetos aninhados e esquemas recursivos, áreas em que as abordagens baseadas em regex não são suficientes.

Um exemplo concreto: geração de um campo de tipo float

Quando o modelo está a gerar "max_discount_percent":, o xgrammar sabe, a partir do esquema, que um float Vem a seguir. A máscara:

A passagem forward pode ter atribuído uma probabilidade elevada à palavra "fifteen". Após a máscara aplicada pelo xgrammar, esse token tem probabilidade 0. O modelo deve gerar dígitos.

Por que “overhead quase nulo”?

Três razões:

  1. Execução paralela. O cálculo da máscara em CPU sobrepõe-se à próxima passagem forward em GPU. Enquanto GPU está a calcular os logits para o token N+1, CPU está a calcular a máscara para o token N.
  2. Armazenamento em cache. A maior parte do trabalho de validação é realizada em tempo de compilação. Runtime consiste essencialmente em consultas ao cache.
  3. Implementação em C++. O caminho mais crítico é feito em C++, e não em Python, sendo que a máscara é aplicada diretamente aos logits.

Em benchmarks, o xgrammar apresenta um overhead praticamente insignificante, e a geração estruturada pode, por vezes, ser mais rápida do que a geração sem restrições, uma vez que o vocabulário limitado torna o processo de amostragem menos dispendioso.


Implementação prática com vLLM

A referência é o sgr – gestor de descontos projeto, uma pequena demonstração que utiliza SGR para definição dinâmica de preços.

Fluxo de Trabalho do Agente

Estrutura do projeto

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

Passo 1: definir os 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.")

Passo 2: um cliente LLM que ativa o 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)

[!NOTA] > guided_json aceita um dicionário JSON Schema. Com guided_decoding_backend: "xgrammar"O LLM só consegue gerar tokens que formem JSON válidos, em conformidade com o seu esquema.

Passo 3: orquestrar o 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}")

Passo 4: executar vLLM com 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

Exemplo de saída

🤖 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.

O registo de auditoria mostra o trabalho efetivo do modelo: este calculou a margem (40 dólares numa compra de 200 dólares a 20%) e definiu limites para o desconto, garantindo que a oferta permanecesse dentro dos parâmetros de lucro estabelecidos.


Boas práticas

Projeto de esquema

  1. Ordene os campos de acordo com o fluxo de raciocínio. Os campos de análise devem preceder os campos de decisão.
  2. Escreva textos descritivos. Field Descrições. Elas orientam a atenção do modelo da mesma forma que o nome do campo.
  3. Restringir com Literal e Annotated. Utilize Literal["a", "b"] para enums e Annotated[int, Ge(1), Le(10)] para delimitar os limites.
  4. Mantenha os esquemas focados. Um esquema por fase de raciocínio, e depois componha-os através de várias chamadas.

Configuração vLLM

  1. Utilize uma temperatura baixa (0,1–0,3) para obter raciocínio determinístico.
  2. Deixe que o xgrammar gere a estrutura; não tente contrariá-lo com instruções de formatação em prompt.
  3. Fique atento ao consumo de tokens. O SGR costuma utilizar menos tokens do que o CoT, pois não contém prosa excessivamente detalhada.

Considerações de produção

  1. Versione os seus esquemas da mesma forma que versiona o APIs.
  2. Mesmo com a utilização de SGR, os erros de rede e servidores continuam a exigir um tratamento adequado.
  3. Registe as saídas brutas de SGR para fins de conformidade e depuração.
  4. Faça testes com casos extremos para garantir que o esquema se mantém funcional nas fronteiras definidas.

Conclusão

SGR é o que permite passar de um cenário de “funciona em demonstração” para um ambiente de “funciona em produção”. Define a topologia de raciocínio em Pydantic, faz com que o xgrammar a aplique no momento da decodificação, e o resultado obtido é:

O sgr – gestor de descontos Teste todos os exemplos de código deste artigo conectando-os a um servidor vLLM real. Clone-o e comece a adaptar os esquemas ao seu próprio fluxo de trabalho.


Principais Conclusões

  1. O Reasoning Orientado por Esquema torna a topologia do raciocínio explícita, em vez de depender que o modelo siga instruções escritas em prosa.
  2. Constrained decoding impede a geração de JSON inválidos durante o processo de criação, o que é mais eficiente do que validar e tentar novamente posteriormente.
  3. Coloque os campos de análise antes dos campos de decisão quando o esquema exigir que o raciocínio ocorra antes da saída.
  4. Utilize SGR sempre que o código subsequente dependa de uma estrutura definida, e não quando o resultado esperado for texto em prosa livre.

Referências

SGR Framework

xgrammar

vLLM

Projeto de Demonstração