Dev.to Security 🔐 Cybersecurity Sun, 02 Aug 2026 01:24:53 +0000 👁 0 📖 6 min read

Securing Haystack Pipelines Against Runaway Costs and PII Leaks

Securing Haystack Pipelines Against Runaway Costs and PII Leaks TealTiger + Haystack | Official Integration | PyPI | Apache 2.0 TL;DR — Add a single governance component to your Haystack pipeline. It scans every docu

Securing Haystack Pipelines Against Runaway Costs and PII Leaks

TealTiger + Haystack | Official Integration | PyPI | Apache 2.0

TL;DR — Add a single governance component to your Haystack pipeline. It scans every document for PII and secrets, enforces cost budgets, and produces compliance-ready audit evidence. <5ms overhead. No LLM calls. No external services.

Once your Haystack pipelines start calling LLMs and retrieving documents from real data sources, three risks appear:

#	Risk	What happens	Impact
1	PII flows to the model	Retrieved doc contains customer SSN/CC	Compliance violation, data breach
2	Secrets in context	API keys in internal docs reach the LLM	Credential exposure
3	Cost runaway	Agent loop iterates indefinitely	$1000+ overnight bill

This tutorial shows how to prevent all three with a deterministic governance component that adds <5ms overhead to your pipeline — no additional LLM calls, no external services.

What we're building

A Haystack RAG pipeline with governance enforcement:

PII detection and redaction before content reaches the LLM
Secret scanning (500+ patterns) on retrieved documents
Per-session cost budget with hard stop
Structured audit trail for compliance evidence

Architecture

How it decides

Prerequisites

Python 3.9+
An OpenAI API key (for the LLM step)
10 minutes

Install

pip install tealtiger-haystack haystack-ai

tealtiger-haystack is an official Haystack community integration.

Step 1: Set up TealTiger governance

import os
from tealtiger_haystack import TealTigerGovernanceComponent

os.environ["OPENAI_API_KEY"] = "your-key-here"

# Zero-config OBSERVE mode — tracks everything, blocks nothing
governance = TealTigerGovernanceComponent()

That's it for observe mode. Every document that passes through this component gets scanned for PII, secrets, and cost — but nothing is blocked. You get visibility first.

Step 2: Build a governed RAG pipeline

from haystack import Pipeline, Document
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.components.builders import ChatPromptBuilder
from haystack.dataclasses import ChatMessage

# Build the pipeline
pipeline = Pipeline()
pipeline.add_component("governance", governance)
pipeline.add_component("prompt_builder", ChatPromptBuilder())
pipeline.add_component("llm", OpenAIChatGenerator(model="gpt-4o-mini"))

# Connect: governance scans documents before they reach the prompt
pipeline.connect("governance.clean_documents", "prompt_builder.documents")
pipeline.connect("prompt_builder", "llm")

The TealTigerGovernanceComponent sits between your retriever and the prompt builder. It receives documents, scans them, and outputs cleaned documents.

Step 3: Test with safe content

safe_docs = [
    Document(content="Q4 revenue was $2.4B, up 12% year-over-year."),
]

result = pipeline.run({
    "governance": {"documents": safe_docs},
    "prompt_builder": {
        "template": [
            ChatMessage.from_user("Based on the documents: {{documents}}, answer: What was Q4 revenue?")
        ]
    },
})

print(result["llm"]["replies"][0].text)
# → "Q4 revenue was $2.4B, up 12% year-over-year."

# Check governance decision
decisions = governance.get_decisions()
print(decisions[-1]["action"])  # "ALLOW"
print(decisions[-1]["findings"])  # [] — nothing found

Clean pass. No PII, no secrets. The governance component allowed it through.

Step 4: Switch to ENFORCE mode

Now let's block dangerous content:

governance_enforced = TealTigerGovernanceComponent(
    mode="ENFORCE",
    pii_categories=["ssn", "credit_card", "email"],
    secret_scan=True,
    budget_per_session_usd=1.00,
)

# Rebuild pipeline with enforced governance
pipeline_enforced = Pipeline()
pipeline_enforced.add_component("governance", governance_enforced)
pipeline_enforced.add_component("prompt_builder", ChatPromptBuilder())
pipeline_enforced.add_component("llm", OpenAIChatGenerator(model="gpt-4o-mini"))
pipeline_enforced.connect("governance.clean_documents", "prompt_builder.documents")
pipeline_enforced.connect("prompt_builder", "llm")

Step 5: Test PII detection

pii_docs = [
    Document(
        content="Customer Jane Doe, SSN: 000-00-0000, has a balance of $4,200.",
        meta={"source": "customer_records"},
    ),
]

result = pipeline_enforced.run({
    "governance": {"documents": pii_docs},
    "prompt_builder": {
        "template": [
            ChatMessage.from_user("Based on the documents: {{documents}}, summarize the customer info.")
        ]
    },
})

# Check what happened
decisions = governance_enforced.get_decisions()
last = decisions[-1]
print(last["action"])      # "DENY" or "REDACT"
print(last["findings"])    # [{"type": "pii", "category": "ssn", ...}]
print(last["reason"])      # "PII detected: ssn in document content"

The SSN was detected and the document was either redacted or blocked before reaching the LLM — depending on your policy configuration.

Step 6: Test secret detection

secret_docs = [
    Document(
        content="The staging server uses password: EXAMPLE_PASSWORD_PLACEHOLDER and connects to db.internal.example.com.",
        meta={"source": "internal_docs"},
    ),
]

result = pipeline_enforced.run({
    "governance": {"documents": secret_docs},
    "prompt_builder": {
        "template": [
            ChatMessage.from_user("Based on the documents: {{documents}}, what's the DB connection info?")
        ]
    },
})

