Implementing guardrails for LLM outputs in production

Why guardrails matter more than prompt engineering alone

Prompt engineering is the first line of defence against unwanted LLM behaviour. It is not the last. A well-crafted system prompt reduces the probability of bad outputs; it does not eliminate it. Models are stochastic, adversarial users are creative, and the surface area of real-world inputs is far larger than any test suite covers.

Guardrails are the safety net that catches failures after the model has already responded. They operate on the output side of the LLM call and enforce properties that the prompt alone cannot guarantee: structural validity, content policy compliance, personally identifiable information (PII) redaction, and factual grounding constraints.

Treating guardrails as an afterthought is a mistake that shows up in production incidents. A customer service bot that occasionally outputs a competitor's name with a positive sentiment, a coding assistant that emits a snippet containing a hardcoded credential, or a summarisation pipeline that silently drops critical caveats—these are the kinds of failures that prompt engineering cannot catch because the failures are emergent from the combination of model behaviour and user input, not from the prompt alone.

This article covers the full stack of guardrails from structural validation at the schema level up to asynchronous human review pipelines, with working TypeScript and Python code you can adapt directly.

Layer 1: Schema enforcement with Zod and structured outputs

The lowest-effort guardrail with the highest return is refusing to accept unstructured text from the model in the first place. When your use case permits it, force the model to return JSON conforming to a schema and validate that schema before any downstream code touches the response.

Modern models support structured output modes. With the Anthropic API you can use tool definitions to coerce the model into returning a specific JSON shape:

import Anthropic from "@anthropic-ai/sdk";
import { z } from "zod";

const client = new Anthropic();

const SummarySchema = z.object({
  headline: z.string().max(120),
  keyPoints: z.array(z.string()).min(1).max(5),
  sentiment: z.enum(["positive", "neutral", "negative"]),
  confidence: z.number().min(0).max(1),
});

type Summary = z.infer<typeof SummarySchema>;

async function summariseDocument(document: string): Promise<Summary> {
  const response = await client.messages.create({
    model: "claude-sonnet-4-5",
    max_tokens: 1024,
    tools: [
      {
        name: "submit_summary",
        description: "Submit the structured summary of the document",
        input_schema: {
          type: "object",
          properties: {
            headline: { type: "string", maxLength: 120 },
            keyPoints: {
              type: "array",
              items: { type: "string" },
              minItems: 1,
              maxItems: 5,
            },
            sentiment: { type: "string", enum: ["positive", "neutral", "negative"] },
            confidence: { type: "number", minimum: 0, maximum: 1 },
          },
          required: ["headline", "keyPoints", "sentiment", "confidence"],
        },
      },
    ],
    tool_choice: { type: "tool", name: "submit_summary" },
    messages: [{ role: "user", content: `Summarise this document:\n\n${document}` }],
  });

  const toolUse = response.content.find((block) => block.type === "tool_use");
  if (!toolUse || toolUse.type !== "tool_use") {
    throw new Error("Model did not return a tool use block");
  }

  // Validate against Zod schema — throws ZodError on violation
  return SummarySchema.parse(toolUse.input);
}

Zod's .parse() throws a detailed error if the model produces output that violates the schema—a confidence field outside [0, 1], a missing headline, or a sentiment value not in the enum. This catches model hallucinations at the structural level before they propagate.

When the model does violate the schema, do not silently retry without logging. Each schema violation is a signal worth capturing for prompt improvement later.

Layer 2: Content classifiers for policy enforcement

Schema enforcement handles structure. Content classifiers handle semantics: is this output appropriate for your use case and user population?

A content classifier is a secondary model call (or a fine-tuned classifier) that receives the LLM output and returns a classification. You can use a fast, cheap model for this—latency and cost matter because this runs on every response.

import anthropic
from dataclasses import dataclass
from enum import Enum

