pipecat-guardrails

Output guardrails for Pipecat voice agents. It checks what your bot is about to say, and stops it before it is spoken.

Status: early development. The rule layer (banned terms, regex, personal data, India pack) and guard.check() are built. The Pipecat processor, classifier and LLM judge are next. See PLAN.md.

Quick start

Everything is opt-in: the guard only runs the rules you give it.

from pipecat_guardrails import Guard
from pipecat_guardrails.packs import india
from pipecat_guardrails.rules import PII, BannedTerms, pii

guard = Guard(
    rules=[
        BannedTerms(["acme corp", "globex"], id="no-competitors"),
        BannedTerms(
            ["dawai band kar do", "chinta mat"],
            id="no-medical-advice",
            word_pattern=india.hinglish_word_pattern,   # also matches "davai", "cheenta", ...
            variants={"dawai": ["दवाई"]},
        ),
        PII(pii.email(), pii.card(), india.aadhaar(), india.phone(), id="pii"),  # redacts
    ],
    fallback="Sorry, I can't help with that.",
)

verdict = await guard.check("Aap davaai band kar do.")
verdict.blocked     # True
verdict.text        # "Sorry, I can't help with that."
verdict.rule_ids    # ("no-medical-advice",)

• Actions: each rule can block (speak the fallback instead), redact (replace the match, e.g. [phone number]), or flag (report only).
• Opt out: pass enabled=False to a rule, or call guard.disable("pii") at runtime.
• Shadow mode: Guard(..., mode="shadow") reports what it would do without changing anything, so you can try new rules on live traffic.
• Errors: a rule that crashes fails closed (blocks) by default. Use on_error="allow" to fail open instead.
• Custom rules: any function works:

  @rule(id="no-urls")
  def no_urls(text: str) -> bool:
      return "http" in text
  

Runnable examples, the same health bot in two languages: examples/whole_reply_english.py and examples/whole_reply_hinglish.py.

The problem

Pipecat has no built-in output guardrail, so every team writes its own processor between the LLM and the TTS. These homemade processors tend to break on user interruptions, keep speaking after a violation, let blocked text into the LLM history, and miss paraphrases and Hindi/Hinglish text.

This package will provide one processor you add to your pipeline:

pipeline = Pipeline([
    transport.input(), stt, context_aggregator.user(), llm,
    OutputGuardProcessor(guard),   # checks each sentence before it is spoken
    tts, transport.output(), context_aggregator.assistant(),
])

There will also be a plain await guard.check(text) for apps that run their own LLM loop.

See BEFORE_AFTER.md for a side-by-side comparison and PLAN.md for the design and roadmap.

Install

pip install pipecat-guardrails          # includes Pipecat
pip install "pipecat-guardrails[jev]"   # also installs the TypeSafe Jev classifier client

Requires Python 3.11+ and Pipecat 1.12 or later (installed automatically).

How checking works

Each sentence goes through up to three layers, and stops at the first one that is sure:

1. Banned words, PII, regex: local and instant.
2. Classifier model (optional), such as TypeSafe Jev: a fast yes/no model with an API key. A confident answer decides; an unsure one moves to layer 3.
3. LLM judge (optional): any OpenAI-compatible LLM gives the final answer.

Layers 2 and 3 use Pipecat's own classifiers (pipecat.classifiers), so any classifier Pipecat supports works here.

Development

uv sync                      # create .venv with dev tools
uv run pytest                # tests
uv run ruff check .          # lint
uv run ruff format .         # format
uv run pyright               # type check
uv run pre-commit install    # run the checks on every commit

License

BSD-2-Clause, the same as Pipecat.