decisions = governance_enforced.get_decisions()
last = decisions[-1]
print(last["action"])      # "DENY"
print(last["findings"])    # [{"type": "secret", "category": "password", ...}]

Step 7: Cost budget enforcement

# After many pipeline runs, the budget is exceeded
for i in range(100):
    try:
        pipeline_enforced.run({
            "governance": {"documents": safe_docs},
            "prompt_builder": {
                "template": [ChatMessage.from_user("Question " + str(i))]
            },
        })
    except Exception as e:
        print(f"Stopped at iteration {i}: {e}")
        break

# The governance component stops the pipeline when cumulative cost exceeds $1.00

Step 8: Inspect the audit trail

Every governance decision is recorded with full context:

{
  "correlation_id": "7f3a2b1c-...",
  "timestamp": "2026-08-01T12:34:56.789Z",
  "action": "DENY",
  "risk_score": 0.92,
  "reason_codes": ["PII_DETECTED"],
  "reason": "SSN found in document content",
  "findings": [{"type": "pii", "category": "ssn"}],
  "evaluation_ms": 1.2,
  "cumulative_cost": 0.0847
}

audit_trail = governance_enforced.get_decisions()

for decision in audit_trail[-3:]:
    print(f"""
    Correlation ID: {decision['correlation_id']}
    Action:         {decision['action']}
    Risk Score:     {decision['risk_score']}
    Findings:       {len(decision.get('findings', []))} issues
    Latency:        {decision['evaluation_time_ms']:.1f}ms
    Cost So Far:    ${decision.get('cumulative_cost', 0):.4f}
    """)

This audit trail is your compliance evidence. Each record links to a specific pipeline run via correlation ID, shows exactly what was found, and proves governance was evaluated — whether the outcome was ALLOW, DENY, or REDACT.

The governance progression

Start conservative, tighten over time:

Mode	Behavior	When to use
`OBSERVE`	Logs everything, blocks nothing	Day 1 — understand what your pipeline sees
`MONITOR`	Logs violations, still allows	Week 1 — validate your policies
`ENFORCE`	Blocks violations	Production — real enforcement

# Start here
TealTigerGovernanceComponent(mode="OBSERVE")

# Then
TealTigerGovernanceComponent(mode="MONITOR")

# Then
TealTigerGovernanceComponent(mode="ENFORCE", pii_categories=["ssn", "credit_card"], budget_per_session_usd=5.0)

Performance

The governance component is deterministic — regex-based pattern matching, no LLM calls in the governance path:

Metric	Value
Evaluation latency	0.5 - 2ms (typical)
Max latency	<5ms
External API calls	Zero — runs in-process
Secret patterns	500+
Method	Deterministic regex (reproducible, auditable)

For comparison: a typical LLM call takes 500-3000ms. TealTiger adds <1% overhead to your pipeline.

What this solves

Risk	❌ Without governance	✅ With TealTiger
PII in documents	Flows to LLM undetected	Detected & redacted/blocked before LLM
Secrets in retrieval	Embedded in model context	Scanned (500+ patterns) and blocked
Cost runaway	Unbounded token spend	Hard budget cap with auto-stop
Compliance audit	Manual logging, no evidence	Structured per-request evidence (JSONL)

What's next?

If you found this useful, TealTiger has integrations for 14+ frameworks:

Python: haystack (this tutorial), ag2, crewai, pydanticai, google-adk, composio, strands, openhands, llamaindex

TypeScript: copilotkit, vercel-ai-sdk, tealtiger-ai-sdk

Observability: phoenix (arize), langfuse, agentops, hindsight (vectorize)

Each follows the same pattern: pip install tealtiger-<framework> or npm install tealtiger-<framework>.

TealTiger is open-source (Apache 2.0), NVIDIA Inception member. No LLM in the governance path — deterministic, auditable, <5ms.

agentguard-ai / tealtiger

Powerful protection for AI agents - Open-source security and cost tracking for AI applications

TealTiger

AI Agent Security & Governance SDK

Deterministic governance, guardrails, cost tracking, and policy management for LLM applications Open source. TypeScript + Python. Works with any provider.

Website · Documentation · Examples · Discord · Contributing

⚡ 60-second quickstart

Install: npm install tealtiger or pip install tealtiger, then wrap one existing OpenAI call:

import { TealOpenAI } from 'tealtiger';
const client = new TealOpenAI({ apiKey: process.env.OPENAI_API_KEY, guardrails: { promptInjection: true } });
const res = await client.chat.completions.create({ model: 'gpt-4o-mini', messages: [{ role: 'user', content: 'Hello!' }] });
console.log(res.security?.decision ?? 'ALLOW');

import os
from tealtiger import TealOpenAI
client = TealOpenAI(api_key=os.environ["OPENAI_API_KEY"],

…

View on GitHub

📰 Read the original article on Dev.to Security

Originally published by Dev.to Security. Aggregated on AIWithGhost for educational purposes — full credit and traffic to the original publisher.

← News 🔐 Cybersecurity 🏠 Home

Securing Haystack Pipelines Against Runaway Costs and PII Leaks

Securing Haystack Pipelines Against Runaway Costs and PII Leaks

What we're building

Architecture

How it decides

Prerequisites

Install

Step 1: Set up TealTiger governance

Step 2: Build a governed RAG pipeline

Step 3: Test with safe content

Step 4: Switch to ENFORCE mode

Step 5: Test PII detection

Step 6: Test secret detection

Step 7: Cost budget enforcement

Step 8: Inspect the audit trail

The governance progression

Performance

What this solves

Links

What's next?

agentguard-ai / tealtiger

Powerful protection for AI agents - Open-source security and cost tracking for AI applications

TealTiger

⚡ 60-second quickstart