I got obsessed with a very specific workflow bug: every time my team exported an OpenAPI spec into Terraform-friendly artifacts, the “glue” logic lived in hand-written Markdown docs and ad-hoc scripts. The docs were the source of truth—but the scripts were the ones people actually ran. That mismatch caused subtle drift: sometimes an endpoint got updated in OpenAPI, but the Markdown examples (and therefore the generated mappings) lagged behind.

So I built a domain-specific LLM pipeline that treats a *particular* kind of Markdown as an input language—specifically, fenced code blocks inside `` sections—and turns them into a deterministic Terraform mapping plan.

The main idea: don’t build a generic “chatbot.” Build a small “compiler” that understands a tiny, boring DSL embedded in Markdown, and only then let the LLM help with the messy bits (schema-to-type mapping, naming conventions, and example normalization).

## What I built: a Markdown-to-Terraform compiler

My input Markdown looks like this:

- It contains sections like:

```markdown

## endpoint: GET /v1/widgets/{widgetId}
resource: aws_lambda_function.widgets_get_widget
request:
  path:
    widgetId: string
response:
  status: 200
  body:
    type: Widget
    schemaRef: components.schemas.Widget
mappingRules:
  - name: widgetId
    from: path.widgetId
    as: var.widget_id
```

- The compiler:
  1. Extracts those sections.
  2. Parses the YAML payload under each `` block.
  3. Uses an LLM to resolve `schemaRef` into Terraform type expressions.
  4. Emits a Terraform mapping file with stable output formatting.

The niche part is the Markdown compiler: most domain-specific LLM efforts stop at RAG (retrieval) or generic codegen. Mine treats Markdown as a formal-ish interface and compiles it like a program.

## The “domain model” I taught the system

I needed an explicit contract so the LLM didn’t improvise.

### Terms
- **OpenAPI**: an API specification format describing endpoints and JSON schemas.
- **Terraform**: infrastructure-as-code tool; it wants typed expressions like `string`, `map(string)`, `object({ ... })`, etc.
- **Domain DSL**: the mini language I defined inside Markdown.

### The strict input/output shape
The Markdown compiler always produces:

- `plan.tf.json` (Terraform JSON form so it’s easy to validate)
- a `diagnostics` list for any unresolved schemas

## End-to-end code (working)

Below is a complete, runnable example in Python that:
- extracts the special Markdown sections
- parses the YAML inside them
- calls an LLM to resolve a schema reference into a Terraform type
- writes a Terraform plan JSON

> This uses OpenAI’s API, but the structure works the same with any chat-completions provider.

### 1) Project layout

```text
md-openapi-terraform-compiler/
  main.py
  sample.md
  schemas.json
  requirements.txt
```

`requirements.txt`:

```text
openai
pyyaml
pydantic
```

Install:

```bash
pip install -r requirements.txt
```

### 2) Sample Markdown input

`sample.md`:

```markdown
# Release Notes


## endpoint: GET /v1/widgets/{widgetId}
resource: aws_lambda_function.widgets_get_widget
request:
  path:
    widgetId: string
response:
  status: 200
  body:
    type: Widget
    schemaRef: components.schemas.Widget
mappingRules:
  - name: widgetId
    from: path.widgetId
    as: var.widget_id


## endpoint: POST /v1/widgets
resource: aws_lambda_function.widgets_post_widget
request:
  body:
    schemaRef: components.schemas.NewWidget
response:
  status: 201
  body:
    schemaRef: components.schemas.WidgetCreated
mappingRules:
  - name: payload
    from: body
    as: var.new_widget
```

### 3) A tiny OpenAPI schemas “index”

I’m simulating the OpenAPI schema registry using a local JSON file that maps schema refs to a simplified JSON schema representation.

`schemas.json`:

```json
{
  "components.schemas.Widget": {
    "type": "object",
    "properties": {
      "id": { "type": "string" },
      "name": { "type": "string" },
      "tags": { "type": "array", "items": { "type": "string" } }
    },
    "required": ["id", "name"]
  },
  "components.schemas.NewWidget": {
    "type": "object",
    "properties": {
      "name": { "type": "string" },
      "tags": { "type": "array", "items": { "type": "string" } }
    },
    "required": ["name"]
  },
  "components.schemas.WidgetCreated": {
    "type": "object",
    "properties": {
      "widget": { "$ref": "components.schemas.Widget" },
      "createdAt": { "type": "string", "format": "date-time" }
    },
    "required": ["widget", "createdAt"]
  }
}
```

