Taming the Poet: Enforcing Strict JSON Schemas for Enterprise Workflows

Large Language Models (LLMs) are incredible poets. They can write sonnets, summarize novels, and draft emails that sound convincingly human.

But in an enterprise workflow, we don’t need a poet.

We need a data entry clerk.

When AI becomes part of a business process—processing invoices, updating a CRM, routing support tickets—the output can’t be creative. It has to be machine-readable.

If your downstream API expects a JSON object and the model responds with:

“Here is the JSON you asked for: { … }”

your system crashes. Not because the model failed, but because the extra text breaks the parser.

This is the reliability gap. Prompt engineering (“please only output JSON”) is statistically unreliable. And at scale, “statistically unreliable” becomes “guaranteed to fail.”

So the goal isn’t to ask more politely.
The goal is to force the format.

Here’s how to implement strict JSON enforcement on Databricks Model Serving.


The Solution: Moving from “Prompts” to “Constraints”

The shift is simple: move the constraint from the prompt (text) to the inference engine (code).

On Databricks, you can use Structured Outputs to restrict what the model is allowed to generate. The model is constrained to produce valid JSON that matches your schema. It can’t “add a friendly sentence” even if it wants to.

Instead of hoping the model follows instructions, you enforce a contract at generation time.


Implementation: Forcing the Schema

When calling a model endpoint (like Llama 3 or Mixtral) on Databricks, you pass a response_format configuration that defines the output contract.

import json
from databricks.sdk import WorkspaceClient

# 1. Define the Schema (The Contract)
INVOICE_SCHEMA = {
  "type": "json_schema",
  "json_schema": {
    "name": "invoice_extraction",
    "schema": {
      "type": "object",
      "properties": {
        "invoice_id": {"type": "string"},
        "vendor": {"type": "string"},
        "total_amount": {"type": "number"},
        "items": {
          "type": "array",
          "items": {
            "type": "object",
            "properties": {
              "description": {"type": "string"},
              "quantity": {"type": "integer"},
              "unit_price": {"type": "number"}
            }
          }
        }
      }
    },
    "strict": True  # <--- The Critical Switch
  }
}

# 2. Call the Model with Enforcement
w = WorkspaceClient()
client = w.serving_endpoints.get_open_ai_client()

response = client.chat.completions.create(
    model="databricks-meta-llama-3-1-70b-instruct",
    messages=[
        {"role": "system", "content": "You are a data extraction engine."},
        {"role": "user", "content": "Extract the invoice details."}
    ],
    response_format=INVOICE_SCHEMA,
    temperature=0
)


With this setting, the model skips the pleasantries and returns raw, parseable JSON shaped exactly like your schema.


The Validation Layer: Catching “Valid” Garbage

Structured Outputs ensure the syntax is correct (valid JSON).
They don’t guarantee the data is correct (valid business logic).

A model can still produce something that parses, but makes no sense. For example:

  • wrong field names or types
  • negative prices
  • missing required values
  • totals that don’t reconcile

That’s why production systems add a second gate: validation.


The Code Contract (Pydantic)

Pydantic turns your schema into a strict contract between the AI and downstream systems. If the output violates the contract, it gets rejected before it can reach your database or ERP.

from pydantic import BaseModel, ValidationError, Field
from typing import List

# 1. Define the Strict Contract
class InvoiceItem(BaseModel):
    description: str
    quantity: int = Field(ge=1)      # Must be at least 1
    unit_price: float = Field(ge=0)  # Must be non-negative

class Invoice(BaseModel):
    invoice_id: str
    total_amount: float
    items: List[InvoiceItem]

# 2. The Validation Wrapper
def process_invoice(raw_text: str):
    # ... call LLM with response_format ...
    json_data = json.loads(llm_response)

    try:
        # 3. Strict Validation
        validated_invoice = Invoice.model_validate(json_data)
        return validated_invoice

    except ValidationError as e:
        # 4. Automatic Retry Logic
        # In production, we catch this and ask the LLM to fix its specific mistake.
        return retry_with_error_message(raw_text, error=str(e))


This pattern ensures that zero invalid records reach your ERP. The AI “fills out a form,” and your system checks the work before it moves downstream.


Bonus: Structured AI Directly in SQL

For teams that live in SQL, Databricks also supports structured enforcement directly in ai_query. That means data engineers can run extraction pipelines at scale without writing Python.

SELECT ai_query(
  'databricks-meta-llama-3-1-70b-instruct',
  'Extract invoice fields from: ' || invoice_text,
  responseFormat => '{"type":"json_object"}'
) AS clean_json
FROM raw_invoices_table;


This democratizes reliability. Analysts and engineers can build structured workflows using familiar SQL primitives.


Managerial Takeaway: Discipline Over Creativity

If you want AI to integrate with legacy IT, treat it like reliability engineering.

  • Stop negotiating. Don’t plead in the prompt—use response_format to enforce machine-readable output.
  • Validate at the gate. Use Pydantic to enforce types and constraints before anything hits your systems.
  • Fail gracefully. Catch validation errors and retry intentionally, instead of passing bad data downstream.

When you tame the poet, you turn Generative AI from a novelty into a dependable component of your enterprise architecture.


Comments

Popular posts from this blog

10 Rules for Professional GenAI Engineering on Databricks

The "CFO-Approved" Deployment: Embedding FinOps into Your CI/CD Pipeline

Zero-Trust RAG: The C-Suite Guide to Secure Multi-Tenant AI | Everstone AI