Building a Voice Agent with Pipecat and AssemblyAI

1
from pipecat.services.assemblyai.stt import AssemblyAISTTService
2
3
# Configure service
4
stt = AssemblyAISTTService(
5
    connection_params=AssemblyAIConnectionParams(
6
        end_of_turn_confidence_threshold=0.7,
7
        min_end_of_turn_silence_when_confident=160,
8
        max_turn_silence=2400,
9
    ),
10
    api_key=os.getenv("ASSEMBLYAI_API_KEY"),
11
    vad_force_turn_endpoint=False
12
)
13
14
# Use in pipeline
15
pipeline = Pipeline([
16
    transport.input(),
17
    stt,
18
    llm,
19
    ...
20
])

1
#
2
# Copyright (c) 2024–2025, Daily
3
#
4
# SPDX-License-Identifier: BSD 2-Clause License
5
#
6
7
import argparse
8
import os
9
10
from dotenv import load_dotenv
11
from loguru import logger
12
13
from pipecat.audio.vad.silero import SileroVADAnalyzer
14
from pipecat.pipeline.pipeline import Pipeline
15
from pipecat.pipeline.runner import PipelineRunner
16
from pipecat.pipeline.task import PipelineParams, PipelineTask
17
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
18
from pipecat.services.cartesia.tts import CartesiaTTSService
19
from pipecat.services.assemblyai.stt import AssemblyAISTTService, AssemblyAIConnectionParams
20
from pipecat.services.openai.llm import OpenAILLMService
21
from pipecat.transports.base_transport import BaseTransport, TransportParams
22
from pipecat.transports.network.fastapi_websocket import FastAPIWebsocketParams
23
from pipecat.transports.services.daily import DailyParams
24
25
load_dotenv(override=True)
26
27
# We store functions so objects (e.g. SileroVADAnalyzer) don't get
28
# instantiated. The function will be called when the desired transport gets
29
# selected.
30
transport_params = {
31
    "daily": lambda: DailyParams(
32
        audio_in_enabled=True,
33
        audio_out_enabled=True,
34
        vad_analyzer=SileroVADAnalyzer(),
35
    ),
36
    "twilio": lambda: FastAPIWebsocketParams(
37
        audio_in_enabled=True,
38
        audio_out_enabled=True,
39
        vad_analyzer=SileroVADAnalyzer(),
40
    ),
41
    "webrtc": lambda: TransportParams(
42
        audio_in_enabled=True,
43
        audio_out_enabled=True,
44
        vad_analyzer=SileroVADAnalyzer(),
45
    ),
46
}
47
48
49
async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_sigint: bool):
50
    logger.info(f"Starting bot")
51
52
    # Configure AssemblyAI STT with advanced turn detection
53
    stt = AssemblyAISTTService(
54
        api_key=os.getenv("ASSEMBLYAI_API_KEY"),
55
        vad_force_turn_endpoint=False,
56
        connection_params=AssemblyAIConnectionParams(
57
            end_of_turn_confidence_threshold=0.7,
58
            min_end_of_turn_silence_when_confident=160,
59
            max_turn_silence=2400,
60
        )
61
    )
62
63
    tts = CartesiaTTSService(
64
        api_key=os.getenv("CARTESIA_API_KEY"),
65
        voice_id="71a7ad14-091c-4e8e-a314-022ece01c121",  # British Reading Lady
66
    )
67
68
    llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
69
70
    messages = [
71
        {
72
            "role": "system",
73
            "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.",
74
        },
75
    ]
76
77
    context = OpenAILLMContext(messages)
78
    context_aggregator = llm.create_context_aggregator(context)
79
80
    pipeline = Pipeline(
81
        [
82
            transport.input(),  # Transport user input
83
            stt,
84
            context_aggregator.user(),  # User responses
85
            llm,  # LLM
86
            tts,  # TTS
87
            transport.output(),  # Transport bot output
88
            context_aggregator.assistant(),  # Assistant spoken responses
89
        ]
90
    )
91
92
    task = PipelineTask(
93
        pipeline,
94
        params=PipelineParams(
95
            allow_interruptions=True,
96
            enable_metrics=True,
97
            enable_usage_metrics=True,
98
            report_only_initial_ttfb=True,
99
        ),
100
    )
101
102
    @transport.event_handler("on_client_connected")
103
    async def on_client_connected(transport, client):
104
        logger.info(f"Client connected")
105
        # Kick off the conversation.
106
        messages.append({"role": "system", "content": "Please introduce yourself to the user."})
107
        await task.queue_frames([context_aggregator.user().get_context_frame()])
108
109
    @transport.event_handler("on_client_disconnected")
110
    async def on_client_disconnected(transport, client):
111
        logger.info(f"Client disconnected")
112
        await task.cancel()
113
114
    runner = PipelineRunner(handle_sigint=handle_sigint)
115
116
    await runner.run(task)
117
118
119
if __name__ == "__main__":
120
    from pipecat.examples.run import main
121
122
    main(run_example, transport_params=transport_params)

1
stt = AssemblyAISTTService(
2
    api_key=os.getenv("ASSEMBLYAI_API_KEY"),
3
    vad_force_turn_endpoint=False,  # Use AssemblyAI's STT-based turn detection
4
    connection_params=AssemblyAIConnectionParams(
5
        end_of_turn_confidence_threshold=0.7,
6
        min_end_of_turn_silence_when_confident=160,  # in ms
7
        max_turn_silence=2400,  # in ms
8
    )
9
)

1
messages = [
2
    {
3
        "role": "system",
4
        "content": "You are a friendly customer service representative. Help users with technical questions and maintain a professional tone. Keep responses under 30 seconds.",
5
    },
6
]

1
messages = [
2
    {
3
        "role": "system",
4
        "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.",
5
    },
6
]