### 4) The compiler (main.py)

```python
import json
import re
from typing import Any, Dict, List

import yaml
from openai import OpenAI


OPENAPI_BLOCK_RE = re.compile(
    r"\s*(.*?)\s*(?=|$)",
    re.DOTALL,
)

ENDPOINT_HEADER_RE = re.compile(r"^##\s*endpoint:\s*(.+?)\s*$", re.MULTILINE)


def extract_blocks(markdown: str) -> List[str]:
    return [m.group(1).strip() for m in OPENAPI_BLOCK_RE.finditer(markdown)]


def parse_block(block: str) -> Dict[str, Any]:
    """
    Each block starts with:
      ## endpoint: GET /v1/widgets/{widgetId}
    then a YAML document with keys like resource, request, response, mappingRules.
    """
    header_match = ENDPOINT_HEADER_RE.search(block)
    if not header_match:
        raise ValueError("Missing '## endpoint: ...' header in openapi-to-terraform block")

    endpoint = header_match.group(1).strip()

    # Remove the markdown header line so the rest is pure YAML
    block_wo_header = re.sub(r"^##\s*endpoint:\s*.+?\s*$", "", block, flags=re.MULTILINE).strip()

    data = yaml.safe_load(block_wo_header) or {}
    data["endpoint"] = endpoint
    return data


def terraform_type_from_json_schema(schema: Dict[str, Any], max_depth: int = 5) -> str:
    """
    This is a deterministic fallback.
    The LLM will handle tricky $ref cases, but we still want a local mapper.
    """
    if max_depth <= 0:
        return "any"

    if "$ref" in schema:
        # Leave unresolved for the LLM.
        return "any"

    t = schema.get("type")
    if t == "string":
        fmt = schema.get("format")
        if fmt == "date-time":
            return "string"  # Terraform doesn't have datetime types; keep as string
        return "string"
    if t == "integer":
        return "number"
    if t == "number":
        return "number"
    if t == "boolean":
        return "bool"
    if t == "array":
        items = schema.get("items", {})
        return f"list({terraform_type_from_json_schema(items, max_depth=max_depth-1)})"
    if t == "object":
        props = schema.get("properties", {})
        required = set(schema.get("required", []))
        # Terraform object({ ... }) is strict about keys; we keep it permissive by making keys optional-like
        # via a map conversion. This avoids brittle plans for missing fields.
        # If you want strict objects, replace with object({ ... }).
        if not props:
            return "map(string)"
        fields = []
        for k, v in props.items():
            fields.append(f"{k}={terraform_type_from_json_schema(v, max_depth=max_depth-1)}")
        # Use object({ ... }) form
        return "object({" + ", ".join(fields) + "})"

    return "any"


def resolve_schema_ref_with_llm(client: OpenAI, schemas_index: Dict[str, Any], schema_ref: str) -> Dict[str, str]:
    """
    Ask the LLM to produce a Terraform type expression for a given OpenAPI schema ref.
    The LLM sees:
      - the schema_ref name
      - the referenced schema JSON
      - already-known rules

    Output is structured JSON with:
      - terraform_type
      - explanation (for diagnostics)
    """
    if schema_ref not in schemas_index:
        return {"terraform_type": "any", "explanation": "schemaRef not found in index"}

    schema = schemas_index[schema_ref]
    fallback = terraform_type_from_json_schema(schema)

    system = (
        "You are a domain-specific compiler for OpenAPI schemas into Terraform type expressions. "
        "Return only valid JSON that matches the output schema."
    )

    user = {
        "schemaRef": schema_ref,
        "openapiSchema": schema,
        "deterministicFallback": fallback,
        "rules": [
            "Map OpenAPI primitive types to Terraform: string->string, integer/number->number, boolean->bool.",
            "Map arrays to list(<itemType>).",
            "Map objects to object({key=type,...}). If schema has no properties, use map(string).",
            "If schema contains $ref, resolve it conceptually to the referenced schema type if possible; otherwise use any.",
            "Prefer correctness over strictness; using 'any' is acceptable when uncertain."
        ],
        "Output format": {
            "terraform_type": "string",
            "explanation": "string"
        }
    }

    # NOTE: replace `gpt-4.1-mini` with your preferred model
    resp = client.chat.completions.create(
        model="gpt-4.1-mini",
        messages=[
            {"role": "system", "content": system},
            {"role": "user", "content": json.dumps(user)}
        ],
        temperature=0.0,
        response_format={"type": "json_object"},
    )

    content = resp.choices[0].message.content
    return json.loads(content)


def compile_markdown_to_terraform_plan(markdown_path: str, schemas_path: str, out_path: str, api_key: str):
    client = OpenAI(api_key=api_key)

    markdown = open(markdown_path, "r", encoding="utf-8").read()
    schemas_index = json.load(open(schemas_path, "r", encoding="utf-8"))

    blocks = extract_blocks(markdown)
    if not blocks:
        raise ValueError("No  blocks found")

    compiled: Dict[str, Any] = {
        "version": 1,
        "mappings": []
    }

    diagnostics: List[Dict[str, str]] = []

    for block in blocks:
        data = parse_block(block)

        endpoint = data["endpoint"]
        resource = data.get("resource")
        request = data.get("request", {}) or {}
        response = data.get("response", {}) or {}

        mapping_entry: Dict[str, Any] = {
            "endpoint": endpoint,
            "resource": resource,
            "request": {},
            "response": {},
            "mappingRules": data.get("mappingRules", [])
        }

        # Resolve schemaRef for request body if present
        req_body = request.get("body") if isinstance(request, dict) else None
        if isinstance(req_body, dict) and "schemaRef" in req_body:
            schema_ref = req_body["schemaRef"]
            resolved = resolve_schema_ref_with_llm(client, schemas_index, schema_ref)
            mapping_entry["request"]["bodyTerraformType"] = resolved["terraform_type"]
            diagnostics.append({
                "endpoint": endpoint,
                "schemaRef": schema_ref,
                "resolution": resolved["terraform_type"],
                "explanation": resolved["explanation"]
            })
        # Path params are already simple primitives in this DSL
        if isinstance(request, dict) and "path" in request:
            mapping_entry["request"]["path"] = request["path"]

        # Resolve schemaRef for response body if present
        resp_body = response.get("body") if isinstance(response, dict) else None
        if isinstance(resp_body, dict) and "schemaRef" in resp_body:
            schema_ref = resp_body["schemaRef"]
            resolved = resolve_schema_ref_with_llm(client, schemas_index, schema_ref)
            mapping_entry["response"]["bodyTerraformType"] = resolved["terraform_type"]
            diagnostics.append({
                "endpoint": endpoint,
                "schemaRef": schema_ref,
                "resolution": resolved["terraform_type"],
                "explanation": resolved["explanation"]
            })

        compiled["mappings"].append(mapping_entry)

    plan = {
        "terraform": {
            # This is a lightweight representation. In real life you might generate
            # HCL or terraform-provider-specific resources.
            "mappingPlan": compiled
        },
        "diagnostics": diagnostics
    }

    with open(out_path, "w", encoding="utf-8") as f:
        json.dump(plan, f, indent=2)
    print(f"Wrote {out_path}")


if __name__ == "__main__":
    # Usage:
    #   export OPENAI_API_KEY="..."
    #   python main.py
    import os
    compile_markdown_to_terraform_plan(
        markdown_path="sample.md",
        schemas_path="schemas.json",
        out_path="plan.tf.json",
        api_key=os.environ["OPENAI_API_KEY"],
    )
```

