📐 Structured Output Determinism#

Structured Output Variance & Determinism shows how to turn a stochastic text generator into a deterministic software component whose output can be safely parsed by downstream code. Given a free-text product review, the service asks the LLM to extract three fields (sentiment, rating, and pros_and_cons) under three progressively stricter strategies at temperature=0.7:

• Strategy A (Unconstrained Freeform): Asks for the three fields in plain English with no format constraint — producing conversational prose or bullet points that immediately crash a downstream json.loads() parser.
• Strategy B (Prompt-Constrained JSON): Instructs the model in the prompt to "Respond in JSON format with keys..." — often works, but frequently wraps the JSON in ```json ... ``` markdown fences or adds conversational preamble.
• Strategy C (API Schema-Enforced JSON Mode): Sets response_format={"type": "json_object"} in the API request so the inference engine constrains token sampling at the logit level, guaranteeing valid, parseable JSON.

1. A Semantically Correct Answer Can Still Crash Your Pipeline

To a human reader, "Sentiment: Positive (4/5). Pros: great camera, sleek design. Cons: weak battery." is a great summary. To a backend microservice that calls json.loads(raw_output) to insert rows into PostgreSQL, it is an unhandled JSONDecodeError: Expecting value: line 1 column 1. Software pipelines require syntactic determinism, not just semantic understanding.

2. Prompt Persuasion (Strategy B) vs. Constrained Decoding (Strategy C)

