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

Update more examples, misc services

Browse Source
This commit is contained in:
Mark Backman
2026-03-04 23:29:50 -05:00
parent 62554a2390
commit 8a203dd98f
36 changed files with 221 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
examples/foundational/03-still-frame.py
View File

@@ -16,7 +16,7 @@ from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineTask
from pipecat.runner.types import RunnerArguments
from pipecat.runner.utils import create_transport
from pipecat.services.fal.image import FalImageGenService
from pipecat.services.fal.image import FalImageGenService, FalImageGenSettings
from pipecat.transports.base_transport import BaseTransport, TransportParams
from pipecat.transports.daily.transport import DailyParams
@@ -45,7 +45,9 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
# Create an HTTP session
async with aiohttp.ClientSession() as session:
imagegen = FalImageGenService(
params=FalImageGenService.InputParams(image_size="square_hd"),
settings=FalImageGenSettings(
image_size="square_hd",
),
aiohttp_session=session,
key=os.getenv("FAL_KEY"),
)

6
examples/foundational/03a-local-still-frame.py
View File

@@ -17,7 +17,7 @@ from pipecat.frames.frames import TextFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineTask
from pipecat.services.fal.image import FalImageGenService
from pipecat.services.fal.image import FalImageGenService, FalImageGenSettings
from pipecat.transports.local.tk import TkLocalTransport, TkTransportParams
load_dotenv(override=True)
@@ -37,7 +37,9 @@ async def main():
)
imagegen = FalImageGenService(
params=FalImageGenService.InputParams(image_size="square_hd"),
settings=FalImageGenSettings(
image_size="square_hd",
),
aiohttp_session=session,
key=os.getenv("FAL_KEY"),
)

6
examples/foundational/05-sync-speech-and-image.py
View File

@@ -28,7 +28,7 @@ from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.runner.types import RunnerArguments
from pipecat.runner.utils import create_transport
from pipecat.services.cartesia.tts import CartesiaHttpTTSService, CartesiaTTSSettings
from pipecat.services.fal.image import FalImageGenService
from pipecat.services.fal.image import FalImageGenService, FalImageGenSettings
from pipecat.services.openai.llm import OpenAILLMService
from pipecat.transports.base_transport import BaseTransport, TransportParams
from pipecat.transports.daily.transport import DailyParams
@@ -104,7 +104,9 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
)
imagegen = FalImageGenService(
params=FalImageGenService.InputParams(image_size="square_hd"),
settings=FalImageGenSettings(
image_size="square_hd",
),
aiohttp_session=session,
key=os.getenv("FAL_KEY"),
)

6
examples/foundational/05a-local-sync-speech-and-image.py
View File

@@ -29,7 +29,7 @@ from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.sentence import SentenceAggregator
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.services.cartesia.tts import CartesiaHttpTTSService, CartesiaTTSSettings
from pipecat.services.fal.image import FalImageGenService
from pipecat.services.fal.image import FalImageGenService, FalImageGenSettings
from pipecat.services.openai.llm import OpenAILLMService
from pipecat.transports.local.tk import TkLocalTransport, TkTransportParams
@@ -104,7 +104,9 @@ async def main():
)
imagegen = FalImageGenService(
params=FalImageGenService.InputParams(image_size="square_hd"),
settings=FalImageGenSettings(
image_size="square_hd",
),
aiohttp_session=session,
key=os.getenv("FAL_KEY"),
)

2
examples/foundational/14o-function-calling-gemini-openai-format.py
View File

@@ -65,7 +65,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
tts = ElevenLabsTTSService(
api_key=os.getenv("ELEVENLABS_API_KEY", ""),
settings=ElevenLabsTTSSettings(
voice_id=os.getenv("ELEVENLABS_VOICE_ID", ""),
voice=os.getenv("ELEVENLABS_VOICE_ID", ""),
),
)

2
examples/foundational/14p-function-calling-gemini-vertex-ai.py
View File

@@ -65,7 +65,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
tts = ElevenLabsTTSService(
api_key=os.getenv("ELEVENLABS_API_KEY", ""),
settings=ElevenLabsTTSSettings(
voice_id=os.getenv("ELEVENLABS_VOICE_ID", ""),
voice=os.getenv("ELEVENLABS_VOICE_ID", ""),
),
)

2
examples/foundational/14r-function-calling-aws.py
View File