class PolicyViolationType(Enum):
    NONE = "none"
    TOXIC = "toxic"
    PII_EXPOSURE = "pii_exposure"
    COMPETITOR_MENTION = "competitor_mention"
    LEGAL_RISK = "legal_risk"
    MEDICAL_ADVICE = "medical_advice"

@dataclass
class ClassificationResult:
    violation_type: PolicyViolationType
    confidence: float
    excerpt: str | None  # The offending text fragment, if any

def classify_output(
    original_user_input: str,
    model_output: str,
    policy_context: str,
) -> ClassificationResult:
    client = anthropic.Anthropic()

    prompt = f"""You are a content policy classifier. Analyse the assistant response below and determine if it violates any of these policies:
- TOXIC: harmful, hateful, or abusive content
- PII_EXPOSURE: exposes personal data (names, emails, phone numbers, addresses)
- COMPETITOR_MENTION: positively mentions competitor products by name
- LEGAL_RISK: makes legally binding statements, guarantees, or gives legal advice
- MEDICAL_ADVICE: provides medical diagnoses or treatment recommendations

Policy context for this deployment: {policy_context}

User input: {original_user_input}

Assistant response to evaluate:
{model_output}

Respond with JSON only: {{"violation": "none|toxic|pii_exposure|competitor_mention|legal_risk|medical_advice", "confidence": 0.0-1.0, "excerpt": "offending text or null"}}"""

    response = client.messages.create(
        model="claude-haiku-4-5",
        max_tokens=256,
        messages=[{"role": "user", "content": prompt}],
    )

    import json
    raw = json.loads(response.content[0].text)
    return ClassificationResult(
        violation_type=PolicyViolationType(raw["violation"]),
        confidence=float(raw["confidence"]),
        excerpt=raw.get("excerpt"),
    )

In production you will want to set a confidence threshold, not a binary pass/fail. Outputs between 0.4 and 0.7 confidence might be logged and allowed through; above 0.7 they are blocked; below 0.4 they pass silently. The right thresholds depend on your risk tolerance and the nature of your application.

A secondary model call adds latency. For latency-sensitive applications, run the classifier asynchronously: return the response immediately to the user, then run the classifier and retroactively flag the conversation if a violation is detected. This is appropriate for logged interactions but not for systems where harmful content appearing even briefly is unacceptable.

Layer 3: PII detection and redaction

PII deserves its own layer because it has different handling requirements. You may need to redact PII before storing a response, before displaying it to a different user, or before passing it to a third-party analytics system.

Regex is often sufficient for well-structured PII patterns. Supplement it with a model-based detector for free-form PII embedded in prose.

interface PIIMatch {
  type: "email" | "phone" | "ssn" | "credit_card" | "ip_address";
  value: string;
  start: number;
  end: number;
}

const PII_PATTERNS: Array<{ type: PIIMatch["type"]; pattern: RegExp }> = [
  {
    type: "email",
    pattern: /\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b/g,
  },
  {
    type: "phone",
    pattern: /\b(\+?1[-.\s]?)?\(?\d{3}\)?[-.\s]?\d{3}[-.\s]?\d{4}\b/g,
  },
  {
    type: "ssn",
    pattern: /\b\d{3}-\d{2}-\d{4}\b/g,
  },
  {
    type: "credit_card",
    pattern: /\b(?:\d{4}[-\s]?){3}\d{4}\b/g,
  },
  {
    type: "ip_address",
    pattern: /\b(?:\d{1,3}\.){3}\d{1,3}\b/g,
  },
];

function detectPII(text: string): PIIMatch[] {
  const matches: PIIMatch[] = [];

  for (const { type, pattern } of PII_PATTERNS) {
    pattern.lastIndex = 0; // reset stateful regex
    let match: RegExpExecArray | null;
    while ((match = pattern.exec(text)) !== null) {
      matches.push({
        type,
        value: match[0],
        start: match.index,
        end: match.index + match[0].length,
      });
    }
  }

  return matches.sort((a, b) => a.start - b.start);
}

