Schema-Guided Reasoning: vLLM, XGrammar und Pydantic

Automatische Übersetzung Dieser Artikel wurde automatisch aus der englischen Originalversion übersetzt.

Dieser Artikel richtet sich an Python Engineers, die Model-Ausgaben benötigen, die nachgelagerter Code validieren kann. Sie lernen, wie Sie ein Pydantic-Schema definieren, Structured Output von vLLM anfordern und zusätzliche Prüfungen für Semantik und Richtlinien in der Anwendung implementieren.

Ein erneuter LLM-Aufruf garantiert kein valides JSON. Das nächste Sample kann auf dieselbe Weise fehlschlagen, und wiederholte Aufrufe erhöhen Latency und Kosten.

Schema-Guided Reasoning (SGR) erzwingt ein Schema, während das Model jeden Token generiert. Sie definieren die erforderlichen Felder mit Pydantic, und die Inference Engine blockiert Tokens, die diese Struktur verletzen würden. Das Ergebnis ist konstruktionsbedingt syntaktisch valide und nicht erst nach einem Retry.

TL;DR. SGR nutzt Constrained Decoding, um die Ausgabe eines LLM innerhalb eines Pydantic-Schemas zu halten. vLLM kann XGrammar verwenden, um die generierte Struktur einzuschränken. Semantische Regeln sollten Sie im Anwendungscode validieren.


Was ist Schema-Guided Reasoning?

Schema-Guided Reasoning ist eine Technik, die Rinat Abdullin im Juli 2025 beschrieben hat. Statt das Model Text frei vervollständigen zu lassen – was inkonsistent oder mehrdeutig sein kann –, geben Sie ihm ein striktes Template vor, das definiert:

  • welche Schritte die Antwort darstellen soll
  • die vorgesehene Reihenfolge dieser Schritte, sodass ein Reviewer den Weg von den Daten zur Entscheidung prüfen kann
  • worauf das Model seine Aufmerksamkeit richten soll

Betrachten Sie es als kognitive Checkliste, der das Model folgen muss.

SGR OverviewSGR Overview

Was das Schema steuert

Felder wie churn_analysis, margin_math und max_discount_percent machen die vorgesehenen Intermediate Outputs explizit. Ein Schema schränkt die zurückgegebene Form ein. Es sorgt nicht automatisch dafür, dass ein Feld von einem anderen abhängt, und beweist auch nicht, dass die Entscheidung korrekt ist.

Das bietet Ihnen:

  • reproduzierbares Reasoning über wiederholte Runs hinweg
  • auditierbare Outputs, bei denen jeder Schritt inspiziert werden kann
  • Intermediate Fields, die Sie gegen ein Test-Dataset bewerten können
  • kleinere Models, die einsetzbar werden, weil das Schema die Struktur liefert, die das Model sonst lernen müsste
  • Abdullin schreibt, dass eine Genauigkeitssteigerung von 5–10 % in den von ihm beobachteten Fällen „nicht ungewöhnlich“ sei; dies ist eine Beobachtung aus der Praxis, kein Benchmark-Ergebnis – messen Sie daher mit Ihrer eigenen Workload

SGR vs. Chain-of-Thought vs. Prompt Engineering

Die drei Ansätze unterscheiden sich vor allem darin, wie stark sie das Model einschränken.

SGR ComparisonSGR Comparison

FeaturePrompt EngineeringChain-of-ThoughtSchema-Guided Reasoning
Output StructureVariabler TextFreie ProsaStarres JSON/Pydantic
Control MechanismSemantische Überredung („Please output JSON“)Heuristisches Prompting („Let’s think step by step“)Constrained Decoding (grammatikbasiert)
Reasoning FlowVom Model bestimmtVom Model bestimmtDer Developer beschreibt eine vorgesehene Topology
AuditabilityGering (Parsing erforderlich)Gering (Lesen der Prosa erforderlich)Hoch (Prüfung auf Feldebene)
IntegrationSchwierig (Regex-Parsing)Schwierig (variables Format)Erfordert Schema-Support und Validation
Error RateHoch (Formatvariabilität)Moderat (Hallucination des Formats)Schema-invalid Output wird blockiert; semantische Fehler bleiben
Model RequirementStarkes Instruction FollowingStarkes Reasoning-FähigkeitFunktioniert auch mit kleineren Models

Prompt Engineering: semantische Überredung

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!

Sie hoffen darauf, dass das Verständnis des Models von „output JSON“ stärker ist als seine Tendenz, konversationell zu antworten. Ein Model-Update, eine Änderung der Temperature oder ein anderes Few-Shot-Beispiel kann Ihren Parser unbrauchbar machen.