@@ -67,7 +67,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
tts = AWSPollyTTSService(
region="us-west-2", # only specific regions support generative TTS
settings=AWSPollyTTSSettings(
voice_id="Joanna",
voice="Joanna",
engine="generative",
rate="1.1",
),

4
examples/foundational/36-user-email-gathering.py
View File

@@ -77,7 +77,9 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
# (see https://docs.rime.ai/api-reference/spell)
# tts = RimeHttpTTSService(
# api_key=os.getenv("RIME_API_KEY", ""),
# voice_id="eva",
# settings=RimeTTSSettings(
# voice="eva",
# ),
# aiohttp_session=session,
# )

9
examples/foundational/55d-update-settings-assemblyai-stt.py
View File

@@ -22,7 +22,6 @@ from pipecat.processors.aggregators.llm_response_universal import (
)
from pipecat.runner.types import RunnerArguments
from pipecat.runner.utils import create_transport
from pipecat.services.assemblyai.models import AssemblyAIConnectionParams
from pipecat.services.assemblyai.stt import AssemblyAISTTService, AssemblyAISTTSettings
from pipecat.services.cartesia.tts import CartesiaTTSService, CartesiaTTSSettings
from pipecat.services.openai.llm import OpenAILLMService
@@ -53,8 +52,8 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
stt = AssemblyAISTTService(
api_key=os.getenv("ASSEMBLYAI_API_KEY"),
connection_params=AssemblyAIConnectionParams(
speech_model="u3-rt-pro",
settings=AssemblyAISTTSettings(
model="u3-rt-pro",
),
)
@@ -111,9 +110,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
await task.queue_frame(
STTUpdateSettingsFrame(
delta=AssemblyAISTTSettings(
connection_params=AssemblyAIConnectionParams(
keyterms_prompt=["Xiomara", "Saoirse", "Krzystof", "Nguyen", "Pipecat"]
)
keyterms_prompt=["Xiomara", "Saoirse", "Krzystof", "Nguyen", "Pipecat"]
)
)
)

2
examples/foundational/55h-update-settings-speechmatics-stt.py
View File

@@ -53,7 +53,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
stt = SpeechmaticsSTTService(
api_key=os.getenv("SPEECHMATICS_API_KEY"),
params=SpeechmaticsSTTService.InputParams(
settings=SpeechmaticsSTTSettings(
enable_diarization=True,
speaker_active_format="<{speaker_id}>{text}</{speaker_id}>",
speaker_passive_format="<PASSIVE><{speaker_id}>{text}</{speaker_id}></PASSIVE>",

2
examples/foundational/55o-update-settings-elevenlabs-http-tts.py
View File

@@ -58,7 +58,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
tts = ElevenLabsHttpTTSService(
api_key=os.getenv("ELEVENLABS_API_KEY"),
voice_id=os.getenv("ELEVENLABS_VOICE_ID"),
settings=ElevenLabsHttpTTSSettings(voice=os.getenv("ELEVENLABS_VOICE_ID")),
aiohttp_session=session,
)

2
examples/foundational/55o-update-settings-elevenlabs-tts.py
View File

@@ -55,7 +55,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
tts = ElevenLabsTTSService(
api_key=os.getenv("ELEVENLABS_API_KEY"),
voice_id=os.getenv("ELEVENLABS_VOICE_ID"),
settings=ElevenLabsTTSSettings(voice=os.getenv("ELEVENLABS_VOICE_ID")),
)
llm = OpenAILLMService(

4
examples/foundational/55u-update-settings-rime-http-tts.py
View File

@@ -57,7 +57,9 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
tts = RimeHttpTTSService(
api_key=os.getenv("RIME_API_KEY"), voice_id="eva", aiohttp_session=session
api_key=os.getenv("RIME_API_KEY"),
settings=RimeTTSSettings(voice="eva"),
aiohttp_session=session,
)
llm = OpenAILLMService(

2
examples/foundational/55u-update-settings-rime-tts.py
View File

@@ -54,7 +54,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
tts = RimeTTSService(
api_key=os.getenv("RIME_API_KEY"),
voice_id="luna",
settings=RimeTTSSettings(voice="luna"),
)
llm = OpenAILLMService(

2
examples/foundational/55v-update-settings-lmnt-tts.py
View File

@@ -54,7 +54,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
tts = LmntTTSService(
api_key=os.getenv("LMNT_API_KEY"),
voice_id="lily",
settings=LmntTTSSettings(voice="lily"),
)
llm = OpenAILLMService(

2
examples/foundational/55w-update-settings-fish-tts.py
View File

@@ -54,7 +54,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
tts = FishAudioTTSService(
api_key=os.getenv("FISH_API_KEY"),
model="4ce7e917cedd4bc2bb2e6ff3a46acaa1", # Barack Obama
settings=FishAudioTTSSettings(voice="4ce7e917cedd4bc2bb2e6ff3a46acaa1"), # Barack Obama
)
llm = OpenAILLMService(

2
examples/foundational/55z-update-settings-hume-tts.py
View File

@@ -54,7 +54,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
tts = HumeTTSService(
api_key=os.getenv("HUME_API_KEY"),
voice_id="f898a92e-685f-43fa-985b-a46920f0650b",
settings=HumeTTSSettings(voice="f898a92e-685f-43fa-985b-a46920f0650b"),
)
llm = OpenAILLMService(

6
examples/foundational/55zc-update-settings-gemini-tts.py
View File

@@ -55,9 +55,9 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
tts = GeminiTTSService(
credentials=os.getenv("GOOGLE_TEST_CREDENTIALS"),
model="gemini-2.5-flash-tts",
voice_id="Charon",
params=GeminiTTSService.InputParams(
settings=GeminiTTSSettings(
model="gemini-2.5-flash-tts",
voice="Charon",
language=Language.EN_US,
prompt="You are a helpful AI assistant. Speak in a natural, conversational tone.",
),

2
examples/foundational/55zh-update-settings-resembleai-tts.py
View File

@@ -54,7 +54,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
tts = ResembleAITTSService(
api_key=os.getenv("RESEMBLE_API_KEY"),
voice_id=os.getenv("RESEMBLE_VOICE_UUID"),
settings=ResembleAITTSSettings(voice=os.getenv("RESEMBLE_VOICE_UUID")),
)
llm = OpenAILLMService(

4
examples/foundational/55zi-update-settings-azure-llm.py
View File

@@ -22,7 +22,7 @@ from pipecat.processors.aggregators.llm_response_universal import (
)
from pipecat.runner.types import RunnerArguments
from pipecat.runner.utils import create_transport
from pipecat.services.azure.llm import AzureLLMService
from pipecat.services.azure.llm import AzureLLMService, AzureLLMSettings
from pipecat.services.cartesia.tts import CartesiaTTSService, CartesiaTTSSettings
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.openai.base_llm import OpenAILLMSettings
@@ -63,7 +63,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
llm = AzureLLMService(
api_key=os.getenv("AZURE_CHATGPT_API_KEY"),
endpoint=os.getenv("AZURE_CHATGPT_ENDPOINT"),
model=os.getenv("AZURE_CHATGPT_MODEL"),
settings=AzureLLMSettings(model=os.getenv("AZURE_CHATGPT_MODEL")),
system_instruction="You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be spoken aloud, so avoid special characters that can't easily be spoken, such as emojis or bullet points. Respond to what the user said in a creative and helpful way.",
)

6
examples/foundational/55zp-update-settings-aws-bedrock-llm.py
View File

@@ -61,8 +61,10 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
llm = AWSBedrockLLMService(
aws_region="us-west-2",
model="us.anthropic.claude-haiku-4-5-20251001-v1:0",
params=AWSBedrockLLMService.InputParams(temperature=0.8),
settings=AWSBedrockLLMSettings(
model="us.anthropic.claude-haiku-4-5-20251001-v1:0",
temperature=0.8,
),
system_instruction="You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be spoken aloud, so avoid special characters that can't easily be spoken, such as emojis or bullet points. Respond to what the user said in a creative and helpful way.",
)

4
examples/foundational/55zv-update-settings-asyncai-http-tts.py
View File

@@ -59,7 +59,9 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
tts = AsyncAIHttpTTSService(
api_key=os.getenv("ASYNCAI_API_KEY", ""),
voice_id=os.getenv("ASYNCAI_VOICE_ID", "e0f39dc4-f691-4e78-bba5-5c636692cc04"),
settings=AsyncAITTSSettings(
voice=os.getenv("ASYNCAI_VOICE_ID", "e0f39dc4-f691-4e78-bba5-5c636692cc04")
),
aiohttp_session=session,
)

4
examples/foundational/55zv-update-settings-asyncai-tts.py
View File

@@ -55,7 +55,9 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
tts = AsyncAITTSService(
api_key=os.getenv("ASYNCAI_API_KEY", ""),
voice_id=os.getenv("ASYNCAI_VOICE_ID", "e0f39dc4-f691-4e78-bba5-5c636692cc04"),
settings=AsyncAITTSSettings(
voice=os.getenv("ASYNCAI_VOICE_ID", "e0f39dc4-f691-4e78-bba5-5c636692cc04")
),
)
llm = OpenAILLMService(

2
examples/foundational/55zw-update-settings-gradium-tts.py
View File

@@ -54,7 +54,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
tts = GradiumTTSService(
api_key=os.getenv("GRADIUM_API_KEY"),
voice_id="YTpq7expH9539ERJ",
settings=GradiumTTSSettings(voice="YTpq7expH9539ERJ"),
url="wss://us.api.gradium.ai/api/speech/tts",
)

4
examples/foundational/55zz-update-settings-fireworks-llm.py
View File

@@ -24,7 +24,7 @@ from pipecat.runner.types import RunnerArguments
from pipecat.runner.utils import create_transport
from pipecat.services.cartesia.tts import CartesiaTTSService, CartesiaTTSSettings
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.fireworks.llm import FireworksLLMService
from pipecat.services.fireworks.llm import FireworksLLMService, FireworksLLMSettings
from pipecat.services.openai.base_llm import OpenAILLMSettings
from pipecat.transports.base_transport import BaseTransport, TransportParams
from pipecat.transports.daily.transport import DailyParams
@@ -62,7 +62,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
llm = FireworksLLMService(
api_key=os.getenv("FIREWORKS_API_KEY"),
model="accounts/fireworks/models/gpt-oss-20b",
settings=FireworksLLMSettings(model="accounts/fireworks/models/gpt-oss-20b"),
system_instruction="You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be spoken aloud, so avoid special characters that can't easily be spoken, such as emojis or bullet points. Respond to what the user said in a creative and helpful way.",
)

4
examples/foundational/55zzb-update-settings-groq-llm.py
View File

@@ -24,7 +24,7 @@ from pipecat.runner.types import RunnerArguments
from pipecat.runner.utils import create_transport
from pipecat.services.cartesia.tts import CartesiaTTSService, CartesiaTTSSettings
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.groq.llm import GroqLLMService
from pipecat.services.groq.llm import GroqLLMService, GroqLLMSettings
from pipecat.services.openai.base_llm import OpenAILLMSettings
from pipecat.transports.base_transport import BaseTransport, TransportParams
from pipecat.transports.daily.transport import DailyParams
@@ -62,7 +62,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
llm = GroqLLMService(
api_key=os.getenv("GROQ_API_KEY"),
model="meta-llama/llama-4-maverick-17b-128e-instruct",
settings=GroqLLMSettings(model="meta-llama/llama-4-maverick-17b-128e-instruct"),
system_instruction="You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be spoken aloud, so avoid special characters that can't easily be spoken, such as emojis or bullet points. Respond to what the user said in a creative and helpful way.",
)

4
examples/foundational/55zzd-update-settings-nvidia-llm.py
View File

@@ -24,7 +24,7 @@ from pipecat.runner.types import RunnerArguments
from pipecat.runner.utils import create_transport
from pipecat.services.cartesia.tts import CartesiaTTSService, CartesiaTTSSettings
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.nvidia.llm import NvidiaLLMService
from pipecat.services.nvidia.llm import NvidiaLLMService, NvidiaLLMSettings
from pipecat.services.openai.base_llm import OpenAILLMSettings
from pipecat.transports.base_transport import BaseTransport, TransportParams
from pipecat.transports.daily.transport import DailyParams
@@ -62,7 +62,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
llm = NvidiaLLMService(
api_key=os.getenv("NVIDIA_API_KEY"),
model="meta/llama-3.1-405b-instruct",
settings=NvidiaLLMSettings(model="meta/llama-3.1-405b-instruct"),
system_instruction="You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be spoken aloud, so avoid special characters that can't easily be spoken, such as emojis or bullet points. Respond to what the user said in a creative and helpful way.",
)

4
examples/foundational/55zze-update-settings-ollama-llm.py
View File

@@ -24,7 +24,7 @@ from pipecat.runner.types import RunnerArguments
from pipecat.runner.utils import create_transport
from pipecat.services.cartesia.tts import CartesiaTTSService, CartesiaTTSSettings
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.ollama.llm import OLLamaLLMService
from pipecat.services.ollama.llm import OLLamaLLMService, OllamaLLMSettings
from pipecat.services.openai.base_llm import OpenAILLMSettings
from pipecat.transports.base_transport import BaseTransport, TransportParams
from pipecat.transports.daily.transport import DailyParams
@@ -61,7 +61,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
)
llm = OLLamaLLMService(
model="llama3.2",
settings=OllamaLLMSettings(model="llama3.2"),
system_instruction="You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be spoken aloud, so avoid special characters that can't easily be spoken, such as emojis or bullet points. Respond to what the user said in a creative and helpful way.",
) # Update to the model you're running locally

4
examples/foundational/55zzh-update-settings-qwen-llm.py
View File

@@ -25,7 +25,7 @@ from pipecat.runner.utils import create_transport
from pipecat.services.cartesia.tts import CartesiaTTSService, CartesiaTTSSettings
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.openai.base_llm import OpenAILLMSettings
from pipecat.services.qwen.llm import QwenLLMService
from pipecat.services.qwen.llm import QwenLLMService, QwenLLMSettings
from pipecat.transports.base_transport import BaseTransport, TransportParams
from pipecat.transports.daily.transport import DailyParams
from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams
@@ -62,7 +62,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
llm = QwenLLMService(
api_key=os.getenv("QWEN_API_KEY"),
model="qwen2.5-72b-instruct",
settings=QwenLLMSettings(model="qwen2.5-72b-instruct"),
system_instruction="You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be spoken aloud, so avoid special characters that can't easily be spoken, such as emojis or bullet points. Respond to what the user said in a creative and helpful way.",
)

4
examples/foundational/55zzj-update-settings-together-llm.py
View File

@@ -25,7 +25,7 @@ from pipecat.runner.utils import create_transport
from pipecat.services.cartesia.tts import CartesiaTTSService, CartesiaTTSSettings
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.openai.base_llm import OpenAILLMSettings
from pipecat.services.together.llm import TogetherLLMService
from pipecat.services.together.llm import TogetherLLMService, TogetherLLMSettings
from pipecat.transports.base_transport import BaseTransport, TransportParams
from pipecat.transports.daily.transport import DailyParams
from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams
@@ -62,7 +62,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
llm = TogetherLLMService(
api_key=os.getenv("TOGETHER_API_KEY"),
model="meta-llama/Meta-Llama-3.1-70B-Instruct-Turbo",
settings=TogetherLLMSettings(model="meta-llama/Meta-Llama-3.1-70B-Instruct-Turbo"),
system_instruction="You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be spoken aloud, so avoid special characters that can't easily be spoken, such as emojis or bullet points. Respond to what the user said in a creative and helpful way.",
)

35
src/pipecat/services/azure/image.py
View File

@@ -12,7 +12,7 @@ using REST endpoints for creating images from text prompts.
import asyncio
import io
from dataclasses import dataclass
from dataclasses import dataclass, field
from typing import AsyncGenerator, Optional
import aiohttp
@@ -20,7 +20,7 @@ from PIL import Image
from pipecat.frames.frames import ErrorFrame, Frame, URLImageRawFrame
from pipecat.services.image_service import ImageGenService
from pipecat.services.settings import ImageGenSettings, _warn_deprecated_param
from pipecat.services.settings import NOT_GIVEN, ImageGenSettings, _NotGiven, _warn_deprecated_param
@dataclass
@@ -29,8 +29,11 @@ class AzureImageGenSettings(ImageGenSettings):
Parameters:
model: Azure image generation model identifier.
image_size: Target size for generated images.
"""
image_size: str | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN)
class AzureImageGenServiceREST(ImageGenService):
"""Azure OpenAI REST-based image generation service.
@@ -40,10 +43,12 @@ class AzureImageGenServiceREST(ImageGenService):
and automatic image download and processing.
"""
_settings: AzureImageGenSettings
def __init__(
self,
*,
image_size: str,
image_size: Optional[str] = None,
api_key: str,
endpoint: str,
model: Optional[str] = None,
@@ -55,6 +60,10 @@ class AzureImageGenServiceREST(ImageGenService):
Args:
image_size: Size specification for generated images (e.g., "1024x1024").
.. deprecated:: 0.0.105
Use ``settings=AzureImageGenSettings(image_size=...)`` instead.
api_key: Azure OpenAI API key for authentication.
endpoint: Azure OpenAI endpoint URL.
model: The image generation model to use.
@@ -67,10 +76,22 @@ class AzureImageGenServiceREST(ImageGenService):
settings: Runtime-updatable settings. When provided alongside deprecated
parameters, ``settings`` values take precedence.
"""
# 1. Initialize default_settings with hardcoded defaults
default_settings = AzureImageGenSettings(
model=None,
image_size=None,
)
# 2. Apply direct init arg overrides (deprecated)
if model is not None:
_warn_deprecated_param("model", AzureImageGenSettings, "model")
default_settings.model = model
default_settings = AzureImageGenSettings(model=model)
if image_size is not None:
_warn_deprecated_param("image_size", AzureImageGenSettings, "image_size")
default_settings.image_size = image_size
# 4. Apply settings delta (canonical API, always wins)
if settings is not None:
default_settings.apply_update(settings)
@@ -79,7 +100,6 @@ class AzureImageGenServiceREST(ImageGenService):
self._api_key = api_key
self._azure_endpoint = endpoint
self._api_version = api_version
self._image_size = image_size
self._aiohttp_session = aiohttp_session
async def run_image_gen(self, prompt: str) -> AsyncGenerator[Frame, None]:
@@ -97,12 +117,13 @@ class AzureImageGenServiceREST(ImageGenService):
headers = {"api-key": self._api_key, "Content-Type": "application/json"}
body = {
# Enter your prompt text here
"prompt": prompt,
"size": self._image_size,
"n": 1,
}
if self._settings.image_size is not None:
body["size"] = self._settings.image_size
async with self._aiohttp_session.post(url, headers=headers, json=body) as submission:
# We never get past this line, because this header isn't
# defined on a 429 response, but something is eating our

70
src/pipecat/services/fal/image.py
View File

@@ -13,8 +13,8 @@ for creating images from text prompts using various AI models.
import asyncio
import io
import os
from dataclasses import dataclass
from typing import AsyncGenerator, Dict, Optional, Union
from dataclasses import dataclass, field
from typing import Any, AsyncGenerator, Dict, Optional, Union
import aiohttp
from loguru import logger
@@ -23,7 +23,7 @@ from pydantic import BaseModel
from pipecat.frames.frames import ErrorFrame, Frame, URLImageRawFrame
from pipecat.services.image_service import ImageGenService
from pipecat.services.settings import ImageGenSettings, _warn_deprecated_param
from pipecat.services.settings import NOT_GIVEN, ImageGenSettings, _NotGiven, _warn_deprecated_param
@dataclass
@@ -32,8 +32,36 @@ class FalImageGenSettings(ImageGenSettings):
Parameters:
model: Fal.ai model identifier.
seed: Random seed for reproducible generation. ``None`` uses a random seed.
num_inference_steps: Number of inference steps for generation.
num_images: Number of images to generate.
image_size: Image dimensions as a string preset or dict with width/height.
expand_prompt: Whether to automatically expand/enhance the prompt.
enable_safety_checker: Whether to enable content safety filtering.
format: Output image format.
"""
seed: int | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN)
num_inference_steps: int | _NotGiven = field(default_factory=lambda: NOT_GIVEN)
num_images: int | _NotGiven = field(default_factory=lambda: NOT_GIVEN)
image_size: str | Dict[str, int] | _NotGiven = field(default_factory=lambda: NOT_GIVEN)
expand_prompt: bool | _NotGiven = field(default_factory=lambda: NOT_GIVEN)
enable_safety_checker: bool | _NotGiven = field(default_factory=lambda: NOT_GIVEN)
format: str | _NotGiven = field(default_factory=lambda: NOT_GIVEN)
def to_api_arguments(self) -> Dict[str, Any]:
"""Build the Fal API arguments dict from settings, excluding None values."""
args: Dict[str, Any] = {}
if self.seed is not None:
args["seed"] = self.seed
args["num_inference_steps"] = self.num_inference_steps
args["num_images"] = self.num_images
args["image_size"] = self.image_size
args["expand_prompt"] = self.expand_prompt
args["enable_safety_checker"] = self.enable_safety_checker
args["format"] = self.format
return args
class FalImageGenService(ImageGenService):
"""Fal's image generation service.
@@ -45,6 +73,9 @@ class FalImageGenService(ImageGenService):
class InputParams(BaseModel):
"""Input parameters for Fal.ai image generation.
.. deprecated:: 0.0.105
Use ``settings=FalImageGenSettings(...)`` instead.
Parameters:
seed: Random seed for reproducible generation. If None, uses random seed.
num_inference_steps: Number of inference steps for generation. Defaults to 8.
@@ -63,10 +94,12 @@ class FalImageGenService(ImageGenService):
enable_safety_checker: bool = True
format: str = "png"
_settings: FalImageGenSettings
def __init__(
self,
*,
params: InputParams,
params: Optional[InputParams] = None,
aiohttp_session: aiohttp.ClientSession,
model: Optional[str] = None,
key: Optional[str] = None,
@@ -77,6 +110,10 @@ class FalImageGenService(ImageGenService):
Args:
params: Input parameters for image generation configuration.
.. deprecated:: 0.0.105
Use ``settings=FalImageGenSettings(...)`` instead.
aiohttp_session: HTTP client session for downloading generated images.
model: The Fal.ai model to use for generation. Defaults to "fal-ai/fast-sdxl".
@@ -89,19 +126,38 @@ class FalImageGenService(ImageGenService):
**kwargs: Additional arguments passed to parent ImageGenService.
"""
# 1. Initialize default_settings with hardcoded defaults
default_settings = FalImageGenSettings(model="fal-ai/fast-sdxl")
default_settings = FalImageGenSettings(
model="fal-ai/fast-sdxl",
seed=None,
num_inference_steps=8,
num_images=1,
image_size="square_hd",
expand_prompt=False,
enable_safety_checker=True,
format="png",
)
# 2. Apply direct init arg overrides (deprecated)
if model is not None:
_warn_deprecated_param("model", FalImageGenSettings, "model")
default_settings.model = model
if params is not None:
_warn_deprecated_param("params", FalImageGenSettings)
if not settings:
default_settings.seed = params.seed
default_settings.num_inference_steps = params.num_inference_steps
default_settings.num_images = params.num_images
default_settings.image_size = params.image_size
default_settings.expand_prompt = params.expand_prompt
default_settings.enable_safety_checker = params.enable_safety_checker
default_settings.format = params.format
# 4. Apply settings delta (canonical API, always wins)
if settings is not None:
default_settings.apply_update(settings)
super().__init__(settings=default_settings, **kwargs)
self._params = params
self._aiohttp_session = aiohttp_session
self._api_key = key or os.getenv("FAL_KEY", "")
if key:
@@ -129,7 +185,7 @@ class FalImageGenService(ImageGenService):
"Authorization": f"Key {self._api_key}",
"Content-Type": "application/json",
}
payload = {"prompt": prompt, **self._params.model_dump(exclude_none=True)}
payload = {"prompt": prompt, **self._settings.to_api_arguments()}
async with self._aiohttp_session.post(
f"https://fal.run/{self._settings.model}",

33
src/pipecat/services/google/image.py
View File

@@ -16,7 +16,7 @@ import os
# Suppress gRPC fork warnings
os.environ["GRPC_ENABLE_FORK_SUPPORT"] = "false"
from dataclasses import dataclass
from dataclasses import dataclass, field
from typing import Any, AsyncGenerator, Optional
from loguru import logger
@@ -26,7 +26,7 @@ from pydantic import BaseModel, Field
from pipecat.frames.frames import ErrorFrame, Frame, URLImageRawFrame
from pipecat.services.google.utils import update_google_client_http_options
from pipecat.services.image_service import ImageGenService
from pipecat.services.settings import ImageGenSettings, _warn_deprecated_param
from pipecat.services.settings import NOT_GIVEN, ImageGenSettings, _NotGiven, _warn_deprecated_param
try:
from google import genai
@@ -43,8 +43,13 @@ class GoogleImageGenSettings(ImageGenSettings):
Parameters:
model: Google Imagen model identifier.
number_of_images: Number of images to generate per request.
negative_prompt: Text describing what not to include in generated images.
"""
number_of_images: int | _NotGiven = field(default_factory=lambda: NOT_GIVEN)
negative_prompt: str | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN)
class GoogleImageGenService(ImageGenService):
"""Google AI image generation service using Imagen models.
@@ -57,6 +62,9 @@ class GoogleImageGenService(ImageGenService):
class InputParams(BaseModel):
"""Configuration parameters for Google image generation.
.. deprecated:: 0.0.105
Use ``settings=GoogleImageGenSettings(...)`` instead.
Parameters:
number_of_images: Number of images to generate (1-8). Defaults to 1.
model: Google Imagen model to use. Defaults to "imagen-3.0-generate-002".
@@ -67,6 +75,8 @@ class GoogleImageGenService(ImageGenService):
model: str = Field(default="imagen-3.0-generate-002")
negative_prompt: Optional[str] = Field(default=None)
_settings: GoogleImageGenSettings
def __init__(
self,
*,
@@ -83,7 +93,7 @@ class GoogleImageGenService(ImageGenService):
params: Configuration parameters for image generation.
.. deprecated:: 0.0.105
Use ``settings=GoogleImageGenSettings(model=...)`` instead.
Use ``settings=GoogleImageGenSettings(...)`` instead.
http_options: HTTP options for the client.
settings: Runtime-updatable settings. When provided alongside deprecated
@@ -91,20 +101,25 @@ class GoogleImageGenService(ImageGenService):
**kwargs: Additional arguments passed to the parent ImageGenService.
"""
# 1. Initialize default_settings with hardcoded defaults
default_settings = GoogleImageGenSettings(model="imagen-3.0-generate-002")
default_settings = GoogleImageGenSettings(
model="imagen-3.0-generate-002",
number_of_images=1,
negative_prompt=None,
)
# 3. Apply params overrides — only if settings not provided
# 2. Apply params overrides (deprecated)
if params is not None:
_warn_deprecated_param("params", GoogleImageGenSettings)
if not settings:
default_settings.model = params.model
default_settings.number_of_images = params.number_of_images
default_settings.negative_prompt = params.negative_prompt
# 4. Apply settings delta (canonical API, always wins)
if settings is not None:
default_settings.apply_update(settings)
super().__init__(settings=default_settings, **kwargs)
self._params = params or GoogleImageGenService.InputParams()
# Add client header
http_options = update_google_client_http_options(http_options)
@@ -137,11 +152,11 @@ class GoogleImageGenService(ImageGenService):
try:
response = await self._client.aio.models.generate_images(
model=self._params.model,
model=self._settings.model,
prompt=prompt,
config=types.GenerateImagesConfig(
number_of_images=self._params.number_of_images,
negative_prompt=self._params.negative_prompt,
number_of_images=self._settings.number_of_images,
negative_prompt=self._settings.negative_prompt,
),
)
await self.stop_ttfb_metrics()

10
src/pipecat/services/heygen/video.py
View File

@@ -12,6 +12,7 @@ audio/video streaming capabilities through the HeyGen API.
"""
import asyncio
from dataclasses import dataclass
from typing import Optional, Union
import aiohttp
@@ -52,6 +53,13 @@ from pipecat.transports.base_transport import TransportParams
AVATAR_VAD_STOP_SECS = 0.35
@dataclass
class HeyGenVideoSettings(ServiceSettings):
"""Settings for the HeyGen video service."""
pass
class HeyGenVideoService(AIService):
"""A service that integrates HeyGen's interactive avatar capabilities into the pipeline.
@@ -81,7 +89,7 @@ class HeyGenVideoService(AIService):
session: aiohttp.ClientSession,
session_request: Optional[Union[LiveAvatarNewSessionRequest, NewSessionRequest]] = None,
service_type: Optional[ServiceType] = None,
settings: Optional[ServiceSettings] = None,
settings: Optional[HeyGenVideoSettings] = None,
**kwargs,
) -> None:
"""Initialize the HeyGen video service.

31
src/pipecat/services/openai/image.py
View File

@@ -11,7 +11,7 @@ for creating images from text prompts.
"""
import io
from dataclasses import dataclass
from dataclasses import dataclass, field
from typing import AsyncGenerator, Literal, Optional
import aiohttp
@@ -25,7 +25,7 @@ from pipecat.frames.frames import (
URLImageRawFrame,
)
from pipecat.services.image_service import ImageGenService
from pipecat.services.settings import ImageGenSettings, _warn_deprecated_param
from pipecat.services.settings import NOT_GIVEN, ImageGenSettings, _NotGiven, _warn_deprecated_param
@dataclass
@@ -34,8 +34,11 @@ class OpenAIImageGenSettings(ImageGenSettings):
Parameters:
model: DALL-E model identifier.
image_size: Target size for generated images.
"""
image_size: str | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN)
class OpenAIImageGenService(ImageGenService):
"""OpenAI DALL-E image generation service.
@@ -45,13 +48,17 @@ class OpenAIImageGenService(ImageGenService):
with configurable quality and style parameters.
"""
_settings: OpenAIImageGenSettings
def __init__(
self,
*,
api_key: str,
base_url: Optional[str] = None,
aiohttp_session: aiohttp.ClientSession,
image_size: Literal["256x256", "512x512", "1024x1024", "1792x1024", "1024x1792"],
image_size: Optional[
Literal["256x256", "512x512", "1024x1024", "1792x1024", "1024x1792"]
] = None,
model: Optional[str] = None,
settings: Optional[OpenAIImageGenSettings] = None,
):
@@ -61,7 +68,11 @@ class OpenAIImageGenService(ImageGenService):
api_key: OpenAI API key for authentication.
base_url: Custom base URL for OpenAI API. If None, uses default.
aiohttp_session: HTTP session for downloading generated images.
image_size: Target size for generated images.
image_size: Target size for generated images. Defaults to "1024x1024".
.. deprecated:: 0.0.105
Use ``settings=OpenAIImageGenSettings(image_size=...)`` instead.
model: DALL-E model to use for generation. Defaults to "dall-e-3".
.. deprecated:: 0.0.105
@@ -71,19 +82,25 @@ class OpenAIImageGenService(ImageGenService):
parameters, ``settings`` values take precedence.
"""
# 1. Initialize default_settings with hardcoded defaults
default_settings = OpenAIImageGenSettings(model="dall-e-3")
default_settings = OpenAIImageGenSettings(
model="dall-e-3",
image_size=None,
)
# 2. Apply direct init arg overrides (deprecated)
if model is not None:
_warn_deprecated_param("model", OpenAIImageGenSettings, "model")
default_settings.model = model
if image_size is not None:
_warn_deprecated_param("image_size", OpenAIImageGenSettings, "image_size")
default_settings.image_size = image_size
# 4. Apply settings delta (canonical API, always wins)
if settings is not None:
default_settings.apply_update(settings)
super().__init__(settings=default_settings)
self._image_size = image_size
self._client = AsyncOpenAI(api_key=api_key, base_url=base_url)
self._aiohttp_session = aiohttp_session
@@ -99,7 +116,7 @@ class OpenAIImageGenService(ImageGenService):
logger.debug(f"Generating image from prompt: {prompt}")
image = await self._client.images.generate(
prompt=prompt, model=self._settings.model, n=1, size=self._image_size
prompt=prompt, model=self._settings.model, n=1, size=self._settings.image_size
)
image_url = image.data[0].url

10
src/pipecat/services/tavus/video.py
View File

@@ -11,6 +11,7 @@ avatar functionality through Tavus's streaming API.
"""
import asyncio
from dataclasses import dataclass
from typing import Optional
import aiohttp
@@ -38,6 +39,13 @@ from pipecat.services.settings import ServiceSettings
from pipecat.transports.tavus.transport import TavusCallbacks, TavusParams, TavusTransportClient
@dataclass
class TavusVideoSettings(ServiceSettings):
"""Settings for the Tavus video service."""
pass
class TavusVideoService(AIService):
"""Service that proxies audio to Tavus and receives audio and video in return.
@@ -58,7 +66,7 @@ class TavusVideoService(AIService):
replica_id: str,
persona_id: str = "pipecat-stream",
session: aiohttp.ClientSession,
settings: Optional[ServiceSettings] = None,
settings: Optional[TavusVideoSettings] = None,
**kwargs,
) -> None:
"""Initialize the Tavus video service.