## What happens when I run it

```bash
export OPENAI_API_KEY="YOUR_KEY"
python main.py
```

It reads `sample.md`, finds both `` blocks, and parses each YAML payload. Then it calls the model for each `schemaRef` it sees:

- `components.schemas.Widget` → generates a Terraform type for an object with `id`, `name`, `tags`
- `components.schemas.NewWidget` → generates Terraform type for request body
- `components.schemas.WidgetCreated` → generates type for response body, possibly using a `$ref` inside

Finally it writes `plan.tf.json`.

A typical output (shape) looks like:

```json
{
  "terraform": {
    "mappingPlan": {
      "version": 1,
      "mappings": [
        {
          "endpoint": "GET /v1/widgets/{widgetId}",
          "resource": "aws_lambda_function.widgets_get_widget",
          "request": {
            "path": {
              "widgetId": "string"
            }
          },
          "response": {
            "bodyTerraformType": "object({id=string, name=string, tags=list(string)})"
          },
          "mappingRules": [
            {
              "name": "widgetId",
              "from": "path.widgetId",
              "as": "var.widget_id"
            }
          ]
        }
      ]
    }
  },
  "diagnostics": [
    {
      "endpoint": "GET /v1/widgets/{widgetId}",
      "schemaRef": "components.schemas.Widget",
      "resolution": "object({id=string, name=string, tags=list(string)})",
      "explanation": "Mapped OpenAPI object properties to Terraform object attributes; array tags to list(string)."
    }
  ]
}
```

