mkdir pipecat-work && cd pipecat-work

mise use python@3.12

cat << 'EOS' >> .mise.toml

[env]
_.python.venv = { path = ".venv", create = true }
EOS

pip install "pipecat-ai[daily,cartesia,openai]"

pip freeze | grep -i pipecat

pipecat-ai==0.0.53

DAILY_SAMPLE_ROOM_URL=https://[サブドメイン].daily.co/[ルーム名]
DAILY_API_KEY=XXXXXXXXXX
CARTESIA_API_KEY=XXXXXXXXXX
CARTESIA_VOICE_ID="2b568345-1d48-4047-b25f-7baccf842eb0"  # 日本語・女性・会話スタイル

#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#

import asyncio
import os
import sys

import aiohttp
from dotenv import load_dotenv
from loguru import logger
from runner import configure

from pipecat.frames.frames import EndFrame, TTSSpeakFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineTask
from pipecat.services.cartesia import CartesiaTTSService, CartesiaHttpTTSService
from pipecat.transports.services.daily import DailyParams, DailyTransport
from pipecat.transcriptions.language import Language
load_dotenv(override=True)

logger.remove(0)
logger.add(sys.stderr, level="DEBUG")


async def main():
    async with aiohttp.ClientSession() as session:
        (room_url, _) = await configure(session)

        # WebRTC通信のためのDailyトランスポートを設定します
        # - room_url: 接続先ルームURL
        # - None: 認証トークンは不要
        # - "テストボット": ボットの表示名
        # - TTS再生のために音声出力を有効にする
        transport = DailyTransport(
            room_url,
            None,
            "テストボット",
            DailyParams(audio_out_enabled=True)
        )

        # Cartesiaのテキスト読み上げサービスを初期化
        # 事前に選択された日本語の女性の声を使用
        # 他の声は以下で確認できる: https://play.cartesia.ai/voices
        # 訳注1:
        # 元のサンプルコードでは、CartesiaTTSServiceが使用されており、これはリアルタイム処理のためだが、
        # その場合、モデルと言語がタイムスタンプに対応している必要がある。Cartesiaの日本語モデルはタイムスタンプに
        # 対応していないため、エラーとなる。そこで非リアルタイムなCartesiaHttpTTSServiceを使用する。
        # 参考: https://docs.cartesia.ai/api-reference/tts/compare-tts-endpoints#if-you-want-to-generate-speech-in-real-time
        # 参考: https://github.com/pipecat-ai/pipecat/issues/513
        # 訳注2:
        # なお、Cartesiaの日本語対応モデルは、Websocketや入力ストリーミングに対応しているので
        # それらを使ってリアルタイムに近い高速TTSレスポンスを生成することは可能。
        # 例えばOpenAI Chat CompletionのストリーミングをそのままCartesiaに投げて高速音声チャットボットができたりする。
        # ここでそれができないのはPipecatでのリアルタイム処理はタイムスタンプ機能が前提となっているからだと思う。
        tts = CartesiaHttpTTSService(
            api_key=os.getenv("CARTESIA_API_KEY"),
            voice_id=os.getenv("CARTESIA_VOICE_ID"),
            model="sonic",
            params=CartesiaHttpTTSService.InputParams(
                language=Language.JA,
            )
        )

        # パイプラインを管理するためのパイプラインランナーを作成
        runner = PipelineRunner()

        # テキストを音声に変換し再生するためのパイプラインを作成
        # 1. tts: テキストメッセージを音声に変換し
        # 2. transport.output(): 音声をDailyルームに送信
        task = PipelineTask(Pipeline([tts, transport.output()]))

        # 最初の参加者がルームに参加したとき:
        # - 参加者情報からユーザー名を抽出
        # - 挨拶メッセージを生成
        # - TTS変換のためにキューに追加
        # - ボットが退出するためのEndFrameをキューに追加（パイプラインをシャットダウンする）
        @transport.event_handler("on_first_participant_joined")
        async def on_first_participant_joined(transport, participant):
            participant_name = participant.get("info", {}).get("userName", "")
            await task.queue_frames(
                [TTSSpeakFrame(f"{participant_name} さん、こんにちは！ようこそお越しくださいました。"), EndFrame()]
            )

        # パイプラインタスクを実行
        await runner.run(task)


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

#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#

import argparse
import os

import aiohttp

from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper


async def configure(aiohttp_session: aiohttp.ClientSession):
    (url, token, _) = await configure_with_args(aiohttp_session)
    return (url, token)


async def configure_with_args(
    aiohttp_session: aiohttp.ClientSession, parser: argparse.ArgumentParser | None = None
):
    if not parser:
        parser = argparse.ArgumentParser(description="Daily AI SDK Bot Sample")
    parser.add_argument(
        "-u", "--url", type=str, required=False, help="URL of the Daily room to join"
    )
    parser.add_argument(
        "-k",
        "--apikey",
        type=str,
        required=False,
        help="Daily API Key (needed to create an owner token for the room)",
    )

    args, unknown = parser.parse_known_args()

    url = args.url or os.getenv("DAILY_SAMPLE_ROOM_URL")
    key = args.apikey or os.getenv("DAILY_API_KEY")

    if not url:
        raise Exception(
            "No Daily room specified. use the -u/--url option from the command line, or set DAILY_SAMPLE_ROOM_URL in your environment to specify a Daily room URL."
        )

    if not key:
        raise Exception(
            "No Daily API key specified. use the -k/--apikey option from the command line, or set DAILY_API_KEY in your environment to specify a Daily API key, available from https://dashboard.daily.co/developers."
        )

    daily_rest_helper = DailyRESTHelper(
        daily_api_key=key,
        daily_api_url=os.getenv("DAILY_API_URL", "https://api.daily.co/v1"),
        aiohttp_session=aiohttp_session,
    )

    # Create a meeting token for the given room with an expiration 1 hour in
    # the future.
    expiry_time: float = 60 * 60

    token = await daily_rest_helper.get_token(url, expiry_time)

    return (url, token, args)

