Pipecat

1
import os
2
from dotenv import load_dotenv
3
from pipecat.services.hume.tts import HUME_SAMPLE_RATE, HumeTTSService
4
from pipecat.services.openai.llm import OpenAILLMService
5
from pipecat.services.deepgram.stt import DeepgramSTTService
6
from pipecat.pipeline.pipeline import Pipeline
7
from pipecat.pipeline.runner import PipelineRunner
8
from pipecat.pipeline.task import PipelineParams, PipelineTask
9
10
load_dotenv(override=True)
11
12
async def run_bot(transport, runner_args):
13
    # 1. Configure the Hume TTS service
14
    tts = HumeTTSService(
15
        api_key=os.getenv("HUME_API_KEY"),
16
        voice_id=os.getenv("HUME_VOICE_ID"),
17
    )
18
19
    # 2. Configure STT and LLM services
20
    stt = # your STT service provider here
21
    llm = # your LLM service provider here
22
23
    # 3. Create your pipeline
24
    pipeline = Pipeline([
25
        transport.input(),
26
        stt,
27
        context_aggregator.user(),
28
        llm,
29
        tts,  # Hume TTS with word timestamps
30
        transport.output(),
31
        context_aggregator.assistant(),
32
    ])
33
34
    # 4. Configure task with matching sample rate
35
    task = PipelineTask(
36
        pipeline,
37
        params=PipelineParams(
38
            enable_metrics=True,
39
            enable_usage_metrics=True,
40
            audio_out_sample_rate=HUME_SAMPLE_RATE,  # 48000 Hz
41
        ),
42
    )
43
44
    # 5. Run the pipeline
45
    runner = PipelineRunner()
46
    await runner.run(task)

1
from pipecat.services.hume.tts import HumeTTSService, HumeTTSService.InputParams
2
3
tts = HumeTTSService(
4
    api_key=os.getenv("HUME_API_KEY"),
5
    voice_id=os.getenv("HUME_VOICE_ID"),
6
    params=HumeTTSService.InputParams(
7
        description="calm, pedagogical",  # Acting instructions
8
        speed=0.8,  # Speaking-rate multiplier (0.5-2.0)
9
        trailing_silence=2.0,  # Seconds of silence to append (0-5)
10
    ),
11
)

1
from pipecat.frames.frames import TTSUpdateSettingsFrame
2
3
# Update voice
4
await task.queue_frames([
5
    TTSUpdateSettingsFrame(settings={"voice_id": "new-voice-id"})
6
])
7
8
# Update synthesis parameters
9
await task.queue_frames([
10
    TTSUpdateSettingsFrame(settings={
11
        "description": "excited, enthusiastic",
12
        "speed": 1.2,
13
    })
14
])

1
from pipecat.observers.loggers.debug_log_observer import (
2
    DebugLogObserver,
3
    FrameEndpoint,
4
)
5
from pipecat.transports.base_output import BaseOutputTransport
6
from pipecat.frames.frames import TTSTextFrame
7
8
task = PipelineTask(
9
    pipeline,
10
    params=PipelineParams(
11
        enable_metrics=True,
12
        enable_usage_metrics=True,
13
        audio_out_sample_rate=HUME_SAMPLE_RATE,
14
    ),
15
    observers=[
16
        DebugLogObserver(
17
            frame_types={
18
                TTSTextFrame: (BaseOutputTransport, FrameEndpoint.SOURCE),
19
            }
20
        ),
21
    ],
22
)