pipecatでSTT・LLM・TTSを組み合わせて音声AIボットを動かしてみる

音声入力 → テキスト文字起こし(STT) → 応答生成(LLM) → 音声合成(TTS) → 音声出力

python3 -m venv .venv
source .venv/bin/activate

brew install portaudio
pip install "pipecat-ai[google,deepgram,elevenlabs,silero,local]"

export GEMINI_API_KEY=your-gemini-api-key
export DEEPGRAM_API_KEY=your-deepgram-api-key
export ELEVENLABS_API_KEY=your-elevenlabs-api-key

# voice_bot.py
import asyncio
import os

from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.frames.frames import LLMRunFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineTask
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response_universal import (
    LLMContextAggregatorPair,
    LLMUserAggregatorParams,
)
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.elevenlabs.tts import ElevenLabsTTSService
from pipecat.services.google.llm import GoogleLLMService
from pipecat.transcriptions.language import Language
from pipecat.transports.local.audio import LocalAudioTransport, LocalAudioTransportParams
from pipecat.turns.user_mute import AlwaysUserMuteStrategy

async def main():

    transport = LocalAudioTransport(
        LocalAudioTransportParams(
            audio_in_enabled=True,
            audio_out_enabled=True,
            vad_analyzer=SileroVADAnalyzer(),
        )
    )

    # 音声 → テキスト
    stt = DeepgramSTTService(
        api_key=os.environ["DEEPGRAM_API_KEY"],
        settings=DeepgramSTTService.Settings(model="nova-2-general", language=Language.JA),
    )

    # テキスト → 応答テキスト
    llm = GoogleLLMService(
        api_key=os.environ["GEMINI_API_KEY"],
        settings=GoogleLLMService.Settings(
            model="gemini-3.5-flash",
            system_instruction="あなたは親しみやすい日本語の音声アシスタントです。短く話してください。",
        ),
    )

    # 応答テキスト → 音声
    tts = ElevenLabsTTSService(
        api_key=os.environ["ELEVENLABS_API_KEY"],
        settings=ElevenLabsTTSService.Settings(
            voice="EXAVITQu4vr4xnSDxMaL",  # ElevenLabs の voice IDを指定することで、好きな音声に切り替えられます
            model="eleven_flash_v2_5",
            language=Language.JA,
        ),
    )

    context = LLMContext()
    aggregator = LLMContextAggregatorPair(
        context,
        user_params=LLMUserAggregatorParams(
            user_mute_strategies=[AlwaysUserMuteStrategy()] # Bot発言中はマイク入力を無視する
        ),
    )

    pipeline = Pipeline(
        [
            transport.input(),       # マイクからの音声
            stt,                     # 音声 → テキスト
            aggregator.user(),
            llm,                     # テキスト → 応答テキスト
            tts,                     # 応答テキスト → 音声
            transport.output(),      # スピーカーへの音声
            aggregator.assistant(),
        ]
    )

    task = PipelineTask(pipeline)

    context.add_message(
        {"role": "user", "content": "日本語で簡単に自己紹介して、挨拶してください。"}
    )
    await task.queue_frames([LLMRunFrame()])

    await PipelineRunner().run(task)

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

pipeline = Pipeline(
    [
        transport.input(),       # マイクからの音声
        stt,                     # 音声 → テキスト
        aggregator.user(),
        llm,                     # テキスト → 応答テキスト
        tts,                     # 応答テキスト → 音声
        transport.output(),      # スピーカーへの音声
        aggregator.assistant(),
    ]
)

python voice_bot.py