[Caller Speech] --> [STT: Pulse] --> [Governance: PII scan] --> [LLM] --> [Governance: output scan] --> [TTS: Lightning] --> [Caller]

pip install tealtiger smallestai openai

from tealtiger import TealEngine, PolicyMode

# Configure governance for voice pipeline
engine = TealEngine(
    policies=[
        {"type": "pii_block", "categories": ["ssn", "credit_card", "phone", "email"]},
        {"type": "cost_limit", "max_per_session": 2.00},
        {"type": "tool_allowlist", "allowed": ["lookup_account", "schedule_*"]},
        {"type": "rate_limit", "max_calls": 20, "window": "5m"},
    ],
    mode=PolicyMode.ENFORCE,
)

# In your Smallest AI agent handler:
async def on_transcript(text: str):
    """Called after STT produces text from caller speech."""
    decision = engine.evaluate(content=text)
    
    if decision.action == "DENY":
        # PII detected — redact before sending to LLM
        redacted = engine.redact(text)
        return redacted
    
    return text  # Clean — pass through to LLM

import asyncio
from smallestai.voice import VoiceAgent
from tealtiger import TealEngine, PolicyMode
from tealtiger.cost import CostTracker, get_model_pricing

# Initialize governance engine
engine = TealEngine(
    policies=[
        {"type": "pii_block", "categories": ["ssn", "credit_card", "phone", "email", "iban"]},
        {"type": "secret_detection"},
        {"type": "cost_limit", "max_per_session": 2.00},
        {"type": "tool_allowlist", "allowed": ["lookup_account", "transfer_*", "schedule_*"]},
        {"type": "rate_limit", "max_calls": 20, "window": "5m"},
    ],
    mode=PolicyMode.ENFORCE,
    agent_id="voice-support-agent",
)

cost_tracker = CostTracker()


async def governed_on_transcript(text: str) -> str:
    """Governance hook after STT, before LLM."""
    decision = engine.evaluate(content=text)
    
    if decision.action == "DENY":
        # Log the violation for compliance
        print(f"[GOVERNANCE] Blocked: {decision.reason_codes}")
        # Redact PII and continue (don't disconnect the caller)
        return engine.redact(text)
    
    return text


async def governed_before_tts(response: str) -> str:
    """Governance hook after LLM, before TTS."""
    decision = engine.evaluate(content=response)
    
    if decision.action == "DENY":
        return engine.redact(response)
    
    return response


# Create governed voice agent
agent = VoiceAgent(
    api_key="your-smallest-api-key",
    on_transcript=governed_on_transcript,
    before_tts=governed_before_tts,
)

asyncio.run(agent.start())

# Start in MONITOR to see what would be caught without blocking callers
engine = TealEngine(policies=[...], mode=PolicyMode.MONITOR)

{
  "decision_id": "550e8400-e29b-41d4-a716-446655440000",
  "action": "DENY",
  "risk_score": 90,
  "reason_codes": ["PII_DETECTED:credit_card"],
  "policy_id": "pii_block",
  "evaluation_time_ms": 1.2,
  "agent_id": "voice-support-agent",
  "timestamp": "2026-08-17T14:30:00Z"
}

engine.export_evidence("call_audit.json")
engine.export_evidence("call_audit.sarif", format="sarif")