【2026年4月】Pipecat完全ガイド｜音声AIエージェント実装

「音声AIエージェントを作りたいけど、WebRTC・STT・LLM・TTSを自前でつなぐのが面倒…」

実際にクライアントのコールセンター自動化プロジェクトを手伝ったとき、最初に直面したのがこの問題でした。Deepgram・OpenAI・Cartesiaそれぞれの公式SDKを組み合わせようとすると、音声ストリームのバッファリングだけで数日かかります。

そのときに採用したのが Pipecat です。Daily社が開発するオープンソースのPythonフレームワークで、STT・LLM・TTS・WebRTCトランスポートを統一的なパイプラインとして組み合わせられます。2026年4月にv1.1.0がリリースされ、Mistral STT（Voxtral Realtime API）への対応など機能が大幅に拡張されました。

この記事では、Pipecat v1.1.0を使った音声AIエージェントの実装をコード付きで解説します。最小構成から始め、本番向けのマルチサービス選択・コスト最適化まで一気にカバーします。

Pipecatとは何か――設計思想と他との違い

Pipecatは 「パイプライン」 という概念でリアルタイム音声処理を抽象化します。STT（音声→テキスト）→LLM（テキスト→テキスト）→TTS（テキスト→音声）という流れを、フレームという単位でストリーミング処理します。

重要なのは「モジュラー性」です。Deepgramを使うかWhisperを使うかをコード1行で切り替えられ、LLMもOpenAI・Anthropic・Geminiを差し替えられます。100以上のAIサービスに公式サポートがあります。

他のフレームワークとの根本的な差

観点	Pipecat	LiveKit Agents	Vapi
ライセンス	OSS（BSD-2）	OSS（Apache 2.0）	マネージドSaaS
カスタマイズ性	★★★★★ フル制御	★★★★☆ 高い	★★☆☆☆ 限定的
セットアップ難度	中（Python知識必要）	中高（インフラ知識も必要）	低（APIキーのみ）
本番コスト（月1万分）	〜$300-500*	〜$400-700*	〜$700-3,300*
レイテンシ目標	<500ms（設定次第）	<500ms	〜700ms-1s
マルチモーダル対応	音声・動画・画像	音声・動画	音声のみ

*コストはSTT・LLM・TTS・インフラ込みの参考値。実際の数値は選択するサービスにより変動します（参照日: 2026-04-28）。

Pipecatが向いているのは「月10万分以上のスケール」「HIPAA/SOC2などコンプライアンス要件あり」「既存システムとの深い統合」が必要なケースです。一方、プロトタイプを1週間で出したい場合はVapiから始める選択も合理的です。

セットアップ：5分で最小パイプラインを動かす

まず最小構成から始めます。必要なのはPython 3.11以上と各サービスのAPIキーです。

インストール

Pipecatはextra方式で必要なサービスのみインストールします：

# 基本インストール（DeepgramSTT + OpenAILLM + CartesiaTTS）
pip install "pipecat-ai[deepgram,openai,cartesia,daily]"

# または全部入り（開発・検証用）
pip install "pipecat-ai[all]"

Python 3.11以上が必須です（推奨: 3.12）。

環境変数の設定

# .env ファイルに記載
DEEPGRAM_API_KEY=your_deepgram_key
OPENAI_API_KEY=your_openai_key
CARTESIA_API_KEY=your_cartesia_key
DAILY_API_KEY=your_daily_key  # WebRTCトランスポートを使う場合

最小パイプラインのコード（音声対話ボット）

以下は最もシンプルなSTT→LLM→TTSパイプラインです。ローカルで動作確認できます。

import asyncio
import os
from dotenv import load_dotenv

from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.openai.llm import OpenAILLMService
from pipecat.services.cartesia.tts import CartesiaTTSService
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.transports.local.audio import LocalAudioTransport
from pipecat.audio.vad.silero import SileroVADAnalyzer

# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。

load_dotenv()