python 01-say-one-thing.py

#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#

import asyncio
import os
import sys

import aiohttp
from dotenv import load_dotenv
from loguru import logger
from runner import configure

from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.frames.frames import EndFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.services.openai import OpenAILLMService, OpenAITTSService
from pipecat.transports.services.daily import DailyParams, DailyTransport, DailyTranscriptionSettings

load_dotenv(override=True)

logger.remove(0)
logger.add(sys.stderr, level="DEBUG")


SYSTEM_PROMPT = """\
あなたは親切なアシスタントで、あなたの仕事はユーザとWebRTC上で音声通話を行うことです。
あなたの回答は音声に変換されるので、回答に特殊文字を含めないでください。
ユーザーの発言に対して、創造的かつ有益な応答をしてください。
"""

async def main():
    async with aiohttp.ClientSession() as session:
        (room_url, token) = await configure(session)

        transport = DailyTransport(
            room_url,
            token,
            "Respond bot",
            DailyParams(
                audio_out_enabled=True,
                transcription_enabled=True,                                       # Daiｌｙではデフォルトで文字起こしを提供していて
                transcription_settings=DailyTranscriptionSettings(language="ja"), # 中身はDeepgramのAPIを使用している様子
                vad_enabled=True,                 # VADを有効にする
                vad_analyzer=SileroVADAnalyzer(), # Silero VADを使用する
            ),
        )

        tts = OpenAITTSService(api_key=os.getenv("OPENAI_API_KEY"), voice="alloy")

        llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o-mini")

        messages = [
            {
                "role": "system",
                "content": SYSTEM_PROMPT,
            },
        ]

        context = OpenAILLMContext(messages)
        context_aggregator = llm.create_context_aggregator(context)

        pipeline = Pipeline(
            [
                transport.input(),               # ユーザの入力を（Daily.co経由で）受信
                context_aggregator.user(),       # ユーザの入力をコンテキストに追加
                llm,                             # LLMを呼び出し
                tts,                             # TTSを呼び出し
                transport.output(),              # ボットの応答を（Daily.co経由で）送信
                context_aggregator.assistant(),  # ボットの応答をコンテキストに追加
            ]
        )

        task = PipelineTask(
            pipeline,
            PipelineParams(
                allow_interruptions=True,
                enable_metrics=True,
                enable_usage_metrics=True,
                report_only_initial_ttfb=True,
            ),
        )

        @transport.event_handler("on_first_participant_joined")
        async def on_first_participant_joined(transport, participant):
            await transport.capture_participant_transcription(participant["id"])
            # 会話の開始時に自己紹介させる
            messages.append({"role": "system", "content": "最初にユーザに自己紹介してください。"})
            await task.queue_frames([context_aggregator.user().get_context_frame()])

        @transport.event_handler("on_participant_left")
        async def on_participant_left(transport, participant, reason):
            await task.queue_frame(EndFrame())

        runner = PipelineRunner()

        await runner.run(task)


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

(snip)
                transcription_enabled=True,                                       # Daiｌｙではデフォルトで文字起こしを提供していて
                transcription_settings=DailyTranscriptionSettings(language="ja"), # 中身はDeepgramのAPIを使用している様子
(snip)

2025-01-28 13:01:39.650 | INFO     | pipecat.transports.services.daily:_start_transcription:363 - Enabling transcription with settings language='ja' tier=None model='nova-2-general' profanity_filter=True redact=False endpointing=True punctuate=True includeRawResponse=True extra={'interim_results': True}

        # Cartesiaのテキスト読み上げサービスを初期化
        # 事前に選択された日本語の女性の声を使用
        # 他の声は以下で確認できる: https://play.cartesia.ai/voices
        # 訳注1:
        # 元のサンプルコードでは、CartesiaTTSServiceが使用されており、これはリアルタイム処理のためだが、
        # その場合、モデルと言語がタイムスタンプに対応している必要がある。Cartesiaの日本語モデルはタイムスタンプに
        # 対応していないため、エラーとなる。そこで非リアルタイムなCartesiaHttpTTSServiceを使用する。
        # 参考: https://docs.cartesia.ai/api-reference/tts/compare-tts-endpoints#if-you-want-to-generate-speech-in-real-time
        # 参考: https://github.com/pipecat-ai/pipecat/issues/513
        # 訳注2:
        # なお、Cartesiaの日本語対応モデルは、Websocketや入力ストリーミングに対応しているので
        # それらを使ってリアルタイムに近い高速TTSレスポンスを生成することは可能。
        # 例えばOpenAI Chat CompletionのストリーミングをそのままCartesiaに投げて高速音声チャットボットができたりする。
        # ここでそれができないのはPipecatでのリアルタイム処理はタイムスタンプ機能が前提となっているからだと思う。
        tts = CartesiaHttpTTSService(
            api_key=os.getenv("CARTESIA_API_KEY"),
            voice_id=os.getenv("CARTESIA_VOICE_ID"),
            model="sonic",
            params=CartesiaHttpTTSService.InputParams(
                language=Language.JA,
            )
        )