Adding Governance Guardrails to Haystack 3.0 Pipelines with TealTiger

Adding Governance Guardrails to Haystack 3.0 Pipelines with TealTiger

Haystack 3.0 redesigned everything around composable pipelines — you connect components like LEGO bricks to build RAG, chat, and agent workflows. But once those pipelines hit production, you need answers to questions like:

• Did the LLM just leak a customer's SSN?
• Is this agent burning $50/hour on GPT-4 calls?
• Which tool calls got blocked and why?

TealTiger is an open-source governance engine that answers these deterministically — no LLM in the governance path, sub-2ms evaluation, structured audit evidence.

This post shows how to wire TealTiger into a Haystack 3.0 pipeline as a custom component.

Why Governance in Haystack?

Haystack pipelines are powerful but trust-everything by default. A ChatGenerator will happily pass PII to OpenAI. A ToolInvoker will execute any tool the LLM requests. In regulated environments (healthcare, finance, government), that's a compliance violation waiting to happen.

TealTiger adds a governance layer that:

• Blocks PII before it reaches the LLM (40+ regex patterns, zero external calls)
• Enforces tool allowlists — only permitted tools execute
• Tracks costs per-request with budget enforcement
• Emits structured audit receipts (TEEC format) for compliance teams

Installation

pip install tealtiger-haystack

No separate adapter package needed — TealTiger works directly as a Haystack custom component.

The Integration Pattern: Custom Component

Haystack 3.0's @component decorator makes this straightforward. We create a TealTigerGuard component that sits in the pipeline between user input and the LLM:

from haystack import component, Pipeline
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.dataclasses import ChatMessage

from tealtiger import TealTiger
from tealtiger.core.engine.types import PolicyMode


@component
class TealTigerGuard:
    """Governance guardrail component for Haystack 3.0 pipelines."""

    def __init__(
        self,
        policies: dict,
        mode: str = "ENFORCE",
        agent_id: str = "haystack-agent",
    ):
        self.engine = TealTiger(
            policies=policies,
            mode=PolicyMode(mode),
            agent_id=agent_id,
        )

    @component.output_types(
        messages=list,  # List[ChatMessage] — passed through if allowed
        blocked=bool,
        decision=dict,
    )
    def run(self, messages: list):
        # Extract text from the last user message
        user_text = ""
        for msg in reversed(messages):
            if msg.role.value == "user":
                user_text = msg.content
                break

        # Evaluate governance
        decision = self.engine.evaluate(
            content=user_text,
            tool_name=None,
            metadata={"pipeline": "haystack", "component": "TealTigerGuard"},
        )

        if decision.action == "DENY":
            return {
                "messages": [],
                "blocked": True,
                "decision": {
                    "action": decision.action,
                    "reason_codes": [str(rc) for rc in decision.reason_codes],
                    "risk_score": decision.risk_score,
                },
            }

        return {
            "messages": messages,
            "blocked": False,
            "decision": {
                "action": "ALLOW",
                "reason_codes": ["POLICY_COMPLIANT"],
                "risk_score": 0,
            },
        }

Building a Governed RAG Pipeline

Here's a complete pipeline that scans user queries before they reach the LLM:

from haystack import Pipeline
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.dataclasses import ChatMessage


# Define governance policies
policies = {
    "pii_block": {
        "enabled": True,
        "categories": ["ssn", "credit_card", "email", "phone"],
    },
    "cost_limit": {
        "enabled": True,
        "max_per_session": 0.50,  # $0.50 per session
    },
    "tool_allowlist": {
        "enabled": True,
        "allowed": ["search", "lookup_*", "calculate"],
    },
}

# Create pipeline
pipe = Pipeline()
pipe.add_component("governance", TealTigerGuard(policies=policies, mode="ENFORCE"))
pipe.add_component("llm", OpenAIChatGenerator(model="gpt-4o-mini"))

