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. Bu
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
-
Try it:
pip install tealtiger haystack-ai - Full docs: docs.tealtiger.ai/integrations
- GitHub: github.com/agentguard-ai/tealtiger (⭐ appreciated!)
- Discord: Join the community
- 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.
Originally published by Dev.to Security. Aggregated on AIWithGhost for educational purposes — full credit and traffic to the original publisher.