DEV Community

Cover image for Securing Haystack Pipelines Against Runaway Costs and PII Leaks
nagasatish chilakamarti
nagasatish chilakamarti

Posted on

Securing Haystack Pipelines Against Runaway Costs and PII Leaks

#haystack #python #ai #security

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

Architecture

How it decides

Decision Flow

Prerequisites

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

Install 

pip install tealtiger-haystack haystack-ai
Enter fullscreen mode Exit fullscreen mode

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()
Enter fullscreen mode Exit fullscreen mode

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")
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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")
Enter fullscreen mode Exit fullscreen mode

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"
Enter fullscreen mode Exit fullscreen mode

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", ...}]
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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
}
Enter fullscreen mode Exit fullscreen mode 
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}
    """)
Enter fullscreen mode Exit fullscreen mode

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:

Governance Modes

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)
Enter fullscreen mode Exit fullscreen mode

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)

Links

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.

GitHub logo agentguard-ai / tealtiger

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

TealTiger

TealTiger Logo

AI Agent Security & Governance SDK

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

npm version PyPI version License: Apache 2.0 Discord GitHub stars Governed by TealTiger OpenSSF Scorecard


NVIDIA Inception Program

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');
Enter fullscreen mode Exit fullscreen mode 
import os
from tealtiger import TealOpenAI
client = TealOpenAI(api_key=os.environ["OPENAI_API_KEY"],
…
Enter fullscreen mode Exit fullscreen mode
View on GitHub

Top comments (0)