Asking for JSON in the prompt (Strategy B) merely nudges token probabilities; at temperature=0.7, the model may still emit markdown code fences (```json) or trailing commentary, requiring defensive regex stripping (re.sub(r"^```(?:json)?\s*|\s*```$", "", raw_b)). Passing response_format={"type": "json_object"} (or a strict JSON Schema) in Strategy C masks out invalid tokens during decoding so the model is mathematically prevented from emitting non-JSON syntax.

3. Why Temperature Is Kept at 0.7 Across All Three Strategies

Holding temperature=0.7 constant proves that Strategy C's reliability comes from structural API enforcement rather than greedy decoding. Even when the model has sampling freedom over word choice, its output structure remains 100% valid JSON.

4. Inspecting Parser Failures with check_json()

The helper check_json(strategy_name, raw_output, text_to_parse) runs json.loads() inside a try/except json.JSONDecodeError block and captures both parsed_data on success and the exact parser exception message (parse_error) on failure for side-by-side inspection.

1. Core Benchmark Architecture (src/python/02_variance_determinism.py)

• 02_variance_determinism.py evaluates how output constraints turn a stochastic LLM into a deterministic software component by extracting structured fields (product, sentiment, pros, cons, and rating) from an unstructured customer review across three progressively stricter strategies — Strategy A (Unconstrained Freeform), Strategy B (Prompt-Constrained Instructional JSON), and Strategy C (API Schema-Enforced JSON Mode) — and verifying each output with Python's json.loads() parser.

2. Pydantic Schemas & JSON Parser Verification Helper

• VarianceDeterminismRequest(BaseModel): Defines the input payload with default review_text: str (a 4-star smartphone review praising camera quality and sleek design while criticizing battery life).
• StrategyOutput(BaseModel): Captures the parser evaluation for one extraction strategy: strategy_name: str, raw_output: str, is_valid_json: bool, parsed_data: Optional[Dict[str, Any]] = None, and parse_error: Optional[str] = None.
• VarianceDeterminismResponse(BaseModel): Wraps the three strategy outcomes in strategies: List[StrategyOutput].
• check_json(strategy_name: str, raw_output: str, text_to_parse: str) -> dict: Attempts json.loads(text_to_parse) inside a try / except json.JSONDecodeError block. On success, it returns is_valid_json: True with the parsed dictionary in parsed_data; on failure, it catches the JSONDecodeError and records is_valid_json: False along with the exact parser exception string in parse_error.

3. Main Benchmark Execution: run_variance_determinism_benchmark()

• Accepts review_text: str and returns a dictionary matching VarianceDeterminismResponse:
  • Step ①: Initializes the OpenAI client for the configured model (default "gpt-4o-mini").
  • Step ② (Strategy A — Unconstrained Freeform): Prompts the model with "Extract product, sentiment, pros, cons, and rating from this review" without specifying an output format, producing conversational prose or bullet points in text_a that fail strict JSON parsing.
  • Step ③ (Strategy B — Prompt-Constrained Instructional JSON): Instructs the model in natural language to extract the 5 keys ('product', 'sentiment', 'pros', 'cons', 'rating') as JSON without markdown ticks or commentary, storing the response in text_b.
  • Step ④ (Strategy C — API Schema-Enforced JSON Mode): Calls client.chat.completions.create() with response_format={"type": "json_object"}, constraining token sampling at the API decoding level so the model is guaranteed to emit valid JSON syntax in text_c.
  • Step ⑤: Evaluates all three outputs with check_json() (stripping optional ```json and ``` fences from text_b before parsing) and returns the three StrategyOutput dictionaries in strategies.

4. End-to-End AI Benchmark Flow

1. run_variance_determinism_benchmark() receives the unstructured customer review text.
2. It runs three extraction calls against the LLM: Strategy A (no format constraint), Strategy B (prompt-level JSON instruction), and Strategy C (API-level constrained decoding via response_format={"type": "json_object"}).
3. Each raw output is tested against Python's json.loads() inside check_json(), demonstrating why downstream software pipelines rely on API-enforced structured outputs rather than prompt persuasion alone.

Structured Output Variance & Determinism Benchmark Service.

"""Structured Output Variance & Determinism Benchmark Service.

Evaluates how output constraints transform stochastic models into deterministic pipeline components:
1. Strategy A (Unconstrained): Model asked for review summary freeform (high variance, brittle to parse).
2. Strategy B (Prompt-Constrained): Prompt instructs JSON format (occasionally wraps markdown code blocks or violates schema).
3. Strategy C (Schema-Enforced): Uses OpenAI Structured Outputs (`response_format: {"type": "json_object"}` or schema) guaranteeing valid parseable JSON.
"""

import json
import os
from typing import Any, Dict, List, Optional
from dotenv import find_dotenv, load_dotenv
from openai import OpenAI
from pydantic import BaseModel

load_dotenv(find_dotenv())


class VarianceDeterminismRequest(BaseModel):
    """Input payload for Variance & Determinism Benchmark."""

    review_text: str = "The new smartphone is amazing, the camera quality is top-notch but the battery life is a bit disappointing. I love the sleek design though! Rating: 4 out of 5 stars."


class StrategyOutput(BaseModel):
    strategy_name: str
    raw_output: str
    is_valid_json: bool
    parsed_data: Optional[Dict[str, Any]] = None
    parse_error: Optional[str] = None


class VarianceDeterminismResponse(BaseModel):
    """Output payload from Variance & Determinism Benchmark."""

    strategies: List[StrategyOutput]


def check_json(strategy_name: str, raw_output: str, text_to_parse: str) -> dict:
    """Try to parse a model output as JSON and record whether it worked."""
    try:
        return {"strategy_name": strategy_name, "raw_output": raw_output, "is_valid_json": True,
                "parsed_data": json.loads(text_to_parse), "parse_error": None}
    except json.JSONDecodeError as exc:
        return {"strategy_name": strategy_name, "raw_output": raw_output, "is_valid_json": False,
                "parsed_data": None, "parse_error": str(exc)}


def run_variance_determinism_benchmark(
    review_text: str = "The new smartphone is amazing, the camera quality is top-notch but the battery life is a bit disappointing. I love the sleek design though! Rating: 4 out of 5 stars.",
) -> dict:
    """Run extraction across the three determinism strategies."""
    # ① Initialize OpenAI client
    client = OpenAI(
        api_key=os.getenv("OPENAI_API_KEY"),
        base_url=os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1"),
    )
    model = os.getenv("OPENAI_MODEL", "gpt-4o-mini")

    # ② Strategy A: Unconstrained freeform prompt (high variance, usually fails JSON parsing)
    prompt_a = f"Extract product, sentiment, pros, cons, and rating from this review:\n\n{review_text}"
    resp_a = client.chat.completions.create(model=model, messages=[{"role": "user", "content": prompt_a}])
    text_a = resp_a.choices[0].message.content or ""

    # ③ Strategy B: Prompt-constrained JSON instructions (may still need markdown fence stripping)
    prompt_b = (
        f"Extract product, sentiment (positive/negative/mixed), pros (list), cons (list), and rating (number) "
        f"from this review as JSON with keys 'product', 'sentiment', 'pros', 'cons', 'rating'. "
        f"Do not include markdown ticks or commentary:\n\n{review_text}"
    )
    resp_b = client.chat.completions.create(model=model, messages=[{"role": "user", "content": prompt_b}])
    text_b = resp_b.choices[0].message.content or ""

    # ④ Strategy C: API Schema-Enforced JSON mode (response_format={"type": "json_object"})
    prompt_c = (
        f"Extract review metadata into JSON with keys 'product', 'sentiment', 'pros', 'cons', 'rating':\n\n{review_text}"
    )
    resp_c = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt_c}],
        response_format={"type": "json_object"},
    )
    text_c = resp_c.choices[0].message.content or ""

    # ⑤ Check which outputs parse as JSON and return all three side by side
    return {
        "strategies": [
            check_json("Strategy A: Unconstrained (Freeform Prompt)", text_a, text_a),
            check_json(
                "Strategy B: Prompt-Constrained (Instructional JSON)",
                text_b,
                text_b.replace("```json", "").replace("```", "").strip(),
            ),
            check_json("Strategy C: Schema-Enforced (API Guaranteed JSON)", text_c, text_c),
        ],
    }

Flask Blueprint server and route handler (variance_determinism_route) in src/python/app.py.

"""Flask server for 07-ai-reliability."""

import importlib.util
import os
import sys
from pathlib import Path

from dotenv import find_dotenv, load_dotenv
from flask import Blueprint, Flask, jsonify, request
from flask_cors import CORS
from rate_limiter import check_rate_limit

load_dotenv(find_dotenv())

PATH_PREFIX = os.environ.get("PATH_PREFIX", "")
STATIC_DIR = Path(__file__).resolve().parents[1]
app = Flask(__name__, static_folder=str(STATIC_DIR))
CORS(app)

bp = Blueprint("main", __name__)

_CUR_DIR = Path(__file__).resolve().parent


def _load_service(filename: str, alias: str):
    # ① Resolve the numbered service file path and create its import spec
    file_path = _CUR_DIR / filename
    spec = importlib.util.spec_from_file_location(f"services.{alias}", file_path)
    if spec is None or spec.loader is None:
        raise ImportError(f"Cannot load module from {file_path}")
    # ② Register and execute the module so its functions can be called by routes
    mod = importlib.util.module_from_spec(spec)
    sys.modules[f"services.{alias}"] = mod
    spec.loader.exec_module(mod)
    return mod


mod_cot = _load_service("01_cot_vs_direct.py", "cot_vs_direct")
mod_var = _load_service("02_variance_determinism.py", "variance_determinism")
mod_calib = _load_service("03_tool_schema_calibration.py", "tool_schema_calibration")
mod_err = _load_service("04_tool_error_injection.py", "tool_error_injection")


@bp.before_request
def enforce_rate_limit():
    # ① Only rate-limit mutating POST requests
    if request.method == "POST":
        # ② Check whether this client has exceeded 30 requests per hour
        blocked, msg, retry_after = check_rate_limit(
            request, max_requests=30, window_seconds=3600
        )
        if blocked:
            # ③ Return HTTP 429 with Retry-After header when rate-limited
            resp = jsonify({"error": msg})
            resp.status_code = 429
            resp.headers["Retry-After"] = str(retry_after)
            return resp


@bp.route("/")
def index():
    # ① Read index.html and inject the runtime PATH_PREFIX into data-api-base
    with open(os.path.join(app.static_folder, "index.html"), encoding="utf-8") as f:
        html = f.read()
    html = html.replace('data-api-base=""', f'data-api-base="{PATH_PREFIX}"')
    # ② Return the patched HTML page
    return app.response_class(html, mimetype="text/html")


@bp.route("/css/<path:filename>")
def css(filename):
    return app.send_static_file(os.path.join("css", filename))


@bp.route("/js/<path:filename>")
def js(filename):
    return app.send_static_file(os.path.join("js", filename))


@bp.route("/explanation/<path:filename>")
@bp.route("/info/<path:filename>")
def explanation_pages(filename):
    return app.send_static_file(os.path.join("explanation", filename))


@bp.route("/theory/<path:filename>")
def theory_pages(filename):
    """Serve the theory documentation pages from src/theory."""
    return app.send_static_file(os.path.join("theory", filename))


@bp.route("/api/cot-vs-direct", methods=["POST"])
def cot_vs_direct_route():
    # ① Parse problem_title from the JSON request body
    data = request.get_json(force=True, silent=True) or {}
    problem_title = data.get("problem_title") or "Spatial Navigation (Grid)"
    try:
        # ② Run Direct vs Chain-of-Thought benchmark in 01_cot_vs_direct.py
        res = mod_cot.run_cot_vs_direct_benchmark(problem_title)
        res["result"] = f"Direct: {'Correct' if res['direct_result']['is_correct'] else 'Failed'} | CoT: {'Correct' if res['cot_result']['is_correct'] else 'Failed'}"
        return jsonify(res)
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


@bp.route("/api/variance-determinism", methods=["POST"])
def variance_determinism_route():
    # ① Parse review_text from the JSON request body and reject empty strings
    data = request.get_json(force=True, silent=True) or {}
    review_text = (data.get("review_text") or data.get("message") or "").strip()
    if not review_text:
        return jsonify({"error": "Review text cannot be empty"}), 400
    try:
        # ② Run the 3-strategy JSON determinism benchmark in 02_variance_determinism.py
        res = mod_var.run_variance_determinism_benchmark(review_text)
        res["result"] = f"Tested {len(res.get('strategies', []))} JSON extraction strategies."
        return jsonify(res)
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


@bp.route("/api/tool-schema-calibration", methods=["POST"])
def tool_schema_calibration_route():
    # ① Parse query from the JSON request body and reject empty strings
    data = request.get_json(force=True, silent=True) or {}
    query = (data.get("query") or data.get("message") or "").strip()
    if not query:
        return jsonify({"error": "Query cannot be empty"}), 400
    try:
        # ② Run the 2x2 tool schema calibration matrix in 03_tool_schema_calibration.py
        res = mod_calib.run_tool_schema_calibration(query)
        res["result"] = res.get("analysis_summary", "Calibration complete.")
        return jsonify(res)
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


@bp.route("/api/tool-error-injection", methods=["POST"])
def tool_error_injection_route():
    # ① Parse question from the JSON request body and reject empty strings
    data = request.get_json(force=True, silent=True) or {}
    question = (data.get("question") or data.get("message") or "").strip()
    if not question:
        return jsonify({"error": "Question cannot be empty"}), 400
    try:
        # ② Run the upstream 503 error injection benchmark in 04_tool_error_injection.py
        res = mod_err.run_tool_error_injection(question)
        res["result"] = f"Evaluated error handling across {len(res.get('scenarios', []))} prompt configurations."
        return jsonify(res)
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


app.register_blueprint(bp, url_prefix=PATH_PREFIX)

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000)

Review these foundational principles for building deterministic, machine-parseable LLM pipelines:

1. Semantic Correctness vs. Syntactic Determinism: An LLM response that accurately summarizes a review in bullet points is semantically useful to a human reader, but causes an immediate json.JSONDecodeError crash when ingested by a downstream software microservice.
2. Why Unconstrained Prompts (Strategy A) Fail in Pipelines: Without explicit structural constraints, instruction-tuned models default to conversational markdown prose, headings, and bullet lists that vary from call to call.
3. Fragility of Prompt-Only JSON Instructions (Strategy B): Asking for JSON in the prompt nudges token probabilities but does not prevent the model from wrapping output in ```json ... ``` fences or adding conversational preambles, requiring defensive string stripping.
4. Constrained Decoding via response_format (Strategy C): Passing response_format={"type": "json_object"} (or a strict JSON Schema) masks invalid tokens at the logit/sampler level during inference, guaranteeing syntactically valid JSON on every call.
5. Prompt Requirement for json_object Mode: OpenAI's json_object response format requires the word "JSON" to appear explicitly in the prompt messages (as in prompt_c) so the model's semantic intent aligns with the constrained grammar sampler.
6. Explicit Parser Boundary Verification (check_json): Always wrap downstream json.loads() calls in a dedicated validation layer (or Pydantic model_validate_json()) that records both the raw string and structured parse errors for observability.
7. Decoupling Creative Sampling from Structural Validity: Constrained decoding allows you to keep non-zero sampling temperature for natural phrasing inside string values while maintaining 100% deterministic JSON syntax at the schema boundary.