# Connect: governance output → LLM input (only if not blocked)
pipe.connect("governance.messages", "llm.messages")

# Run with a safe query
result = pipe.run({
    "governance": {
        "messages": [ChatMessage.from_user("What is the capital of France?")]
    }
})
print(result["llm"]["replies"][0].content)
# → "The capital of France is Paris."

# Run with PII — gets blocked before reaching OpenAI
result = pipe.run({
    "governance": {
        "messages": [ChatMessage.from_user("My SSN is 123-45-6789, look up my records")]
    }
})
print(result["governance"]["blocked"])  # True
print(result["governance"]["decision"]["reason_codes"])  # ["PII_DETECTED"]
# LLM never sees the SSN

Adding Tool Governance

For agentic pipelines where the LLM calls tools, you can wrap the tool execution step:

@component
class TealTigerToolGuard:
    """Guards tool invocations in agent pipelines."""

    def __init__(self, policies: dict, mode: str = "ENFORCE"):
        self.engine = TealTiger(policies=policies, mode=PolicyMode(mode))

    @component.output_types(allowed=bool, decision=dict)
    def run(self, tool_name: str, tool_args: dict):
        decision = self.engine.evaluate(
            content=str(tool_args),
            tool_name=tool_name,
        )

        return {
            "allowed": decision.action == "ALLOW",
            "decision": {
                "action": decision.action,
                "risk_score": decision.risk_score,
                "reason_codes": [str(rc) for rc in decision.reason_codes],
                "tool_name": tool_name,
            },
        }

Governance Modes

TealTiger supports three modes that map to deployment stages:

Mode	Behavior	Use Case
ENFORCE	Blocks violations	Production with strict compliance
MONITOR	Logs violations but allows through	Staging / shadow mode
REPORT_ONLY	Skips evaluation, always allows	Development / dry-run

# Shadow mode — see what would be blocked without breaking anything
guard = TealTigerGuard(policies=policies, mode="MONITOR")

What You Get: Structured Audit Evidence

Every governance decision produces a structured receipt:

{
  "decision_id": "550e8400-e29b-41d4-a716-446655440000",
  "action": "DENY",
  "risk_score": 85,
  "reason_codes": ["PII_DETECTED:ssn"],
  "policy_id": "pii_block",
  "evaluation_time_ms": 0.8,
  "agent_id": "haystack-agent",
  "correlation_id": "trace-abc-123",
  "timestamp": "2026-08-17T10:30:00Z"
}

This feeds directly into SOC2/HIPAA compliance workflows — no manual log parsing.

Performance

TealTiger's governance path is deterministic (regex + fnmatch, no LLM calls):

• PII scan (40 patterns): ~1ms
• Tool allowlist check: <0.1ms
• Cost budget check: <0.1ms
• Total overhead per request: 1-2ms

For comparison, a single LLM call takes 500-3000ms. Governance adds negligible latency.

How This Compares to Haystack's Built-in Options

	TealTiger	Custom validators	No governance
PII detection	40+ patterns, zero config	Write your own regex	❌
Tool allowlisting	Built-in with glob patterns	Write your own	❌
Cost tracking	Per-request with budgets	DIY with token counters	❌
Audit receipts	Structured TEEC format	DIY JSON	❌
Governance modes	ENFORCE/MONITOR/REPORT_ONLY	DIY	❌
Framework lock-in	None (works anywhere)	Haystack-specific	N/A

Next Steps

1. Try it: pip install tealtiger haystack-ai
2. Full docs: docs.tealtiger.ai/integrations
3. GitHub: github.com/agentguard-ai/tealtiger (⭐ appreciated!)
4. Discord: Join the community
5. Haystack Docs:haystack-tealtger integration

TealTiger also integrates with LangChain, AG2, and MLflow — same governance engine, different frameworks.

---

TealTiger is Apache 2.0 licensed. We're an NVIDIA Inception member building deterministic governance for AI agents.