Skip to content

Session 4: Adding Validation and Guard Logic

Synopsis

Introduces practical safeguards such as input checks, output validation, rule-based filters, and fallback behavior. This session prepares learners to build applications that behave more predictably in real-world conditions.

Session Content

Session 4: Adding Validation and Guard Logic

Session Overview

In this session, learners will build safer GenAI applications by adding validation and guard logic around model inputs and outputs. The focus is practical: how to reduce malformed outputs, detect unsafe or irrelevant responses, enforce business rules, and fail gracefully when model behavior is uncertain.

By the end of this session, learners will be able to:

  • Explain why validation and guard logic are essential in GenAI systems
  • Distinguish between input validation, output validation, and behavioral guardrails
  • Implement schema-based response validation
  • Add lightweight rule-based guards for safety and business constraints
  • Build retry-and-repair loops when model outputs do not pass validation
  • Create a reusable Python validation pipeline using the OpenAI Responses API

Learning Objectives

After this session, you should be able to:

  1. Validate user input before sending it to an LLM
  2. Constrain and verify model output using structured formats
  3. Add rule-based checks for safety, relevance, and business rules
  4. Repair or retry invalid outputs automatically
  5. Design a simple guard pipeline for agentic applications

Agenda (~45 minutes)

  • 0–8 min: Why validation and guards matter
  • 8–18 min: Input validation and prompt-level constraints
  • 18–30 min: Output validation with structured responses
  • 30–40 min: Guard logic, retry loops, and fallback handling
  • 40–45 min: Wrap-up and extension ideas

1. Why Validation and Guard Logic Matter

LLMs are powerful but probabilistic. That means they can:

  • produce malformed JSON
  • omit required fields
  • violate business rules
  • answer questions outside scope
  • invent unsupported facts
  • generate unsafe or disallowed content

In a real application, "mostly correct" is not enough.

Common Failure Modes

1. Invalid structure

You expect JSON with specific keys, but receive extra prose or missing fields.

2. Invalid values

The shape is correct, but the values are wrong: - negative price - unsupported category - invalid email - confidence score outside 0–1

3. Scope violations

The model answers questions it should refuse or escalate.

4. Unsafe responses

The application must detect harmful or disallowed outputs.

5. Hallucinated certainty

The model gives a confident answer even when the prompt does not provide enough evidence.

2. Types of Validation and Guard Logic

A robust GenAI app typically uses multiple layers.

A. Input Validation

Validate user input before sending it to the model.

Examples: - required fields present - string length limits - allowed categories only - no empty text - no unsupported file types

B. Prompt Constraints

Tell the model what shape and behavior are expected.

Examples: - “Return only JSON” - “Use one of these labels” - “If uncertain, set needs_review: true”

C. Output Validation

Check the response programmatically after generation.

Examples: - parse JSON - validate schema - validate ranges and enums - reject missing fields

D. Guard Logic

Apply domain-specific rules after validation.

Examples: - refuse if category is “medical” but app is for retail only - flag if confidence < 0.7 - reject if response includes prohibited claims - route to human review if explanation is too short

E. Retry / Repair Logic

If output fails checks: - retry with a stricter prompt - ask model to repair invalid JSON - use a fallback message - escalate to human review

3. Theory: Validation Pipeline Design

A practical validation pipeline often looks like this:

User Input
   ↓
Input Validation
   ↓
LLM Call with Constraints
   ↓
Parse Response
   ↓
Schema Validation
   ↓
Rule-Based Guard Checks
   ↓
Accept / Retry / Repair / Escalate

Design Principles

  • Fail early: Catch bad input before the API call
  • Be explicit: Ask for structured responses and fixed labels
  • Validate independently: Never trust model output blindly
  • Separate concerns: schema validation is different from business rules
  • Provide fallbacks: retries and human-review paths reduce brittleness
  • Log decisions: for debugging and quality improvement

4. Hands-On Exercise 1: Input Validation Before Calling the Model

Goal