## Why this is “domain-specific” (and not generic codegen)

The LLM isn’t free-form. It’s constrained to answer one question:

> “Given this specific `schemaRef`, what Terraform type expression should I use?”

Meanwhile, the compiler logic is deterministic:
- Markdown extraction
- YAML parsing
- plan structure
- stable JSON output

That combination matters. In my experiments, letting the model “write the plan” directly led to unstable key ordering, inconsistent schemas, and occasional hallucinated fields. Keeping the compiler strict made the system reliable even when the LLM was uncertain (it falls back to `any`).

## A small reliability upgrade I added: deterministic fallback

Notice the `terraform_type_from_json_schema` function. It generates a best-effort Terraform type without calling the LLM. I used it as:

- a baseline shown to the model (`deterministicFallback`)
- a guardrail for cases where the `schemaRef` is missing

This pattern is a practical way to keep a domain-specific LLM grounded in local, testable logic.

## Conclusion

I built a niche Domain-Specific LLM system that compiles a very specific Markdown DSL (`` blocks) into a Terraform mapping plan, using the LLM only for the schema-ref-to-typed-expression step. The big win from my tinkering was reliability: deterministic parsing and strict plan structure prevented drift, while the LLM handled the messy schema translation work.

Building A Markdown Compiler Agent For Openapi Terraform Mappers

## Why I got obsessed with “prompt tampering”
I hit a weird problem while building a small generative workflow: two prompts that *looked* identical produced different outputs after a few seemingly harmless changes (whitespace, punctuation, even a different line-ending). Worse, in one setup the prompt content could be “touched” by another system stage—so I started worrying about *accidental* or *malicious* prompt edits.

I wanted a practical, engineering-minded way to detect whether the prompt text I’m about to send is exactly the one I tested—especially when the prompt is dynamically assembled.

So I built a “prompt checksum” mechanism that works like this:

- Generate a normalized representation of the prompt (so line endings and trailing spaces don’t cause false mismatches).
- Compute a stable hash over it.
- Embed the checksum into the prompt itself.
- At runtime, verify the checksum before accepting the model output.

This is not magic. But it’s a surprisingly effective guardrail for systems where determinism matters.

---

## What I mean by “prompt checksum”
A **checksum** is a short fingerprint (like a hash) that changes dramatically if the underlying text changes.

In my case, I used:

- **Normalization**: convert `\r\n` to `\n`, trim trailing whitespace, remove a few predictable variations.
- **Hashing**: SHA-256 of the normalized prompt bytes.
- **Verification**: include the checksum in the prompt and validate it on the way out.

A key detail: the checksum is computed locally from the *exact prompt text being sent*—so it’s not relying on the model to “understand” correctness.

---

## The full working example (Python)

### 1) Prompt normalization + checksum
```python
import hashlib
import json
from dataclasses import dataclass

def normalize_prompt_text(text: str) -> str:
    """
    Normalizes text so accidental formatting differences don't affect the checksum.
    - Converts Windows line endings to Unix.
    - Strips trailing whitespace on each line.
    - Removes leading/trailing whitespace for the whole prompt.
    """
    text = text.replace("\r\n", "\n").replace("\r", "\n")
    lines = text.split("\n")
    lines = [line.rstrip() for line in lines]
    normalized = "\n".join(lines).strip()
    return normalized

def prompt_checksum(prompt_text: str) -> str:
    """
    Produces a deterministic SHA-256 checksum over the normalized prompt.
    Returns hex digest.
    """
    normalized = normalize_prompt_text(prompt_text)
    digest = hashlib.sha256(normalized.encode("utf-8")).hexdigest()
    return digest

@dataclass
class PromptSpec:
    system: str
    user_template: str  # e.g. includes {input}
    variables: dict

    def render_user_prompt(self) -> str:
        return self.user_template.format(**self.variables)

    def full_prompt_text(self) -> str:
        # I keep the checksum tied to the exact serialized structure sent to the model.
        payload = {
            "system": normalize_prompt_text(self.system),
            "user": normalize_prompt_text(self.render_user_prompt()),
        }
        # Stable serialization: sort keys and avoid whitespace changes.
        return json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":"))

    def checksum(self) -> str:
        return prompt_checksum(self.full_prompt_text())
```

