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

Add Inworld Realtime Service (#4140)

Browse Source 
* Add Inworld Realtime LLM service

Adds a WebSocket-based realtime service for Inworld's cascade
STT/LLM/TTS API with semantic VAD, function calling, and streaming
transcription support.

New files:
- src/pipecat/services/inworld/realtime/ (service, events)
- src/pipecat/adapters/services/inworld_realtime_adapter.py
- examples/foundational/19zb-inworld-realtime.py

Also includes:
- websockets dependency for inworld extra in pyproject.toml
- Adapter and settings tests matching OpenAI/Grok realtime patterns
- Fix for double-response when server-side VAD is enabled

* Prefer init-provided system instruction in Inworld Realtime

Adopt _resolve_system_instruction() from BaseLLMAdapter, matching the
pattern applied to OpenAI Realtime, Grok Realtime, Gemini Live, and
Nova Sonic in the pk/realtime-services-init-v-context-system-instructions-cleanup
branch.

* Update changelog entry with PR number

* Fix changelog format to use bullet point

* Polish PR: default model, example cleanup, changelog update

- Change default model from gpt-4.1-nano to gpt-4.1-mini
- Add function calling demo to example
- Remove demo-testing artifact from system instruction
- Mention Router support in changelog

* Address PR review feedback for Inworld Realtime

- Move example to examples/realtime/realtime-inworld.py
- Change initial context role from "user" to "developer"
- Remove explicit sample rates from example; sync them in
  _ensure_audio_config so Inworld gets the transport's actual rates
- Add audio race condition guard in _handle_evt_audio_delta (matches
  OpenAI realtime pattern)
- Convert remaining "system"/"developer" messages to "user" in adapter
- Add clarifying comment for local-VAD vs server-VAD metrics paths

* Simplify example, add provider tracking, remove local VAD path

- Remove function calling from example, switch model to xai/grok-4-1-fast-non-reasoning
- Add pipecat-realtime session key prefix and provider_data metadata
  for Inworld traffic attribution
- Remove local VAD code path (Inworld only supports server-side VAD)
- Use typed InputAudioBufferAppendEvent for audio sends

* Default TTS model to inworld-tts-1.5-max

* Remove dead shimmed tools code, set STT/VAD defaults

- Remove non-functional AdapterType.SHIM custom tools code from adapter
- Default STT model to assemblyai/u3-rt-pro
- Default VAD eagerness to low
This commit is contained in:
Cale Shapera
2026-04-09 10:04:17 -07:00
committed by GitHub
parent 76601944c6
commit ec574edd53
10 changed files with 2561 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
changelog/4140.added.md Normal file
View File

@@ -0,0 +1 @@
- Added Inworld Realtime LLM service with WebSocket-based cascade STT/LLM/TTS, semantic VAD, function calling, and Router support.

162
examples/realtime/realtime-inworld.py Normal file
View File

@@ -0,0 +1,162 @@
#
# Copyright (c) 2024-2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""
Inworld Realtime Example
This example demonstrates using Inworld's Realtime API for real-time voice
conversations. The Inworld Realtime API is OpenAI-compatible and operates
as a cascade STT/LLM/TTS pipeline under the hood, with built-in semantic
voice activity detection for turn management.
Features:
- Real-time audio streaming with low latency
- Built-in semantic VAD (voice activity detection)
- Streaming user transcription
- Text and audio input
Requirements:
- INWORLD_API_KEY environment variable set
- pip install pipecat-ai[inworld]
Usage:
python realtime-inworld.py --transport webrtc
python realtime-inworld.py --transport daily
"""
import os
from dotenv import load_dotenv
from loguru import logger
from pipecat.frames.frames import LLMRunFrame
from pipecat.observers.loggers.transcription_log_observer import (
TranscriptionLogObserver,
)
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response_universal import (
AssistantTurnStoppedMessage,
LLMContextAggregatorPair,
UserTurnStoppedMessage,
)
from pipecat.runner.types import RunnerArguments
from pipecat.runner.utils import create_transport
from pipecat.services.inworld.realtime.llm import InworldRealtimeLLMService
from pipecat.transports.base_transport import BaseTransport, TransportParams
from pipecat.transports.daily.transport import DailyParams
from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams
load_dotenv(override=True)
# --- Transport Configuration ---
# No local VAD needed — Inworld's server-side semantic VAD handles turn detection.
transport_params = {
"daily": lambda: DailyParams(
audio_in_enabled=True,
audio_out_enabled=True,
),
"twilio": lambda: FastAPIWebsocketParams(
audio_in_enabled=True,
audio_out_enabled=True,
),
"webrtc": lambda: TransportParams(
audio_in_enabled=True,
audio_out_enabled=True,
),
}
async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
logger.info("Starting Inworld Realtime bot")
# Create the Inworld Realtime LLM service.
# Common params (llm_model, voice, tts_model, stt_model) are top-level.
# For full control, use settings=InworldRealtimeLLMService.Settings(session_properties=...)
#
# llm_model can be any supported model or an Inworld Router.
# See: https://docs.inworld.ai/router/introduction
llm = InworldRealtimeLLMService(
api_key=os.getenv("INWORLD_API_KEY"),
llm_model="xai/grok-4-1-fast-non-reasoning",
voice="Sarah",
settings=InworldRealtimeLLMService.Settings(
system_instruction="""You are a helpful and friendly AI assistant powered by Inworld.
Your voice and personality should be warm and engaging. Keep your responses
concise and conversational since this is a voice interaction.
Always be helpful and proactive in offering assistance.""",
),
)
# Create context with initial message
context = LLMContext(
[{"role": "developer", "content": "Say hello and introduce yourself!"}],
)
user_aggregator, assistant_aggregator = LLMContextAggregatorPair(context)
# Build the pipeline
pipeline = Pipeline(
[
transport.input(),
user_aggregator,
llm, # Inworld Realtime (handles STT + LLM + TTS)
transport.output(),
assistant_aggregator,
]
)
task = PipelineTask(
pipeline,
params=PipelineParams(
enable_metrics=True,
enable_usage_metrics=True,
),
idle_timeout_secs=runner_args.pipeline_idle_timeout_secs,
observers=[TranscriptionLogObserver()],
)
@transport.event_handler("on_client_connected")
async def on_client_connected(transport, client):
logger.info("Client connected")
await task.queue_frames([LLMRunFrame()])
@transport.event_handler("on_client_disconnected")
async def on_client_disconnected(transport, client):
logger.info("Client disconnected")
await task.cancel()
@user_aggregator.event_handler("on_user_turn_stopped")
async def on_user_turn_stopped(aggregator, strategy, message: UserTurnStoppedMessage):
timestamp = f"[{message.timestamp}] " if message.timestamp else ""
logger.info(f"Transcript: {timestamp}user: {message.content}")
@assistant_aggregator.event_handler("on_assistant_turn_stopped")
async def on_assistant_turn_stopped(aggregator, message: AssistantTurnStoppedMessage):
timestamp = f"[{message.timestamp}] " if message.timestamp else ""
logger.info(f"Transcript: {timestamp}assistant: {message.content}")
runner = PipelineRunner(handle_sigint=runner_args.handle_sigint)
await runner.run(task)
async def bot(runner_args: RunnerArguments):
"""Main bot entry point compatible with Pipecat Cloud."""
transport = await create_transport(runner_args, transport_params)
await run_bot(transport, runner_args)
if __name__ == "__main__":
from pipecat.runner.run import main
main()

2
pyproject.toml
View File

@@ -77,7 +77,7 @@ groq = [ "groq>=0.23.0,<2" ]
gstreamer = [ "pygobject~=3.50.0" ]
heygen = [ "livekit>=1.0.13,<2", "pipecat-ai[websockets-base]" ]
hume = [ "hume>=0.11.2,<1" ]
inworld = []
inworld = [ "pipecat-ai[websockets-base]" ]
koala = [ "pvkoala~=2.0.3" ]
kokoro = [ "kokoro-onnx>=0.5.0,<1", "requests>=2.32.5,<3" ]
langchain = [ "langchain>=1.2.13,<2", "langchain-community>=0.4.1,<1", "langchain-openai>=1.1.12,<2" ]

255
src/pipecat/adapters/services/inworld_realtime_adapter.py Normal file
View File

@@ -0,0 +1,255 @@
#
# Copyright (c) 2024-2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Inworld Realtime LLM adapter for Pipecat.
Converts Pipecat's tool schemas and context into the format required by
Inworld's Realtime API.
"""
import copy
import json
from dataclasses import dataclass
from typing import Any, Dict, List, Optional, TypedDict
from loguru import logger
from pipecat.adapters.base_llm_adapter import BaseLLMAdapter
from pipecat.adapters.schemas.function_schema import FunctionSchema
from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.processors.aggregators.llm_context import LLMContext, LLMContextMessage
from pipecat.services.inworld.realtime import events
class InworldRealtimeLLMInvocationParams(TypedDict):
"""Context-based parameters for invoking Inworld Realtime API.
Attributes:
system_instruction: System prompt/instructions for the session.
messages: List of conversation items formatted for Inworld Realtime.
tools: List of tool definitions.
"""
system_instruction: Optional[str]
messages: List[events.ConversationItem]
tools: List[Dict[str, Any]]
class InworldRealtimeLLMAdapter(BaseLLMAdapter):
"""LLM adapter for Inworld Realtime API.
Converts Pipecat's universal context and tool schemas into the specific
format required by Inworld's Realtime API.
"""
@property
def id_for_llm_specific_messages(self) -> str:
"""Get the identifier used in LLMSpecificMessage instances for Inworld Realtime."""
return "inworld-realtime"
def get_llm_invocation_params(
self, context: LLMContext, *, system_instruction: Optional[str] = None
) -> InworldRealtimeLLMInvocationParams:
"""Get Inworld Realtime-specific LLM invocation parameters from a universal LLM context.
Args:
context: The LLM context containing messages, tools, etc.
system_instruction: Optional system instruction from service settings.
Returns:
Dictionary of parameters for invoking Inworld's Realtime API.
"""
messages = self._from_universal_context_messages(self.get_messages(context))
effective_system = self._resolve_system_instruction(
messages.system_instruction,
system_instruction,
discard_context_system=True,
)
return {
"system_instruction": effective_system,
"messages": messages.messages,
"tools": self.from_standard_tools(context.tools) or [],
}
def get_messages_for_logging(self, context) -> List[Dict[str, Any]]:
"""Get messages from context in a format safe for logging.
Removes or truncates sensitive data like audio content.
Args:
context: The LLM context containing messages.
Returns:
List of messages with sensitive data redacted.
"""
msgs = []
for message in self.get_messages(context):
msg = copy.deepcopy(message)
if "content" in msg:
if isinstance(msg["content"], list):
for item in msg["content"]:
if item.get("type") == "input_audio":
item["audio"] = "..."
if item.get("type") == "audio":
item["audio"] = "..."
msgs.append(msg)
return msgs
@dataclass
class ConvertedMessages:
"""Container for Inworld-formatted messages converted from universal context."""
messages: List[events.ConversationItem]
system_instruction: Optional[str] = None
def _from_universal_context_messages(
self, universal_context_messages: List[LLMContextMessage]
) -> ConvertedMessages:
"""Convert universal context messages to Inworld Realtime format.
Similar to OpenAI Realtime, we pack conversation history into a single
user message since the realtime API doesn't support loading long histories.
Args:
universal_context_messages: List of messages in universal format.
Returns:
ConvertedMessages with Inworld-formatted messages and system instruction.
"""
if not universal_context_messages:
return self.ConvertedMessages(messages=[])
messages = copy.deepcopy(universal_context_messages)
system_instruction = None
# Extract system message as session instructions
if messages[0].get("role") == "system":
system = messages.pop(0)
content = system.get("content")
if isinstance(content, str):
system_instruction = content
elif isinstance(content, list):
system_instruction = content[0].get("text")
if not messages:
return self.ConvertedMessages(messages=[], system_instruction=system_instruction)
# Convert any remaining "system"/"developer" messages to "user"
for msg in messages:
if msg.get("role") in ("system", "developer"):
msg["role"] = "user"
# 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])],
system_instruction=system_instruction,
)
# Pack multiple messages into a single user message
intro_text = """
This is a previously saved conversation. Please treat this conversation history as a
starting point for the current conversation."""
trailing_text = """
This is the end of the previously saved conversation. Please continue the conversation
from here. If the last message is a user instruction or question, act on that instruction
or answer the question. If the last message is an assistant response, simply say that you
are ready to continue the conversation."""
return self.ConvertedMessages(
messages=[
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,
)
def _from_universal_context_message(
self, message: LLMContextMessage
) -> events.ConversationItem:
"""Convert a single universal context message to Inworld format.
Args:
message: Message in universal format.
Returns:
ConversationItem formatted for Inworld Realtime API.
"""
if message.get("role") == "user":
content = message.get("content")
if isinstance(content, list):
text_content = ""
for c in content:
if c.get("type") == "text":
text_content += " " + c.get("text")
else:
logger.error(
f"Unhandled content type in context message: {c.get('type')} - {message}"
)
content = text_content.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]
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}")
@staticmethod
def _to_inworld_function_format(function: FunctionSchema) -> Dict[str, Any]:
"""Convert a function schema to Inworld Realtime function format.
Args:
function: The function schema to convert.
Returns:
Dictionary in Inworld Realtime function format.
"""
return {
"type": "function",
"name": function.name,
"description": function.description,
"parameters": {
"type": "object",
"properties": function.properties,
"required": function.required,
},
}
def to_provider_tools_format(self, tools_schema: ToolsSchema) -> List[Dict[str, Any]]:
"""Convert tool schemas to Inworld Realtime format.
Args:
tools_schema: The tools schema containing functions to convert.
Returns:
List of tool definitions in Inworld Realtime format.
"""
functions_schema = tools_schema.standard_tools
return [self._to_inworld_function_format(func) for func in functions_schema]

