Автоматический перевод

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

Schema-Guided Reasoning в vLLM: структурированные ответы с xgrammar и Pydantic

Цикл повторных попыток — это признание. Он говорит: я ожидал, что модель вернет валидный JSON, она этого не сделала, значит, снова бросаем кости. Большая часть agent-кода для LLM, с которым я работал, сильно опирается на ретраи, и JSON все равно ломается достаточно часто, чтобы кто-то в итоге навесил на это алерт.

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

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

---

Что такое Schema-Guided Reasoning?

Schema-Guided Reasoning — это техника, которую Rinat Abdullin описал в 2024 году. Вместо того чтобы позволять модели свободно дописывать текст, что часто приводит к непоследовательности и неоднозначности, вы даете ей строгий шаблон, который определяет:

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

Думайте об этом как о когнитивном чек-листе, которому модель обязана следовать.

Почему SGR важен

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

Это дает:

• воспроизводимое рассуждение при повторных запусках
• аудируемые ответы, где можно проверить каждый шаг
• промежуточные поля, которые можно оценивать на тестовом датасете
• возможность использовать меньшие модели, потому что схема принудительно задает то, чему иначе модели пришлось бы учиться самой
• прирост точности на 5–10%, который часто наблюдают в production-нагрузках

---

SGR vs Chain of Thought vs prompt engineering

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

Feature	Prompt Engineering	Chain of Thought	Schema-Guided Reasoning
Output Structure	Переменный текст	Свободная проза	Жесткий JSON/Pydantic
Control Mechanism	Семантическое убеждение ("Please output JSON")	Эвристический prompting ("Let's think step by step")	Constrained decoding (на основе grammar)
Reasoning Flow	Определяет модель	Определяет модель	Определяет разработчик (топология схемы)
Auditability	Низкая (нужен парсинг)	Низкая (нужно читать прозу)	Высокая (проверка на уровне полей)
Integration	Сложная (regex parsing)	Сложная (переменный формат)	Тривиальная (нативная десериализация объекта)
Error Rate	Высокая (вариативность формата)	Умеренная (галлюцинации формата)	Почти нулевая (синтаксис задается engine)
Model Requirement	Сильное следование инструкциям	Сильные способности к рассуждению	Работает и с меньшими моделями

Prompt engineering: семантическое убеждение

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!

Вы надеетесь, что понимание моделью фразы "output JSON" перевесит ее склонность отвечать в разговорном стиле. Обновление модели, изменение temperature или другой few-shot пример могут сломать ваш парсер.

Chain of Thought: лучше рассуждение, та же проблема со структурой

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: структурированный chain of thought

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 есть три базовых паттерна, которые можно комбинировать в более крупные workflow.

1. Cascade: последовательные шаги рассуждения

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

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

Подходит для: оценки кандидатов, классификации документов, compliance-анализа, медицинской диагностики.

Модель обязана записать brief_candidate_summary до того, как сможет выставить оценку, и оценку — до того, как сможет рекомендовать. Срезать путь нельзя.

---

2. Routing: семантический switch statement

Routing заставляет модель выбрать один путь из набора вариантов; реализуется это через типы 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]

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

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

---

3. Cycle: повторяющееся рассуждение со списками

Cycle заставляет модель выдавать несколько элементов с ограничениями на их количество.

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

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

Ограничения MinLen и MaxLen принуждают сгенерировать минимум 2 и максимум 4 элемента. В сочетании с Routing это позволяет диспетчеризовать batch вызовов инструментов фиксированной ширины.

---

Как SGR становится рабочим: constrained decoding

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

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

[!TIP] SGR не требует "reasoning models" вроде o1 или DeepSeek-R1. Он нормально работает с instruction-tuned моделями и особенно хорошо — с моделями, дистиллированными из reasoning-моделей.

Облачные провайдеры, которые это поддерживают

Большинство современных LLM-провайдеров предлагают structured outputs через constrained decoding:

Provider	Support
OpenAI	Structured Outputs (включая Azure). GPT-5 uses JSON Schema via llguidance
Google/Gemini	Поддержка JSON Schema с Nov 2025 (Pydantic и Zod)
Mistral	Custom Structured Output
Grok	Structured Outputs для нескольких моделей
Fireworks AI	JSON Schema
Cerebras	Structured Outputs
OpenRouter	Зависит от downstream provider, маппится на JSON Schema

Inference engines, которые это поддерживают

Для self-hosted моделей все основные engines уже имеют backend для constrained decoding:

Engine	Backend
vLLM	xgrammar или guidance
SGLang	Outlines, XGrammar или llguidance
TensorRT-LLM	GuidedDecoding
Ollama	Structured Outputs

Почему в этой статье фокус на vLLM и xgrammar

Причин несколько:

• vLLM — самый широко используемый open-source inference engine для LLM, поэтому то, что вы построите здесь, легко переносится.
• xgrammar реализован на C++ и дает пренебрежимо малую дополнительную latency.
• API vLLM совместим с OpenAI, что удешевляет миграцию с облачных провайдеров.
• xgrammar умеет работать со сложными вложенными схемами, union-типами и рекурсивными структурами.

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

---

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

Эту часть стоит понимать точно, потому что от нее зависит, как вы будете отлаживать и настраивать workflow на SGR.

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

xgrammar модифицирует output logits после forward pass модели и до сэмплирования. Саму модель он не меняет; он фильтрует, какие токены вообще могут быть выбраны.