Chain-of-Thought: nützlicher Reasoning Trace, dasselbe Strukturproblem

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 kann die Task-Genauigkeit verbessern, wenn Prompt, Model, Task und Evaluation dies unterstützen. Das Ergebnis bleibt jedoch Prosa, die sich nur schwer zuverlässig parsen lässt. Möglicherweise benötigen Sie einen zweiten LLM-Aufruf allein zur Extraktion strukturierter Daten.

SGR: Structured Chain-of-Thought

SGR kann Intermediate Fields zur Inspektion und Evaluation verfügbar machen. Ob dadurch die Task-Genauigkeit steigt, hängt vom Model, Prompt, Task und der Verwendung dieser Felder ab; das Schema formalisiert lediglich ihre Form:

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

Das Schema beschreibt diese Felder in dieser Reihenfolge. Ein einzelnes Object-Schema erzeugt keinen separaten Validation-Schritt zwischen ihnen. Verwenden Sie separate Aufrufe oder Application Checks, wenn spätere Entscheidungen von früheren Ergebnissen abhängen müssen.


SGR Patterns

SGR umfasst drei zentrale Patterns, die sich zu größeren Workflows kombinieren lassen.

SGR PatternsSGR Patterns

1. Cascade: sequenzielle Reasoning-Schritte

Cascade stellt eine Reasoning-Reihenfolge in einer strukturierten Antwort dar. Es erzwingt keinen State Transition zwischen Feldern.

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

Geeignet für: Candidate Evaluation, Document Classification, Compliance Analysis und Medical Diagnosis.

Das Model soll brief_candidate_summary, rate_skill_match und final_recommendation in dieser Reihenfolge zurückgeben. Wenn die Reihenfolge eine Policy-Anforderung ist, erzwingen Sie sie mit separaten Aufrufen oder deterministischer Application Logic.


2. Routing: ein semantisches Switch Statement

Routing verpflichtet das Model zu einem Pfad aus einer Gruppe von Optionen, implementiert mit Union-Types.

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]

Geeignet für: Intent Classification, Tool Selection, Support Triage und Multi-Agent Dispatch.

Die branchenspezifischen Literal-Werte helfen der Validation, die Union Members zu unterscheiden. Sie machen das Routing nicht korrekt. Validieren Sie das Ergebnis und führen Sie den Dispatch im Anwendungscode aus. Für einen expliziten Pydantic Discriminator konfigurieren und testen Sie eine Discriminated Union.


3. Cycle: wiederholtes Reasoning mit Listen

Cycle zwingt das Model, mehrere Items mit einer Begrenzung ihrer Anzahl zu erzeugen.

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

Geeignet für: Risk Assessment, Issue Extraction, parallele Tool Calls und Multi-Step Planning.

Die Grenzen MinLen und MaxLen erzwingen mindestens 2 und höchstens 4 Items. In Kombination mit Routing können Sie so einen Batch mit fester Breite von Tool Calls dispatchen.


SGR funktionsfähig machen: Constrained Decoding

Die oben beschriebenen Patterns sind lediglich Pydantic-Schemas. Verbindlich werden sie durch Constrained Decoding, auch Structured Output genannt.

Constrained Decoding verändert den Schritt der Token-Generierung. Statt das Model frei aus seinem Vocabulary samplen zu lassen, wendet die Engine eine Grammar Mask an, die Tokens blockiert, welche das Schema verletzen würden. Dies geschieht in der Inference Engine, nicht in Ihrem Anwendungscode.

[!TIP] SGR erfordert keine „Reasoning Models“ wie o1 oder DeepSeek-R1. Es funktioniert problemlos mit Instruction-Tuned Models und besonders gut mit Models, die aus Reasoning Models distilliert wurden.

Cloud Provider mit Support

Die folgende Provider-Dokumentation beworb zum Zeitpunkt der Aktualisierung dieses Artikels am 2026-06-07 Support für Structured Output. Support, unterstützte Schema-Subsets, Strictness und Failure Behavior unterscheiden sich je nach Model und Endpoint:

ProviderSupport
OpenAIStructured Outputs (einschließlich Azure)
Google/GeminiJSON Schema-Support seit November 2025 (Pydantic und Zod)
MistralCustom Structured Output
GrokStructured Outputs für mehrere Models
Fireworks AIJSON Schema
CerebrasStructured Outputs
OpenRouterHängt vom ausgewählten nachgelagerten Provider und Route ab

Inference Engines mit Support

Für selbst gehostete Models verfügen die großen Engines über ein Backend für Constrained Decoding:

