The Hallucination Firewall: 5 Automated Verification Patterns for Production AI

npm install zod openai @types/node

import OpenAI from 'openai';
import { z } from 'zod';

const openai = new OpenAI();

// 1. Define Strict Business Rules
export const ComplianceAuditSchema = z.object({
  entityName: z.string().min(2),
  jurisdiction: z.enum(['US-SEC', 'EU-GDPR', 'UK-FCA']),
  riskScore: z.number().int().min(1).max(100),
  violationsFound: z.array(z.string()).default([]),
  citationAnchors: z.array(z.string().regex(/^anchor:[a-z0-9_-]+:\d+$/)).nonempty()
});

export type ComplianceAudit = z.infer;

// 2. Execution Function with Single-Turn Reflection Interceptor
export async function executeFirewallExtraction(
  sourceDocument: string,
  docId: string
): Promise {
  const systemPrompt = `You are a compliance extraction engine. 
You must output strictly valid JSON matching the compliance audit schema.
Every claim must cite an anchor formatted as anchor:${docId}:LINE_INDEX from .`;

  const userPrompt = `
${sourceDocument}

Extract compliance audit status.`;

  // First Pass Generation
  let response = await openai.chat.completions.create({
    model: 'gpt-4o-mini',
    temperature: 0.1,
    response_format: { type: 'json_object' },
    messages: [
      { role: 'system', content: systemPrompt },
      { role: 'user', content: userPrompt }
    ]
  });

  let rawOutput = response.choices[0].message.content || '{}';

  // Structural Quarantine
  const parseResult = ComplianceAuditSchema.safeParse(JSON.parse(rawOutput));
  if (parseResult.success) {
    return parseResult.data;
  }

  // Automated 1-Turn Reflection Loop on Schema Failure
  console.warn('Firewall tripwire triggered. Repairing schema failure:', parseResult.error.format());

  const repairResponse = await openai.chat.completions.create({
    model: 'gpt-4o-mini',
    temperature: 0.0,
    response_format: { type: 'json_object' },
    messages: [
      { role: 'system', content: systemPrompt },
      { role: 'user', content: userPrompt },
      { role: 'assistant', content: rawOutput },
      { 
        role: 'user', 
        content: `FIREWALL_SCHEMA_ERROR: Your previous output failed validation:
${JSON.stringify(parseResult.error.flatten().fieldErrors)}
Fix the exact fields above and return valid JSON only.` 
      }
    ]
  });

  const repairedOutput = repairResponse.choices[0].message.content || '{}';
  return ComplianceAuditSchema.parse(JSON.parse(repairedOutput));
}

pip install instructor pydantic openai

import instructor
from openai import OpenAI
from pydantic import BaseModel, Field, field_validator
from typing import List

# Patch the client for automated reflection retries
client = instructor.from_openai(OpenAI())

class FinancialExtraction(BaseModel):
    ticker: str = Field(..., description="Stock symbol in uppercase e.g. NVDA")
    reported_revenue_usd: float = Field(..., gt=0, description="Revenue in USD")
    ebitda_margin: float = Field(..., ge=-1.0, le=1.0, description="EBITDA margin between -100% and +100%")
    bullet_citations: List[str] = Field(..., min_length=1)

    @field_validator("ticker")
    @classmethod
    def validate_ticker(cls, v: str) -> str:
        if not v.isupper() or len(v) > 5:
            raise ValueError("Ticker must be 1-5 capital letters")
        return v

def run_quarantine_pipeline(financial_report_text: str) -> FinancialExtraction:
    # max_retries=2 automatically loops back if Pydantic raises a ValidationError
    result: FinancialExtraction = client.chat.completions.create(
        model="gpt-4o-mini",
        response_model=FinancialExtraction,
        max_retries=2,
        messages=[
            {
                "role": "system",
                "content": "Extract financial metric facts strictly from the provided text. Do not invent numbers."
            },
            {
                "role": "user",
                "content": f"\n{financial_report_text}\n"
            }
        ]
    )
    return result

import outlines
from pydantic import BaseModel, constr

# 1. Define strictly typed data contract
class SecurityIncident(BaseModel):
    cve_id: constr(regex=r"^CVE-\d{4}-\d{4,7}$")
    severity: constr(regex=r"^(LOW|MEDIUM|HIGH|CRITICAL)$")
    affected_subsystem: str
    requires_immediate_patch: bool

# 2. Load model using local transformer engine
model = outlines.models.transformers("Qwen/Qwen2.5-Coder-7B-Instruct")

# 3. Compile schema directly into token logits sampler
generator = outlines.generate.json(model, SecurityIncident)

# 4. Generate guaranteed valid JSON with ZERO retries
prompt = "Analyze system crash logs for unauthorized memory access."
incident: SecurityIncident = generator(prompt)
print(f"Verified CVE: {incident.cve_id} [Severity: {incident.severity}]")

from guardrails import Guard
from guardrails.hub import ProvenanceV1, ValidLength

# Instantiate guardrail with community-vetted hub validators
guard = Guard().use_many(
    ProvenanceV1(threshold=0.85, validation_method="sentence"),
    ValidLength(min=50, max=500, on_fail="reask")
)

# Validates response against retrieved documents in real-time
validated_output = guard(
    llm_api=openai.chat.completions.create,
    model="gpt-4o",
    messages=[{"role": "user", "content": "Explain customer data retention rules."}]
)