Build a simple classifier input validator for a support-ticket triage system.

The app expects: - ticket_text: non-empty string, 10–500 characters - priority_hint: optional, must be one of low, medium, high

What You Will Learn

  • how to validate user inputs in Python
  • how to reject bad requests early
  • how to keep prompts cleaner and more reliable

Code

from typing import Optional


ALLOWED_PRIORITIES = {"low", "medium", "high"}


def validate_ticket_input(ticket_text: str, priority_hint: Optional[str] = None) -> dict:
    """
    Validate user-provided support ticket input before sending it to the LLM.

    Returns:
        A normalized dictionary if valid.

    Raises:
        ValueError: if any validation rule fails.
    """
    if not isinstance(ticket_text, str):
        raise ValueError("ticket_text must be a string.")

    ticket_text = ticket_text.strip()

    if not ticket_text:
        raise ValueError("ticket_text cannot be empty.")

    if len(ticket_text) < 10:
        raise ValueError("ticket_text must be at least 10 characters long.")

    if len(ticket_text) > 500:
        raise ValueError("ticket_text must be 500 characters or fewer.")

    if priority_hint is not None:
        if not isinstance(priority_hint, str):
            raise ValueError("priority_hint must be a string if provided.")
        priority_hint = priority_hint.strip().lower()
        if priority_hint not in ALLOWED_PRIORITIES:
            raise ValueError(f"priority_hint must be one of: {sorted(ALLOWED_PRIORITIES)}")

    return {
        "ticket_text": ticket_text,
        "priority_hint": priority_hint,
    }


if __name__ == "__main__":
    examples = [
        {
            "ticket_text": "I was charged twice for my monthly subscription and need a refund.",
            "priority_hint": "high",
        },
        {
            "ticket_text": "Too short",
            "priority_hint": "medium",
        },
        {
            "ticket_text": "My account page fails to load after login and shows a 500 error.",
            "priority_hint": "urgent",
        },
    ]

    for idx, example in enumerate(examples, start=1):
        print(f"\nExample {idx}:")
        try:
            validated = validate_ticket_input(**example)
            print("Validated input:", validated)
        except ValueError as exc:
            print("Validation error:", exc)

Example Output

Example 1:
Validated input: {'ticket_text': 'I was charged twice for my monthly subscription and need a refund.', 'priority_hint': 'high'}

Example 2:
Validation error: ticket_text must be at least 10 characters long.

Example 3:
Validation error: priority_hint must be one of: ['high', 'low', 'medium']

Exercise Tasks

  1. Add validation for a new field called customer_tier
  2. Allow only free, pro, or enterprise
  3. Normalize all accepted values to lowercase
  4. Test both valid and invalid examples

5. Hands-On Exercise 2: Structured Output with the Responses API

Goal

Call the OpenAI model and request a structured JSON response for ticket triage.

We want the model to return:

  • category: one of billing, technical, account, other
  • priority: one of low, medium, high
  • needs_human_review: boolean
  • short_reply: short message to the customer

What You Will Learn

  • how to use the OpenAI Python SDK with the Responses API
  • how to instruct the model to return strict JSON
  • how to parse model output safely

Code

import json
import os
from openai import OpenAI

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))


def triage_ticket(ticket_text: str, priority_hint: str | None = None) -> dict:
    """
    Send a support ticket to the model and ask for a structured triage response.

    Args:
        ticket_text: The support request text.
        priority_hint: Optional priority hint supplied by the UI.

    Returns:
        Parsed dictionary response from the model.
    """
    instructions = (
        "You are a support ticket triage assistant. "
        "Return ONLY valid JSON with the following keys: "
        "category, priority, needs_human_review, short_reply. "
        "Rules: "
        "- category must be one of: billing, technical, account, other "
        "- priority must be one of: low, medium, high "
        "- needs_human_review must be true or false "
        "- short_reply must be a brief customer-facing response under 30 words "
        "Do not include markdown fences or extra text."
    )

    user_payload = {
        "ticket_text": ticket_text,
        "priority_hint": priority_hint,
    }

    response = client.responses.create(
        model="gpt-5.4-mini",
        instructions=instructions,
        input=json.dumps(user_payload),
    )

    # The SDK provides a convenience property for the aggregated text output.
    raw_text = response.output_text.strip()

    # Parse the JSON returned by the model.
    return json.loads(raw_text)


