[!NOTE] Автоматический перевод Эта статья была автоматически переведена с оригинальной английской версии.

Работа с vLLM с использованием схемы-руководства Ризонинг: Structured Outputs при помощи xgrammar и Pydantic

Повторная попытка выполнения вызова LLM не гарантирует наличия корректного JSON. Следующий образец может снова дать негативный результат, а многократные вызовы приводят к увеличению латентность и дополнительным затратам.

Метод Schema-Guided Ризонинг (SGR) обеспечивает соблюдение схемы, в то время как модель генерирует каждый токен. Необходимые поля определяются с помощью Pydantic, а движок инференс блокирует токены, которые нарушают заданную структуру. Благодаря этому результат автоматически оказывается синтаксически корректным, без необходимости повторных попыток.

Кратко. SGR использует constrained decoding для фиксации вывода LLM в рамках схемы Pydantic, находящейся под вашим контролем. В сочетании с механизмом xgrammar бэкенд из vLLM это обеспечивает генерацию корректных JSON в каждом случае при практически нулевых затратах по ресурсам латентность.


Что такое руководство схемой Ризонинг?

Метод Ризонинг с использованием схемы — это техника, которая Ринат Абдуллин написано в 2024 году. Вместо того чтобы позволять модель свободно заполнять текст (что может привести к несогласованности или двусмысленности), задаётся строгий шаблон, который определяет:

Можно рассматривать это как когнитивный чек-лист, который должен соблюдаться модель.

SGR Обзор

Что контролируется схемой

Поля вроде churn_analysis, margin_math, и max_discount_percent Необходимо явно указывать ожидаемые промежуточные результаты. модель должен заполнить требуемую структуру до того, как будет принято окончательное решение относительно скидки.

Это даёт вам:


SGR против метода последовательного рассуждения против промпт-инжиниринг

Эти три подхода в основном отличаются степенью строгости ограничений, налагаемых на модель.

SGR Сравнение

ФункциональностьПромпт-инжинирингЦепочка мышленияРабота с схемой-руководителем Ризонинг
Структура выводаПеременный текстПроза свободного форматаЖёсткая интеграция JSON/Pydantic
Механизм управленияСемантическое воздействие («Пожалуйста, выведите JSON»)гипертекстуальное формулирование запросов («Давайте будем рассуждать по шагам»)Constrained decoding (основанный на грамматике)
Ризонинг ПотокМодель определяетМодель определяетРазработчик определяет (топологию схемы)
ПроверяемостьНизкий (требует парсинга)Низкий (требует прочтения текста)Высокий уровень (проверка на уровне полей)
ИнтеграцияTrивиальный (десериализация нативных объектов)
Уровень ошибокВысокая (вариативность формата)Умеренная модерация (галлюцинация формата)
Модель ТребованиеСтабильное соблюдение инструкцийВысокая производительность ризонингРаботает также с более мелкими модели.

Промпт-инжиниринг: семантическое убеждение

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!

Вы рассчитываете на то, что понимание JSON «результата вывода» у модель будет превалировать над его склонностью к разговорному стилю формулировок. Обновление модель, изменение параметра температуры или использование другого примера в формате нескольких шотов могут нарушить работу вашего парсера.

Цепочка размышлений: улучшенная ризонинг, та же проблема структуры

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 повышает точность ризонинг, но ухудшает структуру результатов. Получаемый текст имеет непредсказуемую форму, из-за чего его практически невозможно надёжно парсить. Обычно приходится выполнять второй вызов LLM, чтобы извлечь структурированные данные.

SGR: структурированная цепочка мыслей

SGR сохраняет интуицию CoT о том, что использование промежуточных ризонинг повышает точность. Он лишь формализует соответствующие этапы:

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

Этот модель не может выполнять вывод. max_discount_percent до тех пор, пока churn_analysis, financial_analysis, и margin_math Они заполняются. Схема обеспечивает соблюдение порядка ризонинг.


Паттерны SGR

SGR включает три основных шаблона, которые объединяются для формирования более крупных рабочих процессов.

SGR Шаблоны

1. Каскад: последовательные шаги ризонинг

Каскад обеспечивает соблюдение порядка ризонинг: каждое поле должно быть заполнено до перехода к следующему.

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

Хорошие сценарии применения: оценка кандидатов, классификация документов, анализ соответствия стандартам, медицинская диагностика.

Этот модель должен выполнять запись. brief_candidate_summary Прежде чем система сможет оценить качество, необходимо сначала выполнить процедуру оценки, а уже после этого — рекомендацию. Коротких путей здесь нет.


2. Роутинг: семантическое операторное условие

Роутинг обеспечивает то, что модель выбирает один конкретный путь из набора возможных вариантов, при этом реализация осуществляется с использованием 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]

Эффективно применяется для классификации намерений, выбора инструментов, сортировки запросов на поддержку и диспетчеризации multi-agent.

Этот Literal дискриминаторtool_name) заставляет модель выбрать конкретную ветвь и заполнить только те поля, которые необходимы этой ветви.


3. Цикл: многократное выполнение ризонинг с использованием списков

