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

Include examples in type checking

Browse Source 
Remove `examples/` from the `pyrightconfig.json` ignore list and fix
the resulting type errors across all example files. Common fixes:

- Required API keys: `os.getenv("X")` -> `os.environ["X"]` so the
  return type is `str` rather than `str | None`, and misconfiguration
  fails fast.
- Narrow `LLMContextMessage` union members with `isinstance(..., dict)`
  before dict-style access.
- `assert isinstance(params.llm, ...)` before calling service-specific
  methods that aren't on the base `LLMService`.
- Guard optional frame fields (e.g. `LLMSearchResponseFrame.search_result`)
  before use.
This commit is contained in:
Mark Backman
2026-04-20 15:50:49 -04:00
parent 103ced1eaa
commit 58a17c7b1b
293 changed files with 884 additions and 1006 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
examples/realtime/realtime-aws-nova-sonic.py
View File

@@ -116,8 +116,8 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
# Create the AWS Nova Sonic LLM service
llm = AWSNovaSonicLLMService(
secret_access_key=os.getenv("AWS_SECRET_ACCESS_KEY"),
access_key_id=os.getenv("AWS_ACCESS_KEY_ID"),
secret_access_key=os.environ["AWS_SECRET_ACCESS_KEY"],
access_key_id=os.environ["AWS_ACCESS_KEY_ID"],
# as of 2025-12-09, these are the supported regions:
# - Nova 2 Sonic (the default model):
# - us-east-1
@@ -126,7 +126,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
# - Nova Sonic (the older model):
# - us-east-1
# - ap-northeast-1
region=os.getenv("AWS_REGION"),
region=os.environ["AWS_REGION"],
session_token=os.getenv("AWS_SESSION_TOKEN"),
settings=AWSNovaSonicLLMService.Settings(
voice="tiffany",

4
examples/realtime/realtime-azure.py
View File

@@ -112,8 +112,8 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
logger.info(f"Starting bot")
llm = AzureRealtimeLLMService(
api_key=os.getenv("AZURE_REALTIME_API_KEY"),
base_url=os.getenv("AZURE_REALTIME_BASE_URL"),
api_key=os.environ["AZURE_REALTIME_API_KEY"],
base_url=os.environ["AZURE_REALTIME_BASE_URL"],
settings=AzureRealtimeLLMService.Settings(
system_instruction="""You are a helpful and friendly AI.

2
examples/realtime/realtime-gemini-live-files-api.py
View File

@@ -104,7 +104,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
# Initialize Gemini service with File API support
llm = GeminiLiveLLMService(
api_key=os.getenv("GOOGLE_API_KEY"),
api_key=os.environ["GOOGLE_API_KEY"],
settings=GeminiLiveLLMService.Settings(
system_instruction=system_instruction,
voice="Charon", # Aoede, Charon, Fenrir, Kore, Puck

2
examples/realtime/realtime-gemini-live-function-calling.py
View File

@@ -114,7 +114,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
)
llm = GeminiLiveLLMService(
api_key=os.getenv("GOOGLE_API_KEY"),
api_key=os.environ["GOOGLE_API_KEY"],
settings=GeminiLiveLLMService.Settings(
system_instruction=system_instruction,
),

2
examples/realtime/realtime-gemini-live-google-search.py
View File

@@ -67,7 +67,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
# Initialize the Gemini Multimodal Live model
llm = GeminiLiveLLMService(
api_key=os.getenv("GOOGLE_API_KEY"),
api_key=os.environ["GOOGLE_API_KEY"],
settings=GeminiLiveLLMService.Settings(
voice="Puck", # Aoede, Charon, Fenrir, Kore, Puck
system_instruction=system_instruction,

2
examples/realtime/realtime-gemini-live-graceful-end.py
View File

@@ -133,7 +133,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
)
llm = GeminiLiveLLMService(
api_key=os.getenv("GOOGLE_API_KEY"),
api_key=os.environ["GOOGLE_API_KEY"],
settings=GeminiLiveLLMService.Settings(
system_instruction=system_instruction,
),

21
examples/realtime/realtime-gemini-live-grounding-metadata.py
View File

@@ -75,7 +75,8 @@ class GroundingMetadataProcessor(FrameProcessor):
if isinstance(frame, LLMSearchResponseFrame):
self._grounding_count += 1
logger.info(f"\n\n🔍 GROUNDING METADATA RECEIVED #{self._grounding_count}\n")
logger.info(f"📝 Search Result Text: {frame.search_result[:200]}...")
if frame.search_result:
logger.info(f"📝 Search Result Text: {frame.search_result[:200]}...")
if frame.rendered_content:
logger.info(f"🔗 Rendered Content: {frame.rendered_content}")
@@ -101,7 +102,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
)
llm = GeminiLiveLLMService(
api_key=os.getenv("GOOGLE_API_KEY"),
api_key=os.environ["GOOGLE_API_KEY"],
settings=GeminiLiveLLMService.Settings(
system_instruction=SYSTEM_INSTRUCTION,
voice="Charon", # Aoede, Charon, Fenrir, Kore, Puck
@@ -111,16 +112,8 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
# Create a processor to capture grounding metadata
grounding_processor = GroundingMetadataProcessor()
messages = [
{
"role": "user",
"content": "Please introduce yourself and let me know that you can help with current information by searching the web. Ask me what current information I'd like to know about.",
},
]
# Set up conversation context and management
context = LLMContext(messages)
context = LLMContext()
# Server-side VAD is enabled by default; no local VAD is added.
user_aggregator, assistant_aggregator = LLMContextAggregatorPair(context)
@@ -144,6 +137,12 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
async def on_client_connected(transport, client):
logger.info(f"Client connected")
# Kick off the conversation.
context.add_message(
{
"role": "developer",
"content": "Please introduce yourself and let me know that you can help with current information by searching the web. Ask me what current information I'd like to know about.",
}
)
await task.queue_frames([LLMRunFrame()])
@transport.event_handler("on_client_disconnected")

2
examples/realtime/realtime-gemini-live-local-vad.py
View File

@@ -54,7 +54,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
logger.info(f"Starting bot")
llm = GeminiLiveLLMService(
api_key=os.getenv("GOOGLE_API_KEY"),
api_key=os.environ["GOOGLE_API_KEY"],
settings=GeminiLiveLLMService.Settings(
voice="Aoede", # Puck, Charon, Kore, Fenrir, Aoede
vad=GeminiVADParams(disabled=True),

4
examples/realtime/realtime-gemini-live-vertex-function-calling.py
View File

@@ -110,8 +110,8 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
llm = GeminiLiveVertexLLMService(
credentials=os.getenv("GOOGLE_VERTEX_TEST_CREDENTIALS"),
project_id=os.getenv("GOOGLE_CLOUD_PROJECT_ID"),
location=os.getenv("GOOGLE_CLOUD_LOCATION"),
project_id=os.environ["GOOGLE_CLOUD_PROJECT_ID"],
location=os.environ["GOOGLE_CLOUD_LOCATION"],
settings=GeminiLiveVertexLLMService.Settings(
system_instruction=system_instruction,
voice="Puck", # Aoede, Charon, Fenrir, Kore, Puck

2
examples/realtime/realtime-gemini-live-video.py
View File

@@ -47,7 +47,7 @@ transport_params = {
async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
llm = GeminiLiveLLMService(
api_key=os.getenv("GOOGLE_API_KEY"),
api_key=os.environ["GOOGLE_API_KEY"],
settings=GeminiLiveLLMService.Settings(
voice="Aoede", # Puck, Charon, Kore, Fenrir, Aoede
# system_instruction="Talk like a pirate."

2
examples/realtime/realtime-gemini-live.py
View File

@@ -52,7 +52,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
logger.info(f"Starting bot")
llm = GeminiLiveLLMService(
api_key=os.getenv("GOOGLE_API_KEY"),
api_key=os.environ["GOOGLE_API_KEY"],
settings=GeminiLiveLLMService.Settings(
voice="Aoede", # Puck, Charon, Kore, Fenrir, Aoede
# system_instruction="Talk like a pirate."

2
examples/realtime/realtime-grok.py
View File

@@ -179,7 +179,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
# Create the Grok Realtime LLM service
llm = GrokRealtimeLLMService(
api_key=os.getenv("XAI_API_KEY"),
api_key=os.environ["XAI_API_KEY"],
settings=GrokRealtimeLLMService.Settings(
system_instruction="""You are a helpful and friendly AI assistant powered by Grok.

2
examples/realtime/realtime-inworld.py
View File

@@ -84,7 +84,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
# llm_model can be any supported model or an Inworld Router.
# See: https://docs.inworld.ai/router/introduction
llm = InworldRealtimeLLMService(
api_key=os.getenv("INWORLD_API_KEY"),
api_key=os.environ["INWORLD_API_KEY"],
llm_model="xai/grok-4-1-fast-non-reasoning",
voice="Sarah",
settings=InworldRealtimeLLMService.Settings(

6
examples/realtime/realtime-openai-live-video.py
View File

@@ -62,7 +62,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
logger.info(f"Starting bot")
llm = OpenAIRealtimeLLMService(
api_key=os.getenv("OPENAI_API_KEY"),
api_key=os.environ["OPENAI_API_KEY"],
settings=OpenAIRealtimeLLMService.Settings(
system_instruction="""You are a helpful and friendly AI.
@@ -133,8 +133,8 @@ Remember, your responses should be short. Just one or two sentences, usually. Re
async def on_client_connected(transport, client):
logger.info(f"Client connected: {client}")
await maybe_capture_participant_camera(transport, client, framerate=0.5)
await maybe_capture_participant_screen(transport, client, framerate=0.5)
await maybe_capture_participant_camera(transport, client, framerate=1)
await maybe_capture_participant_screen(transport, client, framerate=1)
await task.queue_frames([LLMRunFrame()])

4
examples/realtime/realtime-openai-text.py
View File

@@ -117,7 +117,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
logger.info(f"Starting bot")
llm = OpenAIRealtimeLLMService(
api_key=os.getenv("OPENAI_API_KEY"),
api_key=os.environ["OPENAI_API_KEY"],
settings=OpenAIRealtimeLLMSettings(
system_instruction="""You are a helpful and friendly AI.
@@ -156,7 +156,7 @@ Remember, your responses should be short. Just one or two sentences, usually. Re
)
tts = CartesiaTTSService(
api_key=os.getenv("CARTESIA_API_KEY"),
api_key=os.environ["CARTESIA_API_KEY"],
settings=CartesiaTTSService.Settings(
voice="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady
),

2
examples/realtime/realtime-openai.py
View File

@@ -136,7 +136,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
logger.info(f"Starting bot")
llm = OpenAIRealtimeLLMService(
api_key=os.getenv("OPENAI_API_KEY"),
api_key=os.environ["OPENAI_API_KEY"],
settings=OpenAIRealtimeLLMService.Settings(
system_instruction="""You are a helpful and friendly AI.

2
examples/realtime/realtime-ultravox-text.py
View File

@@ -168,7 +168,7 @@ There is also a secret menu that changes daily. If the user asks about it, use t
llm = UltravoxRealtimeLLMService(
params=OneShotInputParams(
api_key=os.getenv("ULTRAVOX_API_KEY"),
api_key=os.environ["ULTRAVOX_API_KEY"],
system_prompt=system_prompt,
temperature=0.3,
max_duration=datetime.timedelta(minutes=3),

2
examples/realtime/realtime-ultravox.py
View File

@@ -166,7 +166,7 @@ There is also a secret menu that changes daily. If the user asks about it, use t
llm = UltravoxRealtimeLLMService(
params=OneShotInputParams(
api_key=os.getenv("ULTRAVOX_API_KEY"),
api_key=os.environ["ULTRAVOX_API_KEY"],
system_prompt=system_prompt,
temperature=0.3,
max_duration=datetime.timedelta(minutes=3),