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 - 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. Top comments (0)