Цикл принудительно заставляет модель генерировать несколько элементов, при этом существуют ограничения на их количество.

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

Хорошо подходят: оценка рисков, извлечение проблем, параллельный tool calls, многократный планирование.

Этот MinLen и MaxLen Параметр bounds задаёт минимум 2 и максимум 4 элемента. В сочетании с Роутинг именно так осуществляется отправка пакета tool calls фиксированной длины.


Как заставить SGR работать: constrained decoding

Приведённые выше шаблоны представляют собой обычные схемы Pydantic. То, что делает их обязательными к выполнению, — это constrained decoding (также известный как Structured Output).

Constrained decoding влияет на этап генерации токен. Вместо того чтобы позволять образцу модель свободно брать элементы из своего словаря, движок применяет грамматическую маску, которая блокирует токены, способные нарушить установленную схему. Эта обработка происходит внутри движка инференс, а не в коде вашего приложения.

[!TIP] SGR не требует использования ризонинг модели@ вроде o1 или DeepSeek-R1. Он эффективно работает с инструкциями, настроенными для конкретных задач модели, а особенно хорошо справляется с моделями модели, полученными путём дистилляции данных из моделей ризонинг.

Провайдеры облачных сервисов, поддерживающие его

Большинство современных поставщиков LLM предоставляют structured outputs с использованием constrained decoding:

ПровайдерПоддержка
OpenAIStructured Outputs (включая Azure). GPT-5 использует JSON Schema через механизм llguidance
Google/GeminiJSON Schema поддержка с ноября 2025 г. (Pydantic и Zod)
MistralСобственная реализация Structured Output
Structured Outputs для нескольких модели
Фейерверки AIJSON Schema
CerebrasStructured Outputs
OpenRouterЗависит от поставщика нижнего уровня; соответствует JSON Schema

Инференс движки, которые его поддерживают

Для самостоятельно развернутых модели существуют у всех основных движков constrained decoding бэкенд:

двигательБэкенд
vLLMxgrammar или рекомендации
SGLangОсновные положения, XGrammar, или llguidance
TensorRT-LLMGuidedDecoding
OllamaStructured Outputs

Почему в этой статье рассматриваются vLLM и xgrammar

Несколько причин:

В следующем разделе подробно рассматривается, как именно xgrammar обеспечивает соблюдение схемы на уровне токен.


Как xgrammar обеспечивает соблюдение схем

Эту часть крайне важно понимать до мелочей, поскольку она влияет на способ отладки и настройки рабочих процессов SGR.

Обеспечение соблюдения правил xgrammar

Где происходит маскировка

xgrammar модифицирует выходные логиты после выполнения форвардного прохождения модель и перед процедурой сэмплирования. Он не изменяет сам модель, а лишь фильтрует те токены, которые могут быть выбраны.

Стандартный цикл инференс выглядит следующим образом:

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 допускает сбои при переходе между шагами 1 и 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

Модель модель по‑прежнему вычисляет своё полное распределение вероятностей на основе GPU. Затем в системе xgrammar выполняются операции на CPU, и перед генерацией выборки к логитам применяется маска битов. Для недопустимых токены их логиты устанавливаются в определённое значение. -∞, что после применения функции softmax делает их вероятность равной точно 0.

Два этапа

xgrammar разделяет обработку на этапы компиляции и рантайм, и именно это обеспечивает его высокую скорость.

Этап 1: компиляция грамматики — один раз на схему

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

Во время компиляции xgrammar:

  1. Преобразует JSON Schema в грамматику без контекста.
  2. Создаёт автомат с откладыванием (PDA), который представляет собой машина состояний с стеком, позволяющим обрабатывать вложенные структуры. {"a": {"b": {"c": ...}}}.
  3. Заранее вычисляется, какие токены допустимы в каждой позиции грамматики. Результатом является «адаптивная маска токен кэш».
  4. токены классифицируются как «независимые от контекста» (можно хранить в кэше) или «зависимые от контекста» (необходимо проверять в рантайм с учётом состояния стека).

[!NOTE] Около 99% токены оказываются независимыми от контекста и в итоге сохраняются в кэше.Статья XGrammar). Большинство проверок валидности в рантайм представляют собой просто поиски в кэш, и именно по этой причине xgrammar демонстрирует высокую скорость работы.

Этап 2: генерация маски рантайм, каждый токен

На каждом шаге генерации:

  1. Этот GrammarMatcher ! Отслеживает текущее положение в грамматике.
  2. Он запрашивает заранее вычисленную маску для контекстно-независимых токены.
  3. Он запускает алгоритм PDA для проверки оставшихся контекстно-зависимых токены.
  4. Затем эти результаты объединяются в окончательную маску битов, которая применяется к логитам.

Почему именно автоматы с допоміжним стеком, а не регулярные выражения?

Из-за вложенности. Регулярное выражение (конечный машина состояний) не может надёжно совпадать со структурами вроде:

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

