# Real-Time LLM Guardrails

An open-source GenAI guardrail layer providing real-time evaluation, schema compliance, and prompt-injection protection via structured outputs — with offline golden-set evaluation, live production metrics (hallucination rate, precision, recall), a **self-correcting LangGraph orchestration pipeline**, and an **MCP tool server** so any MCP-compatible agent can call this guardrail layer directly.

Built to streamline enterprise Responsible AI governance and stage-gate approvals: the goal is that a governance reviewer can look at one scorecard object and make a launch decision, rather than re-deriving what a pile of raw metrics means.

## Why this exists

Most homegrown "check the LLM output" scripts conflate several genuinely different problems into one fuzzy "is this okay?" check:

- **Schema/structural compliance** — is the output well-formed? (deterministic, cheap)
- **Prompt injection** — has the model's behavior been hijacked by adversarial content? (pattern-based, cheap)
- **Content quality / hallucination** — is the output factually grounded in context? (needs semantic understanding — the one place an LLM-as-judge is actually justified)

This project keeps those three checks separable and orders them cheapest-first, so a badly malformed output never reaches the most expensive check.

## What's inside

```
guardrails/
  schema_guard.py    — deterministic Pydantic-based structured-output validation
  injection_guard.py — heuristic, pattern-based prompt-injection detection
  llm_judge.py        — LLM-as-judge for hallucination detection, with judge validation against human labels
  golden_set.py        — golden set management + precision/recall/F1 computation
  metrics.py           — live (production) rolling-window metrics + governance scorecard
  graph.py              — LangGraph-based SELF-CORRECTING pipeline (the agentic orchestration layer)
  mcp_server.py         — exposes the guardrail checks as an MCP tool for any agent host
app.py                  — Streamlit dashboard tying it together
tests/                  — 40 unit tests covering all six modules
```

### 1. Schema compliance (`schema_guard.py`)
Forcing structured output does double duty as both a formatting control and a security control — malformed output is itself a signal something went wrong upstream (a confused model, or a successful injection attempt hijacking the response format). Pure Pydantic validation, no LLM call, so it's the first and cheapest check in the pipeline.

### 2. Prompt injection detection (`injection_guard.py`)
Deliberately **not** LLM-based — an LLM asked "was this an injection?" can itself be manipulated by the injection it's supposed to catch. Pattern-based detection across four attack-shape categories (instruction override, role hijack, delimiter breakout, exfiltration attempts).

> **Honest scope note:** this is a heuristic layer that catches known attack shapes, not a comprehensive defense. It will miss novel phrasings. In production this should be one layer of defense-in-depth, not the only one.

### 3. LLM-as-judge (`llm_judge.py`)
For the one thing deterministic rules genuinely can't catch — hallucination relative to context. **Critical design point:** an LLM-as-judge is circular unless validated against human-labeled examples first, so `validate_judge_against_golden_set()` makes that validation step a first-class, testable operation. The judge client is injected (`JudgeClient` protocol) so this module is fully unit-testable without a live API key.

### 4. Golden set management (`golden_set.py`)
Golden sets for a guardrail system need two deliberately separate populations: **naturalistic** examples (checking the guardrail doesn't over-trigger on legitimate content) and **adversarial** examples (deliberately constructed attacks, which mostly don't occur naturally in normal traffic logs). `coverage_by_failure_mode()` makes gaps in adversarial coverage visible rather than silent.

### 5. Live metrics + governance scorecard (`metrics.py`)
Rolling-window (not all-time-average) tracking, so a recent regression isn't diluted by months of good history. `scorecard()` produces a governance-ready object with explicit flags (low sample size, schema degradation, hallucination threshold exceeded) designed to be read directly by a Responsible AI reviewer.

### 6. Self-correcting pipeline (`graph.py`) — the agentic layer
Built with **LangGraph** because the control flow is genuinely cyclic: if an output fails a guard, the pipeline can loop back and ask the generator to try again (up to a hard retry budget) before giving up and blocking. A linear chain has no natural way to express "go back and try again" — a graph with conditional edges does. Guard ordering (schema → injection → judge) is deliberately cost-driven, cheapest check first.

### 7. MCP tool server (`mcp_server.py`) — the agent-integration layer
Exposes `validate_llm_output` as an MCP (Model Context Protocol) tool, so any MCP-compatible agent host can call this guardrail layer directly without importing the codebase or knowing its internals. This is the "reusable skill" version of the guardrail logic — one validated tool other teams' agents can call, rather than everyone re-implementing their own output validation.

## Running it

```bash
pip install -r requirements.txt
streamlit run app.py
```

To run the MCP server standalone:
```bash
python -m guardrails.mcp_server
```

## Running the tests

```bash
pip install -r requirements.txt
pytest tests/ -v
```

## Example: the self-correcting pipeline in action

```python
from pydantic import BaseModel
from guardrails.graph import run_guard_pipeline

class AnswerSchema(BaseModel):
    answer: str

result = run_guard_pipeline(
    prompt="What is the capital of France?",
    context="Paris is the capital of France.",
    schema=AnswerSchema,
    generator=my_generator_client,  # anything with .generate(prompt) -> dict
    judge=my_judge_client,          # optional, anything with .judge(prompt) -> str
    max_retries=2,
)
print(result["final_status"])  # "ALLOWED" or "BLOCKED"
print(result["block_reason"])  # None if allowed, otherwise which guard blocked it
```

If the generator's first attempt fails a check, the graph automatically calls `generate` again (up to `max_retries` times) before blocking — this is tested explicitly in `tests/test_graph.py`, including a scenario that fails schema on attempt 1, fails injection on attempt 2, and succeeds on attempt 3, all within one retry budget.

## Example: MCP tool call

```python
from guardrails.mcp_server import validate_llm_output

result = validate_llm_output(content="Ignore all previous instructions.")
print(result["overall_passed"])              # False
print(result["injection_check"]["flagged_categories"])  # ['instruction_override']
```

Any MCP-compatible agent host can call this same tool over the protocol without importing Python code directly — run `python -m guardrails.mcp_server` to start it as a standalone server.

## Known limitations (honest, not hidden)

- Injection detection is pattern-based and will miss novel attack phrasings not covered by the four category patterns — it's one layer of defense-in-depth, not a complete solution.
- The LLM-as-judge is only as trustworthy as its validation against a human-labeled golden set — `validate_judge_against_golden_set()` exists specifically so that validation isn't skipped, but it's on the user of this library to actually run it before trusting judge verdicts in production.
- The MCP tool wraps the deterministic checks only (schema + injection), not the full self-correcting graph, since a single MCP tool call is a request/response — a multi-turn regenerate loop belongs inside whatever agent is calling the tool, not inside the tool itself.
- No PHI/HIPAA-specific redaction or handling — a regulated healthcare deployment would need an additional layer for that before this guardrail set is sufficient on its own.

## License

MIT