**What’s happening and why:**
- `normalize_prompt_text` prevents false positives due to line-ending and trailing whitespace differences.
- `prompt_checksum` hashes the normalized prompt deterministically.
- `PromptSpec.full_prompt_text()` serializes the system/user content into a canonical JSON string. That makes the checksum robust even when prompts are generated in multiple steps.

---

### 2) Embed the checksum and verify it on output
For real LLM APIs, you typically send messages like `{role: "system"}` and `{role: "user"}`. Here I’ll keep it simple and show the “embed + verify” pattern around a plain text response.

```python
import re

CHECKSUM_RE = re.compile(r"PROMPT_CHECKSUM:\s*([0-9a-f]{64})")

def build_checked_prompt(prompt_spec: PromptSpec) -> tuple[str, str]:
    """
    Returns (system_message, user_message) with checksum embedded into the user message.
    """
    system_message = prompt_spec.system

    rendered_user = prompt_spec.render_user_prompt()
    checksum = prompt_spec.checksum()

    user_message = (
        f"{rendered_user}\n\n"
        f"PROMPT_CHECKSUM: {checksum}\n"
        f"Return only the requested output. Do not include the checksum."
    )
    return system_message, user_message

def verify_response_checksum(response_text: str, expected_checksum: str) -> None:
    """
    Ensures the model didn't echo back or alter the checksum.
    In this pattern we expect the model NOT to output the checksum.
    If it does, that's a signal something is off.
    """
    matches = CHECKSUM_RE.findall(response_text or "")
    if matches:
        # If the model printed a checksum, compare it to the expected one.
        # This is a sanity check: models shouldn't include it per our instruction.
        for m in matches:
            if m != expected_checksum:
                raise ValueError(f"Checksum mismatch detected: got {m}, expected {expected_checksum}")
        raise ValueError("Model unexpectedly echoed the checksum.")

def strip_example_output(response_text: str) -> str:
    """
    Example: clean up a hypothetical wrapper the model might include.
    Not required for checksumming, just shows typical post-processing.
    """
    return (response_text or "").strip()
```

**Why this verification approach works (in practice):**
- The checksum is embedded so any downstream system could detect prompt integrity.
- The output verification is intentionally strict: I treat “model echoed checksum” as a failure mode (because I instructed it not to). That catches prompt/format drift early.

---

### 3) Demonstrate tamper detection with a fake “LLM” run
To keep this blog post self-contained, I’ll simulate the model response behavior.

```python
def fake_llm(system_message: str, user_message: str) -> str:
    """
    Simulates an LLM response.
    In a real integration, you'd call an API here.
    """
    # Extract checksum from the user_message to show how tampering could be detected.
    m = CHECKSUM_RE.search(user_message)
    checksum = m.group(1) if m else None

    # Simulate correct behavior: do NOT echo checksum.
    # Pretend the model returned deterministic output.
    return "OUTPUT: {name: 'Ada', score: 42}"

def main():
    system = (
        "You are a data extraction engine. "
        "Produce deterministic JSON-like output."
    )
    user_template = "Extract {field} from the given text: {text}"

    spec = PromptSpec(
        system=system,
        user_template=user_template,
        variables={
            "field": "name",
            "text": "Ada scored 42 points in the contest."
        },
    )

    expected_checksum = spec.checksum()
    system_message, user_message = build_checked_prompt(spec)

    response = fake_llm(system_message, user_message)
    verify_response_checksum(response, expected_checksum)
    print("Verified response:", strip_example_output(response))

    # Now simulate tampering: change whitespace or content in the prompt builder.
    spec_tampered = PromptSpec(
        system=system + " ",  # subtle change
        user_template=user_template,
        variables=spec.variables,
    )
    tampered_checksum = spec_tampered.checksum()
    print("Expected checksum:", expected_checksum)
    print("Tampered checksum:", tampered_checksum)

    if tampered_checksum != expected_checksum:
        print("Tampering detected: checksum changed.")
    else:
        print("No tampering detected (unexpected).")

if __name__ == "__main__":
    main()
```

