Haystack 3.0管道接入TealTiger治理引擎

Haystack 3.0 围绕可组合的管道重新设计了所有内容——你可以像搭乐高积木一样连接各种组件，构建 RAG、聊天和 Agent 工作流。但一旦这些管道进入生产环境，你需要回答类似这样的问题：

LLM 是否刚刚泄露了客户的 SSN？

这个 Agent 是否正在以 $50/小时的速度消耗 GPT-4 调用费用？

哪些工具调用被阻止了，原因是什么？

TealTiger 是一个开源治理引擎，以确定性方式回答这些问题——治理路径中无需 LLM，评估延迟低于 2ms，输出结构化的审计证据。

本文展示如何将 TealTiger 作为自定义组件接入 Haystack 3.0 管道。

为什么要在 Haystack 中加入治理

Haystack 管道功能强大，但默认是信任一切的。ChatGenerator 会欣然将 PII 发送给 OpenAI。ToolInvoker 会执行 LLM 请求的任何工具。在受监管的环境（医疗、金融、政府）中，这是迟早要发生的合规违规。

TealTiger 添加了一个治理层，能够：

在 PII 到达 LLM 之前将其阻止（40+ 正则表达式模式，零外部调用）

强制执行工具白名单——只有允许的工具才能执行

跟踪每个请求的成本并执行预算控制

为合规团队生成结构化审计收据（TEEC 格式）

pip install tealtiger-haystack

无需单独的适配器包——TealTiger 直接作为 Haystack 自定义组件工作。

集成模式：自定义组件

Haystack 3.0 的 @component 装饰器使这变得简单。我们创建一个 TealTigerGuard 组件，放在用户输入和 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,
            },
        }

构建受治理的 RAG 管道

以下是一个完整的管道，在用户查询到达 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

添加工具治理

对于 LLM 调用工具的 Agent 管道，你可以包装工具执行步骤：

@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 支持三种模式，映射到部署阶段：

# Shadow mode — see what would be blocked without breaking anything
guard = TealTigerGuard(policies=policies, mode="MONITOR")

你得到什么：结构化审计证据

每个治理决策都会生成结构化收据：

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

这可以直接接入 SOC2/HIPAA 合规工作流——无需手动解析日志。

TealTiger 的治理路径是确定性的（正则 + fnmatch，无 LLM 调用）：

PII 扫描（40 个模式）：约 1ms

工具白名单检查：<0.1ms

成本预算检查：<0.1ms

每个请求的总开销：1-2ms

作为对比，单次 LLM 调用需要 500-3000ms。治理带来的延迟可以忽略不计。

与 Haystack 内置选项的对比

立即体验：pip install tealtiger haystack-ai

完整文档：docs.tealtiger.ai/integrations

GitHub：github.com/agentguard-ai/tealtiger（欢迎 Star！）

Discord：加入社区

Haystack 文档：haystack-tealtger integration

TealTiger 还与 LangChain、AG2 和 MLflow 集成——同一个治理引擎，不同的框架。

TealTiger 采用 Apache 2.0 许可证。我们是 NVIDIA Inception 成员，致力于为 AI Agent 构建确定性治理方案。