How to Implement LLM Hallucination Detection in Production with Guardrails AI 0.5 and LangChain

\nimport os\nimport logging\nfrom typing import Dict, Optional, List\nfrom dotenv import load_dotenv\nfrom langchain_openai import ChatOpenAI\nfrom langchain.chains import LLMChain\nfrom langchain.prompts import PromptTemplate\nfrom guardrails import Guard, OnFailAction\nfrom guardrails.validators import (\n    ValidChoices,\n    DetectPII,\n    ToxicLanguage,\n    FactualConsistency\n)\nfrom guardrails.hub import HallucinationDetector\nimport openai\nfrom tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type\n\n# Configure logging for production observability\nlogging.basicConfig(\n    level=logging.INFO,\n    format=\"%(asctime)s - %(name)s - %(levelname)s - %(message)s\"\n)\nlogger = logging.getLogger(__name__)\n\n# Load environment variables from .env file\nload_dotenv()\n\n# Configure OpenAI client with retries for rate limits\nopenai.api_key = os.getenv(\"OPENAI_API_KEY\")\nif not openai.api_key:\n    logger.error(\"OPENAI_API_KEY not set in environment variables\")\n    raise ValueError(\"Missing OPENAI_API_KEY\")\n\n@retry(\n    stop=stop_after_attempt(3),\n    wait=wait_exponential(multiplier=1, min=4, max=10),\n    retry=retry_if_exception_type(openai.RateLimitError)\n)\ndef create_llm(temperature: float = 0.0, max_tokens: int = 1024) -> ChatOpenAI:\n    \"\"\"Initialize ChatOpenAI with retry logic for transient failures.\"\"\"\n    try:\n        llm = ChatOpenAI(\n            model=\"gpt-4o\",\n            temperature=temperature,\n            max_tokens=max_tokens,\n            timeout=30,\n            max_retries=2\n        )\n        logger.info(f\"Initialized LLM: {llm.model_name}\")\n        return llm\n    except openai.AuthenticationError as e:\n        logger.error(f\"OpenAI auth failed: {e}\")\n        raise\n    except Exception as e:\n        logger.error(f\"Failed to initialize LLM: {e}\")\n        raise\n\ndef setup_guardrails() -> Guard:\n    \"\"\"Configure Guardrails AI 0.5 with hallucination, PII, and toxicity validators.\"\"\"\n    try:\n        # Initialize Guard with output schema (free text for this example)\n        guard = Guard(\n            name=\"llm_output_validator\",\n            description=\"Validates LLM output for hallucinations, PII, and toxicity\",\n            on_fail=OnFailAction.REASK  # Re-ask LLM if validation fails, up to 2 times\n        )\n\n        # Add hub-based hallucination detector (uses GPT-4o for validation by default)\n        guard.use(\n            HallucinationDetector(\n                llm_api=openai.chat.completions.create,\n                model=\"gpt-4o\",\n                threshold=0.8,  # Fail if hallucination confidence > 0.8\n                on_fail=OnFailAction.REASK\n            )\n        )\n\n        # Add PII detector (checks for emails, phones, SSNs)\n        guard.use(\n            DetectPII(\n                entities=[\"EMAIL_ADDRESS\", \"PHONE_NUMBER\", \"SSN\"],\n                on_fail=OnFailAction.FIX  # Auto-redact PII instead of re-asking\n            )\n        )\n\n        # Add toxicity detector\n        guard.use(\n            ToxicLanguage(\n                threshold=0.9,\n                on_fail=OnFailAction.EXCEPTION\n            )\n        )\n\n        logger.info(\"Guardrails configuration complete with 3 validators\")\n        return guard\n    except Exception as e:\n        logger.error(f\"Failed to setup Guardrails: {e}\")\n        raise\n\ndef create_langchain_chain(llm: ChatOpenAI) -> LLMChain:\n    \"\"\"Create a LangChain chain for factual Q&A with system prompt.\"\"\"\n    prompt = PromptTemplate(\n        template=\"\"\"You are a factual assistant. Answer the following question using only verified information. Do not make up facts.\n        \n        Question: {question}\n        \n        Answer:\"\"\",\n        input_variables=[\"question\"]\n    )\n    chain = LLMChain(llm=llm, prompt=prompt)\n    logger.info(\"LangChain chain created with factual system prompt\")\n    return chain\n\nif __name__ == \"__main__\":\n    # Test the core integration\n    try:\n        llm = create_llm()\n        guard = setup_guardrails()\n        chain = create_langchain_chain(llm)\n\n        test_question = \"What is the capital of France?\"\n        logger.info(f\"Testing core integration with question: {test_question}\")\n\n        raw_output = chain.run(question=test_question)\n        logger.info(f\"Raw LLM output: {raw_output}\")\n\n        # Validate output with Guardrails\n        validated_output = guard(raw_output, metadata={\"question\": test_question})\n        logger.info(f\"Validated output: {validated_output.validated_output}\")\n        logger.info(f\"Validation passed: {validated_output.validation_passed}\")\n\n    except Exception as e:\n        logger.error(f\"Core integration test failed: {e}\")\n        raise\n