**What you’ll see when you run it**
- A verified response printed successfully for the intact prompt.
- Then a checksum difference printed for the “tampered” prompt spec.
- The mechanism does exactly one job: detect prompt changes reliably.

---

## Making it stronger: checksum over *messages* not just strings
In the example above, I hash a canonical JSON string containing:
- normalized `system`
- normalized rendered `user`

In a real pipeline, I’d compute the checksum over the actual message objects I send to the API (after normalization and stable serialization). That prevents a common bug: hashing a prompt *template* while the real runtime prompt includes additional scaffolding (tool instructions, safety blocks, etc.).

A common mistake I made once: hashing only the “user” part but then injecting formatting into the “system” part later. The checksum passed while the effective prompt changed. Hashing the full message set fixed that.

---

## A practical pattern for production pipelines
This is the pattern I ended up using:

1. **Build prompt messages** as explicit variables in code.
2. **Normalize and hash** the exact messages you will send.
3. **Embed checksum** (optional, but helpful for observability).
4. **Verify early**:
   - at prompt-build time (internal assertion)
   - at response time (output sanity checks)

If the checks fail, the system stops before you store “bad provenance” outputs.

---

## Conclusion
I built a deterministic **prompt checksum** flow for generative AI that fingerprints the *exact normalized messages* sent to an LLM using SHA-256, embeds that checksum into the prompt text, and verifies response behavior to catch drift and tampering. The big takeaway from my tinkering is that reliability problems often come from tiny prompt differences and hidden injection steps—hashing the canonical prompt representation turns that uncertainty into a measurable, enforceable guardrail.

Building A “Fuzz-Resistant” Prompt Checksum For Deterministic Llm Outputs

I got curious about a very specific failure mode in enterprise Retrieval-Augmented Generation (RAG): even when the retriever finds the “right” clauses, the final answer can still accidentally quote or summarize clauses the user shouldn’t see—especially when the knowledge base contains many similar contracts.

In my case, the risky pattern was **clause-level leakage**: a contract clause might mention “termination,” but different clauses for different contract types (MSA vs. SOW, enterprise vs. channel) use near-identical wording. So the embeddings retrieved the neighborhood, and the generator filled in the exact clause text—sometimes across the boundary of what the user is authorized to access.

What I built to make this reliable was a **pre-retrieval regex filter**—a lightweight gate that rejects chunks that don’t match an allowlisted clause identity *before* they ever reach the model. Below is the concrete setup I used for a “contract clause RAG guardrail” that I can run in a small pipeline.

---

## The niche problem: “termination” clauses that look the same across contracts

Enterprise contract corpora often store many documents in one index. A user might ask:

> “What is the termination notice period?”

A naive RAG pipeline typically does:

1. Retrieve top-k relevant chunks by embedding similarity  
2. Stuff them into a prompt  
3. Generate an answer

The guardrail failure happens because “termination notice period” is semantically similar across many contracts. Even if authorization checks exist, they may only run **after** generation (too late), or they may be document-level rather than clause-level.

So I needed clause identity to survive the journey from retrieval → generation.

---

## What I implemented: regex-filtered retrieval using clause fingerprints

### Core idea

Before retrieval results are used, I apply a regex-based “fingerprint” filter:

- Each chunk in the index includes lightweight metadata like:
  - `contract_type` (e.g., `MSA`, `SOW`)
  - `clause_id` (e.g., `TERMINATION_NOTICE`)
  - `termination_regex_hint` (a canonical snippet pattern)
- At query time, I:
  1. Detect requested clause intent (“termination notice period” → `TERMINATION_NOTICE`)
  2. Apply an allowlist mapping from clause intent → acceptable clause fingerprints
  3. Filter retrieved chunks by matching `clause_id` and validating the clause text with a regex

This is intentionally old-school: regex isn’t “smart,” but it’s deterministic. Determinism is the point for a guardrail.

### Why pre-retrieval filter (not post-filter)

If I wait until after generation, I can’t reliably recover from leakage. If I gate chunks before they become prompt context, the model can’t output what it never saw.

---

## Working example: a tiny in-memory “enterprise RAG” with guardrails

