wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Rename examples files, update quickstart

Browse Source
This commit is contained in:
Mark Backman
2025-07-27 16:19:45 -04:00
parent 49af1553e0
commit 16c80b2335
12 changed files with 174 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
examples/quickstart/Dockerfile
View File

@@ -1,7 +0,0 @@
FROM dailyco/pipecat-base:latest
COPY ./requirements.txt requirements.txt
RUN pip install --no-cache-dir --upgrade -r requirements.txt
COPY ./cloud-simple-bot.py bot.py

69
examples/quickstart/README.md
View File

@@ -1,14 +1,52 @@
## Quickstart - TBD TBD TBD TBD
# Pipecat Quickstart
### Setup
Run your first Pipecat bot in under 5 minutes. This example creates a voice AI bot that you can talk to in your browser.
1. Set up a venv
## Prerequisites
### Python 3.10+
Pipecat requires Python 3.10 or newer. Check your version:
```bash
python --version
```
If you need to upgrade Python, we recommend using a version manager like `uv` or `pyenv`.
### AI Service API keys
Pipecat orchestrates different AI services in a pipeline, ensuring low latency communication. In this quickstart example, we'll use:
- [Deepgram](https://console.deepgram.com/signup) for Speech-to-Text transcriptions
- [OpenAI](https://auth.openai.com/create-account) for LLM inference
- [Cartesia](https://play.cartesia.ai/sign-up) for Text-to-Speech audio generation
Have your API keys ready. We'll add them to your `.env` shortly.
## Setup
1. Set up a virtual environment
From the root directory of the `pipecat` repo, run:
```bash
cd examples/quickstart
python -m venv .venv
source .venv/bin/activate # On Windows: .venv\Scripts\activate
```
> Using `uv`? Create your venv using: `uv venv && source .venv/bin/activate`.
2. Install packages
pip install "pipecat-ai[webrtc,deepgram,openai,cartesia,silero]" \
"pipecat-ai-small-webrtc-prebuilt" \
"python-dotenv"
Then, install the requirements:
```bash
pip install -r requirements.txt
```
> Using `uv`? Install requirements using: `uv pip install -r requirements.txt`.
3. Configure environment variables
@@ -28,8 +66,25 @@ CARTESIA_API_KEY=your_cartesia_api_key
4. Run the example
Run your bot using:
```bash
python bot.py
```
Connect to your bot in a browser at http://localhost:7860
> Using `uv`? Run your bot using: `uv run bot.py`.
Connect to your bot in a browser at http://localhost:7860.
> 💡 First run note: The initial startup may take ~10 seconds as Pipecat downloads required models, like the Silero VAD model.
## Troubleshooting
- **Browser permissions**: Make sure to allow microphone access when prompted by your browser.
- **Connection issues**: If the WebRTC connection fails, first try a different browser. If that fails, make sure you don't have a VPN or firewall rules blocking traffic. WebRTC uses UDP to communicate.
- **Audio issues**: Check that your microphone and speakers are working and not muted.
## Next Steps
- **Read the docs**: Check out [Pipecat's docs](https://docs.pipecat.ai/) for guides and reference information.
- **Join Discord**: Join [Pipecat's Discord server](https://discord.gg/pipecat) to get help and learn about what others are building.

1
examples/quickstart/local-simple-bot.py → examples/quickstart/bot.py
View File

@@ -90,6 +90,7 @@ async def run_bot(transport: BaseTransport, _: argparse.Namespace, handle_sigint
if __name__ == "__main__":
from pipecat.runner.local import main
# SmallWebRTCTransport for a P2P WebRTC connection
transport_params = {
"webrtc": lambda: TransportParams(
audio_in_enabled=True,

19
examples/quickstart/build.sh
View File

@@ -1,19 +0,0 @@
#!/bin/bash
set -e
VERSION="0.1"
DOCKER_USERNAME="your_docker_username"
AGENT_NAME="cloud-simple-bot"
# Build the Docker image with the correct context
echo "Building Docker image..."
docker build --platform=linux/arm64 -t "$DOCKER_USERNAME/$AGENT_NAME:$VERSION" -t "$DOCKER_USERNAME/$AGENT_NAME:latest" .
# Push the Docker images
echo "Pushing Docker image $DOCKER_USERNAME/$AGENT_NAME:$VERSION..."
docker push "$DOCKER_USERNAME/$AGENT_NAME:$VERSION"
echo "Pushing Docker image $DOCKER_USERNAME/$AGENT_NAME:latest..."
docker push "$DOCKER_USERNAME/$AGENT_NAME:latest"
echo "Successfully built and pushed $DOCKER_USERNAME/$AGENT_NAME:$VERSION and $DOCKER_USERNAME/$AGENT_NAME:latest"

185
examples/quickstart/cloud-multi-bot.py
View File

@@ -1,185 +0,0 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import os
from dotenv import load_dotenv
from loguru import logger
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.openai_llm_context import OpenAILLMContext
from pipecat.processors.frameworks.rtvi import RTVIConfig, RTVIObserver, RTVIProcessor
from pipecat.runner.cloud import SmallWebRTCSessionArguments
from pipecat.services.cartesia.tts import CartesiaTTSService
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.openai.llm import OpenAILLMService
try:
from pipecatcloud.agent import DailySessionArguments, WebSocketSessionArguments
except ImportError:
raise ImportError(
"pipecatcloud package is required for cloud-compatible bots. "
"Install with: pip install pipecat-ai[[pipecatcloud]]"
)
load_dotenv(override=True)
async def run_bot(transport):
"""Main bot logic that works with any transport."""
logger.info(f"Starting bot")
stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
tts = CartesiaTTSService(
api_key=os.getenv("CARTESIA_API_KEY"),
voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady
)
llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
messages = [
{
"role": "system",
"content": "You are a friendly AI assistant. Respond naturally and keep your answers conversational.",
},
]
context = OpenAILLMContext(messages)
context_aggregator = llm.create_context_aggregator(context)
rtvi = RTVIProcessor(config=RTVIConfig(config=[]))
pipeline = Pipeline(
[
transport.input(),
rtvi,
stt,
context_aggregator.user(),
llm,
tts,
transport.output(),
context_aggregator.assistant(),
]
)
task = PipelineTask(
pipeline,
params=PipelineParams(
enable_metrics=True,
enable_usage_metrics=True,
),
observers=[RTVIObserver(rtvi)],
)
@transport.event_handler("on_client_connected")
async def on_client_connected(transport, client):
logger.info("Client connected")
messages.append({"role": "system", "content": "Say hello and briefly introduce yourself."})
await task.queue_frames([context_aggregator.user().get_context_frame()])
@transport.event_handler("on_client_disconnected")
async def on_client_disconnected(transport, client):
logger.info("Client disconnected")
await task.cancel()
runner = PipelineRunner(handle_sigint=False)
await runner.run(task)
async def bot(
session_args: DailySessionArguments | SmallWebRTCSessionArguments | WebSocketSessionArguments,
):
"""Main bot entry point compatible with Pipecat Cloud."""
if isinstance(session_args, DailySessionArguments):
from pipecat.transports.services.daily import DailyParams, DailyTransport
transport = DailyTransport(
session_args.room_url,
session_args.token,
"Pipecat Bot",
params=DailyParams(
audio_in_enabled=True,
audio_out_enabled=True,
vad_analyzer=SileroVADAnalyzer(),
),
)
elif isinstance(session_args, SmallWebRTCSessionArguments):
from pipecat.transports.base_transport import TransportParams
from pipecat.transports.network.small_webrtc import SmallWebRTCTransport
transport = SmallWebRTCTransport(
params=TransportParams(
audio_in_enabled=True,
audio_out_enabled=True,
vad_analyzer=SileroVADAnalyzer(),
),
webrtc_connection=session_args.webrtc_connection,
)
elif isinstance(session_args, WebSocketSessionArguments):
from pipecat.transports.network.fastapi_websocket import (
FastAPIWebsocketParams,
FastAPIWebsocketTransport,
)
# Create base parameters for telephony
params = FastAPIWebsocketParams(
audio_in_enabled=True,
audio_out_enabled=True,
vad_analyzer=SileroVADAnalyzer(),
add_wav_header=False,
)
# Create appropriate serializer based on transport type
transport_type = getattr(session_args, "transport_type", "unknown")
call_info = getattr(session_args, "call_info", {})
if transport_type == "twilio":
from pipecat.serializers.twilio import TwilioFrameSerializer
params.serializer = TwilioFrameSerializer(
stream_sid=call_info["stream_sid"],
call_sid=call_info["call_sid"],
account_sid=os.getenv("TWILIO_ACCOUNT_SID", ""),
auth_token=os.getenv("TWILIO_AUTH_TOKEN", ""),
)
elif transport_type == "telnyx":
from pipecat.serializers.telnyx import TelnyxFrameSerializer
params.serializer = TelnyxFrameSerializer(
stream_id=call_info["stream_id"],
call_control_id=call_info["call_control_id"],
outbound_encoding=call_info["outbound_encoding"],
inbound_encoding="PCMU",
)
elif transport_type == "plivo":
from pipecat.serializers.plivo import PlivoFrameSerializer
params.serializer = PlivoFrameSerializer(
stream_id=call_info["stream_id"],
call_id=call_info["call_id"],
)
else:
raise ValueError(f"Unsupported WebSocket transport type: {transport_type}")
transport = FastAPIWebsocketTransport(websocket=session_args.websocket, params=params)
else:
raise ValueError(f"Unsupported session arguments type: {type(session_args)}")
await run_bot(transport)
if __name__ == "__main__":
from pipecat.runner.cloud import main
main()

161
examples/quickstart/cloud-simple-bot.py
View File

@@ -1,161 +0,0 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import os
from dataclasses import dataclass
from typing import Any, Optional
from dotenv import load_dotenv
from loguru import logger
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.openai_llm_context import OpenAILLMContext
# from pipecat.runner.cloud import SmallWebRTCSessionArguments # Need a release of Pipecat to use this
from pipecat.processors.frameworks.rtvi import RTVIConfig, RTVIObserver, RTVIProcessor
from pipecat.services.cartesia.tts import CartesiaTTSService
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.openai.llm import OpenAILLMService
try:
from pipecatcloud.agent import DailySessionArguments
except ImportError:
raise ImportError(
"pipecatcloud package is required for cloud-compatible bots. "
"Install with: pip install pipecat-ai[[pipecatcloud]]"
)
load_dotenv(override=True)
# For now, we'll just define SmallWebRTCSessionArguments here directly since Pipecat
# isn't released with the pipecat.runner.cloud module yet.
# This saves us from having to build a Docker container from my branch or main to
# deploy to PCC.
@dataclass
class SmallWebRTCSessionArguments:
"""Small WebRTC session arguments for local development.
This will be replaced by pipecatcloud.agent.SmallWebRTCSessionArguments
when WebRTC support is added to Pipecat Cloud.
"""
webrtc_connection: Any
session_id: Optional[str] = None
# Check if we're running locally
IS_LOCAL_RUN = os.environ.get("LOCAL_RUN", "0") == "1"
async def run_bot(transport):
"""Main bot logic that works with any transport."""
logger.info(f"Starting bot")
stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
tts = CartesiaTTSService(
api_key=os.getenv("CARTESIA_API_KEY"),
voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady
)
llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
messages = [
{
"role": "system",
"content": "You are a friendly AI assistant. Respond naturally and keep your answers conversational.",
},
]
context = OpenAILLMContext(messages)
context_aggregator = llm.create_context_aggregator(context)
rtvi = RTVIProcessor(config=RTVIConfig(config=[]))
pipeline = Pipeline(
[
transport.input(),
rtvi,
stt,
context_aggregator.user(),
llm,
tts,
transport.output(),
context_aggregator.assistant(),
]
)
task = PipelineTask(
pipeline,
params=PipelineParams(
enable_metrics=True,
enable_usage_metrics=True,
),
observers=[RTVIObserver(rtvi)],
)
@transport.event_handler("on_client_connected")
async def on_client_connected(transport, client):
logger.info("Client connected")
messages.append({"role": "system", "content": "Say hello and briefly introduce yourself."})
await task.queue_frames([context_aggregator.user().get_context_frame()])
@transport.event_handler("on_client_disconnected")
async def on_client_disconnected(transport, client):
logger.info("Client disconnected")
await task.cancel()
runner = PipelineRunner(handle_sigint=False)
await runner.run(task)
async def bot(session_args: DailySessionArguments | SmallWebRTCSessionArguments):
"""Main bot entry point compatible with Pipecat Cloud."""
if isinstance(session_args, DailySessionArguments):
from pipecat.transports.services.daily import DailyParams, DailyTransport
if not IS_LOCAL_RUN:
from pipecat.audio.filters.krisp_filter import KrispFilter
transport = DailyTransport(
session_args.room_url,
session_args.token,
"Pipecat Bot",
params=DailyParams(
audio_in_enabled=True,
audio_in_filter=None
if IS_LOCAL_RUN
else KrispFilter(), # Only use Krisp in production
audio_out_enabled=True,
vad_analyzer=SileroVADAnalyzer(),
),
)
elif isinstance(session_args, SmallWebRTCSessionArguments):
from pipecat.transports.base_transport import TransportParams
from pipecat.transports.network.small_webrtc import SmallWebRTCTransport
transport = SmallWebRTCTransport(
params=TransportParams(
audio_in_enabled=True,
audio_out_enabled=True,
vad_analyzer=SileroVADAnalyzer(),
),
webrtc_connection=session_args.webrtc_connection,
)
await run_bot(transport)
if __name__ == "__main__":
from pipecat.runner.cloud import main
main()

122
examples/quickstart/local-multi-bot.py
View File

@@ -1,122 +0,0 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import argparse
import os
from dotenv import load_dotenv
from loguru import logger
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.openai_llm_context import OpenAILLMContext
from pipecat.processors.frameworks.rtvi import RTVIConfig, RTVIObserver, RTVIProcessor
from pipecat.services.cartesia.tts import CartesiaTTSService
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.openai.llm import OpenAILLMService
from pipecat.transports.base_transport import BaseTransport, TransportParams
load_dotenv(override=True)
def create_transport_params(transport_name):
"""Create transport parameters based on transport name."""
base_config = {
"audio_in_enabled": True,
"audio_out_enabled": True,
"vad_analyzer": SileroVADAnalyzer(),
}
if transport_name == "daily":
from pipecat.transports.services.daily import DailyParams
return DailyParams(**base_config)
elif transport_name == "livekit":
from pipecat.transports.services.livekit import LiveKitParams
return LiveKitParams(**base_config)
elif transport_name in ["plivo", "telnyx", "twilio"]:
from pipecat.transports.network.fastapi_websocket import FastAPIWebsocketParams
return FastAPIWebsocketParams(**base_config)
else: # webrtc
return TransportParams(**base_config)
async def run_bot(transport: BaseTransport, _: argparse.Namespace, handle_sigint: bool):
logger.info(f"Starting bot")
stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
tts = CartesiaTTSService(
api_key=os.getenv("CARTESIA_API_KEY"),
voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady
)
llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
messages = [
{
"role": "system",
"content": "You are a friendly AI assistant. Respond naturally and keep your answers conversational.",
},
]
context = OpenAILLMContext(messages)
context_aggregator = llm.create_context_aggregator(context)
rtvi = RTVIProcessor(config=RTVIConfig(config=[]))
pipeline = Pipeline(
[
transport.input(), # Transport user input
rtvi,
stt,
context_aggregator.user(), # User responses
llm, # LLM
tts, # TTS
transport.output(), # Transport bot output
context_aggregator.assistant(), # Assistant spoken responses
]
)
task = PipelineTask(
pipeline,
params=PipelineParams(
enable_metrics=True,
enable_usage_metrics=True,
),
observers=[RTVIObserver(rtvi)],
)
@transport.event_handler("on_client_connected")
async def on_client_connected(transport, client):
logger.info(f"Client connected")
# Kick off the conversation.
messages.append({"role": "system", "content": "Say hello and briefly introduce yourself."})
await task.queue_frames([context_aggregator.user().get_context_frame()])
@transport.event_handler("on_client_disconnected")
async def on_client_disconnected(transport, client):
logger.info(f"Client disconnected")
await task.cancel()
runner = PipelineRunner(handle_sigint=handle_sigint)
await runner.run(task)
if __name__ == "__main__":
from pipecat.runner.local import main
transport_params = {
transport: lambda t=transport: create_transport_params(t)
for transport in ["daily", "livekit", "plivo", "telnyx", "twilio", "webrtc"]
}
main(run_bot, transport_params=transport_params)

8
examples/quickstart/pcc-deploy.toml
View File

@@ -1,8 +0,0 @@
agent_name = "cloud-simple-bot"
image = "your_dockerhub_username/cloud-simple-bot:0.1"
image_credentials = "dockerhub-access"
secret_set = "cloud-simple-bot-secrets"
enable_krisp = true
[scaling]
min_agents = 0

6
examples/quickstart/requirements.txt
View File

@@ -1,3 +1,3 @@
pipecatcloud
pipecat-ai[openai,daily,deepgram,cartesia,silero]
python-dotenv
pipecat-ai[webrtc,silero,deepgram,openai,cartesia]
pipecat-ai-small-webrtc-prebuilt
python-dotenv