EngineBackend
vLLMxgrammar oder guidance
SGLangOutlines, XGrammar oder llguidance
TensorRT-LLMGuidedDecoding
OllamaStructured Outputs

Warum sich dieser Artikel auf vLLM und XGrammar konzentriert

Dafür gibt es mehrere Gründe:

  • vLLM ist eine der am weitesten verbreiteten Open-Source-LLM-Inference-Engines, sodass sich das hier Erstellte leicht portieren lässt.
  • XGrammar ist in C++ implementiert, und die zitierten Benchmarks messen den Overhead unter spezifischen Bedingungen. Messen Sie mit Ihrem Model, Schema, Ihrer Hardware und Ihrer Serving-Konfiguration.
  • Die vLLM-API ist OpenAI-kompatibel, wodurch die Migration von Cloud Providern kostengünstig bleibt.
  • XGrammar verarbeitet komplexe verschachtelte Schemas, Unions und rekursive Strukturen.

Wie XGrammar Schemas erzwingt

xgrammar Enforcementxgrammar Enforcement

Wo das Masking stattfindet

XGrammar verändert die Output Logits nach dem Forward Pass des Models und vor dem Sampling. Das Model selbst wird nicht verändert. XGrammar filtert, welche Tokens ausgewählt werden können.

Ein standardmäßiger Inference Loop sieht so aus:

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 wird zwischen Schritt 1 und 2 eingeschoben:

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

Das Model berechnet weiterhin seine vollständige Probability Distribution auf der GPU. XGrammars GrammarMatcher erzeugt die Bitmask auf der CPU. Danach verschiebt die Serving Engine diese Bitmask auf das Device der Logits und wendet sie vor dem Sampling in place an. Bei GPU-Logits verwendet XGrammar dafür einen GPU-Kernel. Die Logits ungültiger Tokens werden auf -∞ gesetzt, wodurch ihre Probability nach Softmax exakt 0 beträgt.

Zwei Phasen

XGrammar teilt die Arbeit in Compile Time und Runtime auf. Dieses Design reduziert wiederholte Grammar-Arbeit.

Phase 1: Grammar Compilation, einmal pro Schema

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

Während der Compilation führt XGrammar folgende Schritte aus:

  1. Konvertiert das JSON Schema in eine Context-Free Grammar.
  2. Erstellt einen Pushdown Automaton (PDA), eine State Machine mit einem Stack, sodass verschachtelte Strukturen wie {"a": {"b": {"c": ...}}} verarbeitet werden können.
  3. Berechnet vorab, welche Tokens an jeder Grammar-Position valide sind. Das Ergebnis ist der „Adaptive Token Mask Cache“.
  4. Kategorisiert Tokens als „context-independent“ (cachebar) oder „context-dependent“ (müssen zur Runtime anhand des Stack-Zustands geprüft werden).

[!NOTE] Das XGrammar-Paper berichtet, dass in seinen Messungen etwa 99 % der Tokens context-independent waren (Paper). Betrachten Sie diese Zahl als Benchmark-spezifisch und nicht als universelles Verhältnis.

Phase 2: Runtime Mask Generation, bei jedem Token

Bei jedem Generation Step:

  1. Der GrammarMatcher verfolgt die aktuelle Position in der Grammar.
  2. Er schlägt die vorberechnete Mask für context-independent Tokens nach.
  3. Er führt den PDA aus, um die verbleibenden context-dependent Tokens zu prüfen.
  4. Er kombiniert beide zu einer finalen Bitmask, verschiebt sie auf das Device der Logits und wendet sie dort an.

Warum Pushdown Automata und keine Regex?

Wegen der Verschachtelung. Ein regulärer Ausdruck (eine Finite State Machine) kann Strukturen wie diese nicht zuverlässig matchen:

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

Die schwierige Stelle sind die schließenden Klammern }}}: Sie müssen sich merken, wie viele öffnende Klammern es gab. Ein Pushdown Automaton besitzt einen Stack, der dies verfolgt, und kann daher beliebige Verschachtelungstiefen verarbeiten. Deshalb kann XGrammar Union Types, verschachtelte Objekte und rekursive Schemas erzwingen, während Regex-basierte Ansätze hier an ihre Grenzen stoßen.

Ein konkretes Beispiel: Generierung eines Float-Felds