\nimport os\nimport logging\nfrom typing import Dict, Optional, Literal\nfrom datetime import datetime\nfrom dotenv import load_dotenv\nfrom fastapi import FastAPI, HTTPException, Request, status\nfrom fastapi.responses import JSONResponse\nfrom pydantic import BaseModel, Field\nfrom prometheus_client import Counter, Histogram, generate_latest, CONTENT_TYPE_LATEST\nfrom fastapi.responses import Response\nfrom langchain_openai import ChatOpenAI\nfrom langchain.chains import LLMChain\nfrom langchain.prompts import PromptTemplate\nfrom guardrails import Guard\nfrom guardrails.hub import HallucinationDetector\nfrom guardrails.validators import DetectPII, ToxicLanguage\nimport time\n\n# Configure logging\nlogging.basicConfig(level=logging.INFO)\nlogger = logging.getLogger(__name__)\nload_dotenv()\n\n# Prometheus metrics\nREQUEST_COUNT = Counter(\n    \"llm_guardrails_requests_total\",\n    \"Total LLM guardrails requests\",\n    [\"endpoint\", \"status\"]\n)\nREQUEST_LATENCY = Histogram(\n    \"llm_guardrails_request_latency_seconds\",\n    \"Request latency in seconds\",\n    [\"endpoint\"]\n)\nVALIDATION_SUCCESS = Counter(\n    \"llm_guardrails_validation_success_total\",\n    \"Total validation successes\",\n    [\"validator\"]\n)\nVALIDATION_FAILURE = Counter(\n    \"llm_guardrails_validation_failure_total\",\n    \"Total validation failures\",\n    [\"validator\", \"reason\"]\n)\n\napp = FastAPI(\n    title=\"LLM Hallucination Detection Service\",\n    description=\"Production service for LLM output validation using Guardrails AI 0.5 and LangChain\",\n    version=\"1.0.0\"\n)\n\n# Pydantic request/response models\nclass PromptRequest(BaseModel):\n    prompt: str = Field(..., min_length=1, max_length=4096, description=\"User prompt for LLM\")\n    temperature: float = Field(0.0, ge=0.0, le=1.0, description=\"LLM temperature parameter\")\n    max_tokens: int = Field(1024, ge=1, le=4096, description=\"Max tokens for LLM response\")\n\nclass ValidatedResponse(BaseModel):\n    output: str = Field(..., description=\"Validated LLM output\")\n    validation_passed: bool = Field(..., description=\"Whether all validations passed\")\n    validation_details: Dict = Field(default_factory=dict, description=\"Per-validator results\")\n    latency_ms: float = Field(..., description=\"Total request latency in milliseconds\")\n    timestamp: datetime = Field(default_factory=datetime.utcnow)\n\nclass ErrorResponse(BaseModel):\n    error: str = Field(..., description=\"Error message\")\n    error_code: Literal[\"VALIDATION_FAILED\", \"LLM_ERROR\", \"RATE_LIMITED\"] = Field(...)\n    remediation: Optional[str] = Field(None, description=\"Suggested fix for the error\")\n    timestamp: datetime = Field(default_factory=datetime.utcnow)\n\n# Initialize LLM and Guardrails (singleton for performance)\nllm: Optional[ChatOpenAI] = None\nguard: Optional[Guard] = None\n\n@app.on_event(\"startup\")\nasync def startup_event():\n    global llm, guard\n    try:\n        llm = ChatOpenAI(\n            model=\"gpt-4o\",\n            temperature=0.0,\n            max_tokens=1024,\n            timeout=30\n        )\n        guard = Guard(name=\"production_guard\")\n        guard.use(\n            HallucinationDetector(\n                llm_api=llm.client.chat.completions.create,\n                model=\"gpt-4o\",\n                threshold=0.8\n            )\n        )\n        guard.use(DetectPII(entities=[\"EMAIL_ADDRESS\", \"PHONE_NUMBER\"], on_fail=\"fix\"))\n        guard.use(ToxicLanguage(threshold=0.9, on_fail=\"exception\"))\n        logger.info(\"Service initialized successfully\")\n    except Exception as e:\n        logger.error(f\"Startup failed: {e}\")\n        raise\n\n@app.post(\"/generate\", response_model=ValidatedResponse, responses={400: {\"model\": ErrorResponse}, 429: {\"model\": ErrorResponse}, 500: {\"model\": ErrorResponse}})\nasync def generate_text(request: PromptRequest, req: Request):\n    start_time = time.time()\n    endpoint = \"/generate\"\n    try:\n        # Run LangChain chain\n        prompt = PromptTemplate(template=\"Answer factually: {prompt}\\nAnswer:\", input_variables=[\"prompt\"])\n        chain = LLMChain(llm=llm, prompt=prompt)\n        raw_output = chain.run(prompt=request.prompt)\n\n        # Validate with Guardrails\n        validation_result = guard(raw_output, metadata={\"prompt\": request.prompt})\n\n        # Record metrics\n        latency = (time.time() - start_time) * 1000\n        REQUEST_LATENCY.labels(endpoint=endpoint).observe(latency / 1000)\n        if validation_result.validation_passed:\n            VALIDATION_SUCCESS.labels(validator=\"all\").inc()\n            REQUEST_COUNT.labels(endpoint=endpoint, status=\"success\").inc()\n        else:\n            VALIDATION_FAILURE.labels(validator=\"guardrails\", reason=\"validation_failed\").inc()\n            REQUEST_COUNT.labels(endpoint=endpoint, status=\"validation_failed\").inc()\n\n        return ValidatedResponse(\n            output=validation_result.validated_output,\n            validation_passed=validation_result.validation_passed,\n            validation_details=validation_result.metadata,\n            latency_ms=latency\n        )\n\n    except Exception as e:\n        latency = (time.time() - start_time) * 1000\n        REQUEST_LATENCY.labels(endpoint=endpoint).observe(latency / 1000)\n        logger.error(f\"Request failed: {e}\")\n        if \"rate limit\" in str(e).lower():\n            REQUEST_COUNT.labels(endpoint=endpoint, status=\"rate_limited\").inc()\n            raise HTTPException(\n                status_code=status.HTTP_429_TOO_MANY_REQUESTS,\n                detail=ErrorResponse(\n                    error=\"Rate limited by LLM provider\",\n                    error_code=\"RATE_LIMITED\",\n                    remediation=\"Retry after 10 seconds\"\n                ).dict()\n            )\n        elif \"validation\" in str(e).lower():\n            REQUEST_COUNT.labels(endpoint=endpoint, status=\"validation_failed\").inc()\n            raise HTTPException(\n                status_code=status.HTTP_400_BAD_REQUEST,\n                detail=ErrorResponse(\n                    error=str(e),\n                    error_code=\"VALIDATION_FAILED\",\n                    remediation=\"Rephrase prompt to avoid unverified claims\"\n                ).dict()\n            )\n        else:\n            REQUEST_COUNT.labels(endpoint=endpoint, status=\"error\").inc()\n            raise HTTPException(\n                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,\n                detail=ErrorResponse(\n                    error=\"Internal server error\",\n                    error_code=\"LLM_ERROR\",\n                    remediation=\"Contact support\"\n                ).dict()\n            )\n\n@app.get(\"/metrics\")\nasync def metrics():\n    return Response(generate_latest(), media_type=CONTENT_TYPE_LATEST)\n\nif __name__ == \"__main__\":\n    import uvicorn\n    uvicorn.run(app, host=\"0.0.0.0\", port=8080)\n