if __name__ == "__main__":
    result = triage_ticket(
        ticket_text="I updated my payment method, but I was still billed on the old card. Please fix this.",
        priority_hint="high",
    )

    print("Structured model response:")
    print(json.dumps(result, indent=2))

Example Output

{
  "category": "billing",
  "priority": "high",
  "needs_human_review": false,
  "short_reply": "Thanks for reporting this. We’re reviewing your billing issue and will help resolve it shortly."
}

Discussion

Prompting for JSON helps, but it does not guarantee correctness. We still need validation after parsing.

6. Hands-On Exercise 3: Output Validation with a Python Schema

Goal

Validate the model output against required fields and business constraints.

What You Will Learn

  • how to validate parsed JSON
  • how to separate model generation from application acceptance
  • how to surface useful errors for retries

Code

import json
from typing import Any


ALLOWED_CATEGORIES = {"billing", "technical", "account", "other"}
ALLOWED_PRIORITIES = {"low", "medium", "high"}


def validate_triage_output(data: dict[str, Any]) -> dict[str, Any]:
    """
    Validate the model-generated triage output.

    Args:
        data: Parsed JSON response from the model.

    Returns:
        The validated data.

    Raises:
        ValueError: if validation fails.
    """
    required_fields = {
        "category": str,
        "priority": str,
        "needs_human_review": bool,
        "short_reply": str,
    }

    if not isinstance(data, dict):
        raise ValueError("Output must be a dictionary.")

    for field_name, field_type in required_fields.items():
        if field_name not in data:
            raise ValueError(f"Missing required field: {field_name}")
        if not isinstance(data[field_name], field_type):
            raise ValueError(
                f"Field '{field_name}' must be of type {field_type.__name__}."
            )

    if data["category"] not in ALLOWED_CATEGORIES:
        raise ValueError(
            f"Invalid category: {data['category']}. Allowed: {sorted(ALLOWED_CATEGORIES)}"
        )

    if data["priority"] not in ALLOWED_PRIORITIES:
        raise ValueError(
            f"Invalid priority: {data['priority']}. Allowed: {sorted(ALLOWED_PRIORITIES)}"
        )

    short_reply = data["short_reply"].strip()

    if not short_reply:
        raise ValueError("short_reply cannot be empty.")

    if len(short_reply.split()) > 30:
        raise ValueError("short_reply must be 30 words or fewer.")

    data["short_reply"] = short_reply
    return data


if __name__ == "__main__":
    valid_output = {
        "category": "billing",
        "priority": "high",
        "needs_human_review": False,
        "short_reply": "We’re sorry about the billing issue and will investigate it right away.",
    }

    invalid_output = {
        "category": "refunds",
        "priority": "urgent",
        "needs_human_review": "maybe",
        "short_reply": "",
    }

    for label, sample in [("VALID", valid_output), ("INVALID", invalid_output)]:
        print(f"\n{label} SAMPLE")
        try:
            validated = validate_triage_output(sample)
            print(json.dumps(validated, indent=2))
        except ValueError as exc:
            print("Validation error:", exc)

Example Output

VALID SAMPLE
{
  "category": "billing",
  "priority": "high",
  "needs_human_review": false,
  "short_reply": "We’re sorry about the billing issue and will investigate it right away."
}

INVALID SAMPLE
Validation error: Field 'needs_human_review' must be of type bool.

Exercise Tasks

  1. Add a new optional field called confidence
  2. Require it to be a float between 0.0 and 1.0
  3. If confidence < 0.6, force needs_human_review = True
  4. Test the validator with multiple examples

