Pipecat

사용자 오디오 → STT → LLM → TTS → 봇 오디오

pip install pipecat-ai-typecast

import os
import aiohttp
from pipecat.pipeline.pipeline import Pipeline
from pipecat_typecast import TypecastTTSService

async with aiohttp.ClientSession() as session:
    # Typecast TTS 초기화
    tts = TypecastTTSService(
        aiohttp_session=session,
        api_key=os.getenv("TYPECAST_API_KEY"),
        voice_id=os.getenv("TYPECAST_VOICE_ID", "tc_672c5f5ce59fac2a48faeaee"),
    )

    # 파이프라인 구축
    pipeline = Pipeline([
        transport.input(),               # 사용자 오디오 입력
        stt,                             # 음성-텍스트 변환
        context_aggregator.user(),       # 문맥에 사용자 텍스트 추가
        llm,                             # LLM이 응답 생성
        tts,                             # Typecast TTS 합성
        transport.output(),              # 사용자에게 오디오 스트리밍
        context_aggregator.assistant(),  # 어시스턴트 응답 저장
    ])

from pipecat_typecast import TypecastTTSService

tts = TypecastTTSService(
    aiohttp_session=session,
    api_key=os.getenv("TYPECAST_API_KEY"),
    voice_id="tc_672c5f5ce59fac2a48faeaee",
    model="ssfm-v30",  # 최신 모델 (기본값)
)

from pipecat_typecast import (
    TypecastTTSService,
    TypecastInputParams,
    PresetPromptOptions,
    OutputOptions,
)

params = TypecastInputParams(
    prompt_options=PresetPromptOptions(
        emotion_preset="happy",      # normal | happy | sad | angry | whisper | toneup | tonedown
        emotion_intensity=1.3,       # 0.0 - 2.0
    ),
    output_options=OutputOptions(
        volume=110,                  # 0 - 200 (퍼센트)
        audio_pitch=2,               # -12 ~ 12 (반음)
        audio_tempo=1.05,            # 0.5 - 2.0 (재생 속도)
    ),
)

tts = TypecastTTSService(
    aiohttp_session=session,
    api_key=os.getenv("TYPECAST_API_KEY"),
    params=params,
)

from pipecat_typecast import (
    TypecastTTSService,
    TypecastInputParams,
    SmartPromptOptions,
)

params = TypecastInputParams(
    prompt_options=SmartPromptOptions(
        previous_text="방금 정말 좋은 소식을 들었어요!",   # 최대 2000자
        next_text="모두와 공유하고 싶어서 기다려지지 않아요!",
    ),
)

tts = TypecastTTSService(
    aiohttp_session=session,
    api_key=os.getenv("TYPECAST_API_KEY"),
    params=params,
)

import os
import aiohttp
from dotenv import load_dotenv

from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.openai.llm import OpenAILLMService
from pipecat.transports.daily.transport import DailyParams, DailyTransport

from pipecat_typecast import TypecastTTSService

load_dotenv()

async def main():
    async with aiohttp.ClientSession() as session:
        # 서비스 초기화
        stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
        llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
        tts = TypecastTTSService(
            aiohttp_session=session,
            api_key=os.getenv("TYPECAST_API_KEY"),
        )

        # 대화 문맥 설정
        messages = [
            {
                "role": "system",
                "content": "당신은 유용한 AI 어시스턴트입니다. 응답은 간결하게 해주세요.",
            },
        ]
        context = LLMContext(messages)
        context_aggregator = LLMContextAggregatorPair(context)

        # 전송 구성
        transport = DailyTransport(
            room_url=os.getenv("DAILY_ROOM_URL"),
            token=os.getenv("DAILY_TOKEN"),
            params=DailyParams(
                audio_in_enabled=True,
                audio_out_enabled=True,
                vad_analyzer=SileroVADAnalyzer(),
            ),
        )

        # 파이프라인 구축 및 실행
        pipeline = Pipeline([
            transport.input(),
            stt,
            context_aggregator.user(),
            llm,
            tts,
            transport.output(),
            context_aggregator.assistant(),
        ])

        task = PipelineTask(pipeline, params=PipelineParams())
        runner = PipelineRunner()
        await runner.run(task)

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