async def main():
    # 1. トランスポート（マイク入力 / スピーカー出力）
    transport = LocalAudioTransport(
        LocalAudioParams(audio_in_enabled=True, audio_out_enabled=True)
    )

    # 2. STT: Deepgram Nova-2（日本語対応）
    stt = DeepgramSTTService(
        api_key=os.environ["DEEPGRAM_API_KEY"],
        language="ja",  # 日本語設定
        model="nova-2",
    )

    # 3. LLM: OpenAI GPT-4o
    llm = OpenAILLMService(
        api_key=os.environ["OPENAI_API_KEY"],
        model="gpt-4o",
    )

    # 4. TTS: Cartesia（低レイテンシ音声合成）
    tts = CartesiaTTSService(
        api_key=os.environ["CARTESIA_API_KEY"],
        voice_id="ja-JP-female-1",  # 日本語音声ID
    )

    # 5. システムプロンプトの設定
    messages = [
        {
            "role": "system",
            "content": "あなたは親切なAIアシスタントです。簡潔に日本語で回答してください。",
        }
    ]
    context = OpenAILLMContext(messages)
    context_aggregator = llm.create_context_aggregator(context)

    # 6. パイプラインの組み立て（入力→STT→LLM→TTS→出力）
    pipeline = Pipeline([
        transport.input(),
        stt,
        context_aggregator.user(),
        llm,
        tts,
        transport.output(),
        context_aggregator.assistant(),
    ])

    # 7. 実行
    task = PipelineTask(pipeline, params=PipelineParams(allow_interruptions=True))
    runner = PipelineRunner()
    await runner.run(task)

if __name__ == "__main__":
    asyncio.run(main())

動作環境: Python 3.11+, pipecat-ai 1.1.0, macOS/Linux
ポイント:

• allow_interruptions=True でユーザーが話し始めると即座に応答を中断する「バージ」機能が有効になります
• SileroVADAnalyzer で音声区間を自動検出します（VAD = Voice Activity Detection）
• パイプラインの各要素は非同期ストリームで接続されるため、STTの結果が出始めた瞬間にLLMへ渡します

実装サンプル：シナリオ別パイプライン構築

ここからが実践的な内容です。ユースケース別に3パターンのパイプラインを紹介します。

パターン1：WebRTCブラウザ対話（Daily.co使用）

実際にクライアントの受付自動化システムで使った構成です。ブラウザから直接音声対話できます。

from pipecat.transports.services.daily import DailyTransport, DailyParams
from pipecat.services.deepgram.stt import DeepgramSTTService, LiveOptions
from pipecat.services.anthropic.llm import AnthropicLLMService
from pipecat.services.elevenlabs.tts import ElevenLabsTTSService

# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。

async def build_webrtc_pipeline(room_url: str, token: str):
    # Daily.co WebRTCトランスポート
    transport = DailyTransport(
        room_url=room_url,
        token=token,
        bot_name="AI受付",
        params=DailyParams(
            audio_in_enabled=True,
            audio_out_enabled=True,
            camera_out_enabled=False,  # 音声のみ
            vad_enabled=True,
            vad_analyzer=SileroVADAnalyzer(),
        ),
    )

    # STT: Deepgram（エンドポイント検出を最適化）
    stt = DeepgramSTTService(
        api_key=os.environ["DEEPGRAM_API_KEY"],
        live_options=LiveOptions(
            model="nova-2",
            language="ja",
            punctuate=True,
            interim_results=True,  # 中間結果を活用してレイテンシ削減
            endpointing=200,  # 200ms無音でユーザー発話終了と判定
        ),
    )

    # LLM: Claude 3.7 Sonnet
    llm = AnthropicLLMService(
        api_key=os.environ["ANTHROPIC_API_KEY"],
        model="claude-sonnet-4-5",
    )

    # TTS: ElevenLabs（日本語品質高）
    tts = ElevenLabsTTSService(
        api_key=os.environ["ELEVENLABS_API_KEY"],
        voice_id="japanese_voice_id",
        model="eleven_flash_v2",  # 低レイテンシモデル
        optimize_streaming_latency=3,
    )

    messages = [{"role": "system", "content": "受付担当AIアシスタントです。丁寧な言葉遣いで応対してください。"}]
    context = AnthropicLLMContext(messages)
    context_aggregator = llm.create_context_aggregator(context)

    pipeline = Pipeline([
        transport.input(),
        stt,
        context_aggregator.user(),
        llm,
        tts,
        transport.output(),
        context_aggregator.assistant(),
    ])

    return PipelineTask(pipeline, params=PipelineParams(allow_interruptions=True))

ポイント: interim_results=True と endpointing=200 の組み合わせで体感レイテンシを大幅に削減できます。エンドポイント値を下げすぎると文の途中で切れるため、200-300msが実用的な範囲です。

パターン2：マルチエージェント構成（ルーティング付き）

Pipecat v1.0以降で使える Subagents機能 を使ったマルチエージェント構成です。複数の専門エージェントに会話を引き継ぎます。

from pipecat.agents.subagent import Subagent, SubagentParams

# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。

# 専門エージェントの定義
class SupportAgent(Subagent):
    """カスタマーサポート専門エージェント"""
    system_prompt = "あなたは製品サポート担当です。技術的な問題を解決します。"

class SalesAgent(Subagent):
    """営業専門エージェント"""
    system_prompt = "あなたは営業担当です。製品の導入提案を行います。"