This demo uses:
- A simple embedding model (via `sentence-transformers`)
- A vector store (in-memory)
- A regex guardrail that filters chunks by clause identity

> You can run this end-to-end locally; it doesn’t call external vector DBs.

### 1) Install dependencies

```bash
pip install sentence-transformers numpy
```

### 2) Create sample clause chunks

```python
import re
from dataclasses import dataclass
from typing import Dict, Any, List

@dataclass
class Chunk:
    text: str
    meta: Dict[str, Any]

chunks: List[Chunk] = [
    Chunk(
        text="TERMINATION NOTICE: Either party may terminate this MSA upon thirty (30) days' written notice.",
        meta={
            "contract_type": "MSA",
            "clause_id": "TERMINATION_NOTICE",
            # Canonical fingerprint: what the clause typically contains
            "termination_regex_hint": r"thirty\s*$\s*30\s*$\s*days'?\s*written\s*notice"
        }
    ),
    Chunk(
        text="TERMINATION FOR CAUSE: Either party may terminate for material breach with fifteen (15) days to cure.",
        meta={
            "contract_type": "MSA",
            "clause_id": "TERMINATION_FOR_CAUSE",
            "termination_regex_hint": r"fifteen\s*$\s*15\s*$\s*days\s*to\s*cure"
        }
    ),
    Chunk(
        text="TERMINATION NOTICE: The SOW may be terminated by the client upon 45 days' written notice.",
        meta={
            "contract_type": "SOW",
            "clause_id": "TERMINATION_NOTICE",
            "termination_regex_hint": r"45\s*days'?\s*written\s*notice"
        }
    ),
    Chunk(
        text="TERMINATION NOTICE: Either party may terminate upon twenty (20) business days' written notice.",
        meta={
            "contract_type": "CHANNEL",
            "clause_id": "TERMINATION_NOTICE",
            "termination_regex_hint": r"twenty\s*$\s*20\s*$\s*business\s*days'?\s*written\s*notice"
        }
    ),
]
```

I intentionally included multiple `TERMINATION_NOTICE` clauses so semantic retrieval can easily grab the “wrong” one.

---

## 3) Build a minimal vector store + embedding

```python
import numpy as np
from sentence_transformers import SentenceTransformer

model = SentenceTransformer("all-MiniLM-L6-v2")

texts = [c.text for c in chunks]
embeddings = model.encode(texts, normalize_embeddings=True)

# simple cosine similarity index
def retrieve(query: str, top_k: int = 3):
    q_emb = model.encode([query], normalize_embeddings=True)[0]
    scores = embeddings @ q_emb
    idxs = np.argsort(-scores)[:top_k]
    return [(chunks[i], float(scores[i])) for i in idxs]
```

---

## 4) Add clause-intent detection and the regex guardrail

### Clause intent mapping

In a real enterprise system this could be:
- a classifier,
- a rules engine,
- or a small LLM prompt.

For the demo, I’ll use deterministic keyword→clause_id mapping.

```python
def detect_clause_intent(query: str) -> str:
    q = query.lower()
    if "termination" in q and "notice" in q:
        return "TERMINATION_NOTICE"
    if "termination" in q and "cure" in q:
        return "TERMINATION_FOR_CAUSE"
    return "UNKNOWN"
```

### Authorization/allowlist filter (clause fingerprint gate)

The guardrail enforces:

- clause identity must match (`clause_id`)
- text must match the canonical regex fingerprint (`termination_regex_hint`)

```python
def guardrail_filter(
    retrieved: List[tuple[Chunk, float]],
    query: str,
    allowed_contract_types: List[str],
) -> List[tuple[Chunk, float]]:
    intent = detect_clause_intent(query)

    filtered = []
    for chunk, score in retrieved:
        if chunk.meta.get("clause_id") != intent:
            continue
        if chunk.meta.get("contract_type") not in allowed_contract_types:
            continue

        # Validate the clause's signature text pattern
        hint = chunk.meta.get("termination_regex_hint")
        if hint:
            if not re.search(hint, chunk.text, flags=re.IGNORECASE):
                continue

        filtered.append((chunk, score))

    return filtered
```

This is the key: even if the retriever gets close, the guardrail can veto chunks that don’t match the specific clause form the user asked for *and* the contract types they’re allowed to see.

---

## 5) Compose the “RAG answer” prompt context (without leaking context)