7. Hands-On Exercise 4: Guard Logic for Business Rules

Goal

Add domain-specific checks beyond structural validation.

Suppose your support bot is not allowed to: - give refund guarantees - make legal claims - classify sensitive cases as fully automated when uncertain

What You Will Learn

  • how to add rule-based guardrails
  • how to combine validation with policy enforcement
  • how to flag risky outputs

Code

from typing import Any


PROHIBITED_PHRASES = [
    "we guarantee a refund",
    "you are legally entitled",
    "this issue is definitely caused by",
]


def apply_guard_rules(data: dict[str, Any]) -> dict[str, Any]:
    """
    Apply business and policy guards to validated triage data.

    This function assumes schema validation already passed.

    Rules:
    - Prohibited phrases are not allowed in short_reply
    - High priority technical tickets should be reviewed by a human
    """
    reply_lower = data["short_reply"].lower()

    for phrase in PROHIBITED_PHRASES:
        if phrase in reply_lower:
            raise ValueError(f"Guard violation: prohibited phrase detected: '{phrase}'")

    if data["category"] == "technical" and data["priority"] == "high":
        data["needs_human_review"] = True

    return data


if __name__ == "__main__":
    sample_ok = {
        "category": "billing",
        "priority": "medium",
        "needs_human_review": False,
        "short_reply": "Thanks for reaching out. We’re reviewing your billing concern now.",
    }

    sample_bad = {
        "category": "billing",
        "priority": "medium",
        "needs_human_review": False,
        "short_reply": "We guarantee a refund and will process it today.",
    }

    sample_technical = {
        "category": "technical",
        "priority": "high",
        "needs_human_review": False,
        "short_reply": "We’re investigating the outage and will update you soon.",
    }

    for label, sample in [
        ("OK", sample_ok),
        ("BAD", sample_bad),
        ("TECHNICAL_HIGH", sample_technical),
    ]:
        print(f"\n{label}")
        try:
            guarded = apply_guard_rules(sample)
            print(guarded)
        except ValueError as exc:
            print("Guard error:", exc)

Example Output

OK
{'category': 'billing', 'priority': 'medium', 'needs_human_review': False, 'short_reply': 'Thanks for reaching out. We’re reviewing your billing concern now.'}

BAD
Guard error: Guard violation: prohibited phrase detected: 'we guarantee a refund'

TECHNICAL_HIGH
{'category': 'technical', 'priority': 'high', 'needs_human_review': True, 'short_reply': 'We’re investigating the outage and will update you soon.'}

8. Hands-On Exercise 5: Retry and Repair Loop

Goal

Build a small pipeline that: 1. validates input 2. calls the model 3. parses output 4. validates schema 5. applies guard logic 6. retries once if validation fails

What You Will Learn

  • how to create a practical LLM safety pipeline
  • how to recover from bad outputs
  • how to design a reusable flow for agentic systems

Full Example

import json
import os
from typing import Any
from openai import OpenAI


client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

ALLOWED_CATEGORIES = {"billing", "technical", "account", "other"}
ALLOWED_PRIORITIES = {"low", "medium", "high"}
PROHIBITED_PHRASES = [
    "we guarantee a refund",
    "you are legally entitled",
    "this issue is definitely caused by",
]


def validate_ticket_input(ticket_text: str, priority_hint: str | None = None) -> dict[str, Any]:
    """
    Validate incoming user input before calling the model.
    """
    if not isinstance(ticket_text, str):
        raise ValueError("ticket_text must be a string.")

    ticket_text = ticket_text.strip()
    if not ticket_text:
        raise ValueError("ticket_text cannot be empty.")
    if len(ticket_text) < 10 or len(ticket_text) > 500:
        raise ValueError("ticket_text must be between 10 and 500 characters.")

    if priority_hint is not None:
        if not isinstance(priority_hint, str):
            raise ValueError("priority_hint must be a string if provided.")
        priority_hint = priority_hint.strip().lower()
        if priority_hint not in ALLOWED_PRIORITIES:
            raise ValueError(f"priority_hint must be one of: {sorted(ALLOWED_PRIORITIES)}")

    return {
        "ticket_text": ticket_text,
        "priority_hint": priority_hint,
    }