# ルーターエージェント（会話内容を判断して専門エージェントに引き渡す）
router_messages = [
    {
        "role": "system",
        "content": """あなたは受付AIです。
        ユーザーの問い合わせ内容を判断し、適切な担当者に引き継ぎます。
        技術的な問題→'support'、製品購入相談→'sales' とだけ返答してください。""",
    }
]

async def route_conversation(user_message: str) -> str:
    """会話内容に基づいてルーティング先を決定"""
    # LLMで分類（実装省略）
    # 実際には llm.generate() で分類結果を取得
    return "support"  # or "sales"

# サブエージェントの登録とハンドオフ
subagents = {
    "support": SupportAgent(params=SubagentParams(llm_service=llm, tts_service=tts)),
    "sales": SalesAgent(params=SubagentParams(llm_service=llm, tts_service=tts)),
}

パターン3：Pipecat Cloud（スケーラブルデプロイ）

Pipecat Cloudを使うと、デプロイ・スケーリングをDaily社のインフラに任せられます。コストはエージェント起動時間あたり$0.01/分（2026年4月時点）。

from pipecat.runner.cloud import CloudRunner

# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。

# Pipecat Cloud設定（pipecat.yaml）
"""
name: my-voice-agent
region: us-east-1
max_instances: 10  # 最大10並列
min_instances: 1   # 常時1インスタンス待機
"""

async def main():
    runner = CloudRunner()
    # ローカルとコードは同じ。CloudRunnerがデプロイ先を切り替える
    await runner.run(task)

Pipecat Cloudは2026年1月にGA（一般提供）になりました。auto-scaling・マルチリージョン・エンタープライズサポートが含まれます。

Daily.co / LiveKit / Vapi 徹底比較

Pipecatはトランスポート（WebRTC基盤）を差し替えられます。主要3つを比較します。

比較軸	Daily.co（Pipecat Cloud）	LiveKit	Vapi
コスト（エージェント分）	$0.01/分	$0.01/分	$0.05/分〜（合計$0.13-0.31/分）
OSS	Pipecat: Yes / Daily: No	Yes	No
セルフホスト	可（Pipecat単体）	可	不可
日本語STT品質	Deepgram選択可	Deepgram/Whisper選択可	Deepgram固定（プランによる）
電話（PSTN）対応	Daily経由で可	SIP Trunk経由で可	標準対応
HIPAA対応	Daily Enterpriseで対応	セルフホストで対応可	有料アドオン（$1,000/月）

最終確認日: 2026-04-28

選定の考え方

• まずPipecatを試したい → Daily.co（Pipecat Cloud）一択。セットアップが最も簡単
• 大規模・フル制御が必要 → LiveKit + Pipecatの組み合わせ
• プロトタイプ1週間で出したい → Vapi（ただし月10万分超えるとコスト逆転）
• コンプライアンス要件（医療・金融） → LiveKitセルフホスト

コスト試算：月1万分・月10万分のシナリオ

実際の運用コストを試算します。サービスの組み合わせによって大きく変わります。

コンポーネント	選択サービス	単価	月1万分コスト	月10万分コスト
STT	Deepgram Nova-2	$0.0059/分	$59	$590
LLM	GPT-4o	〜$0.02/分（平均トークン数想定）	$200	$2,000
TTS	Cartesia Sonic	〜$0.005/分	$50	$500
インフラ	Pipecat Cloud	$0.01/分	$100	$1,000
合計	–	–	〜$409	〜$4,090

同じ構成でVapiを使った場合の合計コストは月1万分で$1,300〜$3,100程度になる計算です。10万分規模では自社ホスト（Pipecat + LiveKit）の選択でコストを80%以上削減できる可能性があります。

注意: LLMコストはユーザーの発話量とシステムプロンプトの長さに大きく依存します。上記はあくまで参考値です。実運用前に必ず自社のトラフィックパターンで試算してください。

よくある失敗パターンと回避策

失敗1：VAD閾値が高すぎてユーザーの発話が切れる

❌ VADをデフォルト設定で使い、短い相槌（「はい」「そうです」）が検出されない
⭕ VADの感度（threshold）を下げ、発話の最小長（min_speech_duration_ms）を設定する

from pipecat.audio.vad.silero import SileroVADAnalyzer, SileroVADParams

vad = SileroVADAnalyzer(params=SileroVADParams(
    threshold=0.4,           # デフォルト0.5より低め
    min_speech_duration_ms=100,  # 100ms以上の音声のみ検出
    min_silence_duration_ms=300, # 300ms無音で発話終了
))

なぜ重要か: 日本語は語尾が短い傾向があります。デフォルトでは「えっと」「あの」などの短い発話が無視され、UXが大幅に低下します。