\nimport os\nimport json\nimport time\nimport logging\nfrom typing import List, Dict, Tuple\nfrom dotenv import load_dotenv\nfrom langchain_openai import ChatOpenAI\nfrom langchain.chains import LLMChain\nfrom langchain.prompts import PromptTemplate\nfrom guardrails import Guard\nfrom guardrails.hub import HallucinationDetector\nfrom guardrails.validators import DetectPII, ToxicLanguage\nfrom tqdm import tqdm  # For progress bar, install with pip install tqdm\n\n# Configure logging\nlogging.basicConfig(level=logging.INFO)\nlogger = logging.getLogger(__name__)\nload_dotenv()\n\n# Benchmark dataset: 1000 factual, 1000 hallucinated prompts (simplified for example)\n# In production, use a curated dataset like TruthfulQA\nFACTUAL_PROMPTS = [\n    \"What is the boiling point of water at sea level?\",\n    \"Who wrote Hamlet?\",\n    \"What is the chemical symbol for gold?\",\n    \"How many continents are there?\",\n    \"What is the speed of light in a vacuum?\"\n] * 200  # 5 * 200 = 1000\n\nHALLUCINATED_PROMPTS = [\n    \"What is the boiling point of water on Mars?\",  # LLM may make up numbers\n    \"Who wrote the 1990 remake of Hamlet?\",  # May hallucinate\n    \"What is the chemical symbol for unobtainium?\",  # Fictional element\n    \"How many moons does Mercury have?\",  # Common hallucination target\n    \"What is the capital of Narnia?\"  # Fictional location\n] * 200  # 5 * 200 = 1000\n\nclass BenchmarkResult:\n    def __init__(self):\n        self.true_positives = 0  # Correctly detected hallucination\n        self.false_positives = 0  # Factual output marked as hallucination\n        self.true_negatives = 0  # Correctly passed factual output\n        self.false_negatives = 0  # Missed hallucination\n        self.total_latency_ms = 0.0\n        self.total_cost_usd = 0.0\n\n    def add_result(self, is_hallucinated: bool, detected_hallucination: bool, latency_ms: float, cost_usd: float):\n        if is_hallucinated and detected_hallucination:\n            self.true_positives += 1\n        elif is_hallucinated and not detected_hallucination:\n            self.false_negatives += 1\n        elif not is_hallucinated and detected_hallucination:\n            self.false_positives += 1\n        else:\n            self.true_negatives += 1\n        self.total_latency_ms += latency_ms\n        self.total_cost_usd += cost_usd\n\n    def to_dict(self) -> Dict:\n        total = self.true_positives + self.false_positives + self.true_negatives + self.false_negatives\n        accuracy = (self.true_positives + self.true_negatives) / total if total > 0 else 0\n        precision = self.true_positives / (self.true_positives + self.false_positives) if (self.true_positives + self.false_positives) > 0 else 0\n        recall = self.true_positives / (self.true_positives + self.false_negatives) if (self.true_positives + self.false_negatives) > 0 else 0\n        return {\n            \"accuracy\": round(accuracy * 100, 2),\n            \"precision\": round(precision * 100, 2),\n            \"recall\": round(recall * 100, 2),\n            \"avg_latency_ms\": round(self.total_latency_ms / total, 2) if total > 0 else 0,\n            \"total_cost_usd\": round(self.total_cost_usd, 2),\n            \"true_positives\": self.true_positives,\n            \"false_positives\": self.false_positives,\n            \"true_negatives\": self.true_negatives,\n            \"false_negatives\": self.false_negatives\n        }\n\ndef run_benchmark(use_guardrails: bool = True) -> BenchmarkResult:\n    \"\"\"Run benchmark with or without Guardrails validation.\"\"\"\n    result = BenchmarkResult()\n    llm = ChatOpenAI(model=\"gpt-4o\", temperature=0.0, max_tokens=256, timeout=30)\n    guard = None\n    if use_guardrails:\n        guard = Guard(name=\"benchmark_guard\")\n        guard.use(HallucinationDetector(llm_api=llm.client.chat.completions.create, model=\"gpt-4o\", threshold=0.8))\n    \n    prompt_template = PromptTemplate(template=\"Answer factually: {prompt}\\nAnswer:\", input_variables=[\"prompt\"])\n    chain = LLMChain(llm=llm, prompt=prompt_template)\n\n    all_prompts: List[Tuple[str, bool]] = []\n    for p in FACTUAL_PROMPTS[:1000]:\n        all_prompts.append((p, False))  # False = not hallucinated\n    for p in HALLUCINATED_PROMPTS[:1000]:\n        all_prompts.append((p, True))  # True = hallucinated\n\n    for prompt, is_hallucinated in tqdm(all_prompts, desc=\"Running benchmark\"):\n        start_time = time.time()\n        try:\n            # Get LLM output\n            raw_output = chain.run(prompt=prompt).strip()\n            # Calculate LLM cost: $0.005 per 1k input tokens, $0.015 per 1k output tokens (GPT-4o pricing)\n            # Simplified: assume 100 input tokens, 50 output tokens per prompt\n            llm_cost = (100 / 1000 * 0.005) + (50 / 1000 * 0.015)  # $0.00095 per request\n\n            detected_hallucination = False\n            if use_guardrails:\n                validation = guard(raw_output, metadata={\"prompt\": prompt})\n                detected_hallucination = not validation.validation_passed\n                # Add Guardrails cost: $0.005 per validation (uses GPT-4o for hallucination check)\n                llm_cost += 0.005\n\n            latency_ms = (time.time() - start_time) * 1000\n            result.add_result(is_hallucinated, detected_hallucination, latency_ms, llm_cost)\n\n        except Exception as e:\n            logger.error(f\"Benchmark failed for prompt {prompt[:50]}: {e}\")\n            continue\n\n    return result\n\nif __name__ == \"__main__\":\n    logger.info(\"Starting benchmark: With Guardrails AI 0.5\")\n    with_guardrails = run_benchmark(use_guardrails=True)\n    logger.info(\"Starting benchmark: Without Guardrails\")\n    without_guardrails = run_benchmark(use_guardrails=False)\n\n    # Save results\n    results = {\n        \"with_guardrails\": with_guardrails.to_dict(),\n        \"without_guardrails\": without_guardrails.to_dict()\n    }\n    with open(\"benchmark_results.json\", \"w\") as f:\n        json.dump(results, f, indent=2)\n    logger.info(f\"Benchmark results saved to benchmark_results.json\")\n    print(json.dumps(results, indent=2))\n