Стандартный inference loop выглядит так:

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 и применяет bitmask к этим logits перед сэмплированием. Для невалидных токенов logits устанавливаются в -∞, что делает их вероятность после softmax ровно 0.

Две фазы

xgrammar разделяет работу на compile-time и runtime, именно это и делает его быстрым.

Фаза 1: компиляция grammar, один раз на схему

# 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 в Context-Free Grammar.
2. Строит Pushdown Automaton (PDA), то есть автомат состояний со стеком, чтобы обрабатывать вложенные структуры вроде {"a": {"b": {"c": ...}}}.
3. Предварительно вычисляет, какие токены валидны в каждой позиции grammar. Результат называется "adaptive token mask cache".
4. Делит токены на "context-independent" (можно кэшировать) и "context-dependent" (нужно проверять во время выполнения по состоянию стека).

[!NOTE] Около 99% токенов оказываются context-independent и попадают в кэш (XGrammar paper). Большинство runtime-проверок валидности — это просто обращения к кэшу, поэтому xgrammar работает быстро.

Фаза 2: генерация mask во время выполнения, для каждого токена

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

1. GrammarMatcher отслеживает текущую позицию в grammar.
2. Он достает заранее вычисленную маску для context-independent токенов.
3. Запускает PDA, чтобы проверить оставшиеся context-dependent токены.
4. Собирает из этого итоговый bitmask и применяет его к logits.

Почему pushdown automata, а не regex?

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

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

Сложность в закрывающих скобках }}}: нужно помнить, сколько открывающих скобок было раньше. Pushdown Automaton хранит это в стеке, поэтому умеет обрабатывать произвольную глубину вложенности. Именно поэтому xgrammar может обеспечивать соблюдение Union-типов, вложенных объектов и рекурсивных схем, где regex-подходы не справляются.

Конкретный пример: генерация float-поля

Когда модель генерирует "max_discount_percent":, xgrammar знает из схемы, что дальше должен идти float. Маска:

• разрешает (вероятность не меняется): 0, 1, 2, ..., 9, ., -
• блокирует (вероятность становится 0): ", {, [, true, false, null и остальной словарь размером 128K+

Forward pass мог присвоить слову "fifteen" высокую вероятность. После применения mask из xgrammar вероятность этого токена становится 0. Модель вынуждена выводить цифры.

Почему "почти без overhead"

Причины три:

1. Параллельное выполнение. Вычисление mask на CPU перекрывается со следующим forward pass на GPU. Пока GPU считает logits для токена N+1, CPU считает mask для токена N.
2. Кэширование. Большая часть работы по проверке валидности делается на этапе компиляции. Во время выполнения это в основном обращения к кэшу.
3. Реализация на C++. Горячий путь написан на C++, а не на Python, и mask применяется к logits in place.

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

---

Практическая реализация с vLLM

В качестве референса используется проект sgr-discount-manager, небольшой демо-проект, который применяет 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 принимает dict с 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%) и ограничила скидку так, чтобы предложение осталось в рамках ограничения по прибыли.

---

Best practices

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

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

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

1. Используйте низкую temperature (0.1-0.3) для детерминированного рассуждения.
2. Пусть структуру задает xgrammar. Не пытайтесь дублировать это инструкциями по форматированию в prompt.
3. Следите за расходом токенов. SGR обычно использует меньше токенов, чем CoT, потому что здесь нет многословной прозы.

Что учитывать в production

1. Версионируйте схемы так же, как версионируете API.
2. Даже с SGR сетевые и серверные ошибки все равно нужно обрабатывать корректно.
3. Логируйте сырые SGR-ответы для compliance и отладки.
4. Тестируйте крайние случаи, чтобы схема удерживалась и на границах.

---

Заключение

SGR — это то, что переводит систему из состояния "работает в демо" в состояние "работает в production". Вы определяете топологию рассуждения в Pydantic, даете xgrammar принудительно обеспечить ее на этапе decode, и на выходе получаете:

• валидный результат каждый раз, без циклов ретраев и сбоев парсинга
• аудируемость на уровне полей
• возможность использовать меньшие модели, потому что им больше не нужно идеально попадать в формат самостоятельно
• более дешевый запуск, потому что вы тратите меньше токенов, меньше ретраев и можете использовать меньшие модели

Демо sgr-discount-manager связывает все примеры кода из этого поста с реальным сервером vLLM. Клонируйте его и начните адаптировать схемы под свой workflow.

---

Ссылки

Фреймворк SGR

• Schema-Guided Reasoning (SGR) — исходный фреймворк Rinat Abdullin
• SGR Patterns — паттерны Cascade, Routing, Cycle

xgrammar

• XGrammar: Flexible and Efficient Structured Generation Engine for Large Language Models — Yixin Dong et al., arXiv:2411.15100 (техническая статья с benchmark'ами)
• xgrammar GitHub — быстрая и гибкая библиотека для структурированной генерации
• xgrammar Documentation — официальная документация с quick start guide
• xgrammar Quick Start — начало работы с xgrammar
• Achieving Efficient Structured Generation with XGrammar — пост в блоге MLC о внутреннем устройстве xgrammar

vLLM

• vLLM Structured Outputs — официальная документация

Демо-проект

• sgr-discount-manager — рабочее демо со всеми примерами кода из этого поста