Merge pull request #4378 from pipecat-ai/pk/more-pyright-fixes

More pyright fixes
This commit is contained in:
kompfner
2026-05-01 14:09:27 -04:00
committed by GitHub
95 changed files with 1031 additions and 489 deletions

View File

@@ -32,7 +32,9 @@ jobs:
run: uv python install 3.12
- name: Install development dependencies
run: uv sync --group dev --extra daily --extra tracing
# `--all-extras` (matching the dev setup in README.md) so pyright can
# resolve types from various optional dependencies.
run: uv sync --group dev --all-extras --no-extra gstreamer --no-extra local
- name: Ruff formatter
id: ruff-format

View File

@@ -10,7 +10,7 @@ Pipecat is an open-source Python framework for building real-time voice and mult
```bash
# Setup development environment
uv sync --group dev --all-extras --no-extra gstreamer
uv sync --group dev --all-extras --no-extra gstreamer --no-extra local
# Install pre-commit hooks
uv run pre-commit install

View File

@@ -6,115 +6,54 @@
"exclude": ["**/*_pb2.py", "**/__pycache__"],
"ignore": [
"tests",
"src/pipecat/adapters/services/anthropic_adapter.py",
"src/pipecat/adapters/services/aws_nova_sonic_adapter.py",
"src/pipecat/adapters/services/bedrock_adapter.py",
"src/pipecat/adapters/services/gemini_adapter.py",
"src/pipecat/adapters/services/grok_realtime_adapter.py",
"src/pipecat/adapters/services/inworld_realtime_adapter.py",
"src/pipecat/adapters/services/open_ai_adapter.py",
"src/pipecat/adapters/services/open_ai_realtime_adapter.py",
"src/pipecat/adapters/services/open_ai_responses_adapter.py",
"src/pipecat/adapters/services/perplexity_adapter.py",
"src/pipecat/audio/dtmf/utils.py",
"src/pipecat/audio/filters/aic_filter.py",
"src/pipecat/audio/filters/krisp_viva_filter.py",
"src/pipecat/audio/filters/rnnoise_filter.py",
"src/pipecat/audio/turn/smart_turn/local_smart_turn_v2.py",
"src/pipecat/audio/turn/smart_turn/local_smart_turn_v3.py",
"src/pipecat/audio/vad/silero.py",
"src/pipecat/processors/aggregators/llm_context.py",
"src/pipecat/processors/aggregators/llm_response_universal.py",
"src/pipecat/processors/frame_processor.py",
"src/pipecat/processors/frameworks/langchain.py",
"src/pipecat/processors/frameworks/rtvi/observer.py",
"src/pipecat/processors/frameworks/rtvi/processor.py",
"src/pipecat/processors/frameworks/strands_agents.py",
"src/pipecat/services/anthropic/llm.py",
"src/pipecat/services/assemblyai/stt.py",
"src/pipecat/services/aws/agent_core.py",
"src/pipecat/services/aws/llm.py",
"src/pipecat/services/aws/nova_sonic/llm.py",
"src/pipecat/services/aws/sagemaker/bidi_client.py",
"src/pipecat/services/aws/stt.py",
"src/pipecat/services/aws/tts.py",
"src/pipecat/services/aws/utils.py",
"src/pipecat/services/azure/stt.py",
"src/pipecat/services/azure/tts.py",
"src/pipecat/services/cartesia/stt.py",
"src/pipecat/services/deepgram/flux/base.py",
"src/pipecat/services/deepgram/flux/sagemaker/stt.py",
"src/pipecat/services/deepgram/flux/stt.py",
"src/pipecat/services/deepgram/sagemaker/stt.py",
"src/pipecat/services/deepgram/sagemaker/tts.py",
"src/pipecat/services/deepgram/tts.py",
"src/pipecat/services/elevenlabs/stt.py",
"src/pipecat/services/elevenlabs/tts.py",
"src/pipecat/services/fish/tts.py",
"src/pipecat/services/gladia/stt.py",
"src/pipecat/services/google/gemini_live/llm.py",
"src/pipecat/services/google/gemini_live/vertex/llm.py",
"src/pipecat/services/google/image.py",
"src/pipecat/services/google/llm.py",
"src/pipecat/services/google/stt.py",
"src/pipecat/services/google/tts.py",
"src/pipecat/services/gradium/stt.py",
"src/pipecat/services/groq/tts.py",
"src/pipecat/services/heygen/api_interactive_avatar.py",
"src/pipecat/services/heygen/base_api.py",
"src/pipecat/services/heygen/client.py",
"src/pipecat/services/heygen/video.py",
"src/pipecat/services/hume/tts.py",
"src/pipecat/services/inworld/realtime/llm.py",
"src/pipecat/services/inworld/tts.py",
"src/pipecat/services/kokoro/tts.py",
"src/pipecat/services/llm_service.py",
"src/pipecat/services/lmnt/tts.py",
"src/pipecat/services/mem0/memory.py",
"src/pipecat/services/mistral/stt.py",
"src/pipecat/services/mistral/tts.py",
"src/pipecat/services/moondream/vision.py",
"src/pipecat/services/neuphonic/tts.py",
"src/pipecat/services/nvidia/stt.py",
"src/pipecat/services/nvidia/tts.py",
"src/pipecat/services/openai/base_llm.py",
"src/pipecat/services/openai/image.py",
"src/pipecat/services/openai/llm.py",
"src/pipecat/services/openai/realtime/llm.py",
"src/pipecat/services/openai/responses/llm.py",
"src/pipecat/services/openai/stt.py",
"src/pipecat/services/openai/tts.py",
"src/pipecat/services/openrouter/llm.py",
"src/pipecat/services/piper/tts.py",
"src/pipecat/services/resembleai/tts.py",
"src/pipecat/services/rime/tts.py",
"src/pipecat/services/sambanova/llm.py",
"src/pipecat/services/sarvam/stt.py",
"src/pipecat/services/sarvam/tts.py",
"src/pipecat/services/simli/video.py",
"src/pipecat/services/smallest/tts.py",
"src/pipecat/services/soniox/stt.py",
"src/pipecat/services/speechmatics/stt.py",
"src/pipecat/services/stt_service.py",
"src/pipecat/services/tavus/video.py",
"src/pipecat/services/tts_service.py",
"src/pipecat/services/ultravox/llm.py",
"src/pipecat/services/websocket_service.py",
"src/pipecat/services/whisper/stt.py",
"src/pipecat/services/xai/realtime/llm.py",
"src/pipecat/services/xtts/tts.py",
"src/pipecat/transports/base_output.py",
"src/pipecat/transports/daily/transport.py",
"src/pipecat/transports/heygen/transport.py",
"src/pipecat/transports/lemonslice/transport.py",
"src/pipecat/transports/livekit/transport.py",
"src/pipecat/transports/smallwebrtc/connection.py",
"src/pipecat/transports/smallwebrtc/request_handler.py",
"src/pipecat/transports/smallwebrtc/transport.py",
"src/pipecat/transports/tavus/transport.py",
"src/pipecat/transports/websocket/client.py",
"src/pipecat/transports/websocket/server.py",
"src/pipecat/transports/whatsapp/client.py"
"src/pipecat/transports/websocket/server.py"
],
"reportMissingImports": false
}

View File

@@ -9,7 +9,7 @@
import copy
import json
from dataclasses import dataclass
from typing import Any, TypedDict, TypeGuard, TypeVar
from typing import Any, TypedDict, TypeGuard, TypeVar, cast
from anthropic import NOT_GIVEN, NotGiven
from anthropic.types.message_param import MessageParam
@@ -121,16 +121,20 @@ class AnthropicLLMAdapter(BaseLLMAdapter[AnthropicLLMInvocationParams]):
messages = self._from_universal_context_messages(self.get_messages(context)).messages
# Sanitize messages for logging
messages_for_logging = []
messages_for_logging: list[dict[str, Any]] = []
for message in messages:
msg = copy.deepcopy(message)
if "content" in msg:
if isinstance(msg["content"], list):
for item in msg["content"]:
if item["type"] == "image":
item["source"]["data"] = "..."
if item["type"] == "thinking" and item.get("signature"):
item["signature"] = "..."
msg: dict[str, Any] = copy.deepcopy(dict(message))
content = msg.get("content")
if isinstance(content, list):
for item in content:
if not isinstance(item, dict):
continue
if item.get("type") == "image":
source = item.get("source")
if isinstance(source, dict):
source["data"] = "..."
if item.get("type") == "thinking" and item.get("signature"):
item["signature"] = "..."
messages_for_logging.append(msg)
return messages_for_logging
@@ -185,8 +189,13 @@ class AnthropicLLMAdapter(BaseLLMAdapter[AnthropicLLMInvocationParams]):
]
if isinstance(next_message["content"], str):
next_message["content"] = [{"type": "text", "text": next_message["content"]}]
# Concatenate the content
current_message["content"].extend(next_message["content"])
# Concatenate the content. MessageParam types content as
# `str | Iterable[...]`, but this codebase assumes it's
# either a str or a list. The str case is handled above, so
# we assume that both are lists here.
cast(list[Any], current_message["content"]).extend(
cast(list[Any], next_message["content"])
)
# Remove the next message from the list
messages.pop(i + 1)
else:
@@ -239,7 +248,7 @@ class AnthropicLLMAdapter(BaseLLMAdapter[AnthropicLLMInvocationParams]):
}
# Fall back to assuming that the message is already in Anthropic format
return copy.deepcopy(message.message)
return cast(MessageParam, copy.deepcopy(message.message))
def _from_standard_message(self, message: LLMStandardMessage) -> MessageParam:
"""Convert standard universal context message to Anthropic format.
@@ -280,20 +289,26 @@ class AnthropicLLMAdapter(BaseLLMAdapter[AnthropicLLMInvocationParams]):
]
}
"""
message = copy.deepcopy(message)
if message["role"] == "tool":
return {
"role": "user",
"content": [
{
"type": "tool_result",
"tool_use_id": message["tool_call_id"],
"content": message["content"],
},
],
}
if message.get("tool_calls"):
tc = message["tool_calls"]
# ChatCompletionMessageParam (input) and MessageParam (output) are
# different TypedDicts — work with the message as a plain dict for the
# transformations below and cast back to MessageParam at return sites.
msg = cast(dict[str, Any], copy.deepcopy(message))
if msg["role"] == "tool":
return cast(
MessageParam,
{
"role": "user",
"content": [
{
"type": "tool_result",
"tool_use_id": msg["tool_call_id"],
"content": msg["content"],
},
],
},
)
if msg.get("tool_calls"):
tc = msg["tool_calls"]
ret = {"role": "assistant", "content": []}
for tool_call in tc:
function = tool_call["function"]
@@ -305,8 +320,8 @@ class AnthropicLLMAdapter(BaseLLMAdapter[AnthropicLLMInvocationParams]):
"input": arguments,
}
ret["content"].append(new_tool_use)
return ret
content = message.get("content")
return cast(MessageParam, ret)
content = msg.get("content")
if isinstance(content, str):
# fix empty text
if content == "":
@@ -354,7 +369,7 @@ class AnthropicLLMAdapter(BaseLLMAdapter[AnthropicLLMInvocationParams]):
image_item = content.pop(img_idx)
content.insert(first_txt_idx, image_item)
return message
return cast(MessageParam, msg)
def _with_cache_control_markers(self, messages: list[MessageParam]) -> list[MessageParam]:
"""Add cache control markers to messages for prompt caching.
@@ -369,7 +384,16 @@ class AnthropicLLMAdapter(BaseLLMAdapter[AnthropicLLMInvocationParams]):
def add_cache_control_marker(message: MessageParam):
if isinstance(message["content"], str):
message["content"] = [{"type": "text", "text": message["content"]}]
message["content"][-1]["cache_control"] = {"type": "ephemeral"}
# Assumptions on the next line:
# - content is a list (str case handled above; this codebase only
# ever constructs content as a str or a list)
# - the list is non-empty (guaranteed by the empty-content
# replacement in `_from_universal_context_messages`)
# - the last item is a dict. The standard-message path enforces
# this via TypedDicts (which are dicts at runtime); the
# LLMSpecificMessage passthrough doesn't, but in practice
# callers use dicts.
cast(list[Any], message["content"])[-1]["cache_control"] = {"type": "ephemeral"}
try:
# Add cache control markers to the most recent two user messages.

View File

@@ -8,9 +8,9 @@
import copy
import json
from dataclasses import dataclass
from dataclasses import asdict, dataclass
from enum import Enum
from typing import Any, TypedDict
from typing import Any, TypedDict, cast
from loguru import logger
@@ -110,7 +110,10 @@ class AWSNovaSonicLLMAdapter(BaseLLMAdapter[AWSNovaSonicLLMInvocationParams]):
Returns:
List of messages in a format ready for logging about AWS Nova Sonic.
"""
return self._from_universal_context_messages(self.get_messages(context)).messages
return [
asdict(m)
for m in self._from_universal_context_messages(self.get_messages(context)).messages
]
@dataclass
class ConvertedMessages:
@@ -123,18 +126,27 @@ class AWSNovaSonicLLMAdapter(BaseLLMAdapter[AWSNovaSonicLLMInvocationParams]):
self, universal_context_messages: list[LLMContextMessage]
) -> ConvertedMessages:
system_instruction = None
messages = []
messages: list[AWSNovaSonicConversationHistoryMessage] = []
# Bail if there are no messages
if not universal_context_messages:
return self.ConvertedMessages()
return self.ConvertedMessages(messages=[])
universal_context_messages = copy.deepcopy(universal_context_messages)
# NOTE: This adapter does not yet handle ``LLMSpecificMessage`` —
# those are filtered out below (the role-extraction and conversion
# logic only applies to standard message dicts). If/when this
# adapter grows a per-provider passthrough like the Anthropic
# adapter has, LLMSpecific items can flow through.
ucm: list[dict[str, Any]] = [
cast(dict[str, Any], m)
for m in copy.deepcopy(universal_context_messages)
if isinstance(m, dict)
]
# If we have a "system" message as our first message,
# pull that out into "instruction"
if universal_context_messages[0].get("role") == "system":
system = universal_context_messages.pop(0)
if ucm and ucm[0].get("role") == "system":
system = ucm.pop(0)
content = system.get("content")
if isinstance(content, str):
system_instruction = content
@@ -145,19 +157,21 @@ class AWSNovaSonicLLMAdapter(BaseLLMAdapter[AWSNovaSonicLLMInvocationParams]):
# Convert any remaining "system"/"developer" messages to "user",
# as Nova Sonic only supports "user" and "assistant" in history.
for msg in universal_context_messages:
for msg in ucm:
if msg.get("role") in ("system", "developer"):
msg["role"] = "user"
# Process remaining messages to fill out conversation history.
for universal_context_message in universal_context_messages:
for universal_context_message in ucm:
message = self._from_universal_context_message(universal_context_message)
if message:
messages.append(message)
return self.ConvertedMessages(messages=messages, system_instruction=system_instruction)
def _from_universal_context_message(self, message) -> AWSNovaSonicConversationHistoryMessage:
def _from_universal_context_message(
self, message: dict[str, Any]
) -> AWSNovaSonicConversationHistoryMessage | None:
"""Convert standard message format to Nova Sonic format.
Args:
@@ -167,17 +181,18 @@ class AWSNovaSonicLLMAdapter(BaseLLMAdapter[AWSNovaSonicLLMInvocationParams]):
Nova Sonic conversation history message, or None if not convertible.
"""
role = message.get("role")
if message.get("role") == "user" or message.get("role") == "assistant":
if role == "user" or role == "assistant":
content = message.get("content")
if isinstance(message.get("content"), list):
content = ""
for c in message.get("content"):
if isinstance(content, list):
text_parts = []
for c in content:
if c.get("type") == "text":
content += " " + c.get("text")
text_parts.append(c.get("text"))
else:
logger.error(
f"Unhandled content type in context message: {c.get('type')} - {message}"
)
content = " ".join(t for t in text_parts if t)
# There won't be content if this is an assistant tool call entry.
# We're ignoring those since they can't be loaded into AWS Nova Sonic conversation
# history

View File

@@ -10,7 +10,7 @@ import base64
import copy
import json
from dataclasses import dataclass
from typing import Any, TypedDict
from typing import Any, TypedDict, cast
from loguru import logger
@@ -68,16 +68,19 @@ class AWSBedrockLLMAdapter(BaseLLMAdapter[AWSBedrockLLMInvocationParams]):
system_instruction,
discard_context_system=True,
)
return {
"system": [{"text": effective_system}] if effective_system else None,
"messages": converted.messages,
# NOTE: LLMContext's tools are guaranteed to be a ToolsSchema (or NOT_GIVEN)
"tools": self.from_standard_tools(context.tools) or [],
# To avoid refactoring in AWSBedrockLLMService, we just pass through tool_choice.
# Eventually (when we don't have to maintain the non-LLMContext code path) we should do
# the conversion to Bedrock's expected format here rather than in AWSBedrockLLMService.
"tool_choice": context.tool_choice,
}
return cast(
AWSBedrockLLMInvocationParams,
{
"system": [{"text": effective_system}] if effective_system else None,
"messages": converted.messages,
# NOTE: LLMContext's tools are guaranteed to be a ToolsSchema (or NOT_GIVEN)
"tools": self.from_standard_tools(context.tools) or [],
# To avoid refactoring in AWSBedrockLLMService, we just pass through tool_choice.
# Eventually (when we don't have to maintain the non-LLMContext code path) we should do
# the conversion to Bedrock's expected format here rather than in AWSBedrockLLMService.
"tool_choice": context.tool_choice,
},
)
def get_messages_for_logging(self, context) -> list[dict[str, Any]]:
"""Get messages from a universal LLM context in a format ready for logging about AWS Bedrock.
@@ -213,35 +216,36 @@ class AWSBedrockLLMAdapter(BaseLLMAdapter[AWSBedrockLLMInvocationParams]):
]
}
"""
message = copy.deepcopy(message)
if message["role"] == "tool":
# ChatCompletionMessageParam (input) and the dict shape Bedrock expects
# are different — work with the deepcopied message as a plain dict for
# the transformations below.
msg = cast(dict[str, Any], copy.deepcopy(message))
if msg["role"] == "tool":
# Try to parse the content as JSON if it looks like JSON
try:
if message["content"].strip().startswith("{") and message[
"content"
].strip().endswith("}"):
content_json = json.loads(message["content"])
if msg["content"].strip().startswith("{") and msg["content"].strip().endswith("}"):
content_json = json.loads(msg["content"])
tool_result_content = [{"json": content_json}]
else:
tool_result_content = [{"text": message["content"]}]
tool_result_content = [{"text": msg["content"]}]
except (json.JSONDecodeError, ValueError, AttributeError):
tool_result_content = [{"text": message["content"]}]
tool_result_content = [{"text": msg["content"]}]
return {
"role": "user",
"content": [
{
"toolResult": {
"toolUseId": message["tool_call_id"],
"toolUseId": msg["tool_call_id"],
"content": tool_result_content,
},
},
],
}
if message.get("tool_calls"):
tc = message["tool_calls"]
ret = {"role": "assistant", "content": []}
if msg.get("tool_calls"):
tc = msg["tool_calls"]
ret: dict[str, Any] = {"role": "assistant", "content": []}
for tool_call in tc:
function = tool_call["function"]
arguments = json.loads(function["arguments"])
@@ -256,12 +260,12 @@ class AWSBedrockLLMAdapter(BaseLLMAdapter[AWSBedrockLLMInvocationParams]):
return ret
# Handle text content
content = message.get("content")
content = msg.get("content")
if isinstance(content, str):
if content == "":
return {"role": message["role"], "content": [{"text": "(empty)"}]}
return {"role": msg["role"], "content": [{"text": "(empty)"}]}
else:
return {"role": message["role"], "content": [{"text": content}]}
return {"role": msg["role"], "content": [{"text": content}]}
elif isinstance(content, list):
new_content = []
for item in content:
@@ -300,9 +304,9 @@ class AWSBedrockLLMAdapter(BaseLLMAdapter[AWSBedrockLLMInvocationParams]):
# Move image before the first text
image_item = new_content.pop(img_idx)
new_content.insert(first_txt_idx, image_item)
return {"role": message["role"], "content": new_content}
return {"role": msg["role"], "content": new_content}
return message
return msg
@staticmethod
def _to_bedrock_function_format(function: FunctionSchema) -> dict[str, Any]:

View File

@@ -9,7 +9,7 @@
import base64
import json
from dataclasses import dataclass, field
from typing import Any, TypedDict
from typing import Any, TypedDict, cast
from loguru import logger
from openai import NotGiven
@@ -154,9 +154,12 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]):
messages = self._from_universal_context_messages(self.get_messages(context)).messages
# Sanitize messages for logging
messages_for_logging = []
messages_for_logging: list[dict[str, Any]] = []
for message in messages:
obj = message.to_json_dict()
# `to_json_dict()` returns `dict[str, object]`; treat as a plain
# dict for the value indexing/mutation below. The broad `except`
# below is the safety net if any item isn't shaped as expected.
obj: dict[str, Any] = cast(dict[str, Any], message.to_json_dict())
try:
if "parts" in obj:
for part in obj["parts"]:
@@ -274,7 +277,8 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]):
# Check if we only have function-related messages (no regular text)
effective_system = extracted_system or system_instruction
has_regular_messages = any(
len(msg.parts) == 1
msg.parts is not None
and len(msg.parts) == 1
and getattr(msg.parts[0], "text", None)
and not getattr(msg.parts[0], "function_call", None)
and not getattr(msg.parts[0], "function_response", None)
@@ -346,8 +350,11 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]):
parts=[Part(function_call=FunctionCall(name="search", args={"query": "test"}))]
)
"""
role = message["role"]
content = message.get("content", [])
# ChatCompletionMessageParam (a union of TypedDicts) doesn't allow
# the dict-style key access used below; treat it as a plain dict.
msg = cast(dict[str, Any], message)
role = msg["role"]
content = msg.get("content", [])
# Convert non-initial system/developer messages to user role,
# as Gemini doesn't support these as input messages.
@@ -359,8 +366,8 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]):
parts = []
tool_call_id_to_name_mapping = {}
if message.get("tool_calls"):
for tc in message["tool_calls"]:
if msg.get("tool_calls"):
for tc in msg["tool_calls"]:
id = tc["id"]
name = tc["function"]["name"]
tool_call_id_to_name_mapping[id] = name
@@ -376,7 +383,7 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]):
elif role == "tool":
role = "user"
try:
response = json.loads(message["content"])
response = json.loads(msg["content"])
if isinstance(response, dict):
response_dict = response
else:
@@ -384,10 +391,10 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]):
except Exception as e:
# Response might not be JSON-deserializable.
# This occurs with a UserImageFrame, for example, where we get a plain "COMPLETED" string.
response_dict = {"value": message["content"]}
response_dict = {"value": msg["content"]}
# Get function name from mapping using tool_call_id, or fallback
tool_call_id = message.get("tool_call_id")
tool_call_id = msg.get("tool_call_id")
function_name = "tool_call_result" # Default fallback
if tool_call_id and tool_call_id in params.tool_call_id_to_name_mapping:
function_name = params.tool_call_id_to_name_mapping[tool_call_id]
@@ -491,7 +498,7 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]):
def is_tool_call_message(msg: Content) -> bool:
"""Check if message contains only function_call parts."""
return (
return bool(
msg.role == "model"
and msg.parts
and all(getattr(part, "function_call", None) for part in msg.parts)
@@ -499,6 +506,8 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]):
def message_has_thought_signature(msg: Content) -> bool:
"""Check if any part in the message has a thought_signature."""
if msg.parts is None:
return False
return any(getattr(part, "thought_signature", None) for part in msg.parts)
merged_messages = []
@@ -564,6 +573,8 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]):
logger.debug(f"Thought signatures to apply: {len(thought_signature_dicts)}")
for ts in thought_signature_dicts:
bookmark = ts.get("bookmark")
if bookmark is None:
continue
if bookmark.get("function_call"):
logger.trace(f" - To function call: {bookmark['function_call']}")
elif bookmark.get("text"):
@@ -665,6 +676,8 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]):
if (
hasattr(part, "inline_data")
and part.inline_data
and part.inline_data.data is not None
and bookmark_inline_data.data is not None
# Comparing length should be good enough for matching inline data,
# especially since we're already matching thought signatures in
# strict message order. Comparing actual data is expensive.

View File

@@ -13,7 +13,7 @@ Grok's Voice Agent API.
import copy
import json
from dataclasses import dataclass
from typing import Any, TypedDict
from typing import Any, TypedDict, cast
from loguru import logger
@@ -85,7 +85,10 @@ class GrokRealtimeLLMAdapter(BaseLLMAdapter):
Returns:
List of messages with sensitive data redacted.
"""
return self.get_messages(context, truncate_large_values=True)
return cast(
list[dict[str, Any]],
self.get_messages(context, truncate_large_values=True),
)
@dataclass
class ConvertedMessages:
@@ -111,11 +114,20 @@ class GrokRealtimeLLMAdapter(BaseLLMAdapter):
if not universal_context_messages:
return self.ConvertedMessages(messages=[])
messages = copy.deepcopy(universal_context_messages)
# NOTE: This adapter does not yet handle ``LLMSpecificMessage`` —
# those are filtered out below. Other adapters (e.g. Anthropic)
# dispatch LLMSpecific items through a per-provider passthrough.
# The pack-into-single-text-message strategy here doesn't compose
# with opaque per-provider payloads.
messages: list[dict[str, Any]] = [
cast(dict[str, Any], m)
for m in copy.deepcopy(universal_context_messages)
if isinstance(m, dict)
]
system_instruction = None
# Extract system message as session instructions
if messages[0].get("role") == "system":
if messages and messages[0].get("role") == "system":
system = messages.pop(0)
content = system.get("content")
if isinstance(content, str):
@@ -133,7 +145,9 @@ class GrokRealtimeLLMAdapter(BaseLLMAdapter):
# Single user message can be sent normally
if len(messages) == 1 and messages[0].get("role") == "user":
return self.ConvertedMessages(
messages=[self._from_universal_context_message(messages[0])],
messages=[
self._from_universal_context_message(cast(LLMContextMessage, messages[0]))
],
system_instruction=system_instruction,
)
@@ -181,26 +195,29 @@ class GrokRealtimeLLMAdapter(BaseLLMAdapter):
Returns:
ConversationItem formatted for Grok Realtime API.
"""
if message.get("role") == "user":
content = message.get("content")
# NOTE: ``LLMSpecificMessage`` is not yet handled here — see the
# corresponding note in `_from_universal_context_messages`.
msg = cast(dict[str, Any], message)
if msg.get("role") == "user":
content = msg.get("content")
if isinstance(content, list):
text_content = ""
text_parts = []
for c in content:
if c.get("type") == "text":
text_content += " " + c.get("text")
text_parts.append(c.get("text"))
else:
logger.error(
f"Unhandled content type in context message: {c.get('type')} - {message}"
f"Unhandled content type in context message: {c.get('type')} - {msg}"
)
content = text_content.strip()
content = " ".join(t for t in text_parts if t).strip()
return events.ConversationItem(
role="user",
type="message",
content=[events.ItemContent(type="input_text", text=content)],
)
if message.get("role") == "assistant" and message.get("tool_calls"):
tc = message.get("tool_calls")[0]
if msg.get("role") == "assistant" and msg.get("tool_calls"):
tc = msg["tool_calls"][0]
return events.ConversationItem(
type="function_call",
call_id=tc["id"],
@@ -208,7 +225,7 @@ class GrokRealtimeLLMAdapter(BaseLLMAdapter):
arguments=tc["function"]["arguments"],
)
logger.error(f"Unhandled message type in _from_universal_context_message: {message}")
raise ValueError(f"Unhandled message type in _from_universal_context_message: {msg}")
@staticmethod
def _to_grok_function_format(function: FunctionSchema) -> dict[str, Any]:

View File

@@ -13,7 +13,7 @@ Inworld's Realtime API.
import copy
import json
from dataclasses import dataclass
from typing import Any, TypedDict
from typing import Any, TypedDict, cast
from loguru import logger
@@ -85,7 +85,10 @@ class InworldRealtimeLLMAdapter(BaseLLMAdapter):
Returns:
List of messages with sensitive data redacted.
"""
return self.get_messages(context, truncate_large_values=True)
return cast(
list[dict[str, Any]],
self.get_messages(context, truncate_large_values=True),
)
@dataclass
class ConvertedMessages:
@@ -111,11 +114,20 @@ class InworldRealtimeLLMAdapter(BaseLLMAdapter):
if not universal_context_messages:
return self.ConvertedMessages(messages=[])
messages = copy.deepcopy(universal_context_messages)
# NOTE: This adapter does not yet handle ``LLMSpecificMessage`` —
# those are filtered out below. Other adapters (e.g. Anthropic)
# dispatch LLMSpecific items through a per-provider passthrough.
# The pack-into-single-text-message strategy here doesn't compose
# with opaque per-provider payloads.
messages: list[dict[str, Any]] = [
cast(dict[str, Any], m)
for m in copy.deepcopy(universal_context_messages)
if isinstance(m, dict)
]
system_instruction = None
# Extract system message as session instructions
if messages[0].get("role") == "system":
if messages and messages[0].get("role") == "system":
system = messages.pop(0)
content = system.get("content")
if isinstance(content, str):
@@ -133,7 +145,9 @@ class InworldRealtimeLLMAdapter(BaseLLMAdapter):
# Single user message can be sent normally
if len(messages) == 1 and messages[0].get("role") == "user":
return self.ConvertedMessages(
messages=[self._from_universal_context_message(messages[0])],
messages=[
self._from_universal_context_message(cast(LLMContextMessage, messages[0]))
],
system_instruction=system_instruction,
)
@@ -181,26 +195,29 @@ class InworldRealtimeLLMAdapter(BaseLLMAdapter):
Returns:
ConversationItem formatted for Inworld Realtime API.
"""
if message.get("role") == "user":
content = message.get("content")
# NOTE: ``LLMSpecificMessage`` is not yet handled here — see the
# corresponding note in `_from_universal_context_messages`.
msg = cast(dict[str, Any], message)
if msg.get("role") == "user":
content = msg.get("content")
if isinstance(content, list):
text_content = ""
text_parts = []
for c in content:
if c.get("type") == "text":
text_content += " " + c.get("text")
text_parts.append(c.get("text"))
else:
logger.error(
f"Unhandled content type in context message: {c.get('type')} - {message}"
f"Unhandled content type in context message: {c.get('type')} - {msg}"
)
content = text_content.strip()
content = " ".join(t for t in text_parts if t).strip()
return events.ConversationItem(
role="user",
type="message",
content=[events.ItemContent(type="input_text", text=content)],
)
if message.get("role") == "assistant" and message.get("tool_calls"):
tc = message.get("tool_calls")[0]
if msg.get("role") == "assistant" and msg.get("tool_calls"):
tc = msg["tool_calls"][0]
return events.ConversationItem(
type="function_call",
call_id=tc["id"],
@@ -208,7 +225,7 @@ class InworldRealtimeLLMAdapter(BaseLLMAdapter):
arguments=tc["function"]["arguments"],
)
logger.error(f"Unhandled message type in _from_universal_context_message: {message}")
raise ValueError(f"Unhandled message type in _from_universal_context_message: {msg}")
@staticmethod
def _to_inworld_function_format(function: FunctionSchema) -> dict[str, Any]:

View File

@@ -127,12 +127,15 @@ class OpenAILLMAdapter(BaseLLMAdapter[OpenAILLMInvocationParams]):
)
if system_instruction:
# Detect initial system message for warning purposes (don't extract)
initial_content = (
messages[0].get("content", "")
if messages and messages[0].get("role") == "system"
else None
)
# Detect initial system message for warning purposes (don't extract).
# ChatCompletionMessageParam.content is `str | Iterable[...]`; we
# only forward it for warning purposes, so coerce non-strings to
# None — the resolver handles None.
initial_content: str | None = None
if messages and messages[0].get("role") == "system":
raw_content = messages[0].get("content", "")
if isinstance(raw_content, str):
initial_content = raw_content
self._resolve_system_instruction(
initial_content,
system_instruction,
@@ -140,12 +143,15 @@ class OpenAILLMAdapter(BaseLLMAdapter[OpenAILLMInvocationParams]):
)
messages = [{"role": "system", "content": system_instruction}] + messages
return {
"messages": messages,
# NOTE; LLMContext's tools are guaranteed to be a ToolsSchema (or NOT_GIVEN)
"tools": self.from_standard_tools(context.tools),
"tool_choice": _openai_from_llm_context_tool_choice(context.tool_choice),
}
return cast(
OpenAILLMInvocationParams,
{
"messages": messages,
# NOTE; LLMContext's tools are guaranteed to be a ToolsSchema (or NOT_GIVEN)
"tools": self.from_standard_tools(context.tools),
"tool_choice": _openai_from_llm_context_tool_choice(context.tool_choice),
},
)
def to_provider_tools_format(self, tools_schema: ToolsSchema) -> list[ChatCompletionToolParam]:
"""Convert function schemas to OpenAI's function-calling format.
@@ -158,13 +164,19 @@ class OpenAILLMAdapter(BaseLLMAdapter[OpenAILLMInvocationParams]):
with ChatCompletion API.
"""
functions_schema = tools_schema.standard_tools
formatted_standard_tools = [
ChatCompletionToolParam(type="function", function=func.to_default_dict())
# `function=...` expects a `FunctionDefinition` TypedDict; the dict
# produced by `to_default_dict()` is structurally compatible. Cast at
# the boundary.
formatted_standard_tools: list[ChatCompletionToolParam] = [
ChatCompletionToolParam(type="function", function=cast(Any, func.to_default_dict()))
for func in functions_schema
]
custom_openai_tools = []
custom_openai_tools: list[ChatCompletionToolParam] = []
if tools_schema.custom_tools:
custom_openai_tools = tools_schema.custom_tools.get(AdapterType.OPENAI, [])
custom_openai_tools = cast(
list[ChatCompletionToolParam],
tools_schema.custom_tools.get(AdapterType.OPENAI, []),
)
return formatted_standard_tools + custom_openai_tools
def get_messages_for_logging(self, context: LLMContext) -> list[dict[str, Any]]:
@@ -178,7 +190,10 @@ class OpenAILLMAdapter(BaseLLMAdapter[OpenAILLMInvocationParams]):
Returns:
List of messages in a format ready for logging about OpenAI.
"""
return self.get_messages(context, truncate_large_values=True)
return cast(
list[dict[str, Any]],
self.get_messages(context, truncate_large_values=True),
)
def _from_universal_context_messages(
self,

View File

@@ -9,7 +9,7 @@
import copy
import json
from dataclasses import dataclass
from typing import Any, TypedDict
from typing import Any, TypedDict, cast
from loguru import logger
@@ -81,7 +81,7 @@ class OpenAIRealtimeLLMAdapter(BaseLLMAdapter):
Returns:
List of messages in a format ready for logging about OpenAI Realtime.
"""
return self.get_messages(context, truncate_large_values=True)
return cast(list[dict[str, Any]], self.get_messages(context, truncate_large_values=True))
@dataclass
class ConvertedMessages:
@@ -101,12 +101,24 @@ class OpenAIRealtimeLLMAdapter(BaseLLMAdapter):
if not universal_context_messages:
return self.ConvertedMessages(messages=[])
messages = copy.deepcopy(universal_context_messages)
# NOTE: This adapter does not yet handle ``LLMSpecificMessage`` — those
# are filtered out below. Other adapters (e.g. Anthropic) dispatch
# LLMSpecific items through a per-provider passthrough. For OpenAI
# Realtime, the strategy here packs a multi-message history into a
# single text message (see comment further down), which doesn't
# compose with opaque per-provider payloads. If/when this adapter
# adopts the per-message strategy, LLMSpecific items can flow
# through `_from_universal_context_message` like in other adapters.
messages: list[dict[str, Any]] = [
cast(dict[str, Any], m)
for m in copy.deepcopy(universal_context_messages)
if isinstance(m, dict)
]
system_instruction = None
# If we have a "system" message as our first message,
# pull that out into session "instructions"
if messages[0].get("role") == "system":
if messages and messages[0].get("role") == "system":
system = messages.pop(0)
content = system.get("content")
if isinstance(content, str):
@@ -124,7 +136,9 @@ class OpenAIRealtimeLLMAdapter(BaseLLMAdapter):
# If we have just a single "user" item, we can just send it normally
if len(messages) == 1 and messages[0].get("role") == "user":
return self.ConvertedMessages(
messages=[self._from_universal_context_message(messages[0])],
messages=[
self._from_universal_context_message(cast(LLMContextMessage, messages[0]))
],
system_instruction=system_instruction,
)
@@ -142,18 +156,18 @@ class OpenAIRealtimeLLMAdapter(BaseLLMAdapter):
return self.ConvertedMessages(
messages=[
{
"role": "user",
"type": "message",
"content": [
{
"type": "input_text",
"text": "\n\n".join(
events.ConversationItem(
role="user",
type="message",
content=[
events.ItemContent(
type="input_text",
text="\n\n".join(
[intro_text, json.dumps(messages, indent=2), trailing_text]
),
}
)
],
}
)
],
system_instruction=system_instruction,
)
@@ -161,31 +175,34 @@ class OpenAIRealtimeLLMAdapter(BaseLLMAdapter):
def _from_universal_context_message(
self, message: LLMContextMessage
) -> events.ConversationItem:
if message.get("role") == "user":
content = message.get("content")
if isinstance(message.get("content"), list):
# NOTE: ``LLMSpecificMessage`` is not yet handled here — see the
# corresponding note in `_from_universal_context_messages`.
msg = cast(dict[str, Any], message)
if msg.get("role") == "user":
content = msg.get("content")
if isinstance(content, list):
content = ""
for c in message.get("content"):
for c in msg.get("content", []):
if c.get("type") == "text":
content += " " + c.get("text")
else:
logger.error(
f"Unhandled content type in context message: {c.get('type')} - {message}"
f"Unhandled content type in context message: {c.get('type')} - {msg}"
)
return events.ConversationItem(
role="user",
type="message",
content=[events.ItemContent(type="input_text", text=content)],
)
if message.get("role") == "assistant" and message.get("tool_calls"):
tc = message.get("tool_calls")[0]
if msg.get("role") == "assistant" and msg.get("tool_calls"):
tc = msg["tool_calls"][0]
return events.ConversationItem(
type="function_call",
call_id=tc["id"],
name=tc["function"]["name"],
arguments=tc["function"]["arguments"],
)
logger.error(f"Unhandled message type in _from_universal_context_message: {message}")
raise ValueError(f"Unhandled message type in _from_universal_context_message: {msg}")
@staticmethod
def _to_openai_realtime_function_format(function: FunctionSchema) -> dict[str, Any]:

View File

@@ -6,7 +6,7 @@
"""OpenAI Responses API adapter for Pipecat."""
from typing import Any, TypedDict
from typing import Any, Required, TypedDict, cast
from openai._types import NotGiven as OpenAINotGiven
from openai.types.responses import FunctionToolParam, ResponseInputItemParam, ToolParam
@@ -23,8 +23,10 @@ from pipecat.processors.aggregators.llm_context import (
class OpenAIResponsesLLMInvocationParams(TypedDict, total=False):
"""Context-based parameters for invoking OpenAI Responses API."""
input: list[ResponseInputItemParam]
tools: list[ToolParam] | OpenAINotGiven
# `input` and `tools` are always populated by `get_llm_invocation_params`;
# `instructions` is only set when a system instruction is present.
input: Required[list[ResponseInputItemParam]]
tools: Required[list[ToolParam] | OpenAINotGiven]
instructions: str
@@ -64,8 +66,11 @@ class OpenAIResponsesLLMAdapter(BaseLLMAdapter[OpenAIResponsesLLMInvocationParam
if system_instruction and messages:
first_msg = messages[0] if not isinstance(messages[0], LLMSpecificMessage) else None
if first_msg and first_msg.get("role") == "system":
# `content` is `str | Iterable[...]`; we only forward it for
# warning purposes. Coerce non-strings to None.
first_content = first_msg.get("content", "")
self._resolve_system_instruction(
first_msg.get("content", ""),
first_content if isinstance(first_content, str) else None,
system_instruction,
discard_context_system=False,
)
@@ -143,7 +148,10 @@ class OpenAIResponsesLLMAdapter(BaseLLMAdapter[OpenAIResponsesLLMInvocationParam
Returns:
List of messages in a format ready for logging.
"""
return self.get_messages(context, truncate_large_values=True)
return cast(
list[dict[str, Any]],
self.get_messages(context, truncate_large_values=True),
)
def _convert_messages_to_input(
self, messages: list[LLMContextMessage]
@@ -169,13 +177,15 @@ class OpenAIResponsesLLMAdapter(BaseLLMAdapter[OpenAIResponsesLLMInvocationParam
content = message.get("content", "")
if isinstance(content, list):
content = self._convert_multimodal_content(content)
result.append({"role": "developer", "content": content})
result.append(
cast(ResponseInputItemParam, {"role": "developer", "content": content})
)
elif role == "user":
content = message.get("content", "")
if isinstance(content, list):
content = self._convert_multimodal_content(content)
result.append({"role": "user", "content": content})
result.append(cast(ResponseInputItemParam, {"role": "user", "content": content}))
elif role == "assistant":
tool_calls = message.get("tool_calls")
@@ -194,7 +204,9 @@ class OpenAIResponsesLLMAdapter(BaseLLMAdapter[OpenAIResponsesLLMInvocationParam
content = message.get("content", "")
if isinstance(content, list):
content = self._convert_multimodal_content(content)
result.append({"role": "assistant", "content": content})
result.append(
cast(ResponseInputItemParam, {"role": "assistant", "content": content})
)
elif role == "tool":
content = message.get("content", "")

View File

@@ -28,6 +28,7 @@ the messages are sent to Perplexity's API.
"""
import copy
from typing import Any, cast
from openai.types.chat import ChatCompletionMessageParam
@@ -116,7 +117,11 @@ class PerplexityLLMAdapter(OpenAILLMAdapter):
if not messages:
return messages
messages = copy.deepcopy(messages)
# ChatCompletionMessageParam is a union of TypedDicts; the
# transformations below mutate by key/index in ways those TypedDicts
# don't permit. Work against a plain-dict view for the duration of
# the transformation and cast back at the return site.
msgs: list[dict[str, Any]] = cast(list[dict[str, Any]], copy.deepcopy(messages))
# Note: "developer" → "user" conversion is handled by the parent adapter
# via the convert_developer_to_user parameter.
@@ -125,10 +130,10 @@ class PerplexityLLMAdapter(OpenAILLMAdapter):
# Perplexity allows system messages at the start, but rejects them
# after any non-system message.
in_initial_system_block = True
for i in range(len(messages)):
if messages[i].get("role") == "system":
for i in range(len(msgs)):
if msgs[i].get("role") == "system":
if not in_initial_system_block:
messages[i]["role"] = "user"
msgs[i]["role"] = "user"
else:
in_initial_system_block = False
@@ -137,9 +142,9 @@ class PerplexityLLMAdapter(OpenAILLMAdapter):
# messages that violate Perplexity's strict alternation requirement.
# Skip consecutive system messages at the start — Perplexity allows those.
i = 0
while i < len(messages) - 1:
current = messages[i]
next_msg = messages[i + 1]
while i < len(msgs) - 1:
current = msgs[i]
next_msg = msgs[i + 1]
if current["role"] == next_msg["role"] == "system":
# Perplexity allows multiple initial system messages, don't merge
i += 1
@@ -154,7 +159,7 @@ class PerplexityLLMAdapter(OpenAILLMAdapter):
next_msg.get("content"), list
):
current["content"].extend(next_msg["content"])
messages.pop(i + 1)
msgs.pop(i + 1)
else:
i += 1
@@ -162,7 +167,7 @@ class PerplexityLLMAdapter(OpenAILLMAdapter):
# Perplexity requires the last message to be "user" or "tool".
# OpenAI appears to silently ignore trailing assistant messages
# server-side, so removing them preserves equivalent behavior.
while messages and messages[-1].get("role") == "assistant":
messages.pop()
while msgs and msgs[-1].get("role") == "assistant":
msgs.pop()
return messages
return cast(list[ChatCompletionMessageParam], msgs)

View File

@@ -14,7 +14,7 @@ in-memory after first load to improve performance on subsequent accesses.
import asyncio
import io
import wave
from importlib.resources import files
from importlib.resources import as_file, files
import aiofiles
@@ -52,10 +52,12 @@ async def load_dtmf_audio(button: KeypadEntry, *, sample_rate: int = 8000) -> by
__DTMF_RESAMPLER__ = create_file_resampler()
dtmf_file_name = __DTMF_FILE_NAME.get(button, f"dtmf-{button.value}.wav")
dtmf_file_path = files("pipecat.audio.dtmf").joinpath(dtmf_file_name)
async with aiofiles.open(dtmf_file_path, "rb") as f:
data = await f.read()
# `as_file` materialises the resource as a real filesystem `Path`,
# which aiofiles can open. (For installed packages this is just the
# bundled file; for zipped distributions it would extract to a temp.)
with as_file(files("pipecat.audio.dtmf").joinpath(dtmf_file_name)) as dtmf_file_path:
async with aiofiles.open(dtmf_file_path, "rb") as f:
data = await f.read()
with io.BytesIO(data) as buffer:
with wave.open(buffer, "rb") as wf:

View File

@@ -60,7 +60,12 @@ class RNNoiseFilter(BaseAudioFilter):
self._sample_rate = sample_rate
try:
# RNNoise always requires 48kHz
# The module-level import sets `RNNoise` to `None` if pyrnnoise
# isn't installed; raise instead of calling `None(...)` so the
# except clause handles it cleanly.
if RNNoise is None:
raise ImportError("pyrnnoise is not installed")
# RNNoise always requires 48kHz.
self._rnnoise = RNNoise(sample_rate=48000)
self._rnnoise_ready = True
except Exception as e:
@@ -107,7 +112,7 @@ class RNNoiseFilter(BaseAudioFilter):
Returns:
Noise-suppressed audio data as bytes.
"""
if not self._rnnoise_ready or not self._filtering:
if not self._rnnoise_ready or not self._filtering or self._rnnoise is None:
return audio
# Resample input if needed

View File

@@ -21,7 +21,7 @@ import io
import wave
from collections.abc import Callable
from dataclasses import dataclass
from typing import Any, TypeAlias, TypeGuard, TypeVar
from typing import Any, TypeAlias, TypeGuard, TypeVar, cast
from loguru import logger
from openai._types import NOT_GIVEN as OPEN_AI_NOT_GIVEN
@@ -129,13 +129,13 @@ class LLMContext:
url: The URL of the image.
text: Optional text to include with the image.
"""
content = []
content: list[dict[str, Any]] = []
if text:
content.append({"type": "text", "text": text})
content.append({"type": "image_url", "image_url": {"url": url}})
return {"role": role, "content": content}
return cast(LLMContextMessage, {"role": role, "content": content})
@staticmethod
async def create_image_message(
@@ -187,7 +187,7 @@ class LLMContext:
audio_frames: List of audio frame objects to include.
text: Optional text to include with the audio.
"""
content = [{"type": "text", "text": text}]
content: list[dict[str, Any]] = [{"type": "text", "text": text}]
def encode_audio():
sample_rate = audio_frames[0].sample_rate
@@ -214,7 +214,7 @@ class LLMContext:
}
)
return {"role": role, "content": content}
return cast(LLMContextMessage, {"role": role, "content": content})
@property
def messages(self) -> list[LLMContextMessage]:
@@ -295,7 +295,10 @@ class LLMContext:
result.append(msg_copy)
continue
msg = copy.deepcopy(message)
# The standard message variant is a union of TypedDicts; the
# mutations below operate on plain dicts at runtime. Treat as
# such for the duration of the redaction loop.
msg: dict[str, Any] = cast(dict[str, Any], copy.deepcopy(message))
content = msg.get("content")
if isinstance(content, list):
for item in content:

View File

@@ -67,9 +67,20 @@ class LangchainProcessor(FrameProcessor):
# The last one by the human is the one we want to send to the LLM.
logger.debug(f"Got transcription frame {frame}")
messages = frame.context.get_messages()
text: str = messages[-1]["content"]
# Historically this processor has only handled plain-text user
# messages; the guards below make that contract explicit for the
# type checker. TODO: maybe handle other message shapes (provider-
# specific messages, multi-modal content lists, etc.).
last_message = messages[-1] if messages else None
if not isinstance(last_message, dict):
await self.push_frame(frame, direction)
return
content = last_message.get("content")
if not isinstance(content, str):
await self.push_frame(frame, direction)
return
await self._ainvoke(text.strip())
await self._ainvoke(content.strip())
else:
await self.push_frame(frame, direction)
@@ -87,7 +98,10 @@ class LangchainProcessor(FrameProcessor):
case str():
return text
case AIMessageChunk():
return text.content
# `content` is `str | list[...]` (multi-modal); stringify if
# it's a list, since downstream consumers want plain text.
content = text.content
return content if isinstance(content, str) else str(content)
case _:
return ""

View File

@@ -102,7 +102,7 @@ class RTVIProcessor(FrameProcessor):
self._client_ready = True
await self._call_event_handler("on_client_ready")
async def set_bot_ready(self, about: Mapping[str, Any] = None):
async def set_bot_ready(self, about: Mapping[str, Any] | None = None):
"""Mark the bot as ready and send the bot-ready message.
Args:
@@ -404,7 +404,7 @@ class RTVIProcessor(FrameProcessor):
)
await self.push_frame(frame)
async def _send_bot_ready(self, about: Mapping[str, Any] = None):
async def _send_bot_ready(self, about: Mapping[str, Any] | None = None):
"""Send the bot-ready message to the client.
Args:

View File

@@ -71,9 +71,15 @@ class StrandsAgentsProcessor(FrameProcessor):
await super().process_frame(frame, direction)
if isinstance(frame, LLMContextFrame):
messages = frame.context.get_messages()
if messages:
last_message = messages[-1]
await self._ainvoke(str(last_message["content"]).strip())
# Historically this processor has only handled plain-text user
# messages; the guards below make that contract explicit for the
# type checker. TODO: handle other message shapes (provider-
# specific messages, multi-modal content lists, etc.).
last_message = messages[-1] if messages else None
if isinstance(last_message, dict):
content = last_message.get("content")
if isinstance(content, str):
await self._ainvoke(content.strip())
else:
await self.push_frame(frame, direction)
@@ -91,6 +97,9 @@ class StrandsAgentsProcessor(FrameProcessor):
await self.start_ttfb_metrics()
if self.graph:
# `__init__` asserts `graph_exit_node` is set whenever `graph`
# is, so this can't be None here.
assert self.graph_exit_node is not None
# Graph does not stream; await full result then emit assistant text
graph_result = await self.graph.invoke_async(text)
if ttfb_tracking:
@@ -115,6 +124,9 @@ class StrandsAgentsProcessor(FrameProcessor):
except Exception as parse_err:
logger.warning(f"Failed to extract messages from GraphResult: {parse_err}")
else:
# `__init__` asserts at least one of `agent`/`graph` is set,
# and we're in the `not self.graph` branch.
assert self.agent is not None
# Agent supports streaming events via async iterator
async for event in self.agent.stream_async(text):
# Push to TTS service

View File

@@ -105,7 +105,7 @@ class AnthropicLLMSettings(LLMSettings):
return instance
class AnthropicLLMService(LLMService):
class AnthropicLLMService(LLMService[AnthropicLLMAdapter]):
"""LLM service for Anthropic's Claude models.
Provides inference capabilities with Claude models including support for
@@ -293,7 +293,7 @@ class AnthropicLLMService(LLMService):
effective_instruction = system_instruction or assert_given(
self._settings.system_instruction
)
adapter: AnthropicLLMAdapter = self.get_llm_adapter()
adapter = self.get_llm_adapter()
invocation_params = adapter.get_llm_invocation_params(
context,
enable_prompt_caching=assert_given(self._settings.enable_prompt_caching),
@@ -328,8 +328,8 @@ class AnthropicLLMService(LLMService):
return next((block.text for block in response.content if hasattr(block, "text")), None)
def _get_llm_invocation_params(self, context: LLMContext) -> AnthropicLLMInvocationParams:
adapter: AnthropicLLMAdapter = self.get_llm_adapter()
params: AnthropicLLMInvocationParams = adapter.get_llm_invocation_params(
adapter = self.get_llm_adapter()
params = adapter.get_llm_invocation_params(
context,
enable_prompt_caching=assert_given(self._settings.enable_prompt_caching),
system_instruction=assert_given(self._settings.system_instruction),

View File

@@ -233,10 +233,11 @@ class AssemblyAISTTService(WebsocketSTTService):
sample_rate = connection_params.sample_rate
encoding = connection_params.encoding
default_settings.model = connection_params.speech_model
default_settings.formatted_finals = connection_params.formatted_finals
default_settings.word_finalization_max_wait_time = (
connection_params.word_finalization_max_wait_time
)
# Note: `formatted_finals` and `word_finalization_max_wait_time`
# were added to Settings after this deprecated input model
# was frozen and have no equivalent on
# AssemblyAIConnectionParams; they are only configurable via
# the canonical `settings=...` API.
default_settings.end_of_turn_confidence_threshold = (
connection_params.end_of_turn_confidence_threshold
)

View File

@@ -42,14 +42,17 @@ except ModuleNotFoundError as e:
raise Exception(f"Missing module: {e}")
def language_to_async_language(language: Language) -> str | None:
def language_to_async_language(language: Language) -> str:
"""Convert a Language enum to Async language code.
Args:
language: The Language enum value to convert.
Returns:
The corresponding Async language code, or None if not supported.
The corresponding service language code. If ``language`` is not in
the verified mapping, falls back to the base language code (e.g.,
``en`` from ``en-US``) and logs a warning (via
``resolve_language(..., use_base_code=True)``).
"""
LANGUAGE_MAP = {
Language.EN: "en",

View File

@@ -201,7 +201,11 @@ class AWSAgentCoreProcessor(FrameProcessor):
if not payload:
return
async with self._aws_session.client("bedrock-agentcore", **self._aws_params) as client:
# aioboto3's `client()` is an async context manager but its stubs don't
# advertise `__aenter__` / `__aexit__` in a way pyright can see.
async with self._aws_session.client( # pyright: ignore[reportGeneralTypeIssues]
"bedrock-agentcore", **self._aws_params
) as client:
# Invoke the AgentCore agent
response = await client.invoke_agent_runtime(
agentRuntimeArn=self._agentArn, payload=payload.encode()

View File

@@ -74,7 +74,7 @@ class AWSBedrockLLMSettings(LLMSettings):
)
class AWSBedrockLLMService(LLMService):
class AWSBedrockLLMService(LLMService[AWSBedrockLLMAdapter]):
"""AWS Bedrock Large Language Model service implementation.
Provides inference capabilities for AWS Bedrock models including Amazon Nova
@@ -282,8 +282,8 @@ class AWSBedrockLLMService(LLMService):
effective_instruction = system_instruction or assert_given(
self._settings.system_instruction
)
adapter: AWSBedrockLLMAdapter = self.get_llm_adapter()
params: AWSBedrockLLMInvocationParams = adapter.get_llm_invocation_params(
adapter = self.get_llm_adapter()
params = adapter.get_llm_invocation_params(
context, system_instruction=effective_instruction
)
messages = params["messages"]
@@ -371,8 +371,8 @@ class AWSBedrockLLMService(LLMService):
}
def _get_llm_invocation_params(self, context: LLMContext) -> AWSBedrockLLMInvocationParams:
adapter: AWSBedrockLLMAdapter = self.get_llm_adapter()
params: AWSBedrockLLMInvocationParams = adapter.get_llm_invocation_params(
adapter = self.get_llm_adapter()
params = adapter.get_llm_invocation_params(
context, system_instruction=assert_given(self._settings.system_instruction)
)
return params

View File

@@ -49,7 +49,7 @@ from pipecat.frames.frames import (
UserStartedSpeakingFrame,
UserStoppedSpeakingFrame,
)
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_context import LLMContext, LLMSpecificMessage
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.aws.nova_sonic.session_continuation import (
SessionContinuationHelper,
@@ -235,7 +235,7 @@ class AWSNovaSonicLLMSettings(LLMSettings):
endpointing_sensitivity: str | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN)
class AWSNovaSonicLLMService(LLMService):
class AWSNovaSonicLLMService(LLMService[AWSNovaSonicLLMAdapter]):
"""AWS Nova Sonic speech-to-speech LLM service.
Provides bidirectional audio streaming, real-time transcription, text generation,
@@ -501,12 +501,18 @@ class AWSNovaSonicLLMService(LLMService):
service, and reconnects with the preserved context.
"""
logger.debug("Resetting conversation")
if self._assistant_is_responding:
self._assistant_is_responding = False
await self._report_assistant_response_ended()
# Grab context to carry through disconnect/reconnect
context = self._context
if context is None:
logger.warning(
"reset_conversation called before an initial context was received; nothing to reset"
)
return
if self._assistant_is_responding:
self._assistant_is_responding = False
await self._report_assistant_response_ended()
await self._disconnect()
await self._start_connecting()
@@ -606,9 +612,18 @@ class AWSNovaSonicLLMService(LLMService):
await self._disconnect()
async def _process_completed_function_calls(self, send_new_results: bool):
if not self._context: # should never happen
return
# Check for set of completed function calls in the context
for message in self._context.get_messages():
if message.get("role") and message.get("content") not in ["IN_PROGRESS", "CANCELLED"]:
# LLMSpecificMessages are opaque provider-specific payloads, not
# standard tool-result messages — skip them.
if isinstance(message, LLMSpecificMessage):
continue
if message.get("role") == "tool" and message.get("content") not in [
"IN_PROGRESS",
"CANCELLED",
]:
tool_call_id = message.get("tool_call_id")
if tool_call_id and tool_call_id not in self._completed_tool_calls:
# Found a newly-completed function call - send the result to the service
@@ -629,7 +644,7 @@ class AWSNovaSonicLLMService(LLMService):
await self._process_completed_function_calls(send_new_results=False)
# Read context
adapter: AWSNovaSonicLLMAdapter = self.get_llm_adapter()
adapter = self.get_llm_adapter()
llm_connection_params = adapter.get_llm_invocation_params(
self._context, system_instruction=assert_given(self._settings.system_instruction)
)
@@ -639,7 +654,7 @@ class AWSNovaSonicLLMService(LLMService):
tools = (
llm_connection_params["tools"]
if llm_connection_params["tools"]
else adapter.from_standard_tools(self._tools)
else (adapter.from_standard_tools(self._tools) or [])
)
logger.debug(f"Using tools: {tools}")
await self._send_prompt_start_event(tools)
@@ -959,7 +974,9 @@ class AWSNovaSonicLLMService(LLMService):
async def open_stream(self, client):
"""Open a bidirectional stream on the given client."""
return await client.invoke_model_with_bidirectional_stream(
InvokeModelWithBidirectionalStreamOperationInput(model_id=self._settings.model)
InvokeModelWithBidirectionalStreamOperationInput(
model_id=assert_given(self._settings.model)
)
)
async def send_event(self, event_json: str, stream):
@@ -1106,16 +1123,18 @@ class AWSNovaSonicLLMService(LLMService):
'''
await self.send_event(event_json, stream)
def get_setup_params(self):
def get_setup_params(self) -> tuple[str | None, list]:
"""Return ``(system_instruction, tools)`` for the next session setup."""
if not self._context:
return None, []
adapter: AWSNovaSonicLLMAdapter = self.get_llm_adapter()
adapter = self.get_llm_adapter()
llm_params = adapter.get_llm_invocation_params(
self._context, system_instruction=self._settings.system_instruction
self._context, system_instruction=assert_given(self._settings.system_instruction)
)
tools = (
llm_params["tools"] if llm_params["tools"] else adapter.from_standard_tools(self._tools)
llm_params["tools"]
if llm_params["tools"]
else (adapter.from_standard_tools(self._tools) or [])
)
return llm_params["system_instruction"], tools

View File

@@ -16,7 +16,7 @@ import random
import string
from collections.abc import AsyncGenerator
from dataclasses import dataclass
from typing import Any
from typing import Any, cast
from loguru import logger
@@ -38,6 +38,7 @@ from pipecat.utils.time import time_now_iso8601
from pipecat.utils.tracing.service_decorators import traced_stt
try:
from websockets import Subprotocol
from websockets.asyncio.client import connect as websocket_connect
from websockets.protocol import State
except ModuleNotFoundError as e:
@@ -314,7 +315,7 @@ class AWSTranscribeSTTService(WebsocketSTTService):
self._websocket = await websocket_connect(
presigned_url,
additional_headers=additional_headers,
subprotocols=["mqtt"],
subprotocols=[Subprotocol("mqtt")],
ping_interval=None,
ping_timeout=None,
compression=None,
@@ -534,7 +535,11 @@ class AWSTranscribeSTTService(WebsocketSTTService):
is_final = not result.get("IsPartial", True)
if transcript:
language = assert_given(self._settings.language)
# Technically `_settings.language` could be a raw string, but
# Language is a StrEnum so downstream handles either.
language = cast(
"Language | None", assert_given(self._settings.language)
)
if is_final:
await self.push_frame(
TranscriptionFrame(

View File

@@ -37,14 +37,16 @@ except ModuleNotFoundError as e:
raise Exception(f"Missing module: {e}")
def language_to_aws_language(language: Language) -> str | None:
def language_to_aws_language(language: Language) -> str:
"""Convert a Language enum to AWS Polly language code.
Args:
language: The Language enum value to convert.
Returns:
The corresponding AWS Polly language code, or None if not supported.
The corresponding service language code. If ``language`` is not in
the verified mapping, falls back to the full language code string and
logs a warning (via ``resolve_language(..., use_base_code=False)``).
"""
LANGUAGE_MAP = {
# Arabic
@@ -343,7 +345,11 @@ class AWSPollyTTSService(TTSService):
# Filter out None values
filtered_params = {k: v for k, v in params.items() if v is not None}
async with self._aws_session.client("polly", **self._aws_params) as polly:
# aioboto3's `client()` is an async context manager but its stubs
# don't advertise `__aenter__` / `__aexit__` to pyright.
async with self._aws_session.client( # pyright: ignore[reportGeneralTypeIssues]
"polly", **self._aws_params
) as polly:
response = await polly.synthesize_speech(**filtered_params)
if "AudioStream" in response:
# Get the streaming body and read it
@@ -351,7 +357,7 @@ class AWSPollyTTSService(TTSService):
audio_data = await stream.read()
else:
logger.error(f"{self} No audio stream in response")
audio_data = None
return
audio_data = await self._resampler.resample(audio_data, 16000, self.sample_rate)

View File

@@ -92,14 +92,18 @@ class AWSTranscribePresignedURL:
"""
def __init__(
self, access_key: str, secret_key: str, session_token: str, region: str = "us-east-1"
self,
access_key: str,
secret_key: str,
session_token: str | None,
region: str = "us-east-1",
):
"""Initialize the presigned URL generator.
Args:
access_key: AWS access key ID.
secret_key: AWS secret access key.
session_token: AWS session token for temporary credentials.
session_token: AWS session token for temporary credentials (optional).
region: AWS region for the service. Defaults to "us-east-1".
"""
self.access_key = access_key
@@ -129,8 +133,8 @@ class AWSTranscribePresignedURL:
sample_rate: int,
language_code: str = "",
media_encoding: str = "pcm",
vocabulary_name: str = "",
vocabulary_filter_name: str = "",
vocabulary_name: str | None = None,
vocabulary_filter_name: str | None = None,
show_speaker_label: bool = False,
enable_channel_identification: bool = False,
number_of_channels: int = 1,

View File

@@ -9,14 +9,16 @@
from pipecat.transcriptions.language import Language, resolve_language
def language_to_azure_language(language: Language) -> str | None:
def language_to_azure_language(language: Language) -> str:
"""Convert a Language enum to Azure language code.
Args:
language: The Language enum value to convert.
Returns:
The corresponding Azure language code, or None if not supported.
The corresponding Azure language code. If ``language`` is not in the
verified mapping, falls back to the full language code string and logs
a warning (via ``resolve_language(..., use_base_code=False)``).
"""
LANGUAGE_MAP = {
# Afrikaans

View File

@@ -13,7 +13,7 @@ Speech SDK for real-time audio transcription.
import asyncio
from collections.abc import AsyncGenerator
from dataclasses import dataclass
from typing import Any
from typing import Any, cast
from loguru import logger
@@ -182,9 +182,9 @@ class AzureSTTService(STTService):
changed = await super()._update_settings(delta)
if "language" in changed:
self._speech_config.speech_recognition_language = (
self._settings.language or language_to_azure_language(Language.EN_US)
)
self._speech_config.speech_recognition_language = assert_given(
self._settings.language
) or language_to_azure_language(Language.EN_US)
if self._audio_stream:
await self._disconnect()
await self._connect()
@@ -280,8 +280,11 @@ class AzureSTTService(STTService):
def _on_handle_recognized(self, event):
if event.result.reason == ResultReason.RecognizedSpeech and len(event.result.text) > 0:
language = getattr(event.result, "language", None) or assert_given(
self._settings.language
# Technically either source could be a raw string, but Language is
# a StrEnum so downstream handles either.
language = cast(
"Language | None",
getattr(event.result, "language", None) or assert_given(self._settings.language),
)
frame = TranscriptionFrame(
event.result.text,
@@ -297,8 +300,11 @@ class AzureSTTService(STTService):
def _on_handle_recognizing(self, event):
if event.result.reason == ResultReason.RecognizingSpeech and len(event.result.text) > 0:
language = getattr(event.result, "language", None) or assert_given(
self._settings.language
# Technically either source could be a raw string, but Language is
# a StrEnum so downstream handles either.
language = cast(
"Language | None",
getattr(event.result, "language", None) or assert_given(self._settings.language),
)
frame = InterimTranscriptionFrame(
event.result.text,

View File

@@ -44,14 +44,17 @@ MODEL_SAMPLE_RATES: dict[str, int] = {
}
def language_to_camb_language(language: Language) -> str | None:
def language_to_camb_language(language: Language) -> str:
"""Convert a Pipecat Language enum to Camb.ai language code.
Args:
language: The Language enum value to convert.
Returns:
The corresponding Camb.ai language code (BCP-47 format), or None if not supported.
The corresponding Camb.ai language code (BCP-47 format). If
``language`` is not in the verified mapping, falls back to the base
language code (e.g., ``en`` from ``en-US``) and logs a warning (via
``resolve_language(..., use_base_code=True)``).
"""
LANGUAGE_MAP = {
Language.EN: "en-us",

View File

@@ -290,6 +290,11 @@ class CartesiaSTTService(WebsocketSTTService):
if not self._websocket or self._websocket.state is not State.OPEN:
await self._connect()
if self._websocket is None:
logger.warning(f"{self}: websocket unavailable after reconnect, dropping audio")
yield None
return
try:
await self._websocket.send(audio)
except Exception as e:

View File

@@ -62,14 +62,17 @@ class GenerationConfig(BaseModel):
emotion: str | None = None
def language_to_cartesia_language(language: Language) -> str | None:
def language_to_cartesia_language(language: Language) -> str:
"""Convert a Language enum to Cartesia language code.
Args:
language: The Language enum value to convert.
Returns:
The corresponding Cartesia language code, or None if not supported.
The corresponding service language code. If ``language`` is not in
the verified mapping, falls back to the base language code (e.g.,
``en`` from ``en-US``) and logs a warning (via
``resolve_language(..., use_base_code=True)``).
"""
LANGUAGE_MAP = {
Language.AR: "ar",

View File

@@ -32,7 +32,7 @@ from pipecat.utils.time import time_now_iso8601
from pipecat.utils.tracing.service_decorators import traced_stt
def language_to_deepgram_flux_language(language: Language) -> str | None:
def language_to_deepgram_flux_language(language: Language) -> str:
"""Convert a Pipecat Language to a Deepgram Flux language code.
Only honored by the ``flux-general-multi`` model. Locale variants
@@ -253,7 +253,7 @@ class DeepgramFluxSTTBase(STTService):
params.append(f"mip_opt_out={str(self._mip_opt_out).lower()}")
# Add keyterm parameters (can have multiple)
for keyterm in self._settings.keyterm:
for keyterm in assert_given(self._settings.keyterm):
params.append(urlencode({"keyterm": keyterm}))
# Add tag parameters (can have multiple)
@@ -536,6 +536,10 @@ class DeepgramFluxSTTBase(STTService):
event = data.get("event")
transcript = data.get("transcript", "")
if not isinstance(event, str):
logger.debug(f"Unhandled TurnInfo event (not a string): {event}")
return
try:
flux_event_type = FluxEventType(event)
except ValueError:
@@ -648,7 +652,11 @@ class DeepgramFluxSTTBase(STTService):
detected_language = self._primary_detected_language(data)
min_confidence = assert_given(self._settings.min_confidence)
if not min_confidence or average_confidence > min_confidence:
# No threshold (None or 0.0) → accept. Otherwise require confidence
# data and compare; drop if data is missing.
if not min_confidence or (
average_confidence is not None and average_confidence > min_confidence
):
# EndOfTurn means Flux has determined the turn is complete,
# so this TranscriptionFrame is always finalized
await self.push_frame(

View File

@@ -145,9 +145,17 @@ class DeepgramFluxSageMakerSTTService(DeepgramFluxSTTBase):
# ------------------------------------------------------------------
async def _transport_send_audio(self, audio: bytes):
if (
self._client is None
): # should never happen — caller should gate on _transport_is_active()
return
await self._client.send_audio_chunk(audio)
async def _transport_send_json(self, message: dict):
if (
self._client is None
): # should never happen — caller should gate on _transport_is_active()
return
await self._client.send_json(message)
def _transport_is_active(self) -> bool:

View File

@@ -240,9 +240,17 @@ class DeepgramFluxSTTService(DeepgramFluxSTTBase, WebsocketService):
# ------------------------------------------------------------------
async def _transport_send_audio(self, audio: bytes):
if (
self._websocket is None
): # should never happen — caller should gate on _transport_is_active()
return
await self._websocket.send(audio)
async def _transport_send_json(self, message: dict):
if (
self._websocket is None
): # should never happen — caller should gate on _transport_is_active()
return
await self._websocket.send(json.dumps(message))
def _transport_is_active(self) -> bool:
@@ -291,14 +299,19 @@ class DeepgramFluxSTTService(DeepgramFluxSTTBase, WebsocketService):
self._connection_established_event.clear()
self._user_is_speaking = False
self._websocket = await websocket_connect(
# `_connect` sets `_websocket_url` before calling us; the assert
# narrows for pyright.
assert self._websocket_url is not None
websocket = await websocket_connect(
self._websocket_url,
additional_headers={"Authorization": f"Token {self._api_key}"},
)
self._websocket = websocket
headers = {
k: v for k, v in self._websocket.response.headers.items() if k.startswith("dg-")
}
# `response` is populated after the handshake completes (which it
# has, since `websocket_connect` already returned).
response_headers = websocket.response.headers if websocket.response else {}
headers = {k: v for k, v in response_headers.items() if k.startswith("dg-")}
logger.debug(f'{self}: Websocket connection initialized: {{"headers": {headers}}}')
# Creating the receiver task

View File

@@ -337,6 +337,10 @@ class DeepgramSageMakerTTSService(TTSService):
the response processor).
"""
logger.debug(f"{self}: Generating TTS [{text}]")
if self._client is None:
logger.warning(f"{self}: client unavailable, skipping TTS")
yield ErrorFrame(error="client unavailable")
return
try:
await self._client.send_json({"type": "Speak", "text": text})
yield None

View File

@@ -232,17 +232,19 @@ class DeepgramTTSService(WebsocketTTSService):
headers = {"Authorization": f"Token {self._api_key}"}
self._websocket = await websocket_connect(url, additional_headers=headers)
websocket = await websocket_connect(url, additional_headers=headers)
self._websocket = websocket
headers = {
k: v for k, v in self._websocket.response.headers.items() if k.startswith("dg-")
}
# `response` is populated after the handshake completes (which it
# has, since `websocket_connect` already returned).
response_headers = websocket.response.headers if websocket.response else {}
headers = {k: v for k, v in response_headers.items() if k.startswith("dg-")}
logger.debug(f'{self}: Websocket connection initialized: {{"headers": {headers}}}')
await self._call_event_handler("on_connected")
except Exception as e:
logger.error(f"{self} exception: {e}")
await self.push_error(ErrorFrame(error=f"{self} error: {e}"))
await self.push_error_frame(ErrorFrame(error=f"{self} error: {e}"))
self._websocket = None
await self._call_event_handler("on_connection_error", f"{e}")
@@ -258,7 +260,7 @@ class DeepgramTTSService(WebsocketTTSService):
await self._websocket.close()
except Exception as e:
logger.error(f"{self} exception: {e}")
await self.push_error(ErrorFrame(error=f"{self} error: {e}"))
await self.push_error_frame(ErrorFrame(error=f"{self} error: {e}"))
finally:
self._websocket = None
await self._call_event_handler("on_disconnected")

View File

@@ -54,7 +54,7 @@ except ModuleNotFoundError as e:
raise Exception(f"Missing module: {e}")
def language_to_elevenlabs_language(language: Language) -> str | None:
def language_to_elevenlabs_language(language: Language) -> str:
"""Convert a Language enum to ElevenLabs language code.
Source:
@@ -64,7 +64,9 @@ def language_to_elevenlabs_language(language: Language) -> str | None:
language: The Language enum value to convert.
Returns:
The corresponding ElevenLabs language code, or None if not supported.
The corresponding service language code. If ``language`` is not in
the verified mapping, falls back to the full language code string and
logs a warning (via ``resolve_language(..., use_base_code=False)``).
"""
LANGUAGE_MAP = {
Language.AF: "afr", # Afrikaans
@@ -739,6 +741,10 @@ class ElevenLabsRealtimeSTTService(WebsocketSTTService):
Args:
silence: Silent 16-bit mono PCM audio bytes.
"""
if (
self._websocket is None
): # should never happen — caller should gate on _is_keepalive_ready()
return
audio_base64 = base64.b64encode(silence).decode("utf-8")
message = {
"message_type": "input_audio_chunk",

View File

@@ -69,14 +69,17 @@ ELEVENLABS_MULTILINGUAL_MODELS = {
}
def language_to_elevenlabs_language(language: Language) -> str | None:
def language_to_elevenlabs_language(language: Language) -> str:
"""Convert a Language enum to ElevenLabs language code.
Args:
language: The Language enum value to convert.
Returns:
The corresponding ElevenLabs language code, or None if not supported.
The corresponding service language code. If ``language`` is not in
the verified mapping, falls back to the base language code (e.g.,
``en`` from ``en-US``) and logs a warning (via
``resolve_language(..., use_base_code=True)``).
"""
LANGUAGE_MAP = {
Language.AR: "ar",
@@ -905,6 +908,11 @@ class ElevenLabsTTSService(WebsocketTTSService):
if not self._websocket or self._websocket.state is State.CLOSED:
await self._connect()
if self._websocket is None:
logger.warning(f"{self}: websocket unavailable after reconnect, skipping TTS")
yield ErrorFrame(error="websocket unavailable")
return
try:
if not self.audio_context_available(context_id):
await self.create_audio_context(context_id)
@@ -915,7 +923,7 @@ class ElevenLabsTTSService(WebsocketTTSService):
self._partial_word_start_time = 0.0
# Initialize context with voice settings and pronunciation dictionaries
msg = {"text": " ", "context_id": context_id}
msg: dict[str, Any] = {"text": " ", "context_id": context_id}
if self._voice_settings:
msg["voice_settings"] = self._voice_settings
if self._pronunciation_dictionary_locators:
@@ -1260,7 +1268,7 @@ class ElevenLabsHttpTTSService(TTSService):
url = f"{self._base_url}/v1/text-to-speech/{self._settings.voice}/stream/with-timestamps"
model_id = assert_given(self._settings.model)
payload: dict[str, str | dict[str, float | bool]] = {
payload: dict[str, Any] = {
"text": text,
"model_id": model_id,
}

View File

@@ -28,14 +28,17 @@ from pipecat.utils.time import time_now_iso8601
from pipecat.utils.tracing.service_decorators import traced_stt
def language_to_fal_language(language: Language) -> str | None:
def language_to_fal_language(language: Language) -> str:
"""Convert a Language enum to Fal's Wizper language code.
Args:
language: The Language enum value to convert.
Returns:
The corresponding Fal Wizper language code, or None if not supported.
The corresponding service language code. If ``language`` is not in
the verified mapping, falls back to the base language code (e.g.,
``en`` from ``en-US``) and logs a warning (via
``resolve_language(..., use_base_code=True)``).
"""
LANGUAGE_MAP = {
Language.AF: "af",

View File

@@ -281,7 +281,8 @@ class FishAudioTTSService(InterruptibleTTSService):
model = assert_given(self._settings.model)
if model is not None:
headers["model"] = model
self._websocket = await websocket_connect(self._base_url, additional_headers=headers)
websocket = await websocket_connect(self._base_url, additional_headers=headers)
self._websocket = websocket
# Send initial start message with ormsgpack
request_settings = {
@@ -300,7 +301,7 @@ class FishAudioTTSService(InterruptibleTTSService):
if self._settings.top_p is not None:
request_settings["top_p"] = self._settings.top_p
start_message = {"event": "start", "request": {"text": "", **request_settings}}
await self._websocket.send(ormsgpack.packb(start_message))
await websocket.send(ormsgpack.packb(start_message))
logger.debug("Sent start message to Fish Audio")
await self._call_event_handler("on_connected")

View File

@@ -56,14 +56,17 @@ except ModuleNotFoundError as e:
raise Exception(f"Missing module: {e}")
def language_to_gladia_language(language: Language) -> str | None:
def language_to_gladia_language(language: Language) -> str:
"""Convert a Language enum to Gladia's language code format.
Args:
language: The Language enum value to convert.
Returns:
The Gladia language code string or None if not supported.
The corresponding Gladia language code. If ``language`` is not in
the verified mapping, falls back to the base language code (e.g.,
``en`` from ``en-US``) and logs a warning (via
``resolve_language(..., use_base_code=True)``).
"""
LANGUAGE_MAP = {
Language.AF: "af",
@@ -361,7 +364,7 @@ class GladiaSTTService(WebsocketSTTService):
language: The Language enum value to convert.
Returns:
The Gladia language code string or None if not supported.
The Gladia language code string, or None if not supported.
"""
return language_to_gladia_language(language)
@@ -539,6 +542,8 @@ class GladiaSTTService(WebsocketSTTService):
logger.debug(f"{self}Connecting to Gladia WebSocket")
if self._session_url is None:
raise RuntimeError(f"{self} session URL is not initialized")
self._websocket = await websocket_connect(self._session_url)
self._connection_active = True

View File

@@ -111,7 +111,7 @@ MAX_CONSECUTIVE_FAILURES = 3
CONNECTION_ESTABLISHED_THRESHOLD = 10.0 # seconds
def language_to_gemini_language(language: Language) -> str | None:
def language_to_gemini_language(language: Language) -> str:
"""Maps a Language enum value to a Gemini Live supported language code.
Source:
@@ -121,7 +121,9 @@ def language_to_gemini_language(language: Language) -> str | None:
language: The language enum value to convert.
Returns:
The Gemini language code string, or None if the language is not supported.
The Gemini language code string. If ``language`` is not in the
verified mapping, falls back to the full language code string and logs
a warning (via ``resolve_language(..., use_base_code=False)``).
"""
LANGUAGE_MAP = {
# Arabic
@@ -351,7 +353,7 @@ class GeminiLiveLLMSettings(LLMSettings):
proactivity: ProactivityConfig | dict | _NotGiven = field(default_factory=lambda: NOT_GIVEN)
class GeminiLiveLLMService(LLMService):
class GeminiLiveLLMService(LLMService[GeminiLLMAdapter]):
"""Provides access to Google's Gemini Live API.
This service enables real-time conversations with Gemini, supporting both
@@ -778,7 +780,7 @@ class GeminiLiveLLMService(LLMService):
# init-provided values). Note that the determination of "effective"
# system instruction is delegated to the adapter, which still
# chooses the init-provided value if there is one.
adapter: GeminiLLMAdapter = self.get_llm_adapter()
adapter = self.get_llm_adapter()
params = adapter.get_llm_invocation_params(
self._context, system_instruction=assert_given(self._system_instruction_from_init)
)
@@ -840,7 +842,7 @@ class GeminiLiveLLMService(LLMService):
async def _process_completed_function_calls(self, send_new_results: bool):
# Check for set of completed function calls in the context
adapter: GeminiLLMAdapter = self.get_llm_adapter()
adapter = self.get_llm_adapter()
messages = adapter.get_llm_invocation_params(self._context).get("messages", [])
for message in messages:
if message.parts:
@@ -1027,7 +1029,7 @@ class GeminiLiveLLMService(LLMService):
# Add system instruction and tools to configuration, if provided.
# These settings from the context take precedence over the ones
# provided at initialization time.
adapter: GeminiLLMAdapter = self.get_llm_adapter()
adapter = self.get_llm_adapter()
system_instruction = None
tools = None
if self._context:
@@ -1333,7 +1335,7 @@ class GeminiLiveLLMService(LLMService):
self._run_llm_when_session_ready = True
return
adapter: GeminiLLMAdapter = self.get_llm_adapter()
adapter = self.get_llm_adapter()
messages = adapter.get_llm_invocation_params(self._context).get("messages", [])
if not messages:
# No messages to seed convo with, so we're ready for realtime input right away
@@ -1392,7 +1394,7 @@ class GeminiLiveLLMService(LLMService):
# Create a throwaway context just for the purpose of getting messages
# in the right format
context = LLMContext(messages=messages_list)
adapter: GeminiLLMAdapter = self.get_llm_adapter()
adapter = self.get_llm_adapter()
messages = adapter.get_llm_invocation_params(context).get("messages", [])
if not messages:

View File

@@ -174,11 +174,17 @@ class GeminiLiveVertexLLMService(GeminiLiveLLMService):
default_settings.temperature = params.temperature
default_settings.top_k = params.top_k
default_settings.top_p = params.top_p
default_settings.modalities = params.modalities
# `params.modalities` and `params.media_resolution` are typed
# `<Enum> | None` on the deprecated InputParams, but None isn't
# a valid setting value (downstream uses call `.value` on
# them). Fall back to the canonical defaults.
default_settings.modalities = params.modalities or GeminiModalities.AUDIO
default_settings.language = (
language_to_gemini_language(params.language) if params.language else "en-US"
)
default_settings.media_resolution = params.media_resolution
default_settings.media_resolution = (
params.media_resolution or GeminiMediaResolution.UNSPECIFIED
)
default_settings.vad = params.vad
default_settings.context_window_compression = (
params.context_window_compression.model_dump()
@@ -233,7 +239,9 @@ class GeminiLiveVertexLLMService(GeminiLiveLLMService):
)
@staticmethod
def _get_credentials(credentials: str | None, credentials_path: str | None) -> str:
def _get_credentials(
credentials: str | None, credentials_path: str | None
) -> service_account.Credentials:
"""Retrieve Credentials using Google service account credentials JSON.
Supports multiple authentication methods:
@@ -246,7 +254,8 @@ class GeminiLiveVertexLLMService(GeminiLiveLLMService):
credentials_path: Path to the service account JSON file.
Returns:
OAuth token for API authentication.
A service-account ``Credentials`` object suitable for the Vertex
AI client (with its access token refreshed).
Raises:
ValueError: If no valid credentials are provided or found.

View File

@@ -30,7 +30,7 @@ from pipecat.services.image_service import ImageGenService
from pipecat.services.settings import NOT_GIVEN, ImageGenSettings, _NotGiven, assert_given
try:
from google import genai
import google.genai as genai
from google.genai import types
except ModuleNotFoundError as e:
logger.error(f"Exception: {e}")
@@ -153,8 +153,12 @@ class GoogleImageGenService(ImageGenService):
await self.start_ttfb_metrics()
try:
model = assert_given(self._settings.model)
if model is None:
yield ErrorFrame("Google image generation model must be specified")
return
response = await self._client.aio.models.generate_images(
model=self._settings.model,
model=model,
prompt=prompt,
config=types.GenerateImagesConfig(
number_of_images=assert_given(self._settings.number_of_images),
@@ -169,8 +173,9 @@ class GoogleImageGenService(ImageGenService):
for img_response in response.generated_images:
# Google returns the image data directly
image_bytes = img_response.image.image_bytes
image = Image.open(io.BytesIO(image_bytes))
if img_response.image is None or img_response.image.image_bytes is None:
continue
image = Image.open(io.BytesIO(img_response.image.image_bytes))
frame = URLImageRawFrame(
url=None, # Google doesn't provide URLs, only image data

View File

@@ -21,7 +21,7 @@ from loguru import logger
from PIL import Image
from pydantic import BaseModel, Field
from pipecat.adapters.services.gemini_adapter import GeminiLLMAdapter, GeminiLLMInvocationParams
from pipecat.adapters.services.gemini_adapter import GeminiLLMAdapter
from pipecat.frames.frames import (
AssistantImageRawFrame,
Frame,
@@ -52,7 +52,7 @@ from pipecat.utils.tracing.service_decorators import traced_llm
os.environ["GRPC_ENABLE_FORK_SUPPORT"] = "false"
try:
from google import genai
import google.genai as genai
from google.api_core.exceptions import DeadlineExceeded
from google.genai.types import (
GenerateContentConfig,
@@ -124,7 +124,7 @@ class GoogleLLMSettings(LLMSettings):
return instance
class GoogleLLMService(LLMService):
class GoogleLLMService(LLMService[GeminiLLMAdapter]):
"""Google AI (Gemini) LLM service implementation.
This class implements inference with Google's AI models, translating internally
@@ -292,7 +292,7 @@ class GoogleLLMService(LLMService):
tools = []
effective_instruction = system_instruction or self._settings.system_instruction
adapter = self.get_llm_adapter()
params: GeminiLLMInvocationParams = adapter.get_llm_invocation_params(
params = adapter.get_llm_invocation_params(
context, system_instruction=effective_instruction
)
messages = params["messages"]
@@ -387,7 +387,7 @@ class GoogleLLMService(LLMService):
async def _stream_content(self, context: LLMContext) -> AsyncIterator[GenerateContentResponse]:
adapter = self.get_llm_adapter()
params: GeminiLLMInvocationParams = adapter.get_llm_invocation_params(
params = adapter.get_llm_invocation_params(
context, system_instruction=assert_given(self._settings.system_instruction)
)

View File

@@ -60,14 +60,17 @@ except ModuleNotFoundError as e:
raise Exception(f"Missing module: {e}")
def language_to_google_stt_language(language: Language) -> str | None:
def language_to_google_stt_language(language: Language) -> str:
"""Maps Language enum to Google Speech-to-Text V2 language codes.
Args:
language: Language enum value.
Returns:
Optional[str]: Google STT language code or None if not supported.
The corresponding Google STT language code. If ``language`` is not
in the verified mapping, falls back to the full language code string
and logs a warning (via
``resolve_language(..., use_base_code=False)``).
"""
LANGUAGE_MAP = {
# Afrikaans
@@ -617,17 +620,20 @@ class GoogleSTTService(STTService):
"""
return True
def language_to_service_language(self, language: Language | list[Language]) -> str | list[str]:
"""Convert Language enum(s) to Google STT language code(s).
def language_to_service_language(self, language: Language) -> str:
"""Convert a Language enum to a Google STT language code.
Narrower return type than the base class's ``str | None``: this
override always returns a string, falling back to ``"en-US"`` for
languages not in the verified mapping (see
:func:`language_to_google_stt_language`).
Args:
language: Single Language enum or list of Language enums.
language: The Language enum value to convert.
Returns:
str | List[str]: Google STT language code(s).
The Google STT language code.
"""
if isinstance(language, list):
return [language_to_google_stt_language(lang) or "en-US" for lang in language]
return language_to_google_stt_language(language) or "en-US"
def _get_language_codes(self) -> list[str]:
@@ -639,8 +645,9 @@ class GoogleSTTService(STTService):
Returns:
List[str]: Google STT language code strings.
"""
if self._settings.languages:
return [self.language_to_service_language(lang) for lang in self._settings.languages]
languages = assert_given(self._settings.languages)
if languages:
return [self.language_to_service_language(lang) for lang in languages]
language_codes = assert_given(self._settings.language_codes)
if language_codes:
return list(language_codes)

View File

@@ -60,7 +60,7 @@ except ModuleNotFoundError as e:
raise Exception(f"Missing module: {e}")
def language_to_google_tts_language(language: Language) -> str | None:
def language_to_google_tts_language(language: Language) -> str:
"""Convert a Language enum to Google TTS language code.
Source:
@@ -70,7 +70,9 @@ def language_to_google_tts_language(language: Language) -> str | None:
language: The Language enum value to convert.
Returns:
The corresponding Google TTS language code, or None if not supported.
The corresponding service language code. If ``language`` is not in
the verified mapping, falls back to the full language code string and
logs a warning (via ``resolve_language(..., use_base_code=False)``).
"""
LANGUAGE_MAP = {
# Arabic
@@ -219,7 +221,7 @@ def language_to_google_tts_language(language: Language) -> str | None:
return resolve_language(language, LANGUAGE_MAP, use_base_code=False)
def language_to_gemini_tts_language(language: Language) -> str | None:
def language_to_gemini_tts_language(language: Language) -> str:
"""Convert a Language enum to Gemini TTS language code.
Source:
@@ -229,7 +231,9 @@ def language_to_gemini_tts_language(language: Language) -> str | None:
language: The Language enum value to convert.
Returns:
The corresponding Gemini TTS language code, or None if not supported.
The corresponding service language code. If ``language`` is not in
the verified mapping, falls back to the full language code string and
logs a warning (via ``resolve_language(..., use_base_code=False)``).
"""
LANGUAGE_MAP = {
# Afrikaans (Preview)
@@ -1413,7 +1417,7 @@ class GeminiTTSService(GoogleBaseTTSService):
if self._settings.multi_speaker and self._settings.speaker_configs:
# Multi-speaker mode
speaker_voice_configs = []
for speaker_config in self._settings.speaker_configs:
for speaker_config in assert_given(self._settings.speaker_configs):
speaker_voice_configs.append(
texttospeech_v1.MultispeakerPrebuiltVoice(
speaker_alias=speaker_config["speaker_alias"],

View File

@@ -15,7 +15,7 @@ import base64
import json
from collections.abc import AsyncGenerator
from dataclasses import dataclass, field
from typing import Any
from typing import Any, cast
from loguru import logger
from pydantic import BaseModel
@@ -79,14 +79,17 @@ def _input_format_from_encoding(encoding: str, sample_rate: int) -> str:
return encoding
def language_to_gradium_language(language: Language) -> str | None:
def language_to_gradium_language(language: Language) -> str:
"""Convert a Language enum to Gradium's language code format.
Args:
language: The Language enum value to convert.
Returns:
The Gradium language code string or None if not supported.
The corresponding Gradium language code. If ``language`` is not in
the verified mapping, falls back to the base language code (e.g.,
``en`` from ``en-US``) and logs a warning (via
``resolve_language(..., use_base_code=True)``).
"""
LANGUAGE_MAP = {
Language.DE: "de",
@@ -383,10 +386,11 @@ class GradiumSTTService(WebsocketSTTService):
"x-api-key": self._api_key,
"x-api-source": "pipecat",
}
self._websocket = await websocket_connect(
websocket = await websocket_connect(
ws_url,
additional_headers=headers,
)
self._websocket = websocket
await self._call_event_handler("on_connected")
setup_msg = {
"type": "setup",
@@ -397,8 +401,10 @@ class GradiumSTTService(WebsocketSTTService):
json_config = {}
if self._json_config:
json_config = json.loads(self._json_config)
language = assert_given(self._settings.language)
if language:
# Technically `_settings.language` could be a raw string, but
# Language is a StrEnum so downstream handles either.
language = cast("Language | None", assert_given(self._settings.language))
if language is not None:
gradium_language = language_to_gradium_language(language)
if gradium_language:
json_config["language"] = gradium_language
@@ -406,8 +412,8 @@ class GradiumSTTService(WebsocketSTTService):
json_config["delay_in_frames"] = self._settings.delay_in_frames
if json_config:
setup_msg["json_config"] = json_config
await self._websocket.send(json.dumps(setup_msg))
ready_msg = await self._websocket.recv()
await websocket.send(json.dumps(setup_msg))
ready_msg = await websocket.recv()
ready_msg = json.loads(ready_msg)
if ready_msg["type"] == "error":
raise Exception(f"received error {ready_msg['message']}")
@@ -478,12 +484,14 @@ class GradiumSTTService(WebsocketSTTService):
"""
self._accumulated_text.append(text)
accumulated = " ".join(self._accumulated_text)
# Technically `_settings.language` could be a raw string, but Language
# is a StrEnum so downstream handles either.
await self.push_frame(
InterimTranscriptionFrame(
text=accumulated,
user_id=self._user_id,
timestamp=time_now_iso8601(),
language=assert_given(self._settings.language),
language=cast("Language | None", assert_given(self._settings.language)),
)
)
await self.stop_processing_metrics()
@@ -515,7 +523,9 @@ class GradiumSTTService(WebsocketSTTService):
text = " ".join(self._accumulated_text)
self._accumulated_text.clear()
logger.debug(f"Final transcription: [{text}]")
language = assert_given(self._settings.language)
# Technically `_settings.language` could be a raw string, but Language
# is a StrEnum so downstream handles either.
language = cast("Language | None", assert_given(self._settings.language))
await self.push_frame(
TranscriptionFrame(
text,

View File

@@ -10,6 +10,7 @@ import io
import wave
from collections.abc import AsyncGenerator
from dataclasses import dataclass, field
from typing import Literal, cast
from loguru import logger
from pydantic import BaseModel
@@ -31,6 +32,19 @@ except ModuleNotFoundError as e:
logger.error("In order to use Groq, you need to `pip install pipecat-ai[groq]`.")
raise Exception(f"Missing module: {e}")
# Hint set for `output_format`. The values mirror the Literal that
# `groq.resources.audio.speech.AsyncSpeech.create` accepts on its
# `response_format` parameter (also visible as the `response_format` field of
# `groq.types.audio.SpeechCreateParams`). The groq SDK does not export this as
# a named alias, so we redeclare it here.
#
# This alias is used in unions like `GroqAudioFormat | str`, so pyright shows
# these values as completion hints without rejecting other strings. If groq
# adds a new format before this list is updated, callers can still pass it and
# we forward it through (with a cast at the API boundary). Keep in sync on a
# best-effort basis when bumping the groq dep.
GroqAudioFormat = Literal["flac", "mp3", "mulaw", "ogg", "wav"]
@dataclass
class GroqTTSSettings(TTSSettings):
@@ -74,7 +88,7 @@ class GroqTTSService(TTSService):
self,
*,
api_key: str,
output_format: str = "wav",
output_format: GroqAudioFormat | str = "wav",
params: InputParams | None = None,
model_name: str | None = None,
voice_id: str | None = None,
@@ -147,7 +161,7 @@ class GroqTTSService(TTSService):
)
self._api_key = api_key
self._output_format = output_format
self._output_format: str = output_format
self._client = AsyncGroq(api_key=self._api_key)
@@ -178,12 +192,18 @@ class GroqTTSService(TTSService):
speed = assert_given(self._settings.speed)
if model is None:
raise ValueError("Groq TTS model must be specified")
if voice is None:
raise ValueError("Groq TTS voice must be specified")
if speed is None:
raise ValueError("Groq TTS speed must be specified")
response = await self._client.audio.speech.create(
model=model,
voice=voice,
response_format=self._output_format,
# Cast satisfies groq's stricter Literal typing while letting
# callers pass any string (e.g. a newer groq format we haven't
# yet added to GroqAudioFormat). If the value is unsupported,
# groq's API will surface a runtime error with a clear message.
response_format=cast(GroqAudioFormat, self._output_format),
# Note: as of 2026-02-25, only a speed of 1.0 is supported, but
# here we pass it for completeness and future-proofing
speed=speed,

View File

@@ -210,11 +210,11 @@ class HeyGenApi(BaseAvatarApi):
"quality": request_data.quality,
"avatar_id": request_data.avatar_id,
"voice": {
"voice_id": request_data.voice.voiceId if request_data.voice else None,
"voice_id": request_data.voice.voice_id if request_data.voice else None,
"rate": request_data.voice.rate if request_data.voice else None,
"emotion": request_data.voice.emotion if request_data.voice else None,
"elevenlabs_settings": (
request_data.voice.elevenlabsSettings if request_data.voice else None
request_data.voice.elevenlabs_settings if request_data.voice else None
),
},
"knowledge_id": request_data.knowledge_id,

View File

@@ -33,8 +33,8 @@ class StandardSessionResponse(BaseModel):
access_token: str
livekit_agent_token: str
livekit_url: str = None
ws_url: str = None
livekit_url: str
ws_url: str
raw_response: Any

View File

@@ -291,6 +291,8 @@ class HeyGenClient:
"""Handle incoming WebSocket messages."""
try:
while self._connected:
if self._websocket is None: # should never happen while _connected is True
break
try:
message = await self._websocket.recv()
parsed_message = json.loads(message)

View File

@@ -19,6 +19,7 @@ from pipecat import version as pipecat_version
from pipecat.frames.frames import (
CancelFrame,
EndFrame,
ErrorFrame,
Frame,
InterruptionFrame,
StartFrame,
@@ -289,10 +290,14 @@ class HumeTTSService(TTSService):
"""
logger.debug(f"{self}: Generating Hume TTS: [{text}]")
voice_id = assert_given(self._settings.voice)
if voice_id is None:
yield ErrorFrame(error="Hume TTS voice must be specified")
return
# Build the request payload
utterance_kwargs: dict[str, Any] = {
"text": text,
"voice": PostedUtteranceVoiceWithId(id=assert_given(self._settings.voice)),
"voice": PostedUtteranceVoiceWithId(id=voice_id),
}
if self._settings.description is not None:
utterance_kwargs["description"] = self._settings.description

View File

@@ -189,7 +189,7 @@ _NON_FATAL_ERROR_CODES = {
}
class InworldRealtimeLLMService(LLMService):
class InworldRealtimeLLMService(LLMService[InworldRealtimeLLMAdapter]):
"""Inworld Realtime LLM service for real-time audio and text communication.
Implements the Inworld Realtime API with WebSocket communication for
@@ -664,7 +664,7 @@ class InworldRealtimeLLMService(LLMService):
async def _send_session_update(self):
"""Update session settings on the server."""
settings = assert_given(self._settings.session_properties)
adapter: InworldRealtimeLLMAdapter = self.get_llm_adapter()
adapter = self.get_llm_adapter()
if self._context:
llm_invocation_params = adapter.get_llm_invocation_params(
@@ -963,7 +963,7 @@ class InworldRealtimeLLMService(LLMService):
self._run_llm_when_api_session_ready = True
return
adapter: InworldRealtimeLLMAdapter = self.get_llm_adapter()
adapter = self.get_llm_adapter()
if self._llm_needs_conversation_setup:
logger.debug(

View File

@@ -794,8 +794,10 @@ class InworldTTSService(WebsocketTTSService):
return word_times
async def _close_context(self, context_id: str):
if context_id and self._websocket:
async def _close_context(self, context_id: str | None):
if not context_id:
return
if self._websocket:
logger.info(f"{self}: Closing context {context_id} due to interruption or completion")
try:
await self._send_close_context(context_id)

View File

@@ -216,6 +216,8 @@ class KokoroTTSService(TTSService):
await self.start_tts_usage_metrics(text)
voice = assert_given(self._settings.voice)
if voice is None:
raise ValueError("Kokoro TTS voice must be specified")
lang = assert_given(self._settings.language)
if lang is None:
raise ValueError("Kokoro TTS language must be specified")

View File

@@ -16,10 +16,13 @@ from collections.abc import Awaitable, Callable, Mapping, Sequence
from dataclasses import dataclass
from typing import (
Any,
Generic,
Protocol,
cast,
)
from loguru import logger
from typing_extensions import TypeVar
from websockets.exceptions import ConnectionClosed
from websockets.protocol import State
@@ -116,7 +119,12 @@ class FunctionCallParams:
function_name: str
tool_call_id: str
arguments: Mapping[str, Any]
llm: LLMService
# `LLMService[Any]` so any concrete subclass (regardless of how — or
# whether — it parameterizes the adapter type) can be assigned here.
# Plain `LLMService` would invoke the TypeVar default and pyright would
# treat it invariantly, rejecting `LLMService[XAdapter]` at the call
# sites that build FunctionCallParams.
llm: LLMService[Any]
context: LLMContext
result_callback: FunctionCallResultCallback
app_resources: Any = None
@@ -190,7 +198,14 @@ class FunctionCallRunnerItem:
group_id: str | None = None
class LLMService(UserTurnCompletionLLMServiceMixin, AIService):
# `default=BaseLLMAdapter` (PEP 696) so that unparameterized subclasses
# (e.g. third-party `class MyService(LLMService):` with no bracket) get
# `TAdapter = BaseLLMAdapter` instead of `Unknown` at type-check time —
# matching the pre-generic behavior of `get_llm_adapter()`.
TAdapter = TypeVar("TAdapter", bound=BaseLLMAdapter, default=BaseLLMAdapter)
class LLMService(UserTurnCompletionLLMServiceMixin, AIService, Generic[TAdapter]):
"""Base class for all LLM services.
Handles function calling registration and execution with support for both
@@ -222,6 +237,7 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService):
"""
_settings: LLMSettings
_adapter: TAdapter
# OpenAILLMAdapter is used as the default adapter since it aligns with most LLM implementations.
# However, subclasses should override this with a more specific adapter when necessary.
@@ -269,7 +285,12 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService):
self._filter_incomplete_user_turns: bool = False
self._async_tool_cancellation_enabled: bool = False
self._base_system_instruction: str | None = None
self._adapter = self.adapter_class()
# `adapter_class` is typed as `type[BaseLLMAdapter]` so subclasses
# don't need to spell out the generic parameter just to subclass
# (backward compatibility for 3rd-party providers outside this repo).
# Cast to TAdapter to keep `_adapter` and `get_llm_adapter()` precisely
# typed for callers that opt into `LLMService[XAdapter]`.
self._adapter = cast(TAdapter, self.adapter_class())
self._functions: dict[str | None, FunctionCallRegistryItem] = {}
self._function_call_tasks: dict[asyncio.Task | None, FunctionCallRunnerItem] = {}
self._sequential_runner_task: asyncio.Task | None = None
@@ -280,7 +301,7 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService):
self._register_event_handler("on_function_calls_cancelled")
self._register_event_handler("on_completion_timeout")
def get_llm_adapter(self) -> BaseLLMAdapter:
def get_llm_adapter(self) -> TAdapter:
"""Get the LLM adapter instance.
Returns:
@@ -1112,7 +1133,7 @@ class WebsocketReconnectedError(Exception):
pass
class WebsocketLLMService(LLMService, WebsocketService):
class WebsocketLLMService(LLMService[TAdapter], WebsocketService, Generic[TAdapter]):
"""Base class for websocket-based LLM services.
Each LLM inference is a discrete request/response exchange: send one
@@ -1160,7 +1181,11 @@ class WebsocketLLMService(LLMService, WebsocketService):
reconnect_on_error: Whether to automatically reconnect on websocket errors.
**kwargs: Additional arguments passed to parent classes.
"""
LLMService.__init__(self, **kwargs)
# pyright stumbles here because the TypeVar default makes
# `LLMService` resolve to `LLMService[BaseLLMAdapter]` invariantly,
# while `self` is `WebsocketLLMService[TAdapter]` for an arbitrary
# TAdapter. The runtime call is fine — generics are erased.
LLMService.__init__(self, **kwargs) # pyright: ignore[reportArgumentType]
WebsocketService.__init__(self, reconnect_on_error=reconnect_on_error, **kwargs)
self._register_event_handler("on_connection_error")
@@ -1240,6 +1265,11 @@ class WebsocketLLMService(LLMService, WebsocketService):
Returns:
The parsed JSON message as a dict.
"""
# Should never happen — `_ensure_connected` (which callers must invoke
# first) raises ConnectionError if it can't establish a websocket.
# Match that contract here.
if self._websocket is None:
raise ConnectionError(f"{self} _ws_recv called without a websocket")
try:
raw = await self._websocket.recv()
return json.loads(raw)

View File

@@ -37,14 +37,17 @@ except ModuleNotFoundError as e:
raise Exception(f"Missing module: {e}")
def language_to_lmnt_language(language: Language) -> str | None:
def language_to_lmnt_language(language: Language) -> str:
"""Convert a Language enum to LMNT language code.
Args:
language: The Language enum value to convert.
Returns:
The corresponding LMNT language code, or None if not supported.
The corresponding service language code. If ``language`` is not in
the verified mapping, falls back to the base language code (e.g.,
``en`` from ``en-US``) and logs a warning (via
``resolve_language(..., use_base_code=True)``).
"""
LANGUAGE_MAP = {
Language.AR: "ar",
@@ -267,10 +270,11 @@ class LmntTTSService(InterruptibleTTSService):
}
# Connect to LMNT's websocket directly
self._websocket = await websocket_connect("wss://api.lmnt.com/v1/ai/speech/stream")
websocket = await websocket_connect("wss://api.lmnt.com/v1/ai/speech/stream")
self._websocket = websocket
# Send initialization message
await self._websocket.send(json.dumps(init_msg))
await websocket.send(json.dumps(init_msg))
await self._call_event_handler("on_connected")
except Exception as e:

View File

@@ -31,14 +31,16 @@ from pipecat.transcriptions.language import Language, resolve_language
from pipecat.utils.tracing.service_decorators import traced_tts
def language_to_minimax_language(language: Language) -> str | None:
def language_to_minimax_language(language: Language) -> str:
"""Convert a Language enum to MiniMax language format.
Args:
language: The Language enum value to convert.
Returns:
The corresponding MiniMax language name, or None if not supported.
The corresponding MiniMax language name. If ``language`` is not in
the verified mapping, falls back to the full language code string and
logs a warning (via ``resolve_language(..., use_base_code=False)``).
"""
LANGUAGE_MAP = {
Language.AF: "Afrikaans",

View File

@@ -12,7 +12,7 @@ Voxtral Realtime transcription API using the Mistral SDK's RealtimeConnection.
from collections.abc import AsyncGenerator
from dataclasses import dataclass
from typing import Any
from typing import Any, cast
from loguru import logger
@@ -30,6 +30,7 @@ from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.settings import STTSettings, assert_given
from pipecat.services.stt_latency import MISTRAL_TTFS_P99
from pipecat.services.stt_service import STTService
from pipecat.transcriptions.language import Language
from pipecat.utils.time import time_now_iso8601
from pipecat.utils.tracing.service_decorators import traced_stt
@@ -132,7 +133,7 @@ class MistralSTTService(STTService):
self._connection: RealtimeConnection | None = None
self._receive_task = None
self._accumulated_text = ""
self._detected_language: str | None = None
self._detected_language: Language | None = None
def can_generate_metrics(self) -> bool:
"""Check if the service can generate processing metrics.
@@ -197,6 +198,13 @@ class MistralSTTService(STTService):
if not self._connection or self._connection.is_closed:
await self._connect()
# `_connect` swallows exceptions and may leave `_connection` unset;
# drop the audio chunk rather than crashing if reconnect failed.
if self._connection is None:
logger.warning(f"{self}: dropping audio chunk — Mistral STT not connected")
yield None
return
await self._connection.send_audio(audio)
yield None
@@ -247,8 +255,13 @@ class MistralSTTService(STTService):
async def _receive_events(self):
"""Background task: iterate connection events and handle them."""
# `_connect` started this task only after assigning `_connection`,
# so it should not be None here; bail out defensively just in case.
connection = self._connection
if connection is None:
return
try:
async for event in self._connection.events():
async for event in connection.events():
if isinstance(event, RealtimeTranscriptionSessionCreated):
logger.debug(f"{self}: Session created: {event.session}")
await self._call_event_handler("on_connected")
@@ -278,7 +291,9 @@ class MistralSTTService(STTService):
self._accumulated_text = ""
elif isinstance(event, TranscriptionStreamLanguage):
self._detected_language = event.audio_language
# Technically the SDK could emit a code we haven't added yet,
# but Language is a StrEnum so downstream handles either.
self._detected_language = cast("Language | None", event.audio_language)
elif isinstance(event, RealtimeTranscriptionError):
error_msg = event.error.message if event.error else "Unknown error"

View File

@@ -153,9 +153,14 @@ class MoondreamService(VisionService):
logger.debug(f"Analyzing image (bytes length: {len(frame.image)})")
def get_image_description(image_bytes: bytes, text: str | None) -> str:
if frame.format is None:
raise ValueError("Cannot decode image bytes without a format")
image = Image.frombytes(frame.format, frame.size, image_bytes)
image_embeds = self._model.encode_image(image)
description = self._model.query(image_embeds, text)["answer"]
# `encode_image` and `query` are custom methods provided by the
# moondream2 model code (via `trust_remote_code=True`) that pyright
# can't see on `AutoModelForCausalLM`'s base type.
image_embeds = self._model.encode_image(image) # pyright: ignore[reportCallIssue]
description = self._model.query(image_embeds, text)["answer"] # pyright: ignore[reportCallIssue]
return description
description = await asyncio.to_thread(get_image_description, frame.image, frame.text)

View File

@@ -44,14 +44,17 @@ except ModuleNotFoundError as e:
raise Exception(f"Missing module: {e}")
def language_to_neuphonic_lang_code(language: Language) -> str | None:
def language_to_neuphonic_lang_code(language: Language) -> str:
"""Convert a Language enum to Neuphonic language code.
Args:
language: The Language enum value to convert.
Returns:
The corresponding Neuphonic language code, or None if not supported.
The corresponding service language code. If ``language`` is not in
the verified mapping, falls back to the base language code (e.g.,
``en`` from ``en-US``) and logs a warning (via
``resolve_language(..., use_base_code=True)``).
"""
LANGUAGE_MAP = {
Language.DE: "de",
@@ -328,7 +331,10 @@ class NeuphonicTTSService(InterruptibleTTSService):
async def _receive_messages(self):
"""Receive and process messages from Neuphonic WebSocket."""
async for message in self._websocket:
websocket = self._websocket
if websocket is None:
return
async for message in websocket:
if isinstance(message, str):
msg = json.loads(message)
if msg.get("data") and msg["data"].get("audio"):

View File

@@ -49,7 +49,7 @@ except ModuleNotFoundError as e:
raise Exception(f"Missing module: {e}")
def language_to_nvidia_nemotron_speech_language(language: Language) -> str | None:
def language_to_nvidia_nemotron_speech_language(language: Language) -> str:
"""Maps Language enum to NVIDIA Nemotron Speech ASR language codes.
Source:
@@ -59,7 +59,9 @@ def language_to_nvidia_nemotron_speech_language(language: Language) -> str | Non
language: Language enum value.
Returns:
str | None: NVIDIA Nemotron Speech language code or None if not supported.
The NVIDIA Nemotron Speech language code. If ``language`` is not in
the verified mapping, falls back to the full language code string and
logs a warning (via ``resolve_language(..., use_base_code=False)``).
"""
LANGUAGE_MAP = {
# Arabic

View File

@@ -26,7 +26,7 @@ from openai._types import NotGiven as OpenAINotGiven
from openai.types.chat import ChatCompletionChunk
from pydantic import BaseModel, Field
from pipecat.adapters.services.open_ai_adapter import OpenAILLMInvocationParams
from pipecat.adapters.services.open_ai_adapter import OpenAILLMAdapter, OpenAILLMInvocationParams
from pipecat.frames.frames import (
Frame,
LLMContextFrame,
@@ -39,7 +39,7 @@ from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.llm_service import FunctionCallFromLLM, LLMService
from pipecat.services.settings import NOT_GIVEN as _NOT_GIVEN
from pipecat.services.settings import LLMSettings, _NotGiven
from pipecat.services.settings import LLMSettings, _NotGiven, assert_given
from pipecat.utils.tracing.service_decorators import traced_llm
@@ -66,12 +66,12 @@ class OpenAILLMSettings(LLMSettings):
)
top_p: float | None | _NotGiven | OpenAINotGiven = field(default_factory=lambda: _NOT_GIVEN)
max_tokens: int | None | _NotGiven | OpenAINotGiven = field(default_factory=lambda: _NOT_GIVEN)
max_completion_tokens: int | _NotGiven | OpenAINotGiven = field(
max_completion_tokens: int | None | _NotGiven | OpenAINotGiven = field(
default_factory=lambda: _NOT_GIVEN
)
class BaseOpenAILLMService(LLMService):
class BaseOpenAILLMService(LLMService[OpenAILLMAdapter]):
"""Base class for all services that use the AsyncOpenAI client.
This service consumes LLMContextFrame frames, which contain a reference to
@@ -297,9 +297,9 @@ class BaseOpenAILLMService(LLMService):
f"{self}: Generating chat from context {adapter.get_messages_for_logging(context)}"
)
params_from_context: OpenAILLMInvocationParams = adapter.get_llm_invocation_params(
params_from_context = adapter.get_llm_invocation_params(
context,
system_instruction=self._settings.system_instruction,
system_instruction=assert_given(self._settings.system_instruction),
convert_developer_to_user=not self.supports_developer_role,
)
@@ -374,7 +374,7 @@ class BaseOpenAILLMService(LLMService):
"""
effective_instruction = system_instruction or self._settings.system_instruction
adapter = self.get_llm_adapter()
invocation_params: OpenAILLMInvocationParams = adapter.get_llm_invocation_params(
invocation_params = adapter.get_llm_invocation_params(
context,
system_instruction=effective_instruction,
convert_developer_to_user=not self.supports_developer_role,

View File

@@ -13,7 +13,7 @@ for creating images from text prompts.
import io
from collections.abc import AsyncGenerator
from dataclasses import dataclass, field
from typing import Literal
from typing import Literal, cast
import aiohttp
from loguru import logger
@@ -28,6 +28,28 @@ from pipecat.frames.frames import (
from pipecat.services.image_service import ImageGenService
from pipecat.services.settings import NOT_GIVEN, ImageGenSettings, _NotGiven, assert_given
# Hint set for the `size` argument to `images.generate`. The values mirror the
# Literal that `openai.resources.images.Images.generate` accepts on its `size`
# parameter (also visible as the `size` field of
# `openai.types.image_generate_params.ImageGenerateParams`). The OpenAI SDK
# does not export this as a named alias, so we redeclare it here.
#
# We cast `_settings.image_size` (a plain `str`) to this Literal at the API
# boundary so callers can still pass any size string (e.g. a newer value the
# SDK accepts before this list is updated). Invalid values surface as an
# OpenAI API error at runtime. Keep in sync on a best-effort basis when
# bumping the openai dep.
OpenAIImageSize = Literal[
"auto",
"1024x1024",
"1536x1024",
"1024x1536",
"256x256",
"512x512",
"1792x1024",
"1024x1792",
]
@dataclass
class OpenAIImageGenSettings(ImageGenSettings):
@@ -116,15 +138,19 @@ class OpenAIImageGenService(ImageGenService):
"""
logger.debug(f"Generating image from prompt: {prompt}")
size = cast(OpenAIImageSize | None, assert_given(self._settings.image_size))
image = await self._client.images.generate(
prompt=prompt,
model=assert_given(self._settings.model),
n=1,
size=assert_given(self._settings.image_size),
size=size,
)
image_url = image.data[0].url
if not image.data:
yield ErrorFrame("Image generation failed: no data returned")
return
image_url = image.data[0].url
if not image_url:
yield ErrorFrame("Image generation failed")
return

View File

@@ -194,7 +194,7 @@ class OpenAIRealtimeLLMSettings(LLMSettings):
return instance
class OpenAIRealtimeLLMService(LLMService):
class OpenAIRealtimeLLMService(LLMService[OpenAIRealtimeLLMAdapter]):
"""OpenAI Realtime LLM service providing real-time audio and text communication.
Implements the OpenAI Realtime API with WebSocket communication for low-latency
@@ -657,7 +657,7 @@ class OpenAIRealtimeLLMService(LLMService):
async def _send_session_update(self):
settings = assert_given(self._settings.session_properties)
adapter: OpenAIRealtimeLLMAdapter = self.get_llm_adapter()
adapter = self.get_llm_adapter()
if self._context:
llm_invocation_params = adapter.get_llm_invocation_params(
@@ -1002,7 +1002,7 @@ class OpenAIRealtimeLLMService(LLMService):
self._run_llm_when_api_session_ready = True
return
adapter: OpenAIRealtimeLLMAdapter = self.get_llm_adapter()
adapter = self.get_llm_adapter()
# Configure the LLM for this session if needed
if self._llm_needs_conversation_setup:

View File

@@ -115,7 +115,7 @@ class OpenAIResponsesLLMSettings(LLMSettings):
# ---------------------------------------------------------------------------
class _BaseOpenAIResponsesLLMService(LLMService):
class _BaseOpenAIResponsesLLMService(LLMService[OpenAIResponsesLLMAdapter]):
"""Shared base for HTTP and WebSocket OpenAI Responses API services.
Contains settings, adapter reference, HTTP client creation, parameter
@@ -294,7 +294,7 @@ class _BaseOpenAIResponsesLLMService(LLMService):
Returns:
The LLM's response as a string, or None if no response is generated.
"""
adapter: OpenAIResponsesLLMAdapter = self.get_llm_adapter()
adapter = self.get_llm_adapter()
effective_instruction = system_instruction or assert_given(
self._settings.system_instruction
)
@@ -353,7 +353,9 @@ class _BaseOpenAIResponsesLLMService(LLMService):
# ---------------------------------------------------------------------------
class OpenAIResponsesLLMService(_BaseOpenAIResponsesLLMService, WebsocketLLMService):
class OpenAIResponsesLLMService(
_BaseOpenAIResponsesLLMService, WebsocketLLMService[OpenAIResponsesLLMAdapter]
):
"""OpenAI Responses API LLM service using WebSocket transport.
Maintains a persistent WebSocket connection to ``wss://api.openai.com/v1/responses``
@@ -747,7 +749,7 @@ class OpenAIResponsesLLMService(_BaseOpenAIResponsesLLMService, WebsocketLLMServ
if self._needs_drain:
await self._drain_cancelled_response()
adapter: OpenAIResponsesLLMAdapter = self.get_llm_adapter()
adapter = self.get_llm_adapter()
logger.debug(
f"{self}: Generating response from universal context "
f"{adapter.get_messages_for_logging(context)}"
@@ -987,7 +989,7 @@ class OpenAIResponsesHttpLLMService(_BaseOpenAIResponsesLLMService):
@traced_llm
async def _process_context(self, context: LLMContext):
adapter: OpenAIResponsesLLMAdapter = self.get_llm_adapter()
adapter = self.get_llm_adapter()
logger.debug(
f"{self}: Generating response from universal context "
f"{adapter.get_messages_for_logging(context)}"

View File

@@ -18,7 +18,7 @@ import base64
import json
from collections.abc import AsyncGenerator
from dataclasses import dataclass, field
from typing import Any, Literal
from typing import Any, Literal, cast
from loguru import logger
@@ -475,6 +475,9 @@ class OpenAIRealtimeSTTService(WebsocketSTTService):
async def _connect_websocket(self):
"""Establish the WebSocket connection to the transcription endpoint."""
try:
# `__init__` raises if websockets isn't installed, so these symbols
# are non-None by the time any method runs.
assert websocket_connect is not None and State is not None
if self._websocket and self._websocket.state is State.OPEN:
return
@@ -534,7 +537,9 @@ class OpenAIRealtimeSTTService(WebsocketSTTService):
"""Send ``session.update`` to configure the transcription session."""
transcription: dict = {"model": self._settings.model}
language = assert_given(self._settings.language)
# Technically `_settings.language` could be a raw string, but Language
# is a StrEnum so downstream handles either.
language = cast("Language | None", assert_given(self._settings.language))
language_code = self._language_to_code(language) if language else None
if language_code:
transcription["language"] = language_code
@@ -611,6 +616,10 @@ class OpenAIRealtimeSTTService(WebsocketSTTService):
Called by ``WebsocketService._receive_task_handler`` which wraps
this method with automatic reconnection on connection errors.
"""
# `_connect` only starts the receive task after `_websocket` is set,
# and reconnects re-establish it before the next iteration, so this
# invariant should always hold when this method runs.
assert self._websocket is not None
async for message in self._websocket:
try:
evt = json.loads(message)

View File

@@ -235,12 +235,22 @@ class OpenAITTSService(TTSService):
Frame: Audio frames containing the synthesized speech data.
"""
logger.debug(f"{self}: Generating TTS [{text}]")
voice = assert_given(self._settings.voice)
if voice is None:
yield ErrorFrame(error="OpenAI TTS voice must be specified")
return
if voice not in VALID_VOICES:
yield ErrorFrame(
error=f"OpenAI TTS voice {voice!r} is not supported "
f"(must be one of: {', '.join(sorted(VALID_VOICES))})"
)
return
try:
# Setup API parameters
create_params = {
"input": text,
"model": self._settings.model,
"voice": VALID_VOICES[assert_given(self._settings.voice)],
"voice": VALID_VOICES[voice],
"response_format": "pcm",
}

View File

@@ -15,6 +15,7 @@ from typing import Any
from loguru import logger
from pipecat.adapters.services.open_ai_adapter import OpenAILLMInvocationParams
from pipecat.services.openai.base_llm import BaseOpenAILLMService
from pipecat.services.openai.llm import OpenAILLMService
from pipecat.services.settings import assert_given
@@ -96,7 +97,9 @@ class OpenRouterLLMService(OpenAILLMService):
logger.debug(f"Creating OpenRouter client with api {base_url}")
return super().create_client(api_key, base_url, **kwargs)
def build_chat_completion_params(self, params_from_context: dict[str, Any]) -> dict[str, Any]:
def build_chat_completion_params(
self, params_from_context: OpenAILLMInvocationParams
) -> dict[str, Any]:
"""Builds chat parameters, handling model-specific constraints.
Args:

View File

@@ -101,6 +101,8 @@ class PiperTTSService(TTSService):
download_dir = download_dir or Path.cwd()
_voice = assert_given(self._settings.voice)
if _voice is None:
raise ValueError("Piper TTS voice must be specified")
model_file = f"{_voice}.onnx"
model_path_resolved = Path(download_dir) / model_file

View File

@@ -409,7 +409,9 @@ class ResembleAITTSService(WebsocketTTSService):
await self.push_frame(TTSStoppedFrame(context_id=context_id))
await self.stop_all_metrics()
await self.push_error(ErrorFrame(error=f"{self} error: {error_name} - {error_msg}"))
await self.push_error_frame(
ErrorFrame(error=f"{self} error: {error_name} - {error_msg}")
)
# Check if this is an unrecoverable error (connection-level failure)
if status_code in [401, 403]:

View File

@@ -216,14 +216,16 @@ def get_speakers_for_model(model: str) -> list[str]:
return list(TTS_MODEL_CONFIGS["bulbul:v2"].speakers)
def language_to_sarvam_language(language: Language) -> str | None:
def language_to_sarvam_language(language: Language) -> str:
"""Convert Pipecat Language enum to Sarvam AI language codes.
Args:
language: The Language enum value to convert.
Returns:
The corresponding Sarvam AI language code, or None if not supported.
The corresponding service language code. If ``language`` is not in
the verified mapping, falls back to the full language code string and
logs a warning (via ``resolve_language(..., use_base_code=False)``).
"""
LANGUAGE_MAP = {
Language.BN: "bn-IN", # Bengali

View File

@@ -51,14 +51,17 @@ class SmallestTTSModel(StrEnum):
LIGHTNING_V3_1 = "lightning-v3.1"
def language_to_smallest_tts_language(language: Language) -> str | None:
def language_to_smallest_tts_language(language: Language) -> str:
"""Convert a Language enum to a Smallest TTS language string.
Args:
language: The Language enum value to convert.
Returns:
The Smallest language code string, or None if unsupported.
The corresponding Smallest language code. If ``language`` is not in
the verified mapping, falls back to the base language code (e.g.,
``en`` from ``en-US``) and logs a warning (via
``resolve_language(..., use_base_code=True)``).
"""
LANGUAGE_MAP = {
Language.AR: "ar",

View File

@@ -648,4 +648,7 @@ class SonioxSTTService(WebsocketSTTService):
Args:
silence: Silent PCM audio bytes (unused, Soniox uses a protocol message).
"""
if self._websocket is None:
logger.warning(f"{self}: websocket unavailable, skipping keepalive")
return
await self._websocket.send(KEEPALIVE_MESSAGE)

View File

@@ -269,7 +269,10 @@ class STTService(AIService):
language: The language to convert.
Returns:
The service-specific language identifier, or None if not supported.
The service-specific language identifier. Return ``None`` to
indicate an unsupported language. This optional return is an
extension hook for future or third-party subclasses; as of
2026-04-28, first-party services return a string.
"""
return Language(language)
@@ -859,6 +862,10 @@ class WebsocketSTTService(STTService, WebsocketService):
Args:
silence: Silent 16-bit mono PCM audio bytes.
"""
if (
self._websocket is None
): # should never happen — caller should gate on _is_keepalive_ready()
return
await self._websocket.send(silence)
async def _report_error(self, error: ErrorFrame):

View File

@@ -467,7 +467,10 @@ class TTSService(AIService):
language: The language to convert.
Returns:
The service-specific language identifier, or None if not supported.
The service-specific language identifier. Return ``None`` to
indicate an unsupported language. This optional return is an
extension hook for future or third-party subclasses; as of
2026-04-28, first-party services return a string.
"""
return Language(language)
@@ -609,8 +612,10 @@ class TTSService(AIService):
"""Handle the completion of a turn."""
# For HTTP services they emit the frames synchronously, so close the audio context here
# once all frames (including TTSTextFrame above) have been enqueued.
if self._is_yielding_frames_synchronously and self.audio_context_available(
self._turn_context_id
if (
self._is_yielding_frames_synchronously
and self._turn_context_id is not None
and self.audio_context_available(self._turn_context_id)
):
if self._push_stop_frames:
await self.append_to_audio_context(
@@ -1171,16 +1176,18 @@ class TTSService(AIService):
logger.trace(f"{self} created audio context {context_id}")
async def append_to_audio_context(
self, context_id: str, frame: Frame | _WordTimestampEntry | None
self, context_id: str | None, frame: Frame | _WordTimestampEntry | None
):
"""Append a frame or word-timestamp entry to an existing audio context queue.
Passing ``None`` signals end-of-context (used by remove_audio_context to mark
the queue for deletion). If the context no longer exists but the context_id
Passing a ``frame`` of ``None`` signals end-of-context (used by remove_audio_context
to mark the queue for deletion). If the context no longer exists but the context_id
matches the active turn, the context is transparently recreated before appending.
Args:
context_id: The context to append to.
context_id: The context to append to. ``None`` is accepted as a no-op
(with a debug log) so callers can pass through values from
``get_active_audio_context_id()`` without an explicit guard.
frame: The frame, word-timestamp entry, or ``None`` (end-of-context sentinel)
to append.
"""
@@ -1201,12 +1208,17 @@ class TTSService(AIService):
else:
logger.debug(f"{self} unable to append audio to context {context_id}")
async def remove_audio_context(self, context_id: str):
async def remove_audio_context(self, context_id: str | None):
"""Remove an existing audio context.
Args:
context_id: The context to remove.
context_id: The context to remove. ``None`` is accepted as a
no-op (logged) so callers can pass through values from
``get_active_audio_context_id()`` without an explicit guard.
"""
if not context_id:
logger.debug(f"{self} unable to remove audio context: no context ID provided")
return
if self.audio_context_available(context_id):
# We just mark the audio context for deletion by appending
# None. Once we reach None while handling audio we know we can

View File

@@ -42,7 +42,7 @@ class WebsocketService(ABC):
reconnect_on_error: Whether to automatically reconnect on connection errors.
**kwargs: Additional arguments (unused, for compatibility).
"""
self._websocket: websockets.WebSocketClientProtocol | None = None
self._websocket: websockets.WebSocketClientProtocol | None = None # pyright: ignore[reportAttributeAccessIssue]
self._reconnect_on_error = reconnect_on_error
self._reconnect_in_progress: bool = False
self._disconnecting: bool = False
@@ -120,12 +120,17 @@ class WebsocketService(ABC):
async def send_with_retry(self, message, report_error: Callable[[ErrorFrame], Awaitable[None]]):
"""Attempt to send a message, retrying after reconnect if necessary."""
try:
# If websocket isn't connected/present, treat as a send failure —
# the broad `except Exception` below will trigger a reconnect
# attempt.
if self._websocket is None:
raise ConnectionError(f"{self} no websocket connected")
await self._websocket.send(message)
except Exception as e:
logger.error(f"{self} send failed: {e}, will try to reconnect")
# Try to reconnect before retrying
success = await self._try_reconnect(report_error=report_error)
if success:
if success and self._websocket is not None:
logger.info(f"{self} reconnected successfully, will retry send the message")
# trying to send the message one more time
await self._websocket.send(message)

View File

@@ -40,7 +40,7 @@ class BaseWhisperSTTSettings(STTSettings):
temperature: float | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN)
def language_to_whisper_language(language: Language) -> str | None:
def language_to_whisper_language(language: Language) -> str:
"""Maps pipecat Language enum to Whisper API language codes.
Language support for Whisper API.
@@ -50,7 +50,10 @@ def language_to_whisper_language(language: Language) -> str | None:
language: A Language enum value representing the input language.
Returns:
str or None: The corresponding Whisper language code, or None if not supported.
The corresponding service language code. If ``language`` is not in
the verified mapping, falls back to the base language code (e.g.,
``en`` from ``en-US``) and logs a warning (via
``resolve_language(..., use_base_code=True)``).
"""
LANGUAGE_MAP = {
Language.AF: "af",
@@ -235,7 +238,7 @@ class BaseWhisperSTTService(SegmentedSTTService):
language: The Language enum value to convert.
Returns:
str or None: The corresponding service language code, or None if not supported.
The corresponding service language code, or None if not supported.
"""
return language_to_whisper_language(language)

View File

@@ -97,14 +97,17 @@ class MLXModel(Enum):
LARGE_V3_TURBO_Q4 = "mlx-community/whisper-large-v3-turbo-q4"
def language_to_whisper_language(language: Language) -> str | None:
def language_to_whisper_language(language: Language) -> str:
"""Maps pipecat Language enum to Whisper language codes.
Args:
language: A Language enum value representing the input language.
Returns:
str or None: The corresponding Whisper language code, or None if not supported.
The corresponding service language code. If ``language`` is not in
the verified mapping, falls back to the base language code (e.g.,
``en`` from ``en-US``) and logs a warning (via
``resolve_language(..., use_base_code=True)``).
Note:
Only includes languages officially supported by Whisper.
@@ -300,7 +303,7 @@ class WhisperSTTService(SegmentedSTTService):
language: The Language enum value to convert.
Returns:
str or None: The corresponding Whisper language code, or None if not supported.
The corresponding Whisper language code, or None if not supported.
"""
return language_to_whisper_language(language)

View File

@@ -179,7 +179,7 @@ class GrokRealtimeLLMSettings(LLMSettings):
return instance
class GrokRealtimeLLMService(LLMService):
class GrokRealtimeLLMService(LLMService[GrokRealtimeLLMAdapter]):
"""Grok Realtime Voice Agent LLM service providing real-time audio and text communication.
Implements the Grok Voice Agent API with WebSocket communication for low-latency
@@ -596,7 +596,7 @@ class GrokRealtimeLLMService(LLMService):
async def _send_session_update(self):
"""Update session settings on the server."""
settings = assert_given(self._settings.session_properties)
adapter: GrokRealtimeLLMAdapter = self.get_llm_adapter()
adapter = self.get_llm_adapter()
if self._context:
llm_invocation_params = adapter.get_llm_invocation_params(
@@ -871,7 +871,7 @@ class GrokRealtimeLLMService(LLMService):
self._run_llm_when_api_session_ready = True
return
adapter: GrokRealtimeLLMAdapter = self.get_llm_adapter()
adapter = self.get_llm_adapter()
if self._llm_needs_conversation_setup:
logger.debug(

View File

@@ -44,7 +44,7 @@ except ModuleNotFoundError as e:
raise Exception(f"Missing module: {e}")
def language_to_xai_stt_language(language: Language) -> str | None:
def language_to_xai_stt_language(language: Language) -> str:
"""Convert a Language enum to the xAI STT language code.
xAI STT accepts two-letter language codes (e.g. ``en``, ``fr``, ``de``,
@@ -54,7 +54,10 @@ def language_to_xai_stt_language(language: Language) -> str | None:
language: The Language enum value to convert.
Returns:
The corresponding xAI STT language code, or None if not supported.
The corresponding service language code. If ``language`` is not in
the verified mapping, falls back to the base language code (e.g.,
``en`` from ``en-US``) and logs a warning (via
``resolve_language(..., use_base_code=True)``).
"""
LANGUAGE_MAP = {
Language.AR: "ar",

View File

@@ -49,14 +49,17 @@ except ModuleNotFoundError as e:
raise Exception(f"Missing module: {e}")
def language_to_xai_language(language: Language) -> str | None:
def language_to_xai_language(language: Language) -> str:
"""Convert a Language enum to xAI language code.
Args:
language: The Language enum value to convert.
Returns:
The corresponding xAI language code, or None if not supported.
The corresponding service language code. If ``language`` is not in
the verified mapping, falls back to the base language code (e.g.,
``en`` from ``en-US``) and logs a warning (via
``resolve_language(..., use_base_code=True)``).
"""
LANGUAGE_MAP = {
Language.AR: "ar-EG",

View File

@@ -37,14 +37,17 @@ from pipecat.utils.tracing.service_decorators import traced_tts
# https://github.com/coqui-ai/xtts-streaming-server
def language_to_xtts_language(language: Language) -> str | None:
def language_to_xtts_language(language: Language) -> str:
"""Convert a Language enum to XTTS language code.
Args:
language: The Language enum value to convert.
Returns:
The corresponding XTTS language code, or None if not supported.
The corresponding service language code. If ``language`` is not in
the verified mapping, falls back to the base language code (e.g.,
``en`` from ``en-US``) and logs a warning (via
``resolve_language(..., use_base_code=True)``).
"""
LANGUAGE_MAP = {
Language.CS: "cs",
@@ -211,7 +214,11 @@ class XTTSService(TTSService):
logger.error(f"{self} no studio speakers available")
return
embeddings = self._studio_speakers[assert_given(self._settings.voice)]
voice = assert_given(self._settings.voice)
if voice is None:
yield ErrorFrame(error="XTTS voice must be specified")
return
embeddings = self._studio_speakers[voice]
url = self._base_url + "/tts_stream"

View File

@@ -771,13 +771,16 @@ class BaseOutputTransport(FrameProcessor):
await self._bot_stopped_speaking()
async def with_mixer(vad_stop_secs: float) -> AsyncGenerator[Frame, None]:
# Caller below only invokes this when `self._mixer` is set.
mixer = self._mixer
assert mixer is not None
last_frame_time = 0
silence = b"\x00" * self._audio_chunk_size
while True:
try:
frame = self._audio_queue.get_nowait()
if isinstance(frame, OutputAudioRawFrame):
frame.audio = await self._mixer.mix(frame.audio)
frame.audio = await mixer.mix(frame.audio)
last_frame_time = time.time()
yield frame
self._audio_queue.task_done()
@@ -788,7 +791,7 @@ class BaseOutputTransport(FrameProcessor):
await self._bot_stopped_speaking()
# Generate an audio frame with only the mixer's part.
frame = OutputAudioRawFrame(
audio=await self._mixer.mix(silence),
audio=await mixer.mix(silence),
sample_rate=self._sample_rate,
num_channels=self._params.audio_out_channels,
)
@@ -927,6 +930,11 @@ class BaseOutputTransport(FrameProcessor):
"""
def resize_frame(frame: OutputImageRawFrame) -> OutputImageRawFrame:
# Without a format we can't decode the bytes, so leave the
# frame as-is and let the transport pass it through unchanged.
if frame.format is None:
return frame
desired_size = (self._params.video_out_width, self._params.video_out_height)
# TODO: we should refactor in the future to support dynamic resolutions

View File

@@ -239,8 +239,9 @@ class HeyGenOutputTransport(BaseOutputTransport):
logger.warning("self._event_id is already defined!")
self._event_id = str(frame.id)
elif isinstance(frame, BotStoppedSpeakingFrame):
await self._client.agent_speak_end(self._event_id)
self._event_id = None
if self._event_id is not None:
await self._client.agent_speak_end(self._event_id)
self._event_id = None
await super().push_frame(frame, direction)
async def process_frame(self, frame: Frame, direction: FrameDirection):
@@ -261,7 +262,8 @@ class HeyGenOutputTransport(BaseOutputTransport):
"""
await super().process_frame(frame, direction)
if isinstance(frame, InterruptionFrame):
await self._client.interrupt(self._event_id)
if self._event_id is not None:
await self._client.interrupt(self._event_id)
await self.push_frame(frame, direction)
if isinstance(frame, UserStartedSpeakingFrame):
await self._client.start_agent_listening()
@@ -281,6 +283,11 @@ class HeyGenOutputTransport(BaseOutputTransport):
audio = frame.audio
if frame.sample_rate != HEY_GEN_SAMPLE_RATE:
audio = await self._resampler.resample(audio, frame.sample_rate, HEY_GEN_SAMPLE_RATE)
if self._event_id is None:
# No active bot-speech event — drop the chunk rather than send a
# message the HeyGen API will reject.
logger.warning(f"{self}: dropping audio frame because no event_id is set")
return False
await self._client.agent_speak(bytes(audio), self._event_id)
return True

View File

@@ -312,6 +312,9 @@ class LemonSliceTransportClient:
Args:
frame: The message frame to send.
"""
if self._daily_transport_client is None:
return
await self._daily_transport_client.send_message(frame)
@property

View File

@@ -224,6 +224,8 @@ class SmallWebRTCRequestHandler:
)
answer = pipecat_connection.get_answer()
if answer is None:
raise RuntimeError("SmallWebRTC connection produced no SDP answer")
if self._esp32_mode:
from pipecat.runner.utils import smallwebrtc_sdp_munging

View File

@@ -360,6 +360,9 @@ class TavusTransportClient:
Args:
frame: The message frame to send.
"""
if self._client is None:
return
await self._client.send_message(frame)
@property
@@ -416,6 +419,7 @@ class TavusTransportClient:
"""
if not self._client:
return False
return await self._client.write_audio_frame(frame)
async def register_audio_destination(self, destination: str, auto_silence: bool | None = True):

View File

@@ -64,9 +64,18 @@ class WebsocketClientCallbacks(BaseModel):
on_message: Called when a message is received from the WebSocket.
"""
on_connected: Callable[[websockets.WebSocketClientProtocol], Awaitable[None]]
on_disconnected: Callable[[websockets.WebSocketClientProtocol], Awaitable[None]]
on_message: Callable[[websockets.WebSocketClientProtocol, websockets.Data], Awaitable[None]]
on_connected: Callable[
[websockets.WebSocketClientProtocol], # pyright: ignore[reportAttributeAccessIssue]
Awaitable[None],
]
on_disconnected: Callable[
[websockets.WebSocketClientProtocol], # pyright: ignore[reportAttributeAccessIssue]
Awaitable[None],
]
on_message: Callable[
[websockets.WebSocketClientProtocol, websockets.Data], # pyright: ignore[reportAttributeAccessIssue]
Awaitable[None],
]
class WebsocketClientSession:
@@ -98,7 +107,7 @@ class WebsocketClientSession:
self._leave_counter = 0
self._task_manager: BaseTaskManager | None = None
self._websocket: websockets.WebSocketClientProtocol | None = None
self._websocket: websockets.WebSocketClientProtocol | None = None # pyright: ignore[reportAttributeAccessIssue]
@property
def task_manager(self) -> BaseTaskManager:
@@ -192,6 +201,10 @@ class WebsocketClientSession:
async def _client_task_handler(self):
"""Handle incoming messages from the WebSocket connection."""
# `connect()` only starts this task after `_websocket` is assigned, and
# `disconnect()` cancels the task before clearing `_websocket`, so this
# invariant should always hold when this method runs.
assert self._websocket is not None
try:
# Handle incoming messages
async for message in self._websocket:

View File

@@ -226,7 +226,7 @@ class WebsocketServerInputTransport(BaseInputTransport):
# Notify disconnection
await self._callbacks.on_client_disconnected(websocket)
await self._websocket.close()
await websocket.close()
self._websocket = None
logger.info(f"Client {websocket.remote_address} disconnected")

View File

@@ -154,8 +154,17 @@ class WhatsAppClient:
return int(challenge)
async def _validate_whatsapp_webhook_request(self, raw_body: bytes, sha256_signature: str):
async def _validate_whatsapp_webhook_request(
self, raw_body: bytes | None, sha256_signature: str | None
):
"""Common handler for both /start and /connect endpoints."""
# Callers gate on `self._whatsapp_secret`, so the assert holds.
assert self._whatsapp_secret is not None
if raw_body is None:
raise Exception("Missing raw request body")
if not sha256_signature:
raise Exception("Missing X-Hub-Signature-256 header")
# Compute HMAC SHA256 using your App Secret
expected_signature = hmac.new(
key=self._whatsapp_secret.encode("utf-8"),
@@ -164,8 +173,6 @@ class WhatsAppClient:
).hexdigest()
# Extract signature from header (strip 'sha256=' prefix)
if not sha256_signature:
raise Exception("Missing X-Hub-Signature-256 header")
received_signature = sha256_signature.split("sha256=")[-1]
# Compare signatures securely
@@ -306,7 +313,12 @@ class WhatsAppClient:
# Create and initialize WebRTC connection
pipecat_connection = SmallWebRTCConnection(self._ice_servers)
await pipecat_connection.initialize(sdp=call.session.sdp, type=call.session.sdp_type)
sdp_answer = pipecat_connection.get_answer().get("sdp")
answer = pipecat_connection.get_answer()
if answer is None:
raise RuntimeError("SmallWebRTC connection produced no SDP answer")
sdp_answer = answer.get("sdp")
if sdp_answer is None:
raise RuntimeError("SmallWebRTC SDP answer missing 'sdp' field")
sdp_answer = self._filter_sdp_for_whatsapp(sdp_answer)
logger.debug(f"SDP answer generated for call {call.id}")

View File

@@ -7,6 +7,8 @@
import unittest
from unittest.mock import AsyncMock, patch
from pipecat.adapters.base_llm_adapter import BaseLLMAdapter
from pipecat.adapters.services.open_ai_adapter import OpenAILLMAdapter
from pipecat.frames.frames import (
FunctionCallFromLLM,
FunctionCallInProgressFrame,
@@ -39,6 +41,25 @@ class MockLLMService(LLMService):
super().__init__(settings=settings, **kwargs)
class TestUnparameterizedSubclass(unittest.TestCase):
"""Backward-compat coverage: third-party providers subclass LLMService
without specifying a generic adapter parameter. That should keep working
after LLMService became `Generic[TAdapter]`.
"""
def test_unparameterized_subclass_instantiates(self):
# MockLLMService is declared as `class MockLLMService(LLMService):`
# — no generic bracket. The TypeVar's `bound=BaseLLMAdapter` should
# resolve TAdapter to BaseLLMAdapter for callers that don't opt in.
service = MockLLMService()
adapter = service.get_llm_adapter()
# Default adapter_class is OpenAILLMAdapter; the runtime instance
# should reflect that, regardless of how generics are erased.
self.assertIsInstance(adapter, OpenAILLMAdapter)
self.assertIsInstance(adapter, BaseLLMAdapter)
class TestLLMService(unittest.IsolatedAsyncioTestCase):
async def _run_function_calls_inline(self, service: MockLLMService):
async def run_inline(runner_items):