# Adding Governance Guardrails to Haystack 3.0 Pipelines with TealTiger

> Source: <https://dev.to/nagasatish_chilakamarti_2/adding-governance-guardrails-to-haystack-30-pipelines-with-tealtiger-113p>
> Published: 2026-08-18 08:54:50+00:00

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:

[TealTiger](https://github.com/agentguard-ai/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.

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:

```
pip install tealtiger-haystack
```

No separate adapter package needed — TealTiger works directly as a Haystack 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:

``` python
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,
            },
        }
```

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

``` python
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
```

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,
            },
        }
```

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")
```

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.

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

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

| 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 |

`pip install tealtiger haystack-ai`

TealTiger also integrates with [LangChain](https://pypi.org/project/langchain-tealtiger/), [AG2](https://docs.ag2.ai/extensions/tealtiger), and [MLflow](https://github.com/agentguard-ai/tealtiger/tree/main/packages/mlflow-tealtiger) — same governance engine, different frameworks.

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