失敗2：LLMのストリーミングを使わない

❌ LLMの出力を全文待ってからTTSに渡す → 2-5秒の沈黙が発生
⭕ ストリーミング出力を有効にしてトークン単位でTTSに渡す

# OpenAILLMServiceはデフォルトでストリーミング有効
# stream=True が自動設定されている（明示的な設定不要）
llm = OpenAILLMService(
    api_key=os.environ["OPENAI_API_KEY"],
    model="gpt-4o",
    # stream=True がデフォルト
)

ポイント: Pipecatは内部的にLLMのストリーミング出力をTTSに順次渡します。文節区切りを検出してTTSを起動するため、体感レイテンシを500ms以下に抑えられます。

失敗3：エラーハンドリングなしで本番投入する

❌ APIタイムアウト時に沈黙したままになる
⭕ フォールバックメッセージとリトライロジックを実装する

from pipecat.processors.frameworks.langchain import LangchainProcessor

# エラー時のフォールバック
@pipeline.on_error
async def handle_error(error: Exception):
    # TTSでフォールバックメッセージを再生
    await tts.say("申し訳ありません。少しお待ちください。")
    # 再接続ロジック（省略）

失敗4：ログ・トレースを設定しない

❌ 本番で問題が起きても何が起きたかわからない
⭕ Langfuse等のオブザーバビリティツールと連携する

PipecatはLangfuseとの公式連携があり、各フレームの処理時間・エラー・LLMコールをトレースできます。

FAQ：よくある疑問

Q: Pipecatは日本語に対応していますか？

A: Pipecat自体は言語非依存です。STTにDeepgram Nova-2（language="ja"）、TTSにElevenLabsやNijivoice（日本語音声特化）を選べば日本語対応できます。ただし日本語対応の品質はSTT/TTSサービスの選択に大きく依存します。

Q: 電話（固定電話・携帯電話）と繋げられますか？

A: Pipecat単体では電話回線に直接つながりません。Twilio・Daily（PSTN経由）・LiveKit SIP Trunkと組み合わせることで実現できます。電話対応が主要ユースケースの場合はVapiが標準対応しているため、そちらを検討するのも選択肢です。

Q: Pipecat Cloudを使わずにセルフホストできますか？

A: できます。PipecatはOSSなので、AWS・GCP・自社サーバー上にデプロイ可能です。Pipecat CloudはDaily社の管理インフラを利用する有料オプションですが、コア機能はクラウド不要で動作します。

Q: GPT-4o以外のLLMは使えますか？

A: はい。Anthropic Claude・Google Gemini・Mistral・ローカルモデル（Ollama経由）など100以上のサービスに対応しています。サービスごとのモジュールが用意されており、インポート先を変えるだけで切り替えられます。

Q: 開発環境でWebRTCなしにテストできますか？

A: LocalAudioTransport（マイク・スピーカー直結）またはWebSocketTransportを使えばWebRTCなしでテストできます。Daily.co等のアカウントなしにローカルで動作確認できるのがPipecatの強みです。

参考・出典

• Pipecat公式ドキュメント — Getting Started（参照日: 2026-04-28）
• pipecat-ai PyPI — v1.1.0リリース情報（参照日: 2026-04-28）
• Pipecat Cloud Pricing — Daily.co（参照日: 2026-04-28）
• LiveKit Pricing（参照日: 2026-04-28）
• Vapi Pricing（参照日: 2026-04-28）
• Open Source Observability for Pipecat — Langfuse（参照日: 2026-04-28）

---

まとめ：今日から始める3つのアクション

1. 今日: pip install "pipecat-ai[deepgram,openai,cartesia]" を実行し、最小パイプラインをローカルで動かしてみる
2. 今週中: 自社のユースケースに合わせてSTT/LLM/TTSの組み合わせを試し、レイテンシとコストを計測する
3. 今月中: Pipecat Cloudまたはセルフホストで本番デプロイし、オブザーバビリティ（Langfuse等）を設定する

---

あわせて読みたい:

• 【2026年4月】OpenAI Agents SDK完全ガイド｜実装とコスト試算 — マルチエージェント設計の基礎
• 【2026年4月】E2B・Daytona・Modal完全比較｜AI実行環境選定 — エージェント実行環境の選び方

---

---

著者: 佐藤傑（さとう・すぐる）
株式会社Uravation代表取締役。X（@SuguruKun_ai）フォロワー10万人超。
100社以上の企業向けAI研修・導入支援。著書累計3万部突破。
SoftBank IT連載7回執筆（NewsPicks最大1,125ピックス）。
ご質問・ご相談は お問い合わせフォーム からお気軽にどうぞ。