In a real system I’d call an LLM. Here I’ll show the exact assembled context and then extract the notice period deterministically to keep the demo fully runnable.

```python
def extract_notice_period(text: str) -> str:
    # Very small extraction logic for the demo
    patterns = [
        r"thirty\s*$\s*30\s*$\s*days'?\s*written\s*notice",
        r"(\d+)\s*days'?\s*written\s*notice",
        r"twenty\s*$\s*20\s*$\s*business\s*days'?\s*written\s*notice",
    ]
    for p in patterns:
        m = re.search(p, text, flags=re.IGNORECASE)
        if m:
            return m.group(0)
    return "UNKNOWN"

def rag_answer(query: str, allowed_contract_types: List[str], top_k: int = 4):
    retrieved = retrieve(query, top_k=top_k)
    filtered = guardrail_filter(retrieved, query, allowed_contract_types)

    # If nothing survives the guardrail, return a safe refusal/empty answer
    if not filtered:
        return {
            "intent": detect_clause_intent(query),
            "context_used": [],
            "answer": "No authorized matching clause found.",
        }

    # Use top surviving chunk(s)
    filtered_sorted = sorted(filtered, key=lambda x: -x[1])
    context = [c.text for c, _ in filtered_sorted]

    # Demo answer extraction from the best matching clause
    best_text = context[0]
    notice = extract_notice_period(best_text)

    return {
        "intent": detect_clause_intent(query),
        "retrieved": [(c.meta, s) for c, s in retrieved],
        "context_used": context,
        "answer": f"Termination notice clause found: {notice}",
    }
```

---

## 6) Run it: showing exactly what happens

### Case A: Allowed only for `MSA`

```python
result = rag_answer(
    query="What is the termination notice period?",
    allowed_contract_types=["MSA"],
    top_k=4
)
print(result["retrieved"])
print("----")
print(result["context_used"])
print("----")
print(result["answer"])
```

Expected behavior:
- Retriever may return chunks from `SOW` or `CHANNEL` because they’re semantically similar (“termination notice”).
- Guardrail keeps only `MSA` chunks and validates the regex fingerprint.
- Final answer uses the `30 days` notice clause.

### Case B: Allowed only for `SOW`

```python
result = rag_answer(
    query="What is the termination notice period?",
    allowed_contract_types=["SOW"],
    top_k=4
)
print("----")
print(result["context_used"])
print("----")
print(result["answer"])
```

Expected behavior:
- The system should ignore the `MSA` notice chunk.
- It should use the `SOW` notice clause (`45 days`).

### Case C: Allowed for none

```python
result = rag_answer(
    query="What is the termination notice period?",
    allowed_contract_types=[],
    top_k=4
)
print(result["answer"])
```

Expected behavior:
- No chunk survives the filter.
- The pipeline refuses safely: “No authorized matching clause found.”

---

## Why this works in the real world

This pattern is effective when:
- You can define a **clause identity** (like `TERMINATION_NOTICE`)
- You can define a **signature** (regex fingerprint, a canonical phrase template, or even a checksum of a “clause heading”)
- You need deterministic rejection before the LLM sees the text

It’s not trying to replace embeddings. It’s complementing them with a gate that protects against cross-clause and cross-contract leakage.

---

## Productionizing notes I learned the hard way

1. **Regex hints must be specific but not brittle.**  
   Too strict, and legitimate clauses fail. Too loose, and leakage slips through.

2. **Store fingerprint metadata alongside chunks.**  
   Keeping `clause_id` and `regex_hint` per chunk avoids guessing at query time.

3. **Filter before prompt assembly.**  
   Once the model sees restricted text, “forgetting” isn’t guaranteed.

4. **Log both stages.**  
   I logged `retrieved` vs `context_used`. That single diff made debugging much faster.

---

I built a contract clause RAG guardrail by adding a deterministic regex-based fingerprint filter that runs before generation. Instead of trusting semantic similarity alone, I used clause intent detection plus allowlisted contract types and a clause signature check to prevent prompt context from containing unauthorized or mismatched termination clauses.

Artificial Intelligence Posts

Building A Markdown Compiler Agent For Openapi Terraform Mappers

Building A “Fuzz-Resistant” Prompt Checksum For Deterministic Llm Outputs

Building A Contract Clause Rag Guardrail Using Regex-Filter Before Retrieval