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:

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:

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):
        user_text = ""
        for msg in reversed(messages):
            if msg.role.value == "user":
                user_text = msg.content
                break

        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:

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

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"],
    },
}

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

pipe.connect("governance.messages", "llm.messages")

result = pipe.run({
    "governance": {
        "messages": [ChatMessage.from_user("What is the capital of France?")]
    }
})
print(result["llm"]["replies"][0].content)

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

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:

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.

pip install tealtiger haystack-ai

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.