def request_triage(payload: dict[str, Any], strict_retry: bool = False) -> dict[str, Any]:
    """
    Request triage JSON from the model.

    If strict_retry is True, the prompt becomes more explicit about formatting.
    """
    instructions = (
        "You are a support ticket triage assistant. "
        "Return ONLY valid JSON with keys: "
        "category, priority, needs_human_review, short_reply. "
        "Allowed category values: billing, technical, account, other. "
        "Allowed priority values: low, medium, high. "
        "needs_human_review must be a boolean. "
        "short_reply must be under 30 words and must not promise refunds or legal outcomes. "
        "No markdown. No backticks. No extra text."
    )

    if strict_retry:
        instructions += (
            " This is a retry because the previous answer failed validation. "
            "Be extra careful to output exactly one JSON object and nothing else."
        )

    response = client.responses.create(
        model="gpt-5.4-mini",
        instructions=instructions,
        input=json.dumps(payload),
    )

    return json.loads(response.output_text.strip())


def validate_triage_output(data: dict[str, Any]) -> dict[str, Any]:
    """
    Validate response structure and value constraints.
    """
    if not isinstance(data, dict):
        raise ValueError("Output must be a dictionary.")

    required_fields = {
        "category": str,
        "priority": str,
        "needs_human_review": bool,
        "short_reply": str,
    }

    for field_name, field_type in required_fields.items():
        if field_name not in data:
            raise ValueError(f"Missing field: {field_name}")
        if not isinstance(data[field_name], field_type):
            raise ValueError(f"Field '{field_name}' must be {field_type.__name__}")

    if data["category"] not in ALLOWED_CATEGORIES:
        raise ValueError(f"Invalid category: {data['category']}")
    if data["priority"] not in ALLOWED_PRIORITIES:
        raise ValueError(f"Invalid priority: {data['priority']}")

    short_reply = data["short_reply"].strip()
    if not short_reply:
        raise ValueError("short_reply cannot be empty")
    if len(short_reply.split()) > 30:
        raise ValueError("short_reply must be 30 words or fewer")

    data["short_reply"] = short_reply
    return data


def apply_guard_rules(data: dict[str, Any]) -> dict[str, Any]:
    """
    Enforce policy and business rules after schema validation.
    """
    reply_lower = data["short_reply"].lower()

    for phrase in PROHIBITED_PHRASES:
        if phrase in reply_lower:
            raise ValueError(f"Prohibited phrase found: {phrase}")

    if data["category"] == "technical" and data["priority"] == "high":
        data["needs_human_review"] = True

    return data


def triage_with_validation(ticket_text: str, priority_hint: str | None = None) -> dict[str, Any]:
    """
    End-to-end pipeline:
    - validate input
    - call model
    - parse output
    - validate structure
    - apply business guards
    - retry once if needed
    """
    payload = validate_ticket_input(ticket_text, priority_hint)

    errors = []

    for attempt in range(2):
        try:
            raw_output = request_triage(payload, strict_retry=(attempt == 1))
            validated = validate_triage_output(raw_output)
            guarded = apply_guard_rules(validated)
            return guarded
        except (json.JSONDecodeError, ValueError) as exc:
            errors.append(str(exc))

    return {
        "category": "other",
        "priority": "medium",
        "needs_human_review": True,
        "short_reply": "Thanks for your message. A support specialist will review this request shortly.",
        "fallback_reason": errors,
    }


if __name__ == "__main__":
    ticket = "Our team cannot log in after the latest deployment, and production access seems broken."

    result = triage_with_validation(ticket_text=ticket, priority_hint="high")

    print("Final triage result:")
    print(json.dumps(result, indent=2))

Example Output