Самая сложная часть — это закрывающие скобки. }}}: необходимо запоминать, сколько таких элементов было открыто. Автомат с дополнительной памятью использует стек для отслеживания этого количества, что позволяет ему обрабатывать объекты любой глубины вложенности. Именно по этой причине xgrammar способен поддерживать типы объединения, вложенные объекты и рекурсивные схемы, в то время как подходы, основанные на регулярных выражениях, оказываются недостаточными.

Конкретный пример: генерация поля с числом с плавающей точкой

Когда модель осуществляет генерацию "max_discount_percent":, xgrammar определяет на основе схемы, что а float Следующий элемент — маска:

Во время прямой передачи данных к слову могла быть присвоена высокая вероятность. "fifteen". После применения маски xgrammar вероятность значения токен равна 0. Значение модель должно представлять собой цифры.

Почему «почти нулевая нагрузка»

Три причины:

  1. Параллельная обработка. Вычисление маски для CPU происходит одновременно с следующей передачей данных по сети на GPU. Пока GPU вычисляет логиты для токен N+1, CPU формирует маску для токен N.
  2. Кэширование. Большая часть работ по определению срока действия кэша выполняется на этапе компиляции. Рантайм в основном представляет собой операции поиска в кэш.
  3. Реализация на C++. Основной путь выполнения кода — это C++, а не Python, причём маска применяется к логитам непосредственно на месте.

В бенчмарки xgrammar накладные расходы минимальны, а структурированное генерирование иногда бывает быстрее, чем неограниченное, поскольку ограниченный словарь снижает затраты на выбор экземпляров.


Практическая реализация с использованием vLLM

Ссылка представляет собой sgr — механизм управления скидками проект — небольшая демонстрационная версия, в которой для реализации динамического ценообразования используется SGR.

Рабочий процесс агента

Структура проекта

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

Шаг 1: определение схем

# 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.")

Шаг 2: клиент LLM, включающий реализацию 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 принимает словарь JSON Schema. При этом guided_decoding_backend: "xgrammar", LLM может генерировать только токены, которые соответствуют вашей схеме и представляют собой допустимые JSON.

Шаг 3: организация работы агента

# 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}")

Шаг 4: запустить vLLM с использованием 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

Пример вывода

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

Лог аудита отражает фактическую работу модель: он рассчитал маржу (40 долларов для корзины стоимостью 200 долларов при проценте 20%) и задал предел скидке, чтобы предложение оставалось в рамках требований к прибыли.


Рекомендуемые практики

Проектирование схемы

  1. Сортируйте поля в соответствии с потоком ризонинг. Поля анализа должны располагаться перед полями принятия решений.
  2. Формулируйте описания. Field Описания. Они оказывают не меньшее влияние на аттеншн объекта модель, чем сами имена полей.
  3. Ограничение с помощью Literal и Annotated. Используйте Literal["a", "b"] для перечислений и Annotated[int, Ge(1), Le(10)] для определения границ.
  4. Сохраняйте схемы лаконичными. Одна схема на каждый этап ризонинг, после чего объединяйте результаты с помощью нескольких вызовов.

Конфигурация vLLM

  1. Для обеспечения детерминированности ризонинг следует использовать низкую температуру (0,1–0,3).
  2. Пусть xgrammar самостоятельно обрабатывает структуру — не пытайтесь вмешиваться в его работу с помощью инструкций по форматированию в промпт.
  3. Следите за объёмом использования токен. Обычно SGR требует меньше токены по сравнению с CoT, поскольку в нём отсутствует избыточный описательный текст.

Аспекты, связанные с эксплуатацией в продакшене

  1. Версионируйте схемы таким же образом, как вы версионируете APIs.
  2. Даже при наличии SGR, сетевые и серверные ошибки по-прежнему требуют корректной обработки.
  3. Фиксируйте необработанные результаты работы SGR в логах для соблюдения стандартов и отладки.
  4. Проводите тестирование с использованием крайних случаев, чтобы убедиться в стабильности схемы на границах допустимых значений.

Заключение

SGR — именно это позволяет перейти от ситуации, когда решение «работает в демо-среде», к ситуации, когда оно стабильно функционирует в производственной среде. Вы определяете топологию ризонинг в Pydantic, после чего библиотека xgrammar обеспечивает её соблюдение во время декодирования, в результате чего получается следующий формат данных:

Этот sgr — механизм управления скидками В демо-версии каждый пример кода из этой статьи тестируется на реальном сервере vLLM. Клонируйте его и начните адаптировать схемы под собственный рабочий процесс.


Основные выводы

  1. Руководство схемой Ризонинг позволяет явно определить топологию ризонинг, вместо того чтобы рассчитывать на то, что модель будет следовать инструкциям, изложенным в простом тексте.
  2. Constrained decoding предотвращает появление недопустимых JSON на этапе генерации, что делает процесс более чистым по сравнению с последующей проверкой и повторной попыткой.
  3. При условии, что схема требует формирования ризонинг перед выводом, необходимо размещать поля анализа перед полями принятия решений.
  4. Использовать SGR следует тогда, когда код на последующих этапах зависит от определённой структуры, а не в случаях, когда результатом работы является свободный текст.

Список литературы

SGR Фреймворк

xgrammar

vLLM

Проект-демо