Build a Voice Agent with Pipecat and Deepgram

1
# bot.py
2
3
import os
4
5
from dotenv import load_dotenv
6
7
from pipecat.audio.vad.silero import SileroVADAnalyzer
8
from pipecat.frames.frames import LLMRunFrame
9
from pipecat.pipeline.pipeline import Pipeline
10
from pipecat.pipeline.runner import PipelineRunner
11
from pipecat.pipeline.task import PipelineParams, PipelineTask
12
from pipecat.processors.aggregators.llm_context import LLMContext
13
from pipecat.processors.aggregators.llm_response_universal import (
14
    LLMContextAggregatorPair,
15
    LLMUserAggregatorParams,
16
)
17
from pipecat.runner.types import RunnerArguments
18
from pipecat.runner.utils import create_transport
19
from pipecat.services.deepgram.stt import DeepgramSTTService
20
from pipecat.services.deepgram.tts import DeepgramTTSService
21
from pipecat.services.openai.llm import OpenAILLMService
22
from pipecat.transports.base_transport import BaseTransport, TransportParams
23
24
load_dotenv(override=True)
25
26
transport_params = {
27
    "webrtc": lambda: TransportParams(
28
        audio_in_enabled=True,
29
        audio_out_enabled=True,
30
    ),
31
}
32
33
34
async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
35
    stt = DeepgramSTTService(
36
        api_key=os.getenv("DEEPGRAM_API_KEY"),
37
        settings=DeepgramSTTService.Settings(
38
            model="nova-3-general",
39
            language="en",
40
            punctuate=True,
41
            smart_format=True,
42
        ),
43
    )
44
45
    llm = OpenAILLMService(
46
        api_key=os.getenv("OPENAI_API_KEY"),
47
        settings=OpenAILLMService.Settings(
48
            model="gpt-4o",
49
            system_instruction=(
50
                "You are a friendly, helpful voice assistant. "
51
                "Keep your responses concise — aim for 1-3 sentences "
52
                "unless the user asks for detail. "
53
                "Your responses will be spoken aloud, so avoid emojis, "
54
                "bullet points, or other formatting that can't be spoken."
55
            ),
56
        ),
57
    )
58
59
    tts = DeepgramTTSService(
60
        api_key=os.getenv("DEEPGRAM_API_KEY"),
61
        settings=DeepgramTTSService.Settings(
62
            voice="aura-2-thalia-en",
63
        ),
64
    )
65
66
    context = LLMContext()
67
68
    user_aggregator, assistant_aggregator = LLMContextAggregatorPair(
69
        context,
70
        user_params=LLMUserAggregatorParams(
71
            vad_analyzer=SileroVADAnalyzer(),
72
        ),
73
    )
74
75
    pipeline = Pipeline([
76
        transport.input(),
77
        stt,
78
        user_aggregator,
79
        llm,
80
        tts,
81
        transport.output(),
82
        assistant_aggregator,
83
    ])
84
85
    task = PipelineTask(
86
        pipeline,
87
        params=PipelineParams(
88
            enable_metrics=True,
89
            enable_usage_metrics=True,
90
        ),
91
    )
92
93
    @transport.event_handler("on_client_connected")
94
    async def on_client_connected(transport, client):
95
        context.add_message(
96
            {"role": "developer", "content": "Greet the user and ask how you can help."}
97
        )
98
        await task.queue_frames([LLMRunFrame()])
99
100
    @transport.event_handler("on_client_disconnected")
101
    async def on_client_disconnected(transport, client):
102
        await task.cancel()
103
104
    runner = PipelineRunner(handle_sigint=runner_args.handle_sigint)
105
    await runner.run(task)
106
107
108
async def bot(runner_args: RunnerArguments):
109
    transport = await create_transport(runner_args, transport_params)
110
    await run_bot(transport, runner_args)
111
112
113
if __name__ == "__main__":
114
    from pipecat.runner.run import main
115
116
    main()