{
  "category": "technical",
  "priority": "high",
  "needs_human_review": true,
  "short_reply": "We’re sorry for the disruption. A support specialist is reviewing this urgent issue now."
}

Extension Tasks

  1. Add logging for each failed validation attempt
  2. Add a third path: escalate to a human if the ticket includes words like outage, security, or breach
  3. Store the original raw model output for debugging
  4. Add confidence scoring and threshold-based escalation

9. Applying This to Agentic Systems

Validation and guard logic become even more important in agentic workflows.

Why?

Agents often: - decide what action to take - call tools - transform intermediate data - generate final user-facing responses

A single invalid step can cascade into bigger failures.

Example Guard Points in Agentic Flows

Before tool calls

Validate: - required tool arguments - allowed parameter ranges - user authorization

After tool results

Validate: - result schema - null or empty values - stale or inconsistent data

Before final answer

Validate: - no unsupported claims - no hidden internal data leakage - response matches business policy

Mini Pattern

def agent_step(user_input: str) -> dict:
    # 1. Validate input
    # 2. Ask model for next action
    # 3. Validate chosen action
    # 4. Execute tool
    # 5. Validate tool output
    # 6. Generate final response
    # 7. Apply final guards
    return {}

The key lesson: guard each boundary.

10. Best Practices Checklist

Use this checklist when building GenAI applications:

Input Safety

  • [ ] check required fields
  • [ ] enforce length and type constraints
  • [ ] normalize values
  • [ ] reject unsupported requests early

Prompt Design

  • [ ] request structured output
  • [ ] define allowed values explicitly
  • [ ] tell the model what to do when uncertain
  • [ ] prohibit extra commentary when machine parsing is required

Output Validation

  • [ ] parse carefully
  • [ ] validate schema
  • [ ] validate enums and ranges
  • [ ] validate text length and formatting

Guard Logic

  • [ ] encode business rules separately from schema rules
  • [ ] detect prohibited claims or phrases
  • [ ] route risky cases to human review
  • [ ] log violations for analysis

Reliability

  • [ ] retry invalid generations with stricter instructions
  • [ ] use safe fallback responses
  • [ ] keep raw outputs for debugging
  • [ ] monitor validation failure rates

11. Common Mistakes to Avoid

Mistake 1: Trusting JSON-shaped output without checking values

Correct shape does not mean correct content.

Mistake 2: Mixing prompt logic with business policy

Keep policy rules in code where they are testable and auditable.

Mistake 3: No fallback path

A production system must know what to do when validation fails.

Mistake 4: Overly complex guard logic inside one function

Split into: - input validation - schema validation - business guards - escalation logic

Mistake 5: Ignoring observability

Without logs, you cannot improve your pipeline.

12. Quick Recap

In this session, you learned how to:

  • validate user input before calling a model
  • request structured responses using the OpenAI Responses API
  • parse and validate model output in Python
  • apply business rule guards after schema checks
  • build retry and fallback logic for reliability

This is a core skill for safe and production-ready GenAI systems.

13. Practice Challenge

Build a product review moderation assistant with the following requirements:

Input

  • review_text: 20–300 characters
  • product_category: one of books, electronics, clothing

Model Output

  • sentiment: positive, neutral, negative
  • contains_abuse: boolean
  • publish: boolean
  • summary: under 20 words

Guard Rules

  • If contains_abuse = true, then publish = false
  • Reject summaries that contain insults
  • If output is invalid, retry once
  • If still invalid, return a safe fallback

Try implementing: 1. input validation 2. model call 3. schema validation 4. guard rules 5. retry logic

Useful Resources

End-of-Session Summary

Validation and guard logic are essential for turning LLM demos into dependable applications. Prompting helps, but code-level enforcement is what makes systems safer and more predictable. As your applications become more agentic, these checks should be added at every step where data crosses a boundary: user input, model output, tool arguments, tool results, and final responses.

In the next session, these patterns can be extended into more advanced agent workflows with tool usage, state handling, and decision control.

Back to Chapter | Back to Master Plan | Previous Session