5
src/pipecat/services/inworld/realtime/__init__.py Normal file
View File

@@ -0,0 +1,5 @@
#
# Copyright (c) 2024-2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#

868
src/pipecat/services/inworld/realtime/events.py Normal file
View File

@@ -0,0 +1,868 @@
#
# Copyright (c) 2024-2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Event models and data structures for Inworld Realtime API communication.
Based on Inworld's Realtime API documentation:
https://docs.inworld.ai/api-reference/realtimeAPI/realtime/realtime-websocket
"""
import json
import uuid
from typing import Any, Dict, List, Literal, Optional, Union
from pydantic import BaseModel, ConfigDict, Field
from pipecat.adapters.schemas.tools_schema import ToolsSchema
#
# Audio format configuration
#
# Inworld supports configurable sample rates for PCM audio
SUPPORTED_SAMPLE_RATES = Literal[8000, 16000, 24000, 32000, 44100, 48000]
class AudioFormat(BaseModel):
"""Base class for audio format configuration."""
type: str
class PCMAudioFormat(AudioFormat):
"""PCM audio format configuration with configurable sample rate.
Parameters:
type: Audio format type, always "audio/pcm".
rate: Sample rate in Hz. Defaults to 24000.
"""
type: Literal["audio/pcm"] = "audio/pcm"
rate: SUPPORTED_SAMPLE_RATES = 24000
class PCMUAudioFormat(AudioFormat):
"""PCMU (G.711 mu-law) audio format configuration.
Fixed at 8000 Hz sample rate.
Parameters:
type: Audio format type, always "audio/pcmu".
"""
type: Literal["audio/pcmu"] = "audio/pcmu"
class PCMAAudioFormat(AudioFormat):
"""PCMA (G.711 A-law) audio format configuration.
Fixed at 8000 Hz sample rate.
Parameters:
type: Audio format type, always "audio/pcma".
"""
type: Literal["audio/pcma"] = "audio/pcma"
#
# Turn detection configuration (lives inside audio.input)
#
class TurnDetection(BaseModel):
"""Server-side voice activity detection configuration.
Parameters:
type: Detection type. "server_vad" for standard VAD, "semantic_vad"
for semantic-based detection.
eagerness: How eagerly to detect end of turn. Options: "low", "medium", "high".
create_response: Whether to automatically create a response on turn end.
interrupt_response: Whether user speech interrupts the current response.
"""
type: Optional[Literal["server_vad", "semantic_vad"]] = "semantic_vad"
eagerness: Optional[str] = None
create_response: Optional[bool] = None
interrupt_response: Optional[bool] = None
class InputTranscription(BaseModel):
"""Configuration for input audio transcription.
Parameters:
model: The STT model to use for transcription.
"""
model: Optional[str] = None
#
# Audio configuration
#
class AudioInput(BaseModel):
"""Audio input configuration.
Parameters:
format: The format configuration for input audio.
transcription: Configuration for input audio transcription.
turn_detection: Configuration for turn detection.
"""
format: Optional[Union[PCMAudioFormat, PCMUAudioFormat, PCMAAudioFormat]] = None
transcription: Optional[InputTranscription] = None
turn_detection: Optional[TurnDetection] = None
class AudioOutput(BaseModel):
"""Audio output configuration.
Parameters:
format: The format configuration for output audio.
model: The TTS model to use (e.g. "inworld-tts-1.5-max").
voice: The voice ID to use (e.g. "Sarah", "Clive").
"""
format: Optional[Union[PCMAudioFormat, PCMUAudioFormat, PCMAAudioFormat]] = None
model: Optional[str] = None
voice: Optional[str] = None
class AudioConfiguration(BaseModel):
"""Audio configuration for input and output.
Parameters:
input: Configuration for input audio.
output: Configuration for output audio.
"""
input: Optional[AudioInput] = None
output: Optional[AudioOutput] = None
#
# Tool definitions
#
class FunctionTool(BaseModel):
"""Custom function tool configuration.
Parameters:
type: Tool type, always "function".
name: Name of the function.
description: Description of what the function does.
parameters: JSON schema for function parameters.
"""
type: Literal["function"] = "function"
name: str
description: str
parameters: Dict[str, Any]
# Union type for Inworld tools
InworldTool = Union[FunctionTool, Dict[str, Any]]
#
# Session properties
#
class SessionProperties(BaseModel):
"""Configuration properties for an Inworld Realtime session.
Parameters:
type: Session type, always "realtime".
model: The LLM model to use (e.g. "openai/gpt-4.1-nano").
instructions: System instructions for the assistant.
output_modalities: Output modalities (e.g. ["audio", "text"]).
audio: Audio configuration including input (transcription, turn detection)
and output (TTS model, voice).
tools: Available tools for the assistant.
"""
# Needed to support ToolSchema in tools field.
model_config = ConfigDict(arbitrary_types_allowed=True)
type: Optional[str] = "realtime"
model: Optional[str] = None
instructions: Optional[str] = None
temperature: Optional[float] = None
output_modalities: Optional[List[str]] = None
audio: Optional[AudioConfiguration] = None
# Tools can be ToolsSchema when provided by user, or list of dicts for API
tools: Optional[ToolsSchema | List[InworldTool]] = None
provider_data: Optional[Dict[str, Any]] = None
#
# Conversation items
#
class ItemContent(BaseModel):
"""Content within a conversation item.
Parameters:
type: Content type (input_text, input_audio, text, audio).
text: Text content for text-based items.
audio: Base64-encoded audio data for audio items.
transcript: Transcribed text for audio items.
"""
type: Literal["text", "audio", "input_text", "input_audio", "output_text", "output_audio"]
text: Optional[str] = None
audio: Optional[str] = None # base64-encoded audio
transcript: Optional[str] = None
class ConversationItem(BaseModel):
"""A conversation item in the realtime session.
Parameters:
id: Unique identifier for the item, auto-generated if not provided.
object: Object type identifier for the realtime API.
type: Item type (message, function_call, or function_call_output).
status: Current status of the item.
role: Speaker role for message items (user, assistant, or system).
content: Content list for message items.
call_id: Function call identifier for function_call items.
name: Function name for function_call items.
arguments: Function arguments as JSON string for function_call items.
output: Function output as JSON string for function_call_output items.
"""
id: str = Field(default_factory=lambda: str(uuid.uuid4().hex))
object: Optional[Literal["realtime.item"]] = None
type: Literal["message", "function_call", "function_call_output"]
status: Optional[Literal["completed", "in_progress", "incomplete"]] = None
role: Optional[Literal["user", "assistant", "system", "tool"]] = None
content: Optional[List[ItemContent]] = None
call_id: Optional[str] = None
name: Optional[str] = None
arguments: Optional[str] = None
output: Optional[str] = None
class RealtimeConversation(BaseModel):
"""A realtime conversation session.
Parameters:
id: Unique identifier for the conversation.
object: Object type identifier, always "realtime.conversation".
"""
id: str
object: Literal["realtime.conversation"]
class ResponseProperties(BaseModel):
"""Properties for configuring assistant responses.
Parameters:
modalities: Output modalities for the response (text, audio, or both).
"""
modalities: Optional[List[Literal["text", "audio"]]] = ["text", "audio"]
#
# Error class
#
class RealtimeError(BaseModel):
"""Error information from the realtime API.
Parameters:
type: Error type identifier.
code: Specific error code.
message: Human-readable error message.
param: Parameter name that caused the error, if applicable.
event_id: Event ID associated with the error, if applicable.
"""
type: Optional[str] = None
code: Optional[str] = ""
message: str
param: Optional[str] = None
event_id: Optional[str] = None
#
# Client Events (sent to Inworld)
#
class ClientEvent(BaseModel):
"""Base class for client events sent to the realtime API.
Parameters:
event_id: Unique identifier for the event, auto-generated if not provided.
"""
event_id: str = Field(default_factory=lambda: str(uuid.uuid4()))
class SessionUpdateEvent(ClientEvent):
"""Event to update session properties.
Parameters:
type: Event type, always "session.update".
session: Updated session properties.
"""
type: Literal["session.update"] = "session.update"
session: SessionProperties
class InputAudioBufferAppendEvent(ClientEvent):
"""Event to append audio data to the input buffer.
Parameters:
type: Event type, always "input_audio_buffer.append".
audio: Base64-encoded audio data to append.
"""
type: Literal["input_audio_buffer.append"] = "input_audio_buffer.append"
audio: str # base64-encoded audio
class InputAudioBufferCommitEvent(ClientEvent):
"""Event to commit the current input audio buffer.
Used when turn_detection is null (manual mode).
Parameters:
type: Event type, always "input_audio_buffer.commit".
"""
type: Literal["input_audio_buffer.commit"] = "input_audio_buffer.commit"
class InputAudioBufferClearEvent(ClientEvent):
"""Event to clear the input audio buffer.
Parameters:
type: Event type, always "input_audio_buffer.clear".
"""
type: Literal["input_audio_buffer.clear"] = "input_audio_buffer.clear"
class ConversationItemCreateEvent(ClientEvent):
"""Event to create a new conversation item.
Parameters:
type: Event type, always "conversation.item.create".
previous_item_id: ID of the item to insert after, if any.
item: The conversation item to create.
"""
type: Literal["conversation.item.create"] = "conversation.item.create"
previous_item_id: Optional[str] = None
item: ConversationItem
class ResponseCreateEvent(ClientEvent):
"""Event to create a new assistant response.
Parameters:
type: Event type, always "response.create".
response: Optional response configuration properties.
"""
type: Literal["response.create"] = "response.create"
response: Optional[ResponseProperties] = None
class ResponseCancelEvent(ClientEvent):
"""Event to cancel the current assistant response.
Parameters:
type: Event type, always "response.cancel".
"""
type: Literal["response.cancel"] = "response.cancel"
#
# Server Events (received from Inworld)
#
class ServerEvent(BaseModel):
"""Base class for server events received from the realtime API.
Parameters:
event_id: Unique identifier for the event.
type: Type of the server event.
"""
model_config = ConfigDict(arbitrary_types_allowed=True)
event_id: str
type: str
class SessionCreatedEvent(ServerEvent):
"""Event indicating a session has been created.
This is the first event received after connecting.
Parameters:
type: Event type, always "session.created".
session: The initial session properties.
"""
type: Literal["session.created"]
session: Optional[SessionProperties] = None
class SessionUpdatedEvent(ServerEvent):
"""Event indicating a session has been updated.
Parameters:
type: Event type, always "session.updated".
session: The updated session properties.
"""
type: Literal["session.updated"]
session: SessionProperties
class ConversationCreated(ServerEvent):
"""Event indicating a conversation has been created.
This is the first message received after connecting.
Parameters:
type: Event type, always "conversation.created".
conversation: The created conversation.
"""
type: Literal["conversation.created"]
conversation: RealtimeConversation
class ConversationItemAdded(ServerEvent):
"""Event indicating a conversation item has been added.
Parameters:
type: Event type, always "conversation.item.added".
previous_item_id: ID of the previous item, if any.
item: The added conversation item.
"""
type: Literal["conversation.item.added"]
previous_item_id: Optional[str] = None
item: ConversationItem
class ConversationItemInputAudioTranscriptionCompleted(ServerEvent):
"""Event indicating input audio transcription is complete.
Parameters:
type: Event type, always "conversation.item.input_audio_transcription.completed".
item_id: ID of the conversation item that was transcribed.
transcript: Complete transcription text.
"""
type: Literal["conversation.item.input_audio_transcription.completed"]
item_id: str
transcript: str
class ConversationItemInputAudioTranscriptionDelta(ServerEvent):
"""Event containing incremental input audio transcription.
Parameters:
type: Event type, always "conversation.item.input_audio_transcription.delta".
item_id: ID of the conversation item being transcribed.
content_index: Index of the content part.
delta: Incremental transcription text.
"""
type: Literal["conversation.item.input_audio_transcription.delta"]
item_id: str
content_index: Optional[int] = None
delta: str
class InputAudioBufferSpeechStarted(ServerEvent):
"""Event indicating speech has started in the input audio buffer.
Only sent when turn_detection is "server_vad".
Parameters:
type: Event type, always "input_audio_buffer.speech_started".
item_id: ID of the associated conversation item.
"""
type: Literal["input_audio_buffer.speech_started"]
item_id: str
class InputAudioBufferSpeechStopped(ServerEvent):
"""Event indicating speech has stopped in the input audio buffer.
Only sent when turn_detection is "server_vad".
Parameters:
type: Event type, always "input_audio_buffer.speech_stopped".
item_id: ID of the associated conversation item.
"""
type: Literal["input_audio_buffer.speech_stopped"]
item_id: str
class InputAudioBufferCommitted(ServerEvent):
"""Event indicating the input audio buffer has been committed.
Parameters:
type: Event type, always "input_audio_buffer.committed".
previous_item_id: ID of the previous item, if any.
item_id: ID of the committed conversation item.
"""
type: Literal["input_audio_buffer.committed"]
previous_item_id: Optional[str] = None
item_id: str
class InputAudioBufferCleared(ServerEvent):
"""Event indicating the input audio buffer has been cleared.
Parameters:
type: Event type, always "input_audio_buffer.cleared".
"""
type: Literal["input_audio_buffer.cleared"]
class ResponseCreated(ServerEvent):
"""Event indicating an assistant response has been created.
Parameters:
type: Event type, always "response.created".
response: The created response object.
"""
type: Literal["response.created"]
response: "Response"
class ResponseOutputItemAdded(ServerEvent):
"""Event indicating an output item has been added to a response.
Parameters:
type: Event type, always "response.output_item.added".
response_id: ID of the response.
output_index: Index of the output item.
item: The added conversation item.
"""
type: Literal["response.output_item.added"]
response_id: str
output_index: int
item: ConversationItem
class ResponseAudioTranscriptDelta(ServerEvent):
"""Event containing incremental audio transcript from a response.
Parameters:
type: Event type, always "response.output_audio_transcript.delta".
response_id: ID of the response.
item_id: ID of the conversation item.
delta: Incremental transcript text.
"""
type: Literal["response.output_audio_transcript.delta"]
response_id: str
item_id: str
delta: str
class ResponseAudioTranscriptDone(ServerEvent):
"""Event indicating audio transcript is complete.
Parameters:
type: Event type, always "response.output_audio_transcript.done".
response_id: ID of the response.
item_id: ID of the conversation item.
"""
type: Literal["response.output_audio_transcript.done"]
response_id: str
item_id: str
class ResponseAudioDelta(ServerEvent):
"""Event containing incremental audio data from a response.
Parameters:
type: Event type, always "response.output_audio.delta".
response_id: ID of the response.
item_id: ID of the conversation item.
output_index: Index of the output item.
content_index: Index of the content part.
delta: Base64-encoded incremental audio data.
"""
type: Literal["response.output_audio.delta"]
response_id: str
item_id: str
output_index: int
content_index: int
delta: str # base64-encoded audio
class ResponseAudioDone(ServerEvent):
"""Event indicating audio content is complete.
Parameters:
type: Event type, always "response.output_audio.done".
response_id: ID of the response.
item_id: ID of the conversation item.
"""
type: Literal["response.output_audio.done"]
response_id: str
item_id: str
class ResponseFunctionCallArgumentsDelta(ServerEvent):
"""Event containing incremental function call arguments.
Parameters:
type: Event type, always "response.function_call_arguments.delta".
response_id: ID of the response.
item_id: ID of the conversation item.
call_id: ID of the function call.
delta: Incremental function arguments as JSON.
previous_item_id: ID of the previous item, if any.
"""
type: Literal["response.function_call_arguments.delta"]
response_id: Optional[str] = None
item_id: Optional[str] = None
call_id: str
delta: str
previous_item_id: Optional[str] = None
class ResponseFunctionCallArgumentsDone(ServerEvent):
"""Event indicating function call arguments are complete.
Parameters:
type: Event type, always "response.function_call_arguments.done".
call_id: ID of the function call.
name: Name of the function being called. Optional — Inworld may omit
this; the name can be resolved from the tracked function call item.
arguments: Complete function arguments as JSON string.
"""
type: Literal["response.function_call_arguments.done"]
call_id: str
name: Optional[str] = None
arguments: str
class Usage(BaseModel):
"""Token usage statistics for a response.
Parameters:
total_tokens: Total number of tokens used.
input_tokens: Number of input tokens used.
output_tokens: Number of output tokens used.
"""
total_tokens: Optional[int] = None
input_tokens: Optional[int] = None
output_tokens: Optional[int] = None
class Response(BaseModel):
"""A complete assistant response.
Parameters:
id: Unique identifier for the response.
object: Object type, always "realtime.response".
status: Current status of the response.
output: List of conversation items in the response.
usage: Token usage statistics for the response.
"""
id: str
object: Literal["realtime.response"]
status: Literal["completed", "in_progress", "incomplete", "cancelled", "failed"]
status_details: Optional[Any] = None
output: List[ConversationItem]
usage: Optional[Usage] = None
class ResponseDone(ServerEvent):
"""Event indicating an assistant response is complete.
Parameters:
type: Event type, always "response.done".
response: The completed response object.
usage: Token usage (also available at top level).
"""
type: Literal["response.done"]
response: Response
usage: Optional[Usage] = None
class ResponseOutputItemDone(ServerEvent):
"""Event indicating an output item is complete.
Parameters:
type: Event type, always "response.output_item.done".
response_id: ID of the response.
output_index: Index of the output item.
item: The completed conversation item.
"""
type: Literal["response.output_item.done"]
response_id: str
output_index: int
item: ConversationItem
class ContentPart(BaseModel):
"""A content part within a response.
Parameters:
type: Type of the content part (audio, text).
transcript: Transcript text if applicable.
"""
type: str
transcript: Optional[str] = None
class ResponseContentPartAdded(ServerEvent):
"""Event indicating a content part has been added to a response.
Parameters:
type: Event type, always "response.content_part.added".
response_id: ID of the response.
item_id: ID of the conversation item.
content_index: Index of the content part.
output_index: Index of the output item.
part: The added content part.
"""
type: Literal["response.content_part.added"]
response_id: str
item_id: str
content_index: int
output_index: int
part: ContentPart
class ResponseContentPartDone(ServerEvent):
"""Event indicating a content part is complete.
Parameters:
type: Event type, always "response.content_part.done".
response_id: ID of the response.
item_id: ID of the conversation item.
content_index: Index of the content part.
output_index: Index of the output item.
"""
type: Literal["response.content_part.done"]
response_id: str
item_id: str
content_index: int
output_index: int
class PingEvent(ServerEvent):
"""Keep-alive ping event from the server.
Parameters:
type: Event type, always "ping".
timestamp: Server timestamp in milliseconds.
"""
type: Literal["ping"]
timestamp: int
class ErrorEvent(ServerEvent):
"""Event indicating an error occurred.
Parameters:
type: Event type, always "error".
error: Error details.
"""
type: Literal["error"]
error: RealtimeError
#
# Event parsing
#
_server_event_types = {
"error": ErrorEvent,
"ping": PingEvent,
"session.created": SessionCreatedEvent,
"session.updated": SessionUpdatedEvent,
"conversation.created": ConversationCreated,
"conversation.item.added": ConversationItemAdded,
"conversation.item.input_audio_transcription.completed": ConversationItemInputAudioTranscriptionCompleted,
"conversation.item.input_audio_transcription.delta": ConversationItemInputAudioTranscriptionDelta,
"input_audio_buffer.speech_started": InputAudioBufferSpeechStarted,
"input_audio_buffer.speech_stopped": InputAudioBufferSpeechStopped,
"input_audio_buffer.committed": InputAudioBufferCommitted,
"input_audio_buffer.cleared": InputAudioBufferCleared,
"response.created": ResponseCreated,
"response.output_item.added": ResponseOutputItemAdded,
"response.output_item.done": ResponseOutputItemDone,
"response.content_part.added": ResponseContentPartAdded,
"response.content_part.done": ResponseContentPartDone,
"response.output_audio_transcript.delta": ResponseAudioTranscriptDelta,
"response.output_audio_transcript.done": ResponseAudioTranscriptDone,
"response.output_audio.delta": ResponseAudioDelta,
"response.output_audio.done": ResponseAudioDone,
"response.function_call_arguments.delta": ResponseFunctionCallArgumentsDelta,
"response.function_call_arguments.done": ResponseFunctionCallArgumentsDone,
"response.done": ResponseDone,
}
def parse_server_event(data: str):
"""Parse a server event from JSON string.
Args:
data: JSON string containing the server event.
Returns:
Parsed server event object of the appropriate type.
Raises:
Exception: If the event type is unimplemented or parsing fails.
"""
try:
event = json.loads(data)
event_type = event["type"]
if event_type not in _server_event_types:
raise Exception(f"Unimplemented server event type: {event_type}")
return _server_event_types[event_type].model_validate(event)
except Exception as e:
raise Exception(f"{e} \n\n{data}")

1039
src/pipecat/services/inworld/realtime/llm.py Normal file
View File

File diff suppressed because it is too large Load Diff

27
tests/test_function_calling_adapters.py
View File

@@ -13,6 +13,7 @@ from pipecat.adapters.schemas.tools_schema import AdapterType, ToolsSchema
from pipecat.adapters.services.anthropic_adapter import AnthropicLLMAdapter
from pipecat.adapters.services.bedrock_adapter import AWSBedrockLLMAdapter
from pipecat.adapters.services.gemini_adapter import GeminiLLMAdapter
from pipecat.adapters.services.inworld_realtime_adapter import InworldRealtimeLLMAdapter
from pipecat.adapters.services.open_ai_adapter import OpenAILLMAdapter
from pipecat.adapters.services.open_ai_realtime_adapter import OpenAIRealtimeLLMAdapter
@@ -143,6 +144,32 @@ class TestFunctionAdapters(unittest.TestCase):
]
assert OpenAIRealtimeLLMAdapter().to_provider_tools_format(self.tools_def) == expected
def test_inworld_realtime_adapter(self):
"""Test Inworld Realtime adapter format transformation."""
expected = [
{
"type": "function",
"name": "get_weather",
"description": "Get the weather in a given location",
"parameters": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "The city, e.g. San Francisco",
},
"format": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
"description": "The temperature unit to use.",
},
},
"required": ["location", "format"],
},
}
]
assert InworldRealtimeLLMAdapter().to_provider_tools_format(self.tools_def) == expected
def test_gemini_adapter_with_custom_tools(self):
"""Test Gemini adapter format transformation."""
search_tool = {"google_search": {}}

199
tests/test_settings.py
View File

@@ -10,6 +10,8 @@ from unittest.mock import patch
from pipecat.services.deepgram.sagemaker.stt import DeepgramSageMakerSTTSettings
from pipecat.services.deepgram.stt import DeepgramSTTService, DeepgramSTTSettings
from pipecat.services.inworld.realtime import events as inworld_events
from pipecat.services.inworld.realtime.llm import InworldRealtimeLLMSettings
from pipecat.services.openai.realtime import events
from pipecat.services.openai.realtime.llm import OpenAIRealtimeLLMSettings
from pipecat.services.settings import (
@@ -979,3 +981,200 @@ class TestGrokRealtimeSettingsFromMapping:
assert store.session_properties.instructions == "Be concise."
assert store.session_properties.voice == "Eve"
assert store.system_instruction == "Be concise."
# ---------------------------------------------------------------------------
# InworldRealtimeLLMSettings: apply_update with bidirectional sync
# ---------------------------------------------------------------------------
class TestInworldRealtimeSettingsApplyUpdate:
def _make_store(self, **kwargs) -> InworldRealtimeLLMSettings:
"""Helper to build a store-mode InworldRealtimeLLMSettings."""
defaults = dict(
model="openai/gpt-4.1-nano",
system_instruction=None,
temperature=None,
max_tokens=None,
top_p=None,
top_k=None,
frequency_penalty=None,
presence_penalty=None,
seed=None,
filter_incomplete_user_turns=False,
user_turn_completion_config=None,
session_properties=inworld_events.SessionProperties(),
)
defaults.update(kwargs)
return InworldRealtimeLLMSettings(**defaults)
def test_top_level_model_syncs_to_sp(self):
"""Updating top-level model should propagate to session_properties.model."""
store = self._make_store()
delta = InworldRealtimeLLMSettings(model="openai/gpt-4.1")
changed = store.apply_update(delta)
assert "model" in changed
assert store.model == "openai/gpt-4.1"
assert store.session_properties.model == "openai/gpt-4.1"
def test_top_level_system_instruction_syncs_to_sp(self):
"""Updating top-level system_instruction should propagate to session_properties.instructions."""
store = self._make_store()
delta = InworldRealtimeLLMSettings(system_instruction="Be helpful.")
changed = store.apply_update(delta)
assert "system_instruction" in changed
assert store.system_instruction == "Be helpful."
assert store.session_properties.instructions == "Be helpful."
def test_sp_replaces_wholesale(self):
"""session_properties in delta replaces the entire stored SP."""
store = self._make_store(
session_properties=inworld_events.SessionProperties(
output_modalities=["audio", "text"],
instructions="Old instructions.",
),
system_instruction="Old instructions.",
)
new_sp = inworld_events.SessionProperties(output_modalities=["text"])
delta = InworldRealtimeLLMSettings(session_properties=new_sp)
changed = store.apply_update(delta)
assert "session_properties" in changed
assert store.session_properties.output_modalities == ["text"]
# model is synced from top-level
assert store.session_properties.model == "openai/gpt-4.1-nano"
def test_sp_model_syncs_to_top_level(self):
"""session_properties.model should sync to top-level model."""
store = self._make_store()
new_sp = inworld_events.SessionProperties(model="openai/gpt-4.1")
delta = InworldRealtimeLLMSettings(session_properties=new_sp)
changed = store.apply_update(delta)
assert "model" in changed
assert store.model == "openai/gpt-4.1"
assert store.session_properties.model == "openai/gpt-4.1"
def test_sp_instructions_syncs_to_top_level(self):
"""session_properties.instructions should sync to top-level system_instruction."""
store = self._make_store()
new_sp = inworld_events.SessionProperties(instructions="New instructions.")
delta = InworldRealtimeLLMSettings(session_properties=new_sp)
changed = store.apply_update(delta)
assert "system_instruction" in changed
assert store.system_instruction == "New instructions."
assert store.session_properties.instructions == "New instructions."
def test_top_level_model_takes_precedence_over_sp_model(self):
"""When both model and session_properties.model are in the delta, top-level wins."""
store = self._make_store()
new_sp = inworld_events.SessionProperties(model="sp-model")
delta = InworldRealtimeLLMSettings(model="top-model", session_properties=new_sp)
store.apply_update(delta)
assert store.model == "top-model"
assert store.session_properties.model == "top-model"
def test_top_level_si_takes_precedence_over_sp_instructions(self):
"""When both system_instruction and SP.instructions are in delta, top-level wins."""
store = self._make_store()
new_sp = inworld_events.SessionProperties(instructions="sp instructions")
delta = InworldRealtimeLLMSettings(
system_instruction="top instructions",
session_properties=new_sp,
)
store.apply_update(delta)
assert store.system_instruction == "top instructions"
assert store.session_properties.instructions == "top instructions"
def test_non_synced_field_update_does_not_affect_sp(self):
"""Updating a non-synced field like temperature shouldn't touch session_properties."""
store = self._make_store(
session_properties=inworld_events.SessionProperties(instructions="Keep me."),
system_instruction="Keep me.",
)
original_sp = store.session_properties
delta = InworldRealtimeLLMSettings(temperature=0.5)
changed = store.apply_update(delta)
assert "temperature" in changed
assert store.temperature == 0.5
# SP should be untouched (same object)
assert store.session_properties is original_sp
assert store.session_properties.instructions == "Keep me."
# ---------------------------------------------------------------------------
# InworldRealtimeLLMSettings: from_mapping
# ---------------------------------------------------------------------------
class TestInworldRealtimeSettingsFromMapping:
def test_sp_keys_route_to_session_properties(self):
"""SessionProperties fields (instructions, output_modalities) route into nested SP."""
delta = InworldRealtimeLLMSettings.from_mapping(
{"instructions": "Be concise.", "output_modalities": ["text"]}
)
assert is_given(delta.session_properties)
assert delta.session_properties.instructions == "Be concise."
assert delta.session_properties.output_modalities == ["text"]
def test_model_routes_to_top_level(self):
"""model should go to the top-level field, not session_properties."""
delta = InworldRealtimeLLMSettings.from_mapping({"model": "openai/gpt-4.1"})
assert delta.model == "openai/gpt-4.1"
# No session_properties should be created since no SP keys were present
assert not is_given(delta.session_properties)
def test_unknown_keys_go_to_extra(self):
"""Unrecognized keys should land in extra."""
delta = InworldRealtimeLLMSettings.from_mapping({"unknown_param": 42})
assert not is_given(delta.model)
assert not is_given(delta.session_properties)
assert delta.extra == {"unknown_param": 42}
def test_mixed_keys(self):
"""model + SP keys + unknown keys are routed correctly."""
delta = InworldRealtimeLLMSettings.from_mapping(
{
"model": "openai/gpt-4.1",
"instructions": "Be helpful.",
"unknown": "val",
}
)
assert delta.model == "openai/gpt-4.1"
assert is_given(delta.session_properties)
assert delta.session_properties.instructions == "Be helpful."
assert delta.extra == {"unknown": "val"}
def test_roundtrip_from_mapping_apply_update(self):
"""Simulate dict-style update: from_mapping -> apply_update."""
store = InworldRealtimeLLMSettings(
model="openai/gpt-4.1-nano",
system_instruction=None,
temperature=None,
max_tokens=None,
top_p=None,
top_k=None,
frequency_penalty=None,
presence_penalty=None,
seed=None,
filter_incomplete_user_turns=False,
user_turn_completion_config=None,
session_properties=inworld_events.SessionProperties(),
)
raw = {"instructions": "Be concise.", "output_modalities": ["text"]}
delta = InworldRealtimeLLMSettings.from_mapping(raw)
changed = store.apply_update(delta)
assert "session_properties" in changed
assert store.session_properties.instructions == "Be concise."
assert store.session_properties.output_modalities == ["text"]
assert store.system_instruction == "Be concise."

4
uv.lock generated
View File

@@ -4248,6 +4248,9 @@ heygen = [
hume = [
{ name = "hume" },
]
inworld = [
{ name = "websockets" },
]
koala = [
{ name = "pvkoala" },
]
@@ -4470,6 +4473,7 @@ requires-dist = [
{ name = "pipecat-ai", extras = ["websockets-base"], marker = "extra == 'google'" },
{ name = "pipecat-ai", extras = ["websockets-base"], marker = "extra == 'gradium'" },
{ name = "pipecat-ai", extras = ["websockets-base"], marker = "extra == 'heygen'" },
{ name = "pipecat-ai", extras = ["websockets-base"], marker = "extra == 'inworld'" },
{ name = "pipecat-ai", extras = ["websockets-base"], marker = "extra == 'lmnt'" },
{ name = "pipecat-ai", extras = ["websockets-base"], marker = "extra == 'neuphonic'" },
{ name = "pipecat-ai", extras = ["websockets-base"], marker = "extra == 'openai'" },