Haystack Integration

Add deterministic governance to your Haystack pipelines with custom components. PII detection, tool authorization, cost limits, secret scanning, and structured audit evidence — no LLM in the governance path.

Package: tealtiger-haystack on PyPI
Haystack Version: v3 and above
License: Apache-2.0
Latency: Sub-2ms per evaluation
Source: GitHub

Installation

pip install tealtiger-haystack

This installs the TealTiger Haystack adapter with the governance engine as a dependency.

Quick Start

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


# Simplest approach: observe() wraps any callable
@observe(
    policies=[
        {"type": "pii_block", "categories": ["ssn", "credit_card", "email"]},
        {"type": "cost_limit", "max_per_session": 1.00},
    ],
    mode="ENFORCE",
)
def governed_chat(messages):
    generator = OpenAIChatGenerator(model="gpt-4o-mini")
    return generator.run(messages=messages)


result = governed_chat(messages=[ChatMessage.from_user("What is AI governance?")])

Custom Component Pattern

For full pipeline integration, create a TealTigerGuard component that slots into any Haystack pipeline:

from haystack import component, Pipeline
from haystack.dataclasses import ChatMessage
from tealtiger import TealTiger
from tealtiger.core.engine.types import PolicyMode, DecisionAction


@component
class TealTigerGuard:
    """Governance guardrail component for Haystack pipelines.
    
    Evaluates governance policies on user messages before they reach
    downstream components (LLMs, retrievers, tool invokers).
    """

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

    @component.output_types(
        messages=list,
        blocked=bool,
        decision=dict,
    )
    def run(self, messages: list):
        """Evaluate governance on incoming messages.
        
        Args:
            messages: List of ChatMessage objects from the pipeline
            
        Returns:
            messages: Passed through if allowed, empty list if blocked
            blocked: Whether the request was blocked
            decision: Structured governance decision
        """
        # 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 policies
        decision = self.client.evaluate(
            content=user_text,
            metadata={"pipeline": "haystack", "component": "TealTigerGuard"},
        )

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

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

Governed RAG Pipeline

A complete retrieval-augmented generation pipeline with governance:

from haystack import Pipeline
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.components.retrievers.in_memory import InMemoryBM25Retriever
from haystack.components.builders import ChatPromptBuilder
from haystack.dataclasses import ChatMessage

# Define governance policies
policies = [
    {"type": "pii_block", "categories": ["ssn", "credit_card", "email", "phone"]},
    {"type": "secret_detection"},
    {"type": "cost_limit", "max_per_session": 0.50},
    {"type": "tool_allowlist", "allowed": ["search", "lookup_*", "calculate"]},
]

# Build pipeline
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")

# Safe query — passes governance
result = pipe.run({
    "governance": {
        "messages": [ChatMessage.from_user("What are best practices for AI governance?")]
    }
})
print(result["llm"]["replies"][0].content)

# PII in query — blocked before reaching OpenAI
result = pipe.run({
    "governance": {
        "messages": [ChatMessage.from_user("Look up records for SSN 123-45-6789")]
    }
})
print(result["governance"]["blocked"])  # True
print(result["governance"]["decision"]["reason_codes"])  # ["PII_DETECTED:ssn"]
# The LLM never sees the SSN

Tool Governance Component

For agentic Haystack pipelines with tool use:

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

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

    @component.output_types(allowed=bool, decision=dict)
    def run(self, tool_name: str, tool_args: dict):
        """Evaluate governance on a tool call before execution.
        
        Args:
            tool_name: Name of the tool being invoked
            tool_args: Arguments being passed to the tool
            
        Returns:
            allowed: Whether the tool call is permitted
            decision: Structured governance decision
        """
        decision = self.client.evaluate(
            content=str(tool_args),
            tool_name=tool_name,
        )

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

Governance Capabilities

PII Detection

Scans messages for 40+ PII patterns before they leave your infrastructure:

policies = [
    {
        "type": "pii_block",
        "categories": ["ssn", "credit_card", "email", "phone", "iban"],
        "action": "BLOCK",  # or "REDACT"
    }
]

Detected patterns: SSN, credit cards, IBAN, phone numbers, email addresses, IP addresses, AWS keys, passport numbers, dates of birth, and 30+ more.

Secret Scanning

Catches leaked credentials before they reach external APIs:

policies = [
    {"type": "secret_detection"},  # 500+ patterns: AWS keys, GitHub tokens, DB URLs...
]

Tool Authorization

Control which tools agents can invoke:

policies = [
    {"type": "tool_allowlist", "allowed": ["search", "read_*", "calculator"]},
]

Supports glob patterns — read_* matches read_file, read_database, etc.

Cost Governance

Hard budget enforcement per session:

policies = [
    {
        "type": "cost_limit",
        "max_per_session": 1.00,
        "max_daily": 50.00,
    }
]

Prompt Injection Detection

Deterministic pattern matching for injection attacks:

policies = [
    {"type": "prompt_injection", "action": "BLOCK", "sensitivity": "high"}
]

Governance Modes

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