Wenn das Model "max_discount_percent": generiert, weiß XGrammar aus dem Schema, dass als Nächstes ein float folgt. Die Menge valider Tokens hängt vom Parser-Zustand und vom Tokenizer ab. Am Anfang der Zahl kann die Mask eine Ziffer oder ein Minuszeichen zulassen; nach einer Ziffer sind Fortsetzungen wie eine weitere Ziffer, ein Dezimalpunkt oder ein Exponent-Marker möglich.

  • Ein Anführungszeichen, {, [, true, false oder null kann diese Zahl nicht beginnen; die Mask blockiert deren Tokens in diesem Zustand.
  • Der Forward Pass kann dem Token für "fifteen" eine hohe Probability zugewiesen haben. Da dieses Token das numerische Feld nicht fortsetzen kann, entfernt die Mask es, und das Model muss eine valide numerische Fortsetzung auswählen.

Was den Overhead beeinflusst

Drei Gründe sind ausschlaggebend:

  1. Mask Generation und Transfer. XGrammar erzeugt Masks auf der CPU, und die Serving Engine überträgt sie zur In-Place-Anwendung auf das Device der Logits. Wie stark dies überlappt, hängt von der Serving-Implementierung und der Workload ab.
  2. Caching. Der größte Teil der Validitätsprüfung erfolgt zur Compile Time. Zur Runtime finden hauptsächlich Cache Lookups statt.
  3. C++-Implementierung. Der Hot Path ist in C++ und nicht in Python implementiert; die Mask wird in place auf die Logits angewendet.

Die zitierten Benchmarks berichten für die getesteten Grammars, Tokenizers, Hardware-Konfigurationen und Workloads einen geringen Overhead. Diese Ergebnisse stellen keine allgemeine Garantie für Latency oder Throughput dar.


Praktische Implementierung mit vLLM

Das Projekt sgr-discount-manager ist eine illustrative externe Demo. Dieser Artikel pinnt weder dessen Commit noch behauptet er, dass die folgenden Snippets in diesem Repository ausgeführt wurden.

Agent WorkflowAgent Workflow

Projektstruktur

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

Schritt 1: Schemas definieren

# 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):
    """
    Structured response for dynamic pricing. The fields record an intended 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. Model-proposed decision; application code approves it.
    max_discount_percent: float = Field(...,
        description="Proposed discount percentage. Application code enforces policy.")

Schritt 2: ein LLM-Client, der XGrammar aktiviert

# sgr/utils/llm_client.py
import json
from typing import TypeVar

from openai import OpenAI
from pydantic import BaseModel

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"):
        # Local vLLM commonly has no authentication. EMPTY is not authentication.
        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 structured outputs to constrain the JSON shape at generation time.
        """
        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 v0.12+ structured outputs. Configure the backend on the server.
        completion = self.client.chat.completions.create(
            model=self.model,
            messages=enhanced_messages,
            temperature=0.1,  # Reduces sampling variation; it is not deterministic.
            extra_body={"structured_outputs": {"json": schema_dict}},
        )

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

[!NOTE] Im aktuellen vLLM fordert structured_outputs: {"json": schema_dict} JSON an, das dem Schema entspricht. Konfigurieren Sie bei Bedarf das Structured-Output-Backend mit der --structured-outputs-config.backend-Option des Servers. Dies ist eine softwarebasierte Durchsetzung in der Inference Server und keine hardwarebasierte Durchsetzung.

Schritt 3: den Agent orchestrieren

# sgr/agent.py
from decimal import Decimal

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 approve_discount(offer: PricingLogic, context: dict) -> Decimal:
    """Enforce the pricing policy independently of the model's explanation."""
    cart_value = Decimal(str(context["current_cart_value"]))
    margin = Decimal(str(context["cart_profit_margin"]))
    proposed = Decimal(str(offer.max_discount_percent))

    if cart_value <= 0 or not Decimal("0") <= margin <= Decimal("1"):
        raise ValueError("Invalid pricing context")

    gross_profit = cart_value * margin
    discount_cost = cart_value * proposed / Decimal("100")
    policy_cap = min(margin * Decimal("100"), Decimal("20"))
    if not Decimal("0") <= proposed <= policy_cap or discount_cost > gross_profit:
        raise ValueError("Proposed discount violates pricing policy")

    return proposed.quantize(Decimal("0.01"))

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: model proposal, then deterministic policy enforcement ---
        print("🧠 Proposing Offer (Schema Enforced)...")
        offer = llm.run_sgr(history, PricingLogic)
        approved_discount = approve_discount(offer, context)

        # Audit log: reasoning is inspectable; pricing is application-enforced.
        print(f"   [Audit] Math: {offer.margin_math}")
        print(f"   [Audit] Approved Discount: {approved_discount}%")

        return (
            "We value your loyalty! Here's a special "
            f"{approved_discount}% discount with code SAVE{approved_discount:.0f}."
        )

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

Schritt 4: vLLM mit XGrammar ausführen

# Start vLLM server with XGrammar backend
vllm serve \
    --model Qwen/Qwen2.5-7B-Instruct \
    --port 8000 \
    --structured-outputs-config.backend xgrammar

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

Beispielhafte Ausgabe

🤖 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%
🧠 Proposing Offer (Schema Enforced)...
   [Audit] Math: Cart $200 * 0.20 Margin = $40
   [Audit] Approved Discount: 15.00%

💬 Final Reply: We value your loyalty! Here's a special 15.00% discount
   with code SAVE15.

Diese Ausgabe dient der Illustration. Das Schema schränkt die Form ein, aber die Anwendung muss vor der Verwendung des Angebots prüfen, ob Arithmetik und Discount Policy korrekt sind.


Schema, vLLM und Production Checklist

Schema Design

  1. Ordnen Sie Felder entsprechend dem vorgesehenen Reasoning Flow an. Behandeln Sie diese Reihenfolge als Dokumentation, sofern separate Aufrufe oder Application Checks sie nicht erzwingen.
  2. Verfassen Sie beschreibende Field-Descriptions. Sie lenken die Aufmerksamkeit des Models ebenso stark wie der Field Name.
  3. Schränken Sie mit Literal und Annotated ein. Verwenden Sie Literal["a", "b"] für Enums und Annotated[int, Ge(1), Le(10)] für Bounds.
  4. Halten Sie Schemas fokussiert. Verwenden Sie ein Schema pro Reasoning Phase und kombinieren Sie diese anschließend mit mehreren Aufrufen.

vLLM-Konfiguration

  1. Verwenden Sie eine niedrige Temperature (0,1–0,3), um Sampling-Variationen zu reduzieren. Für reproduzierbare Tests fixieren Sie Model und Serving-Konfiguration, verwenden Sie den deterministischen Modus des Servers, sofern vorhanden, und verifizieren Sie das Ergebnis.
  2. Überlassen Sie die Struktur XGrammar. Versuchen Sie nicht, XGrammar mit Formatting Instructions im Prompt zu umgehen.
  3. Messen Sie den Token-Verbrauch mit demselben Prompt, Model, Schema und Task. SGR kann mehr oder weniger Tokens ausgeben als eine freie CoT-Antwort.

Überlegungen für Production

  1. Versionieren Sie Ihre Schemas genauso wie Ihre APIs.
  2. Auch bei SGR müssen Netzwerk- und Serverfehler Graceful Handling erhalten.
  3. Loggen Sie rohe SGR-Outputs für Compliance und Debugging und anschließend die deterministische Policy-Entscheidung separat.
  4. Berechnen Sie Preise, Limits, Berechtigungen und andere semantische Regeln vor der Verwendung erneut im Anwendungscode; testen Sie Grenzwerte, die das Schema allein nicht ablehnen kann.

Fazit

Sie beschreiben eine Reasoning Topology in Pydantic und lassen Constrained Decoding die Output-Form erzwingen. Das Ergebnis ist:

  • konstruktionsbedingt syntaktisch valide, wodurch der Retry-and-Reparse-Loop entfällt
  • auf Feldebene auditierbar
  • mit kleineren Models nutzbar, weil diese das Format nicht mehr allein korrekt treffen müssen
  • potenziell günstiger im Betrieb, wenn Validation Retries reduziert oder das gewählte Schema und Model weniger Output Tokens verwenden

Die sgr-discount-manager-Demo ist ein externer Ausgangspunkt. Prüfen Sie die gepinnten Dependencies und die aktuelle vLLM-Kompatibilität, bevor Sie sie als ausführbare Referenz behandeln.


Wichtigste Erkenntnisse

  1. Schema-Guided Reasoning macht eine vorgesehene Reasoning Topology explizit, statt sich ausschließlich auf Prose Instructions zu verlassen.
  2. Constrained Decoding verhindert bereits während der Generierung ungültiges JSON und ist damit sauberer, als anschließend zu validieren und zu retrien.
  3. Platzieren Sie Analysis Fields vor Decision Fields, wenn die Antwort diesen Reasoning Path dokumentieren soll. Verwenden Sie separate Aufrufe oder Application Checks, wenn die Reihenfolge erzwungen werden muss.
  4. Verwenden Sie SGR, wenn nachgelagerter Code von der Struktur abhängt, nicht wenn freie Prosa das eigentliche Produkt ist.

Referenzen

SGR Framework

XGrammar

vLLM

Demo Project

  • sgr-discount-manager – ältere illustrative Demo; sie enthält nicht jedes Codebeispiel dieses Posts