function redactPII(text: string, matches: PIIMatch[]): string {
  // Process in reverse order to preserve indices
  let result = text;
  for (const match of [...matches].reverse()) {
    const placeholder = `[REDACTED_${match.type.toUpperCase()}]`;
    result = result.slice(0, match.start) + placeholder + result.slice(match.end);
  }
  return result;
}

// Usage
const output = "Contact Sarah at [email protected] or call 555-123-4567.";
const piiMatches = detectPII(output);
const redacted = redactPII(output, piiMatches);
// "Contact Sarah at [REDACTED_EMAIL] or call [REDACTED_PHONE]."

For prose-embedded PII ("the customer's name is John Smith and he lives at..."), regex falls short. In that case, pass the output through the classifier described in Layer 2 with PII detection as the primary target, or use a dedicated NER (named entity recognition) model.

Layer 4: Factual grounding checks for RAG systems

If your LLM is answering questions from a retrieved document set (a RAG architecture), the model's response should be grounded in the retrieved context. A grounding check verifies that claims in the response can be traced back to the source documents.

This is especially important for high-stakes domains: legal, medical, financial, and compliance applications where a model hallucinating a fact with authoritative confidence can cause real harm.

from dataclasses import dataclass

@dataclass
class GroundingResult:
    is_grounded: bool
    ungrounded_claims: list[str]
    grounding_score: float  # 0.0–1.0

def check_grounding(
    retrieved_context: str,
    model_response: str,
) -> GroundingResult:
    client = anthropic.Anthropic()

    prompt = f"""You are a grounding checker. Your task is to verify that all factual claims in the assistant response below are supported by the retrieved context.

Retrieved context:
<context>
{retrieved_context}
</context>

Assistant response:
<response>
{model_response}
</response>

Identify any claims in the response that are NOT supported by the context. Output JSON:
{{
  "is_grounded": true/false,
  "ungrounded_claims": ["claim1", "claim2"],
  "grounding_score": 0.0-1.0
}}

grounding_score should be 1.0 if all claims are supported, 0.0 if no claims are supported."""

    response = client.messages.create(
        model="claude-haiku-4-5",
        max_tokens=512,
        messages=[{"role": "user", "content": prompt}],
    )

    import json
    raw = json.loads(response.content[0].text)
    return GroundingResult(
        is_grounded=raw["is_grounded"],
        ungrounded_claims=raw.get("ungrounded_claims", []),
        grounding_score=float(raw["grounding_score"]),
    )

A grounding score below 0.7 in a RAG system is a strong signal that the model is confabulating. The appropriate response depends on your system: some applications show the response with a confidence warning; others block the response and ask the user to rephrase.

Layer 5: Human-in-the-loop escalation

Not every failure mode can be caught automatically. Some outputs fall into ambiguous territory where the classifier is uncertain, the grounding is partial, or the stakes are high enough that a human should review before the response reaches the user.

Design a human review queue that fits into your system's latency budget. For synchronous applications, this means falling back to a "let me check with our team" response when guardrails fire. For asynchronous applications (document processing pipelines, batch summarisation), route flagged outputs to a review UI.

interface ReviewQueueItem {
  id: string;
  timestamp: Date;
  userId: string;
  sessionId: string;
  userInput: string;
  modelOutput: string;
  guardrailTrigger: {
    layer: "schema" | "classifier" | "pii" | "grounding";
    details: Record<string, unknown>;
  };
  status: "pending" | "approved" | "rejected" | "edited";
  reviewedBy?: string;
  reviewedAt?: Date;
  editedOutput?: string;
}

async function enqueueForReview(
  item: Omit<ReviewQueueItem, "id" | "timestamp" | "status">
): Promise<string> {
  const queueItem: ReviewQueueItem = {
    ...item,
    id: crypto.randomUUID(),
    timestamp: new Date(),
    status: "pending",
  };

  // Store to your database — Firestore, Postgres, etc.
  await db.collection("review_queue").doc(queueItem.id).set(queueItem);

  // Notify the review team
  await notifyReviewers(queueItem);

  return queueItem.id;
}