# Start in MONITOR to observe, graduate to ENFORCE
guard = TealTigerGuard(policies=policies, mode="MONITOR")

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

Access decisions programmatically:

guard = TealTigerGuard(policies=policies)
# After pipeline runs...
for decision in guard.client.decisions:
    print(f"{decision['action']} — {decision['reason_codes']} (risk: {decision['risk_score']})")

Performance

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

Check	Latency
PII scan (40 patterns)	~1ms
Tool allowlist	under 0.1ms
Cost budget	under 0.1ms
Secret detection	~1ms
Total per request	1–2ms

For comparison, a single LLM call takes 500–3000ms. Governance adds negligible overhead.

Why Not Just Validate in Code?

	TealTiger	Custom validators	No governance
PII detection	40+ patterns, zero config	Write your own regex	❌
Tool allowlisting	Glob patterns, FREEZE rules	Manual if/else	❌
Cost tracking	Per-request with budgets	DIY token counting	❌
Audit evidence	Structured TEEC format	DIY JSON logging	❌
Governance modes	ENFORCE/MONITOR/REPORT_ONLY	Feature flags	❌
Secret scanning	500+ patterns	Build your own	❌
Framework lock-in	None (works anywhere)	Haystack-specific	N/A

OWASP AI Security Coverage

TealTiger addresses 7/10 OWASP Agentic Security Issues:

OWASP ASI	Coverage	How
ASI-01: Excessive Agency	✅	Tool allowlist + FREEZE rules
ASI-02: Inadequate Sandboxing	✅	Argument validation
ASI-03: Prompt Injection	✅	Deterministic pattern detection
ASI-04: Sensitive Info Disclosure	✅	PII detection + redaction
ASI-05: Improper Output Handling	✅	Output scanning
ASI-07: Inadequate Error Handling	✅	Circuit breaker
ASI-09: Insufficient Monitoring	✅	Full audit trail

Complete Example

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


@component
class TealTigerGuard:
    def __init__(self, policies, mode="ENFORCE", agent_id="haystack-agent"):
        self.client = TealTiger(
            policies=policies, mode=PolicyMode(mode), agent_id=agent_id
        )

    @component.output_types(messages=list, blocked=bool, decision=dict)
    def run(self, messages: list):
        user_text = next(
            (m.content for m in reversed(messages) if m.role.value == "user"), ""
        )
        decision = self.client.evaluate(content=user_text)

        if decision.action == DecisionAction.DENY:
            return {"messages": [], "blocked": True, "decision": {
                "action": "DENY",
                "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
        }}


# Build governed pipeline
pipe = Pipeline()
pipe.add_component("guard", TealTigerGuard(
    policies=[
        {"type": "pii_block", "categories": ["ssn", "credit_card", "email"]},
        {"type": "secret_detection"},
        {"type": "cost_limit", "max_per_session": 1.00},
        {"type": "prompt_injection", "action": "BLOCK"},
    ],
    mode="ENFORCE",
))
pipe.add_component("llm", OpenAIChatGenerator(model="gpt-4o-mini"))
pipe.connect("guard.messages", "llm.messages")

# Run
result = pipe.run({"guard": {"messages": [
    ChatMessage.from_user("Research AI governance frameworks for 2026")
]}})
print(result["llm"]["replies"][0].content)

Next Steps

LangChain Integration — Full-lifecycle governance with LangChain middleware
AG2 Integration — Built-in governance extension for AG2 agents
Policy Authoring Guide — Full policy configuration reference
OWASP Policy Pack — Map TealTiger to OWASP Agentic Security Issues

Getting Started

Core Concepts

Governance Domains

Policy Authoring

Architecture

Integrations

Deployment

API Reference

Cookbook

Compliance & GRC

Guides

Reference

Versions

Playground

Dashboard

Blog

Haystack Integration

Haystack Integration

Installation

Quick Start

Custom Component Pattern

Governed RAG Pipeline

Tool Governance Component

Governance Capabilities

PII Detection

Secret Scanning

Tool Authorization

Cost Governance

Prompt Injection Detection

Governance Modes

Audit Evidence

Performance

Why Not Just Validate in Code?

OWASP AI Security Coverage

Complete Example

Next Steps

​Haystack Integration

​Installation

​Quick Start

​Custom Component Pattern

​Governed RAG Pipeline

​Tool Governance Component

​Governance Capabilities

​PII Detection

​Secret Scanning

​Tool Authorization

​Cost Governance

​Prompt Injection Detection

​Governance Modes

​Audit Evidence

​Performance

​Why Not Just Validate in Code?

​OWASP AI Security Coverage

​Complete Example

​Next Steps

Haystack Integration

Installation

Quick Start

Custom Component Pattern

Governed RAG Pipeline

Tool Governance Component

Governance Capabilities

PII Detection

Secret Scanning

Tool Authorization

Cost Governance

Prompt Injection Detection

Governance Modes

Audit Evidence

Performance

Why Not Just Validate in Code?

OWASP AI Security Coverage

Complete Example

Next Steps