\n# Tune hallucination threshold for high-stakes use case\nfrom guardrails.hub import HallucinationDetector\nfrom guardrails import Guard\n\nguard = Guard(name=\"tuned_guard\")\nguard.use(\n    HallucinationDetector(\n        llm_api=llm.client.chat.completions.create,\n        model=\"gpt-4o\",\n        threshold=0.7,  # Lower threshold for healthcare/finance\n        on_fail=OnFailAction.REASK,\n        max_reasks=1  # Reduce re-asks to limit latency\n    )\n)\n

\n# Use FIX instead of REASK for PII detection\nfrom guardrails.validators import DetectPII\nfrom guardrails import Guard, OnFailAction\n\nguard = Guard(name=\"pii_guard\")\nguard.use(\n    DetectPII(\n        entities=[\"EMAIL_ADDRESS\", \"PHONE_NUMBER\", \"SSN\", \"CREDIT_CARD_NUMBER\"],\n        on_fail=OnFailAction.FIX,  # Auto-redact instead of re-asking\n        redact_string=\"[PII REDACTED]\"  # Custom redaction label\n    )\n)\n

\nimport hashlib\nimport json\nimport redis\nfrom guardrails import Guard\n\n# Initialize Redis client\nr = redis.Redis(host=\"localhost\", port=6379, db=0, decode_responses=True)\n\ndef get_cache_key(prompt: str, output: str) -> str:\n    \"\"\"Generate cache key from prompt and output hash.\"\"\"\n    combined = f\"{prompt}{output}\"\n    return hashlib.sha256(combined.encode()).hexdigest()\n\ndef cached_validate(guard: Guard, output: str, prompt: str, ttl: int = 3600) -> dict:\n    \"\"\"Validate output with Redis caching.\"\"\"\n    cache_key = get_cache_key(prompt, output)\n    cached = r.get(cache_key)\n    if cached:\n        return json.loads(cached)\n    # Run validation if no cache hit\n    result = guard(output, metadata={\"prompt\": prompt})\n    r.setex(cache_key, ttl, json.dumps(result.dict()))  # 1 hour TTL by default\n    return result\n

\nlangchain-hallucination-demo/\n├── app/\n│   ├── __init__.py\n│   ├── main.py          # FastAPI service from Step 2\n│   ├── chain.py         # LangChain chain setup\n│   ├── guardrails.py    # Guardrails configuration\n│   ├── models.py        # Pydantic request/response models\n│   └── metrics.py       # Prometheus metrics setup\n├── benchmarks/\n│   ├── __init__.py\n│   ├── run_benchmark.py # Benchmark script from Step 3\n│   ├── dataset.py       # Benchmark dataset loader\n│   └── results/         # Benchmark result JSON files\n├── tests/\n│   ├── __init__.py\n│   ├── test_chain.py    # Unit tests for LangChain chain\n│   ├── test_guardrails.py # Unit tests for Guardrails validators\n│   └── test_api.py      # Integration tests for FastAPI service\n├── .env.example         # Example environment variables\n├── requirements.txt     # Python dependencies\n├── Dockerfile           # Containerization config\n├── docker-compose.yml   # Local dev environment\n└── README.md            # Setup and usage instructions\n