// Fallback response shown to user while item is queued
const ESCALATION_RESPONSE =
  "I want to make sure I give you accurate information here. " +
  "Let me connect you with our team who can help with this specific question.";

The review queue serves two purposes: it catches failures in real time, and it builds a labelled dataset you can use to improve your classifiers and fine-tune models over time.

Assembling the guardrail pipeline

In production, these layers run in sequence (or in parallel where latency allows). Here is a pipeline that combines all five:

interface GuardrailResult {
  passed: boolean;
  output: string; // possibly redacted
  violations: string[];
  requiresReview: boolean;
}

async function runGuardrails(
  userInput: string,
  rawModelOutput: string,
  retrievedContext: string | null,
  policyContext: string,
): Promise<GuardrailResult> {
  const violations: string[] = [];
  let output = rawModelOutput;
  let requiresReview = false;

  // Layer 1: Schema validation is already enforced at call time
  // If we reach here, the schema passed

  // Layer 3: PII redaction (fast, do this first before logging anything)
  const piiMatches = detectPII(output);
  if (piiMatches.length > 0) {
    output = redactPII(output, piiMatches);
    violations.push(`PII detected and redacted: ${piiMatches.map((m) => m.type).join(", ")}`);
  }

  // Layer 2 + Layer 4: Run classifier and grounding check in parallel
  const [classificationResult, groundingResult] = await Promise.all([
    classify_output(userInput, output, policyContext),
    retrievedContext ? check_grounding(retrievedContext, output) : null,
  ]);

  if (classificationResult.violation_type !== "none") {
    if (classificationResult.confidence > 0.7) {
      return {
        passed: false,
        output: "",
        violations: [`Policy violation: ${classificationResult.violation_type}`],
        requiresReview: true,
      };
    } else if (classificationResult.confidence > 0.4) {
      requiresReview = true;
      violations.push(
        `Uncertain policy violation (${classificationResult.confidence.toFixed(2)}): ${classificationResult.violation_type}`,
      );
    }
  }

  if (groundingResult && groundingResult.grounding_score < 0.7) {
    requiresReview = true;
    violations.push(
      `Low grounding score: ${groundingResult.grounding_score.toFixed(2)}`,
    );
  }

  return { passed: true, output, violations, requiresReview };
}

Monitoring and continuous improvement

Guardrails without monitoring are half-finished. Every guardrail trigger is a data point. Track:

• Trigger rate per layer: a sudden spike in schema violations suggests a prompt regression; a spike in PII detections may indicate a new user pattern or a dataset issue.
• False positive rate: sample approved outputs and have humans review them to estimate the rate at which the guardrails blocked or flagged valid content. High false positive rates degrade user experience.
• Coverage gaps: regularly red-team your system by sending adversarial inputs designed to bypass each layer and verify that the guardrail catches them.

Set up structured logging so each guardrail decision is queryable:

import structlog

log = structlog.get_logger()

def log_guardrail_decision(
    session_id: str,
    layer: str,
    passed: bool,
    violation_type: str | None,
    confidence: float | None,
    latency_ms: float,
) -> None:
    log.info(
        "guardrail_decision",
        session_id=session_id,
        layer=layer,
        passed=passed,
        violation_type=violation_type,
        confidence=confidence,
        latency_ms=latency_ms,
    )

Feed this log into your observability stack (Datadog, Grafana, or Cloud Logging) and alert on trigger rate anomalies. A 10× spike in classifier triggers at 2 AM is worth waking someone up for.

The final point worth emphasising: guardrails are not a substitute for responsible design decisions upstream. Choosing an appropriate model, writing clear system prompts, scoping the use case narrowly, and giving the model access only to the data it needs are all decisions that reduce the load on your guardrail layer. Guardrails exist for the residual risk that remains after good design—not as a cleanup crew for a poorly constrained system.