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

Merge branch 'main' into add-cartesia-generation-config

Browse Source
This commit is contained in:
Roshan
2025-10-29 18:31:09 -07:00
committed by GitHub
parent abf34bcccf 56eb633390
commit e86c546a1a
34 changed files with 1445 additions and 440 deletions
Download Patch File Download Diff File Expand all files Collapse all files

169
CHANGELOG.md
View File

@@ -15,8 +15,155 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
and `emotion` (60+ options) parameters for fine-grained speech generation
control.
- Expanded support for univeral `LLMContext` to `OpenAIRealtimeLLMService`.
As a reminder, the context-setup pattern when using `LLMContext` is:
```python
context = LLMContext(messages, tools)
context_aggregator = LLMContextAggregatorPair(
context,
# This part is `OpenAIRealtimeLLMService`-specific.
# `expect_stripped_words=False` needed when OpenAI Realtime used with
# "audio" modality (the default).
assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False),
)
```
(Note that even though `OpenAIRealtimeLLMService` now supports the universal
`LLMContext`, it is not meant to be swapped out for another LLM service at
runtime with `LLMSwitcher`.)
Note: `TranscriptionFrame`s and `InterimTranscriptionFrame`s now go upstream
from `OpenAIRealtimeLLMService`, so if you're using `TranscriptProcessor`,
say, you'll want to adjust accordingly:
```python
pipeline = Pipeline(
[
transport.input(),
context_aggregator.user(),
# BEFORE
llm,
transcript.user(),
# AFTER
transcript.user(),
llm,
transport.output(),
transcript.assistant(),
context_aggregator.assistant(),
]
)
```
Also worth noting: whether or not you use the new context-setup pattern with
`OpenAIRealtimeLLMService`, some types have changed under the hood:
```python
## BEFORE:
# Context aggregator type
context_aggregator: OpenAIContextAggregatorPair
# Context frame type
frame: OpenAILLMContextFrame
# Context type
context: OpenAIRealtimeLLMContext
# or
context: OpenAILLMContext
## AFTER:
# Context aggregator type
context_aggregator: LLMContextAggregatorPair
# Context frame type
frame: LLMContextFrame
# Context type
context: LLMContext
```
Also note that `RealtimeMessagesUpdateFrame` and
`RealtimeFunctionCallResultFrame` have been deprecated, since they're no
longer used by `OpenAIRealtimeLLMService`. OpenAI Realtime now works more
like other LLM services in Pipecat, relying on updates to its context, pushed
by context aggregators, to update its internal state. Listen for
`LLMContextFrame`s for context updates.
Finally, `LLMTextFrame`s are no longer pushed from `OpenAIRealtimeLLMService`
when it's configured with `output_modalities=['audio']`. If you need
to process its output, listen for `TTSTextFrame`s instead.
- Expanded support for universal `LLMContext` to `GeminiLiveLLMService`.
As a reminder, the context-setup pattern when using `LLMContext` is:
```python
context = LLMContext(messages, tools)
context_aggregator = LLMContextAggregatorPair(
context,
# This part is `GeminiLiveLLMService`-specific.
# `expect_stripped_words=False` needed when Gemini Live used with AUDIO
# modality (the default).
assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False),
)
```
(Note that even though `GeminiLiveLLMService` now supports the universal
`LLMContext`, it is not meant to be swapped out for another LLM service at
runtime with `LLMSwitcher`.)
Worth noting: whether or not you use the new context-setup pattern with
`GeminiLiveLLMService`, some types have changed under the hood:
```python
## BEFORE:
# Context aggregator type
context_aggregator: GeminiLiveContextAggregatorPair
# Context frame type
frame: OpenAILLMContextFrame
# Context type
context: GeminiLiveLLMContext
# or
context: OpenAILLMContext
## AFTER:
# Context aggregator type
context_aggregator: LLMContextAggregatorPair
# Context frame type
frame: LLMContextFrame
# Context type
context: LLMContext
```
Also note that `LLMTextFrame`s are no longer pushed from `GeminiLiveLLMService`
when it's configured with `modalities=GeminiModalities.AUDIO`. If you need
to process its output, listen for `TTSTextFrame`s instead.
### Changed
- `DailyTransport` triggers `on_error` event if transcription can't be started
or stopped.
- `DailyTransport` updates: `start_dialout()` now returns two values:
`session_id` and `error`. `start_recording()` now returns two values:
`stream_id` and `error`.
- Updated `daily-python` to 0.21.0.
- `SimliVideoService` now accepts `api_key` and `face_id` parameters directly,
with optional `params` for `max_session_length` and `max_idle_time`
configuration, aligning with other Pipecat service patterns.
- Updated the default model to `sonic-3` for `CartesiaTTSService` and
`CartesiaHttpTTSService`.
@@ -26,6 +173,19 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
- Upgraded `aws_sdk_bedrock_runtime` to v0.1.1 to resolve potential CPU issues
when running `AWSNovaSonicLLMService`.
### Deprecated
- The `send_transcription_frames` argument to `OpenAIRealtimeLLMService` is
deprecated. Transcription frames are now always sent. They go upstream, to be
handled by the user context aggregator. See "Added" section for details.
- Types in `pipecat.services.openai.realtime.context` and
`pipecat.services.openai.realtime.frames` are deprecated, as they're no
longer used by `OpenAIRealtimeLLMService`. See "Added" section for details.
- `SimliVideoService` `simli_config` parameter is deprecated. Use `api_key` and
`face_id` parameters instead.
### Removed
- Removed the `aiohttp_session` arg from `SarvamTTSService` as it's no longer
@@ -42,6 +202,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
- Fixed an issue in `ServiceSwitcher` where the `STTService`s would result in
all STT services producing `TranscriptionFrame`s.
- Fixed an issue in `HumeTTSService` that was only using Octave 2, which does not support the `description` field. Now, if a description is provided, it switches to Octave 1.
## [0.0.91] - 2025-10-21
### Added
@@ -63,7 +225,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
(Note that even though `AWSNovaSonicLLMService` now supports the universal
`LLMContext`, it is not meant to be swapped out for another LLM service at
runtime.)
runtime with `LLMSwitcher`.)
Worth noting: whether or not you use the new context-setup pattern with
`AWSNovaSonicLLMService`, some types have changed under the hood:
@@ -142,8 +304,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
deprecated. Transcription frames are now always sent. They go upstream, to be
handled by the user context aggregator. See "Added" section for details.
- Types in `pipecat.services.aws.nova_sonic.context` have been deprecated due
to changes to support `LLMContext`. See "Changed" section for details.
- Types in `pipecat.services.aws.nova_sonic.context` are deprecated, as they're
no longer used by `AWSNovaSonicLLMService`. See "Added" section for
details.
### Fixed

49
examples/foundational/19-openai-realtime.py
View File

@@ -5,6 +5,7 @@
#
import asyncio
import os
from datetime import datetime
@@ -14,12 +15,14 @@ from loguru import logger
from pipecat.adapters.schemas.function_schema import FunctionSchema
from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.frames.frames import LLMRunFrame, TranscriptionMessage
from pipecat.frames.frames import LLMRunFrame, LLMSetToolsFrame, TranscriptionMessage
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.openai_llm_context import OpenAILLMContext
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response import LLMAssistantAggregatorParams
from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
from pipecat.processors.transcript_processor import TranscriptProcessor
from pipecat.runner.types import RunnerArguments
from pipecat.runner.utils import create_transport
@@ -52,6 +55,18 @@ async def fetch_weather_from_api(params: FunctionCallParams):
)
async def get_news(params: FunctionCallParams):
await params.result_callback(
{
"news": [
"Massive UFO currently hovering above New York City",
"Stock markets reach all-time highs",
"Living dinosaur species discovered in the Amazon rainforest",
],
}
)
async def fetch_restaurant_recommendation(params: FunctionCallParams):
await params.result_callback({"name": "The Golden Dragon"})
@@ -73,6 +88,13 @@ weather_function = FunctionSchema(
required=["location", "format"],
)
get_news_function = FunctionSchema(
name="get_news",
description="Get the current news.",
properties={},
required=[],
)
restaurant_function = FunctionSchema(
name="get_restaurant_recommendation",
description="Get a restaurant recommendation",
@@ -140,10 +162,6 @@ even if you're asked about them.
You are participating in a voice conversation. Keep your responses concise, short, and to the point
unless specifically asked to elaborate on a topic.
You have access to the following tools:
- get_current_weather: Get the current weather for a given location.
- get_restaurant_recommendation: Get a restaurant recommendation for a given location.
Remember, your responses should be short. Just one or two sentences, usually. Respond in English.""",
)
@@ -157,25 +175,31 @@ Remember, your responses should be short. Just one or two sentences, usually. Re
# llm.register_function(None, fetch_weather_from_api)
llm.register_function("get_current_weather", fetch_weather_from_api)
llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation)
llm.register_function("get_news", get_news)
transcript = TranscriptProcessor()
# Create a standard OpenAI LLM context object using the normal messages format. The
# OpenAIRealtimeLLMService will convert this internally to messages that the
# openai WebSocket API can understand.
context = OpenAILLMContext(
context = LLMContext(
[{"role": "user", "content": "Say hello!"}],
tools,
)
context_aggregator = llm.create_context_aggregator(context)
context_aggregator = LLMContextAggregatorPair(
context,
# `expect_stripped_words=False` needed when OpenAI Realtime used with
# "audio" modality (the default)
assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False),
)
pipeline = Pipeline(
[
transport.input(), # Transport user input
context_aggregator.user(),
transcript.user(), # LLM pushes TranscriptionFrames upstream
llm, # LLM
transcript.user(), # Placed after the LLM, as LLM pushes TranscriptionFrames downstream
transport.output(), # Transport bot output
transcript.assistant(), # After the transcript output, to time with the audio output
context_aggregator.assistant(),
@@ -198,6 +222,13 @@ Remember, your responses should be short. Just one or two sentences, usually. Re
# Kick off the conversation.
await task.queue_frames([LLMRunFrame()])
# Add a new tool at runtime after a delay.
await asyncio.sleep(15)
new_tools = ToolsSchema(
standard_tools=[weather_function, restaurant_function, get_news_function]
)
await task.queue_frames([LLMSetToolsFrame(tools=new_tools)])
@transport.event_handler("on_client_disconnected")
async def on_client_disconnected(transport, client):
logger.info(f"Client disconnected")

15
examples/foundational/19a-azure-realtime.py
View File

@@ -18,7 +18,9 @@ from pipecat.frames.frames import LLMRunFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response import LLMAssistantAggregatorParams
from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
from pipecat.runner.types import RunnerArguments
from pipecat.runner.utils import create_transport
from pipecat.services.azure.realtime.llm import AzureRealtimeLLMService
@@ -155,10 +157,10 @@ Remember, your responses should be short. Just one or two sentences, usually. Re
llm.register_function("get_current_weather", fetch_weather_from_api)
llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation)
# Create a standard OpenAI LLM context object using the normal messages format. The
# Create a standard LLM context object using the normal messages format. The
# OpenAIRealtimeBetaLLMService will convert this internally to messages that the
# openai WebSocket API can understand.
context = OpenAILLMContext(
context = LLMContext(
[{"role": "user", "content": "Say hello!"}],
# [{"role": "user", "content": [{"type": "text", "text": "Say hello!"}]}],
# [
@@ -173,7 +175,12 @@ Remember, your responses should be short. Just one or two sentences, usually. Re
tools,
)
context_aggregator = llm.create_context_aggregator(context)
context_aggregator = LLMContextAggregatorPair(
context,
# `expect_stripped_words=False` needed when OpenAI Realtime used with
# "audio" modality (the default)
assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False),
)
pipeline = Pipeline(
[

9
examples/foundational/19b-openai-realtime-text.py
View File

@@ -18,7 +18,8 @@ from pipecat.frames.frames import LLMRunFrame, TranscriptionMessage
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
from pipecat.processors.transcript_processor import TranscriptProcessor
from pipecat.runner.types import RunnerArguments
from pipecat.runner.utils import create_transport
@@ -169,20 +170,20 @@ Remember, your responses should be short. Just one or two sentences, usually. Re
# Create a standard OpenAI LLM context object using the normal messages format. The
# OpenAIRealtimeLLMService will convert this internally to messages that the
# openai WebSocket API can understand.
context = OpenAILLMContext(
context = LLMContext(
[{"role": "user", "content": "Say hello!"}],
tools,
)
context_aggregator = llm.create_context_aggregator(context)
context_aggregator = LLMContextAggregatorPair(context)
pipeline = Pipeline(
[
transport.input(), # Transport user input
context_aggregator.user(),
transcript.user(), # LLM pushes TranscriptionFrames upstream
llm, # LLM
tts, # TTS
transcript.user(), # Placed after the LLM, as LLM pushes TranscriptionFrames downstream
transport.output(), # Transport bot output
transcript.assistant(), # After the transcript output, to time with the audio output
context_aggregator.assistant(),

89
examples/foundational/20b-persistent-context-openai-realtime.py
View File

@@ -13,14 +13,15 @@ from datetime import datetime
from dotenv import load_dotenv
from loguru import logger
from pipecat.adapters.schemas.function_schema import FunctionSchema
from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.frames.frames import LLMRunFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.openai_llm_context import (
OpenAILLMContext,
)
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
from pipecat.runner.types import RunnerArguments
from pipecat.runner.utils import create_transport
from pipecat.services.deepgram.stt import DeepgramSTTService
@@ -69,11 +70,11 @@ async def save_conversation(params: FunctionCallParams):
timestamp = datetime.now().strftime("%Y-%m-%d_%H:%M:%S")
filename = f"{BASE_FILENAME}{timestamp}.json"
logger.debug(
f"writing conversation to {filename}\n{json.dumps(params.context.messages, indent=4)}"
f"writing conversation to {filename}\n{json.dumps(params.context.get_messages(), indent=4)}"
)
try:
with open(filename, "w") as file:
messages = params.context.get_messages_for_persistent_storage()
messages = params.context.get_messages()
# remove the last message, which is the instruction we just gave to save the conversation
messages.pop()
json.dump(messages, file, indent=2)
@@ -90,6 +91,10 @@ async def load_conversation(params: FunctionCallParams):
with open(filename, "r") as file:
params.context.set_messages(json.load(file))
await params.llm.reset_conversation()
# NOTE: we manually create a response here rather than relying
# on the function callback to trigger one since we've reset the
# conversation so the remote service doesn't know about the
# in-progress tool call.
await params.llm._create_response()
except Exception as e:
await params.result_callback({"success": False, "error": str(e)})
@@ -97,14 +102,12 @@ async def load_conversation(params: FunctionCallParams):
asyncio.create_task(_reset())
tools = [
{
"type": "function",
"name": "get_current_weather",
"description": "Get the current weather",
"parameters": {
"type": "object",
"properties": {
tools = ToolsSchema(
standard_tools=[
FunctionSchema(
name="get_current_weather",
description="Get the current weather",
properties={
"location": {
"type": "string",
"description": "The city and state, e.g. San Francisco, CA",
@@ -115,45 +118,33 @@ tools = [
"description": "The temperature unit to use. Infer this from the users location.",
},
},
"required": ["location", "format"],
},
},
{
"type": "function",
"name": "save_conversation",
"description": "Save the current conversatione. Use this function to persist the current conversation to external storage.",
"parameters": {
"type": "object",
"properties": {},
"required": [],
},
},
{
"type": "function",
"name": "get_saved_conversation_filenames",
"description": "Get a list of saved conversation histories. Returns a list of filenames. Each filename includes a date and timestamp. Each file is conversation history that can be loaded into this session.",
"parameters": {
"type": "object",
"properties": {},
"required": [],
},
},
{
"type": "function",
"name": "load_conversation",
"description": "Load a conversation history. Use this function to load a conversation history into the current session.",
"parameters": {
"type": "object",
"properties": {
required=["location", "format"],
),
FunctionSchema(
name="save_conversation",
description="Save the current conversatione. Use this function to persist the current conversation to external storage.",
properties={},
required=[],
),
FunctionSchema(
name="get_saved_conversation_filenames",
description="Get a list of saved conversation histories. Returns a list of filenames. Each filename includes a date and timestamp. Each file is conversation history that can be loaded into this session.",
properties={},
required=[],
),
FunctionSchema(
name="load_conversation",
description="Load a conversation history. Use this function to load a conversation history into the current session.",
properties={
"filename": {
"type": "string",
"description": "The filename of the conversation history to load.",
}
},
"required": ["filename"],
},
},
]
required=["filename"],
),
]
)
# We store functions so objects (e.g. SileroVADAnalyzer) don't get
@@ -224,8 +215,8 @@ Remember, your responses should be short. Just one or two sentences, usually."""
llm.register_function("get_saved_conversation_filenames", get_saved_conversation_filenames)
llm.register_function("load_conversation", load_conversation)
context = OpenAILLMContext([], tools)
context_aggregator = llm.create_context_aggregator(context)
context = LLMContext([{"role": "user", "content": "Say hello!"}], tools)
context_aggregator = LLMContextAggregatorPair(context)
pipeline = Pipeline(
[

13
examples/foundational/26a-gemini-live-transcription.py
View File

@@ -16,7 +16,9 @@ from pipecat.frames.frames import LLMRunFrame, TranscriptionMessage
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response import LLMAssistantAggregatorParams
from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
from pipecat.processors.transcript_processor import TranscriptProcessor
from pipecat.runner.types import RunnerArguments
from pipecat.runner.utils import create_transport
@@ -72,7 +74,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
# inference_on_context_initialization=False,
)
context = OpenAILLMContext(
context = LLMContext(
[
{
"role": "user",
@@ -90,7 +92,12 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
# },
],
)
context_aggregator = llm.create_context_aggregator(context)
context_aggregator = LLMContextAggregatorPair(
context,
# `expect_stripped_words=False` needed when Gemini Live used with AUDIO
# modality (the default)
assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False),
)
transcript = TranscriptProcessor()

13
examples/foundational/26b-gemini-live-function-calling.py
View File

@@ -19,7 +19,9 @@ from pipecat.frames.frames import LLMRunFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response import LLMAssistantAggregatorParams
from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
from pipecat.runner.types import RunnerArguments
from pipecat.runner.utils import create_transport
from pipecat.services.google.gemini_live.llm import GeminiLiveLLMService
@@ -139,10 +141,15 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
llm.register_function("get_current_weather", fetch_weather_from_api)
llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation)
context = OpenAILLMContext(
context = LLMContext(
[{"role": "user", "content": "Say hello."}],
)
context_aggregator = llm.create_context_aggregator(context)
context_aggregator = LLMContextAggregatorPair(
context,
# `expect_stripped_words=False` needed when Gemini Live used with AUDIO
# modality (the default)
assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False),
)
pipeline = Pipeline(
[

13
examples/foundational/26c-gemini-live-video.py
View File

@@ -17,7 +17,9 @@ from pipecat.frames.frames import LLMRunFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response import LLMAssistantAggregatorParams
from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
from pipecat.runner.types import RunnerArguments
from pipecat.runner.utils import (
create_transport,
@@ -65,7 +67,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
# inference_on_context_initialization=False,
)
context = OpenAILLMContext(
context = LLMContext(
[
{
"role": "user",
@@ -73,7 +75,12 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
},
],
)
context_aggregator = llm.create_context_aggregator(context)
context_aggregator = LLMContextAggregatorPair(
context,
# `expect_stripped_words=False` needed when Gemini Live used with AUDIO
# modality (the default)
assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False),
)
pipeline = Pipeline(
[

7
examples/foundational/26d-gemini-live-text.py
View File

@@ -16,7 +16,8 @@ from pipecat.frames.frames import LLMRunFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
from pipecat.runner.types import RunnerArguments
from pipecat.runner.utils import create_transport
from pipecat.services.cartesia.tts import CartesiaTTSService
@@ -109,8 +110,8 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
# Set up conversation context and management
# The context_aggregator will automatically collect conversation context
context = OpenAILLMContext(messages)
context_aggregator = llm.create_context_aggregator(context)
context = LLMContext(messages)
context_aggregator = LLMContextAggregatorPair(context)
pipeline = Pipeline(
[

13
examples/foundational/26e-gemini-live-google-search.py
View File

@@ -16,7 +16,9 @@ from pipecat.frames.frames import LLMRunFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response import LLMAssistantAggregatorParams
from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
from pipecat.runner.types import RunnerArguments
from pipecat.runner.utils import create_transport
from pipecat.services.google.gemini_live.llm import GeminiLiveLLMService
@@ -90,7 +92,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
tools=tools,
)
context = OpenAILLMContext(
context = LLMContext(
[
{
"role": "user",
@@ -98,7 +100,12 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
}
],
)
context_aggregator = llm.create_context_aggregator(context)
context_aggregator = LLMContextAggregatorPair(
context,
# `expect_stripped_words=False` needed when Gemini Live used with AUDIO
# modality (the default)
assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False),
)
pipeline = Pipeline(
[

15
examples/foundational/26f-gemini-live-files-api.py
View File

@@ -16,7 +16,9 @@ from pipecat.frames.frames import LLMRunFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response import LLMAssistantAggregatorParams
from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
from pipecat.runner.types import RunnerArguments
from pipecat.runner.utils import create_transport
from pipecat.services.google.gemini_live.llm import GeminiLiveLLMService
@@ -129,7 +131,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
mime_type = "text/plain"
# Create context with file reference
context = OpenAILLMContext(
context = LLMContext(
[
{
"role": "user",
@@ -152,7 +154,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
except Exception as e:
logger.error(f"Error uploading file: {e}")
# Continue with a basic context if file upload fails
context = OpenAILLMContext(
context = LLMContext(
[
{
"role": "user",
@@ -162,7 +164,12 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
)
# Create context aggregator
context_aggregator = llm.create_context_aggregator(context)
context_aggregator = LLMContextAggregatorPair(
context,
# `expect_stripped_words=False` needed when Gemini Live used with AUDIO
# modality (the default)
assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False),
)
# Build the pipeline
pipeline = Pipeline(

13
examples/foundational/26g-gemini-live-groundingMetadata.py
View File

@@ -10,7 +10,9 @@ from pipecat.frames.frames import Frame, LLMRunFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineTask
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response import LLMAssistantAggregatorParams
from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.runner.types import RunnerArguments
from pipecat.runner.utils import create_transport
@@ -124,8 +126,13 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
]
# Set up conversation context and management
context = OpenAILLMContext(messages)
context_aggregator = llm.create_context_aggregator(context)
context = LLMContext(messages)
context_aggregator = LLMContextAggregatorPair(
context,
# `expect_stripped_words=False` needed when Gemini Live used with AUDIO
# modality (the default)
assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False),
)
pipeline = Pipeline(
[

17
examples/foundational/26h-gemini-live-vertex-function-calling.py
View File

@@ -9,21 +9,21 @@ import os
from datetime import datetime
from dotenv import load_dotenv
from google.genai.types import HttpOptions
from loguru import logger
from pipecat.adapters.schemas.function_schema import FunctionSchema
from pipecat.adapters.schemas.tools_schema import AdapterType, ToolsSchema
from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.audio.vad.vad_analyzer import VADParams
from pipecat.frames.frames import LLMRunFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response import LLMAssistantAggregatorParams
from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
from pipecat.runner.types import RunnerArguments
from pipecat.runner.utils import create_transport
from pipecat.services.google.gemini_live.llm import GeminiLiveLLMService
from pipecat.services.google.gemini_live.llm_vertex import GeminiLiveVertexLLMService
from pipecat.services.llm_service import FunctionCallParams
from pipecat.transports.base_transport import BaseTransport, TransportParams
@@ -139,10 +139,13 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
llm.register_function("get_current_weather", fetch_weather_from_api)
llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation)
context = OpenAILLMContext(
[{"role": "user", "content": "Say hello."}],
context = LLMContext([{"role": "user", "content": "Say hello."}])
context_aggregator = LLMContextAggregatorPair(
context,
# `expect_stripped_words=False` needed when Gemini Live used with AUDIO
# modality (the default)
assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False),
)
context_aggregator = llm.create_context_aggregator(context)
pipeline = Pipeline(
[

15
examples/foundational/26i-gemini-live-graceful-end.py
View File

@@ -18,7 +18,9 @@ from pipecat.frames.frames import EndTaskFrame, LLMRunFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response import LLMAssistantAggregatorParams
from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
from pipecat.processors.frame_processor import FrameDirection
from pipecat.runner.types import RunnerArguments
from pipecat.runner.utils import create_transport
@@ -62,7 +64,7 @@ You have three tools available to you:
After you've responded to the user three times, do two things, in order:
1. Politely let them know that that's all the time you have today and say goodbye.
2. Call the end_conversation tool to gracefully end the conversation.
2. *WITHOUT WAITING FOR THE USER TO RESPOND*, call the end_conversation tool to gracefully end the conversation.
"""
@@ -152,10 +154,15 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation)
llm.register_function("end_conversation", end_conversation)
context = OpenAILLMContext(
context = LLMContext(
[{"role": "user", "content": "Say hello."}],
)
context_aggregator = llm.create_context_aggregator(context)
context_aggregator = LLMContextAggregatorPair(
context,
# `expect_stripped_words=False` needed when Gemini Live used with AUDIO
# modality (the default)
assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False),
)
pipeline = Pipeline(
[

6
examples/foundational/27-simli-layer.py
View File

@@ -9,7 +9,6 @@ import os
from dotenv import load_dotenv
from loguru import logger
from simli import SimliConfig
from pipecat.audio.turn.smart_turn.base_smart_turn import SmartTurnParams
from pipecat.audio.turn.smart_turn.local_smart_turn_v3 import LocalSmartTurnAnalyzerV3
@@ -66,11 +65,12 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
tts = CartesiaTTSService(
api_key=os.getenv("CARTESIA_API_KEY"),
voice_id="a167e0f3-df7e-4d52-a9c3-f949145efdab",
voice_id="71a7ad14-091c-4e8e-a314-022ece01c121",
)
simli_ai = SimliVideoService(
SimliConfig(os.getenv("SIMLI_API_KEY"), os.getenv("SIMLI_FACE_ID")),
api_key=os.getenv("SIMLI_API_KEY"),
face_id="cace3ef7-a4c4-425d-a8cf-a5358eb0c427",
)
llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o-mini")

13
examples/foundational/46-video-processing.py
View File

@@ -15,7 +15,9 @@ from pipecat.frames.frames import Frame, InputImageRawFrame, LLMRunFrame, Output
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response import LLMAssistantAggregatorParams
from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.processors.frameworks.rtvi import RTVIObserver, RTVIProcessor
from pipecat.runner.types import RunnerArguments
@@ -108,8 +110,13 @@ async def run_bot(pipecat_transport):
}
]
context = OpenAILLMContext(messages)
context_aggregator = llm.create_context_aggregator(context)
context = LLMContext(messages)
context_aggregator = LLMContextAggregatorPair(
context,
# `expect_stripped_words=False` needed when Gemini Live used with AUDIO
# modality (the default)
assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False),
)
# RTVI events for Pipecat client UI
rtvi = RTVIProcessor()

2
pyproject.toml
View File

@@ -55,7 +55,7 @@ azure = [ "azure-cognitiveservices-speech~=1.42.0"]
cartesia = [ "cartesia~=2.0.3", "pipecat-ai[websockets-base]" ]
cerebras = []
deepseek = []
daily = [ "daily-python~=0.20.0" ]
daily = [ "daily-python~=0.21.0" ]
deepgram = [ "deepgram-sdk~=4.7.0" ]
elevenlabs = [ "pipecat-ai[websockets-base]" ]
fal = [ "fal-client~=0.5.9" ]

3
src/pipecat/adapters/schemas/tools_schema.py
View File

@@ -22,9 +22,12 @@ class AdapterType(Enum):
Parameters:
GEMINI: Google Gemini adapter - currently the only service supporting custom tools.
SHIM: Backward compatibility shim for creating ToolsSchemas from lists of tools in
any format, used by LLMContext.from_openai_context.
"""
GEMINI = "gemini" # that is the only service where we are able to add custom tools for now
SHIM = "shim" # for use as backward compatibility shim for creating ToolsSchemas from list of tools in any format
class ToolsSchema:

18
src/pipecat/adapters/services/aws_nova_sonic_adapter.py
View File

@@ -16,7 +16,7 @@ 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.adapters.schemas.tools_schema import AdapterType, ToolsSchema
from pipecat.processors.aggregators.llm_context import LLMContext, LLMContextMessage
@@ -210,4 +210,18 @@ class AWSNovaSonicLLMAdapter(BaseLLMAdapter[AWSNovaSonicLLMInvocationParams]):
List of dictionaries in AWS Nova Sonic function format.
"""
functions_schema = tools_schema.standard_tools
return [self._to_aws_nova_sonic_function_format(func) for func in functions_schema]
standard_tools = [
self._to_aws_nova_sonic_function_format(func) for func in functions_schema
]
# For backward compatibility, AWS Nova Sonic can still be used with
# tools in dict format, even though it always uses `LLMContext` under
# the hood (via `LLMContext.from_openai_context()`).
# To support this behavior, we use "shimmed" custom tools here.
# (We maintain this backward compatibility because users aren't
# *knowingly* opting into the new `LLMContext`.)
shimmed_tools = []
if tools_schema.custom_tools:
shimmed_tools = tools_schema.custom_tools.get(AdapterType.SHIM, [])
return standard_tools + shimmed_tools

28
src/pipecat/adapters/services/gemini_adapter.py
View File

@@ -24,13 +24,7 @@ from pipecat.processors.aggregators.llm_context import (
)
try:
from google.genai.types import (
Blob,
Content,
FunctionCall,
FunctionResponse,
Part,
)
from google.genai.types import Blob, Content, FileData, FunctionCall, FunctionResponse, Part
except ModuleNotFoundError as e:
logger.error(f"Exception: {e}")
logger.error("In order to use Google AI, you need to `pip install pipecat-ai[google]`.")
@@ -309,6 +303,7 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]):
parts.append(
Part(
function_call=FunctionCall(
id=id,
name=name,
args=json.loads(tc["function"]["arguments"]),
)
@@ -334,9 +329,12 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]):
function_name = params.tool_call_id_to_name_mapping[tool_call_id]
parts.append(
Part.from_function_response(
name=function_name,
response=response_dict,
Part(
function_response=FunctionResponse(
id=tool_call_id,
name=function_name,
response=response_dict,
)
)
)
elif isinstance(content, str):
@@ -358,6 +356,16 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]):
input_audio = c["input_audio"]
audio_bytes = base64.b64decode(input_audio["data"])
parts.append(Part(inline_data=Blob(mime_type="audio/wav", data=audio_bytes)))
elif c["type"] == "file_data":
file_data = c["file_data"]
parts.append(
Part(
file_data=FileData(
mime_type=file_data.get("mime_type"),
file_uri=file_data.get("file_uri"),
)
)
)
return self.MessageConversionResult(
content=Content(role=role, parts=parts),

161
src/pipecat/adapters/services/open_ai_realtime_adapter.py
View File

@@ -6,12 +6,18 @@
"""OpenAI Realtime LLM adapter for Pipecat."""
from typing import Any, Dict, List, TypedDict
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
from pipecat.adapters.schemas.tools_schema import AdapterType, ToolsSchema
from pipecat.processors.aggregators.llm_context import LLMContext, LLMContextMessage
from pipecat.services.openai.realtime import events
class OpenAIRealtimeLLMInvocationParams(TypedDict):
@@ -20,7 +26,9 @@ class OpenAIRealtimeLLMInvocationParams(TypedDict):
This is a placeholder until support for universal LLMContext machinery is added for OpenAI Realtime.
"""
pass
system_instruction: Optional[str]
messages: List[events.ConversationItem]
tools: List[Dict[str, Any]]
class OpenAIRealtimeLLMAdapter(BaseLLMAdapter):
@@ -33,7 +41,7 @@ class OpenAIRealtimeLLMAdapter(BaseLLMAdapter):
@property
def id_for_llm_specific_messages(self) -> str:
"""Get the identifier used in LLMSpecificMessage instances for OpenAI Realtime."""
raise NotImplementedError("Universal LLMContext is not yet supported for OpenAI Realtime.")
return "openai-realtime"
def get_llm_invocation_params(self, context: LLMContext) -> OpenAIRealtimeLLMInvocationParams:
"""Get OpenAI Realtime-specific LLM invocation parameters from a universal LLM context.
@@ -46,7 +54,13 @@ class OpenAIRealtimeLLMAdapter(BaseLLMAdapter):
Returns:
Dictionary of parameters for invoking OpenAI Realtime's API.
"""
raise NotImplementedError("Universal LLMContext is not yet supported for OpenAI Realtime.")
messages = self._from_universal_context_messages(self.get_messages(context))
return {
"system_instruction": messages.system_instruction,
"messages": messages.messages,
# NOTE: LLMContext's tools are guaranteed to be a ToolsSchema (or NOT_GIVEN)
"tools": self.from_standard_tools(context.tools) or [],
}
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 OpenAI Realtime.
@@ -61,7 +75,124 @@ class OpenAIRealtimeLLMAdapter(BaseLLMAdapter):
Returns:
List of messages in a format ready for logging about OpenAI Realtime.
"""
raise NotImplementedError("Universal LLMContext is not yet supported for OpenAI Realtime.")
# NOTE: this is the same as in OpenAIAdapter, as that's what it was
# prior to a refactor. Worth noting that for OpenAI Realtime
# specifically, not everything handled here is necessarily supported
# (or supported yet).
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["type"] == "image_url":
if item["image_url"]["url"].startswith("data:image/"):
item["image_url"]["url"] = "data:image/..."
if item["type"] == "input_audio":
item["input_audio"]["data"] = "..."
if "mime_type" in msg and msg["mime_type"].startswith("image/"):
msg["data"] = "..."
msgs.append(msg)
return msgs
@dataclass
class ConvertedMessages:
"""Container for OpenAI-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:
# We can't load a long conversation history into the openai realtime api yet. (The API/model
# forgets that it can do audio, if you do a series of `conversation.item.create` calls.) So
# our general strategy until this is fixed is just to put everything into a first "user"
# message as a single input.
if not universal_context_messages:
return self.ConvertedMessages(messages=[])
messages = copy.deepcopy(universal_context_messages)
system_instruction = None
# If we have a "system" message as our first message, let's pull that out into 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)
# 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])],
system_instruction=system_instruction,
)
# Otherwise, let's pack everything into a single "user" message with a bit of
# explanation for the LLM
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, simple say that you
are ready to continue the conversation."""
return self.ConvertedMessages(
messages=[
{
"role": "user",
"type": "message",
"content": [
{
"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:
if message.get("role") == "user":
content = message.get("content")
if isinstance(message.get("content"), list):
content = ""
for c in message.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}"
)
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_openai_realtime_function_format(function: FunctionSchema) -> Dict[str, Any]:
@@ -94,4 +225,18 @@ class OpenAIRealtimeLLMAdapter(BaseLLMAdapter):
List of function definitions in OpenAI Realtime format.
"""
functions_schema = tools_schema.standard_tools
return [self._to_openai_realtime_function_format(func) for func in functions_schema]
standard_tools = [
self._to_openai_realtime_function_format(func) for func in functions_schema
]
# For backward compatibility, OpenAI Realtime can still be used with
# tools in dict format, even though it always uses `LLMContext` under
# the hood (via `LLMContext.from_openai_context()`).
# To support this behavior, we use "shimmed" custom tools here.
# (We maintain this backward compatibility because users aren't
# *knowingly* opting into the new `LLMContext`.)
shimmed_tools = []
if tools_schema.custom_tools:
shimmed_tools = tools_schema.custom_tools.get(AdapterType.SHIM, [])
return standard_tools + shimmed_tools

40
src/pipecat/processors/aggregators/llm_context.py
View File

@@ -15,7 +15,6 @@ service-specific adapter.
"""
import base64
import copy
import io
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any, List, Optional, TypeAlias, Union
@@ -29,7 +28,7 @@ from openai.types.chat import (
)
from PIL import Image
from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.adapters.schemas.tools_schema import AdapterType, ToolsSchema
from pipecat.frames.frames import AudioRawFrame
if TYPE_CHECKING:
@@ -83,9 +82,17 @@ class LLMContext:
Returns:
New LLMContext instance with converted messages and settings.
"""
# Convert tools to ToolsSchema if needed.
# If the tools are already a ToolsSchema, this is a no-op.
# Otherwise, we wrap them in a shim ToolsSchema.
converted_tools = openai_context.tools
if isinstance(converted_tools, list):
converted_tools = ToolsSchema(
standard_tools=[], custom_tools={AdapterType.SHIM: converted_tools}
)
return LLMContext(
messages=openai_context.get_messages(),
tools=openai_context.tools,
tools=converted_tools,
tool_choice=openai_context.tool_choice,
)
@@ -119,6 +126,33 @@ class LLMContext:
"""
return self.get_messages()
def get_messages_for_persistent_storage(self) -> List[LLMContextMessage]:
"""Get messages suitable for persistent storage.
NOTE: the only reason this method exists is because we're "silently"
switching from OpenAILLMContext to LLMContext under the hood in some
services and don't want to trip up users who may have been relying on
this method, which is part of the public API of OpenAILLMContext but
doesn't need to be for LLMContext.
.. deprecated::
Use `get_messages()` instead.
Returns:
List of conversation messages.
"""
import warnings
with warnings.catch_warnings():
warnings.simplefilter("always")
warnings.warn(
"get_messages_for_persistent_storage() is deprecated, use get_messages() instead.",
DeprecationWarning,
stacklevel=2,
)
return self.get_messages()
def get_messages(self, llm_specific_filter: Optional[str] = None) -> List[LLMContextMessage]:
"""Get the current messages list.

6
src/pipecat/processors/aggregators/llm_response_universal.py
View File

@@ -290,6 +290,12 @@ class LLMUserAggregator(LLMContextAggregator):
await self._handle_llm_messages_update(frame)
elif isinstance(frame, LLMSetToolsFrame):
self.set_tools(frame.tools)
# Push the LLMSetToolsFrame as well, since speech-to-speech LLM
# services (like OpenAI Realtime) may need to know about tool
# changes; unlike text-based LLM services they won't just "pick up
# the change" on the next LLM run, as the LLM is continuously
# running.
await self.push_frame(frame, direction)
elif isinstance(frame, LLMSetToolChoiceFrame):
self.set_tool_choice(frame.tool_choice)
elif isinstance(frame, SpeechControlParamsFrame):

2
src/pipecat/services/azure/realtime/llm.py
View File

@@ -38,7 +38,7 @@ class AzureRealtimeLLMService(OpenAIRealtimeLLMService):
Args:
api_key: The API key for the Azure OpenAI service.
base_url: The full Azure WebSocket endpoint URL including api-version and deployment.
Example: "wss://my-project.openai.azure.com/openai/realtime?api-version=2024-10-01-preview&deployment=my-realtime-deployment"
Example: "wss://my-project.openai.azure.com/openai/realtime?api-version=2025-04-01-preview&deployment=my-realtime-deployment"
**kwargs: Additional arguments passed to parent OpenAIRealtimeLLMService.
"""
super().__init__(base_url=base_url, api_key=api_key, **kwargs)

250
src/pipecat/services/google/gemini_live/llm.py
View File

@@ -17,6 +17,7 @@ import json
import random
import time
import uuid
import warnings
from dataclasses import dataclass
from enum import Enum
from typing import Any, Dict, List, Optional, Union
@@ -56,10 +57,12 @@ from pipecat.frames.frames import (
UserStoppedSpeakingFrame,
)
from pipecat.metrics.metrics import LLMTokenUsage
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response import (
LLMAssistantAggregatorParams,
LLMUserAggregatorParams,
)
from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
from pipecat.processors.aggregators.openai_llm_context import (
OpenAILLMContext,
OpenAILLMContextFrame,
@@ -219,6 +222,10 @@ class GeminiLiveContext(OpenAILLMContext):
Provides Gemini-specific context management including system instruction
extraction and message format conversion for the Live API.
.. deprecated:: 0.0.92
Gemini Live no longer uses `GeminiLiveContext` under the hood.
It now uses `LLMContext`.
"""
@staticmethod
@@ -231,6 +238,22 @@ class GeminiLiveContext(OpenAILLMContext):
Returns:
The upgraded Gemini context instance.
"""
# This warning is here rather than `__init__` since `upgrade()` was the
# "main" way that GeminiLiveContext instances were created.
# Almost no users should be seeing this message anyway, as
# GeminiLiveContext instances were typically created under the hood:
# the user would pass an OpenAILLMContext instance, which would be
# upgraded without them necessarily knowing.
with warnings.catch_warnings():
warnings.simplefilter("always")
warnings.warn(
"GeminiLiveContext is deprecated. "
"Gemini Live no longer uses GeminiLiveContext under the hood. "
"It now uses LLMContext.",
DeprecationWarning,
stacklevel=2,
)
if isinstance(obj, OpenAILLMContext) and not isinstance(obj, GeminiLiveContext):
logger.debug(f"Upgrading to Gemini Live Context: {obj}")
obj.__class__ = GeminiLiveContext
@@ -328,8 +351,28 @@ class GeminiLiveUserContextAggregator(OpenAIUserContextAggregator):
Extends OpenAI user aggregator to handle Gemini-specific message passing
while maintaining compatibility with the standard aggregation pipeline.
.. deprecated:: 0.0.92
Gemini Live no longer expects a `GeminiLiveUserContextAggregator`.
It now expects a `LLMUserAggregator`.
"""
def __init__(self, *args, **kwargs):
"""Initialize Gemini Live user context aggregator."""
# Almost no users should be seeing this message, as
# `GeminiLiveUserContextAggregator`` instances were typically created
# under the hood, as part of `llm.create_context_aggregator()`.
with warnings.catch_warnings():
warnings.simplefilter("always")
warnings.warn(
"GeminiLiveUserContextAggregator is deprecated. "
"Gemini Live no longer expects a GeminiLiveUserContextAggregator. "
"It now expects a LLMUserAggregator.",
DeprecationWarning,
stacklevel=2,
)
super().__init__(*args, **kwargs)
async def process_frame(self, frame, direction):
"""Process incoming frames for user context aggregation.
@@ -349,8 +392,28 @@ class GeminiLiveAssistantContextAggregator(OpenAIAssistantContextAggregator):
Handles assistant response aggregation while filtering out LLMTextFrames
to prevent duplicate context entries, as Gemini Live pushes both
LLMTextFrames and TTSTextFrames.
.. deprecated:: 0.0.92
Gemini Live no longer uses `GeminiLiveAssistantContextAggregator` under the hood.
It now uses `LLMAssistantAggregator`.
"""
def __init__(self, *args, **kwargs):
"""Initialize Gemini Live assistant context aggregator."""
# Almost no users should be seeing this message, as
# `GeminiLiveAssistantContextAggregator` instances were typically
# created under the hood, as part of `llm.create_context_aggregator()`.
with warnings.catch_warnings():
warnings.simplefilter("always")
warnings.warn(
"GeminiLiveAssistantContextAggregator is deprecated. "
"Gemini Live no longer uses GeminiLiveAssistantContextAggregator under the hood. "
"It now uses LLMAssistantAggregator.",
DeprecationWarning,
stacklevel=2,
)
super().__init__(*args, **kwargs)
async def process_frame(self, frame: Frame, direction: FrameDirection):
"""Process incoming frames for assistant context aggregation.
@@ -380,6 +443,10 @@ class GeminiLiveAssistantContextAggregator(OpenAIAssistantContextAggregator):
class GeminiLiveContextAggregatorPair:
"""Pair of user and assistant context aggregators for Gemini Live.
.. deprecated:: 0.0.92
`GeminiLiveContextAggregatorPair` is deprecated.
Use `LLMContextAggregatorPair` instead.
Parameters:
_user: The user context aggregator instance.
_assistant: The assistant context aggregator instance.
@@ -388,6 +455,19 @@ class GeminiLiveContextAggregatorPair:
_user: GeminiLiveUserContextAggregator
_assistant: GeminiLiveAssistantContextAggregator
def __post_init__(self):
# Almost no users should be seeing this message, as
# `GeminiLiveContextAggregatorPair` instances were typically created
# under the hood, with `llm.create_context_aggregator()`.
with warnings.catch_warnings():
warnings.simplefilter("always")
warnings.warn(
"GeminiLiveContextAggregatorPair is deprecated. "
"Use LLMContextAggregatorPair instead.",
DeprecationWarning,
stacklevel=2,
)
def user(self) -> GeminiLiveUserContextAggregator:
"""Get the user context aggregator.
@@ -609,7 +689,7 @@ class GeminiLiveLLMService(LLMService):
self._run_llm_when_session_ready = False
self._user_is_speaking = False
self._bot_is_speaking = False
self._bot_is_responding = False
self._user_audio_buffer = bytearray()
self._user_transcription_buffer = ""
self._last_transcription_sent = ""
@@ -665,6 +745,9 @@ class GeminiLiveLLMService(LLMService):
# Initialize the API client. Subclasses can override this if needed.
self.create_client()
# Bookkeeping for tool calls
self._completed_tool_calls = set()
def create_client(self):
"""Create the Gemini API client instance. Subclasses can override this."""
self._client = Client(api_key=self._api_key, http_options=self._http_options)
@@ -787,9 +870,13 @@ class GeminiLiveLLMService(LLMService):
#
async def _handle_interruption(self):
await self._set_bot_is_speaking(False)
await self.push_frame(TTSStoppedFrame())
await self.push_frame(LLMFullResponseEndFrame())
if self._bot_is_responding:
await self._set_bot_is_responding(False)
if self._settings.get("modalities") == GeminiModalities.AUDIO:
await self.push_frame(TTSStoppedFrame())
# Do not send LLMFullResponseEndFrame here - an interruption
# already tells the assistant context aggregator that the response
# is over.
async def _handle_user_started_speaking(self, frame):
self._user_is_speaking = True
@@ -807,7 +894,6 @@ class GeminiLiveLLMService(LLMService):
#
# frame processing
#
# StartFrame, StopFrame, CancelFrame implemented in base class
#
@@ -820,7 +906,7 @@ class GeminiLiveLLMService(LLMService):
"""
# Defer EndFrame handling until after the bot turn is finished
if isinstance(frame, EndFrame):
if self._bot_is_speaking:
if self._bot_is_responding:
logger.debug("Deferring handling EndFrame until bot turn is finished")
self._end_frame_pending_bot_turn_finished = frame
return
@@ -829,22 +915,13 @@ class GeminiLiveLLMService(LLMService):
if isinstance(frame, TranscriptionFrame):
await self.push_frame(frame, direction)
elif isinstance(frame, OpenAILLMContextFrame):
context: GeminiLiveContext = GeminiLiveContext.upgrade(frame.context)
# For now, we'll only trigger inference here when either:
# 1. We have not seen a context frame before
# 2. The last message is a tool call result
if not self._context:
self._context = context
if frame.context.tools:
self._tools = frame.context.tools
await self._create_initial_response()
elif context.messages and context.messages[-1].get("role") == "tool":
# Support just one tool call per context frame for now
tool_result_message = context.messages[-1]
await self._tool_result(tool_result_message)
elif isinstance(frame, LLMContextFrame):
raise NotImplementedError("Universal LLMContext is not yet supported for Gemini Live.")
elif isinstance(frame, (LLMContextFrame, OpenAILLMContextFrame)):
context = (
frame.context
if isinstance(frame, LLMContextFrame)
else LLMContext.from_openai_context(frame.context)
)
await self._handle_context(context)
elif isinstance(frame, InputTextRawFrame):
await self._send_user_text(frame.text)
await self.push_frame(frame, direction)
@@ -883,13 +960,48 @@ class GeminiLiveLLMService(LLMService):
else:
await self.push_frame(frame, direction)
async def _set_bot_is_speaking(self, speaking: bool):
if self._bot_is_speaking == speaking:
async def _handle_context(self, context: LLMContext):
if not self._context:
# We got our initial context
self._context = context
if context.tools:
self._tools = context.tools
# Initialize our bookkeeping of already-completed tool calls in
# the context
await self._process_completed_function_calls(send_new_results=False)
await self._create_initial_response()
else:
# We got an updated context.
# This may contain a new user message or tool call result.
self._context = context
# Send results for newly-completed function calls, if any.
await self._process_completed_function_calls(send_new_results=True)
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()
messages = adapter.get_llm_invocation_params(self._context).get("messages", [])
for message in messages:
if message.parts:
for part in message.parts:
if part.function_response:
tool_call_id = part.function_response.id
tool_name = part.function_response.name
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
if send_new_results:
await self._tool_result(
tool_call_id, tool_name, part.function_response.response
)
self._completed_tool_calls.add(tool_call_id)
async def _set_bot_is_responding(self, responding: bool):
if self._bot_is_responding == responding:
return
self._bot_is_speaking = speaking
self._bot_is_responding = responding
if not self._bot_is_speaking and self._end_frame_pending_bot_turn_finished:
if not self._bot_is_responding and self._end_frame_pending_bot_turn_finished:
await self.queue_frame(self._end_frame_pending_bot_turn_finished)
self._end_frame_pending_bot_turn_finished = None
@@ -1116,6 +1228,7 @@ class GeminiLiveLLMService(LLMService):
if self._session:
await self._session.close()
self._session = None
self._completed_tool_calls = set()
self._disconnecting = False
except Exception as e:
logger.error(f"{self} error disconnecting: {e}")
@@ -1195,7 +1308,8 @@ class GeminiLiveLLMService(LLMService):
self._run_llm_when_session_ready = True
return
messages = self._context.get_messages_for_initializing_history()
adapter: GeminiLLMAdapter = self.get_llm_adapter()
messages = adapter.get_llm_invocation_params(self._context).get("messages", [])
if not messages:
return
@@ -1223,8 +1337,9 @@ class GeminiLiveLLMService(LLMService):
# Create a throwaway context just for the purpose of getting messages
# in the right format
context = GeminiLiveContext.upgrade(OpenAILLMContext(messages=messages_list))
messages = context.get_messages_for_initializing_history()
context = LLMContext(messages=messages_list)
adapter: GeminiLLMAdapter = self.get_llm_adapter()
messages = adapter.get_llm_invocation_params(context).get("messages", [])
if not messages:
return
@@ -1239,17 +1354,16 @@ class GeminiLiveLLMService(LLMService):
await self._handle_send_error(e)
@traced_gemini_live(operation="llm_tool_result")
async def _tool_result(self, tool_result_message):
async def _tool_result(
self, tool_call_id: str, tool_name: str, tool_result_message: Dict[str, Any]
):
"""Send tool result back to the API."""
if self._disconnecting or not self._session:
return
# For now we're shoving the name into the tool_call_id field, so this
# will work until we revisit that.
id = tool_result_message.get("tool_call_id")
name = tool_result_message.get("tool_call_name")
result = json.loads(tool_result_message.get("content") or "")
response = FunctionResponse(name=name, id=id, response=result)
response = FunctionResponse(name=tool_name, id=tool_call_id, response=tool_result_message)
try:
await self._session.send_tool_response(function_responses=response)
@@ -1277,7 +1391,10 @@ class GeminiLiveLLMService(LLMService):
# part.text is added when `modalities` is set to TEXT; otherwise, it's None
text = part.text
if text:
if not self._bot_text_buffer:
if not self._bot_is_responding:
# Update bot responding state and send service start frame
# (AUDIO modality case)
await self._set_bot_is_responding(True)
await self.push_frame(LLMFullResponseStartFrame())
self._bot_text_buffer += text
@@ -1288,6 +1405,8 @@ class GeminiLiveLLMService(LLMService):
if msg.server_content and msg.server_content.grounding_metadata:
self._accumulated_grounding_metadata = msg.server_content.grounding_metadata
# If we have no audio, stop here.
# All logic below this point pertains to the AUDIO modality.
inline_data = part.inline_data
if not inline_data:
return
@@ -1313,8 +1432,10 @@ class GeminiLiveLLMService(LLMService):
if not audio:
return
if not self._bot_is_speaking:
await self._set_bot_is_speaking(True)
# Update bot responding state and send service start frames
# (AUDIO modality case)
if not self._bot_is_responding:
await self._set_bot_is_responding(True)
await self.push_frame(TTSStartedFrame())
await self.push_frame(LLMFullResponseStartFrame())
@@ -1354,7 +1475,6 @@ class GeminiLiveLLMService(LLMService):
@traced_gemini_live(operation="llm_response")
async def _handle_msg_turn_complete(self, message: LiveServerMessage):
"""Handle the turn complete message."""
await self._set_bot_is_speaking(False)
text = self._bot_text_buffer
# Trace the complete LLM response (this will be handled by the decorator)
@@ -1373,13 +1493,15 @@ class GeminiLiveLLMService(LLMService):
self._search_result_buffer = ""
self._accumulated_grounding_metadata = None
# Only push the TTSStoppedFrame if the bot is outputting audio
# when text is found, modalities is set to TEXT and no audio
# is produced.
if not text:
await self.push_frame(TTSStoppedFrame())
await self.push_frame(LLMFullResponseEndFrame())
if self._bot_is_responding:
await self._set_bot_is_responding(False)
if not text:
# AUDIO modality case
await self.push_frame(TTSStoppedFrame())
await self.push_frame(LLMFullResponseEndFrame())
else:
# TEXT modality case
await self.push_frame(LLMFullResponseEndFrame())
@traced_stt
async def _handle_user_transcription(
@@ -1442,8 +1564,8 @@ class GeminiLiveLLMService(LLMService):
return
# This is the output transcription text when modalities is set to AUDIO.
# In this case, we push LLMTextFrame and TTSTextFrame to be handled by the
# downstream assistant context aggregator.
# In this case, we push TTSTextFrame to be handled by the downstream
# assistant context aggregator.
text = message.server_content.output_transcription.text
if not text:
@@ -1458,7 +1580,17 @@ class GeminiLiveLLMService(LLMService):
# Collect text for tracing
self._llm_output_buffer += text
await self.push_frame(LLMTextFrame(text=text))
# NOTE: Shoot. When using Vertex AI, output transcription messages
# arrive *before* the model_turn messages with audio, so we need to
# handle sending TTSStartedFrame and LLMFullResponseStartFrame here as
# well. These messages also contain much *more* text (it looks further
# ahead). That means that on an interruption our recorded context will
# contain some text that was actually never spoken.
if not self._bot_is_responding:
await self._set_bot_is_responding(True)
await self.push_frame(TTSStartedFrame())
await self.push_frame(LLMFullResponseStartFrame())
await self.push_frame(TTSTextFrame(text=text))
async def _handle_msg_grounding_metadata(self, message: LiveServerMessage):
@@ -1557,26 +1689,26 @@ class GeminiLiveLLMService(LLMService):
*,
user_params: LLMUserAggregatorParams = LLMUserAggregatorParams(),
assistant_params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(),
) -> GeminiLiveContextAggregatorPair:
) -> LLMContextAggregatorPair:
"""Create an instance of GeminiLiveContextAggregatorPair from an OpenAILLMContext.
Constructor keyword arguments for both the user and assistant aggregators can be provided.
NOTE: this method exists only for backward compatibility. New code
should instead do:
context = LLMContext(...)
context_aggregator = LLMContextAggregatorPair(context)
Args:
context: The LLM context to use.
user_params: User aggregator parameters. Defaults to LLMUserAggregatorParams().
assistant_params: Assistant aggregator parameters. Defaults to LLMAssistantAggregatorParams().
Returns:
GeminiLiveContextAggregatorPair: A pair of context
aggregators, one for the user and one for the assistant,
encapsulated in an GeminiLiveContextAggregatorPair.
A pair of user and assistant context aggregators.
"""
context.set_llm_adapter(self.get_llm_adapter())
GeminiLiveContext.upgrade(context)
user = GeminiLiveUserContextAggregator(context, params=user_params)
context = LLMContext.from_openai_context(context)
assistant_params.expect_stripped_words = False
assistant = GeminiLiveAssistantContextAggregator(context, params=assistant_params)
return GeminiLiveContextAggregatorPair(_user=user, _assistant=assistant)
return LLMContextAggregatorPair(
context, user_params=user_params, assistant_params=assistant_params
)

6
src/pipecat/services/hume/tts.py
View File

@@ -184,11 +184,15 @@ class HumeTTSService(TTSService):
# Hume emits mono PCM at 48 kHz; downstream can resample if needed.
# We buffer audio bytes before sending to prevent glitches.
self._audio_bytes = b""
# Use version "2" by default if no description is provided
# Version "1" is needed when description is used
version = "1" if self._params.description is not None else "2"
async for chunk in self._client.tts.synthesize_json_streaming(
utterances=[utterance],
format=pcm_fmt,
instant_mode=True,
version="2",
version=version,
):
audio_b64 = getattr(chunk, "audio", None)
if not audio_b64:

80
src/pipecat/services/openai/realtime/context.py
View File

@@ -4,7 +4,85 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""OpenAI Realtime LLM context and aggregator implementations."""
"""OpenAI Realtime LLM context and aggregator implementations.
.. deprecated:: 0.0.92
OpenAI Realtime no longer uses types from this module under the hood.
It now uses `LLMContext` and `LLMContextAggregatorPair`.
Using the new patterns should allow you to not need types from this module.
BEFORE:
```
# Setup
context = OpenAILLMContext(messages, tools)
context_aggregator = llm.create_context_aggregator(context)
# Context aggregator type
context_aggregator: OpenAIContextAggregatorPair
# Context frame type
frame: OpenAILLMContextFrame
# Context type
context: OpenAIRealtimeLLMContext
# or
context: OpenAILLMContext
```
AFTER:
```
# Setup
context = LLMContext(messages, tools)
context_aggregator = LLMContextAggregatorPair(context)
# Context aggregator type
context_aggregator: LLMContextAggregatorPair
# Context frame type
frame: LLMContextFrame
# Context type
context: LLMContext
```
"""
import warnings
with warnings.catch_warnings():
warnings.simplefilter("always")
warnings.warn(
"Types in pipecat.services.openai.realtime.llm (or "
"pipecat.services.openai_realtime.llm) are deprecated. \n"
"OpenAI Realtime no longer uses types from this module under the hood. \n"
"It now uses `LLMContext` and `LLMContextAggregatorPair`. \n"
"Using the new patterns should allow you to not need types from this module.\n\n"
"BEFORE:\n"
"```\n"
"# Setup\n"
"context = OpenAILLMContext(messages, tools)\n"
"context_aggregator = llm.create_context_aggregator(context)\n\n"
"# Context aggregator type\n"
"context_aggregator: OpenAIContextAggregatorPair\n\n"
"# Context frame type\n"
"frame: OpenAILLMContextFrame\n\n"
"# Context type\n"
"context: OpenAIRealtimeLLMContext\n"
"# or\n"
"context: OpenAILLMContext\n\n"
"```\n\n"
"AFTER:\n"
"```\n"
"# Setup\n"
"context = LLMContext(messages, tools)\n"
"context_aggregator = LLMContextAggregatorPair(context)\n\n"
"# Context aggregator type\n"
"context_aggregator: LLMContextAggregatorPair\n\n"
"# Context frame type\n"
"frame: LLMContextFrame\n\n"
"# Context type\n"
"context: LLMContext\n\n"
"```\n",
)
import copy
import json

23
src/pipecat/services/openai/realtime/frames.py
View File

@@ -4,7 +4,28 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Custom frame types for OpenAI Realtime API integration."""
"""Custom frame types for OpenAI Realtime API integration.
.. deprecated:: 0.0.92
OpenAI Realtime no longer uses types from this module under the hood.
It now works more like most LLM services in Pipecat, relying on updates to
its context, pushed by context aggregators, to update its internal state.
Listen for `LLMContextFrame`s for context updates.
"""
import warnings
with warnings.catch_warnings():
warnings.simplefilter("always")
warnings.warn(
"Types in pipecat.services.openai.realtime.frames are deprecated. \n"
"OpenAI Realtime no longer uses types from this module under the hood. \n\n"
"It now works more like other LLM services in Pipecat, relying on updates to \n"
"its context, pushed by context aggregators, to update its internal state.\n\n"
"Listen for `LLMContextFrame`s for context updates.\n"
)
from dataclasses import dataclass
from typing import TYPE_CHECKING

280
src/pipecat/services/openai/realtime/llm.py
View File

@@ -14,7 +14,9 @@ from typing import Optional
from loguru import logger
from pipecat.adapters.services.open_ai_realtime_adapter import OpenAIRealtimeLLMAdapter
from pipecat.adapters.services.open_ai_realtime_adapter import (
OpenAIRealtimeLLMAdapter,
)
from pipecat.frames.frames import (
BotStoppedSpeakingFrame,
CancelFrame,
@@ -41,10 +43,12 @@ from pipecat.frames.frames import (
UserStoppedSpeakingFrame,
)
from pipecat.metrics.metrics import LLMTokenUsage
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response import (
LLMAssistantAggregatorParams,
LLMUserAggregatorParams,
)
from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
from pipecat.processors.aggregators.openai_llm_context import (
OpenAILLMContext,
OpenAILLMContextFrame,
@@ -57,12 +61,6 @@ from pipecat.utils.time import time_now_iso8601
from pipecat.utils.tracing.service_decorators import traced_openai_realtime, traced_stt
from . import events
from .context import (
OpenAIRealtimeAssistantContextAggregator,
OpenAIRealtimeLLMContext,
OpenAIRealtimeUserContextAggregator,
)
from .frames import RealtimeFunctionCallResultFrame, RealtimeMessagesUpdateFrame
try:
from websockets.asyncio.client import connect as websocket_connect
@@ -108,22 +106,39 @@ class OpenAIRealtimeLLMService(LLMService):
base_url: str = "wss://api.openai.com/v1/realtime",
session_properties: Optional[events.SessionProperties] = None,
start_audio_paused: bool = False,
send_transcription_frames: bool = True,
send_transcription_frames: Optional[bool] = None,
**kwargs,
):
"""Initialize the OpenAI Realtime LLM service.
Args:
api_key: OpenAI API key for authentication.
model: OpenAI model name. Defaults to "gpt-4o-realtime-preview-2025-06-03".
model: OpenAI model name. Defaults to "gpt-realtime".
base_url: WebSocket base URL for the realtime API.
Defaults to "wss://api.openai.com/v1/realtime".
session_properties: Configuration properties for the realtime session.
If None, uses default SessionProperties.
start_audio_paused: Whether to start with audio input paused. Defaults to False.
send_transcription_frames: Whether to emit transcription frames. Defaults to True.
send_transcription_frames: Whether to emit transcription frames.
.. deprecated:: 0.0.92
This parameter is deprecated and will be removed in a future version.
Transcription frames are always sent.
**kwargs: Additional arguments passed to parent LLMService.
"""
if send_transcription_frames is not None:
import warnings
with warnings.catch_warnings():
warnings.simplefilter("always")
warnings.warn(
"`send_transcription_frames` is deprecated and will be removed in a future version. "
"Transcription frames are always sent.",
DeprecationWarning,
stacklevel=2,
)
full_url = f"{base_url}?model={model}"
super().__init__(base_url=full_url, **kwargs)
@@ -135,10 +150,11 @@ class OpenAIRealtimeLLMService(LLMService):
session_properties or events.SessionProperties()
)
self._audio_input_paused = start_audio_paused
self._send_transcription_frames = send_transcription_frames
self._websocket = None
self._receive_task = None
self._context = None
self._context: LLMContext = None
self._llm_needs_conversation_setup = True
self._disconnecting = False
self._api_session_ready = False
@@ -148,8 +164,8 @@ class OpenAIRealtimeLLMService(LLMService):
self._current_audio_response = None
self._messages_added_manually = {}
self._user_and_response_message_tuple = None
self._pending_function_calls = {} # Track function calls by call_id
self._completed_tool_calls = set()
self._register_event_handler("on_conversation_item_created")
self._register_event_handler("on_conversation_item_updated")
@@ -347,22 +363,13 @@ class OpenAIRealtimeLLMService(LLMService):
if isinstance(frame, TranscriptionFrame):
pass
elif isinstance(frame, OpenAILLMContextFrame):
context: OpenAIRealtimeLLMContext = OpenAIRealtimeLLMContext.upgrade_to_realtime(
elif isinstance(frame, (LLMContextFrame, OpenAILLMContextFrame)):
context = (
frame.context
if isinstance(frame, LLMContextFrame)
else LLMContext.from_openai_context(frame.context)
)
if not self._context:
self._context = context
elif frame.context is not self._context:
# If the context has changed, reset the conversation
self._context = context
await self.reset_conversation()
# Run the LLM at next opportunity
await self._create_response()
elif isinstance(frame, LLMContextFrame):
raise NotImplementedError(
"Universal LLMContext is not yet supported for OpenAI Realtime."
)
await self._handle_context(context)
elif isinstance(frame, InputAudioRawFrame):
if not self._audio_input_paused:
await self._send_user_audio(frame)
@@ -376,29 +383,33 @@ class OpenAIRealtimeLLMService(LLMService):
await self._handle_bot_stopped_speaking()
elif isinstance(frame, LLMMessagesAppendFrame):
await self._handle_messages_append(frame)
elif isinstance(frame, RealtimeMessagesUpdateFrame):
self._context = frame.context
elif isinstance(frame, LLMUpdateSettingsFrame):
self._session_properties = events.SessionProperties(**frame.settings)
await self._update_settings()
elif isinstance(frame, LLMSetToolsFrame):
await self._update_settings()
elif isinstance(frame, RealtimeFunctionCallResultFrame):
await self._handle_function_call_result(frame.result_frame)
await self.push_frame(frame, direction)
async def _handle_context(self, context: LLMContext):
if not self._context:
# We got our initial context
self._context = context
# Initialize our bookkeeping of already-completed tool calls in
# the context
await self._process_completed_function_calls(send_new_results=False)
# Run the LLM at next opportunity
await self._create_response()
else:
# We got an updated context.
# This may contain a new user message or tool call result.
self._context = context
# Send results for newly-completed function calls, if any.
await self._process_completed_function_calls(send_new_results=True)
async def _handle_messages_append(self, frame):
logger.error("!!! NEED TO IMPLEMENT MESSAGES APPEND")
async def _handle_function_call_result(self, frame):
item = events.ConversationItem(
type="function_call_output",
call_id=frame.tool_call_id,
output=json.dumps(frame.result),
)
await self.send_client_event(events.ConversationItemCreateEvent(item=item))
#
# websocket communication
#
@@ -439,16 +450,21 @@ class OpenAIRealtimeLLMService(LLMService):
if self._receive_task:
await self.cancel_task(self._receive_task, timeout=1.0)
self._receive_task = None
self._completed_tool_calls = set()
self._disconnecting = False
except Exception as e:
logger.error(f"{self} error disconnecting: {e}")
async def _ws_send(self, realtime_message):
try:
if self._websocket:
if not self._disconnecting and self._websocket:
await self._websocket.send(json.dumps(realtime_message))
except Exception as e:
if self._disconnecting:
if self._disconnecting or not self._websocket:
# We're in the process of disconnecting.
# (If not self._websocket, that could indicate that we
# somehow *started* the websocket send attempt while we still
# had a connection)
return
logger.error(f"Error sending message to websocket: {e}")
# In server-to-server contexts, a WebSocket error should be quite rare. Given how hard
@@ -459,13 +475,20 @@ class OpenAIRealtimeLLMService(LLMService):
async def _update_settings(self):
settings = self._session_properties
# tools given in the context override the tools in the session properties
if self._context and self._context.tools:
settings.tools = self._context.tools
# instructions in the context come from an initial "system" message in the
# messages list, and override instructions in the session properties
if self._context and self._context._session_instructions:
settings.instructions = self._context._session_instructions
if self._context:
adapter: OpenAIRealtimeLLMAdapter = self.get_llm_adapter()
llm_invocation_params = adapter.get_llm_invocation_params(self._context)
# tools given in the context override the tools in the session properties
if llm_invocation_params["tools"]:
settings.tools = llm_invocation_params["tools"]
# instructions in the context come from an initial "system" message in the
# messages list, and override instructions in the session properties
if llm_invocation_params["system_instruction"]:
settings.instructions = llm_invocation_params["system_instruction"]
await self.send_client_event(events.SessionUpdateEvent(session=settings))
#
@@ -571,12 +594,7 @@ class OpenAIRealtimeLLMService(LLMService):
del self._messages_added_manually[evt.item.id]
return
if evt.item.role == "user":
# We need to wait for completion of both user message and response message. Then we'll
# add both to the context. User message is complete when we have a "transcript" field
# that is not None. Response message is complete when we get a "response.done" event.
self._user_and_response_message_tuple = (evt.item, {"done": False, "output": []})
elif evt.item.role == "assistant":
if evt.item.role == "assistant":
self._current_assistant_response = evt.item
await self.push_frame(LLMFullResponseStartFrame())
@@ -587,11 +605,11 @@ class OpenAIRealtimeLLMService(LLMService):
# For now, no additional logic needed beyond the event handler call
async def _handle_evt_input_audio_transcription_delta(self, evt):
if self._send_transcription_frames:
await self.push_frame(
# no way to get a language code?
InterimTranscriptionFrame(evt.delta, "", time_now_iso8601(), result=evt)
)
await self.push_frame(
# no way to get a language code?
InterimTranscriptionFrame(evt.delta, "", time_now_iso8601(), result=evt),
direction=FrameDirection.UPSTREAM,
)
@traced_stt
async def _handle_user_transcription(
@@ -608,22 +626,12 @@ class OpenAIRealtimeLLMService(LLMService):
"""
await self._call_event_handler("on_conversation_item_updated", evt.item_id, None)
if self._send_transcription_frames:
await self.push_frame(
# no way to get a language code?
TranscriptionFrame(evt.transcript, "", time_now_iso8601(), result=evt)
)
await self._handle_user_transcription(evt.transcript, True, Language.EN)
pair = self._user_and_response_message_tuple
if pair:
user, assistant = pair
user.content[0].transcript = evt.transcript
if assistant["done"]:
self._user_and_response_message_tuple = None
self._context.add_user_content_item_as_message(user)
else:
# User message without preceding conversation.item.created. Bug?
logger.warning(f"Transcript for unknown user message: {evt}")
await self.push_frame(
# no way to get a language code?
TranscriptionFrame(evt.transcript, "", time_now_iso8601(), result=evt),
FrameDirection.UPSTREAM,
)
await self._handle_user_transcription(evt.transcript, True, Language.EN)
async def _handle_conversation_item_retrieved(self, evt: events.ConversationItemRetrieved):
futures = self._retrieve_conversation_item_futures.pop(evt.item.id, None)
@@ -653,26 +661,17 @@ class OpenAIRealtimeLLMService(LLMService):
# response content
for item in evt.response.output:
await self._call_event_handler("on_conversation_item_updated", item.id, item)
pair = self._user_and_response_message_tuple
if pair:
user, assistant = pair
assistant["done"] = True
assistant["output"] = evt.response.output
if user.content[0].transcript is not None:
self._user_and_response_message_tuple = None
self._context.add_user_content_item_as_message(user)
else:
# Response message without preceding user message (standalone response)
# Function calls in this response were already processed immediately when arguments were complete
logger.debug(f"Handling standalone response: {evt.response.id}")
async def _handle_evt_text_delta(self, evt):
# We receive text deltas (as opposed to audio transcript deltas) when
# the output modality is "text"
if evt.delta:
await self.push_frame(LLMTextFrame(evt.delta))
async def _handle_evt_audio_transcript_delta(self, evt):
# We receive audio transcript deltas (as opposed to text deltas) when
# the output modality is "audio" (the default)
if evt.delta:
await self.push_frame(LLMTextFrame(evt.delta))
await self.push_frame(TTSTextFrame(evt.delta))
async def _handle_evt_function_call_arguments_done(self, evt):
@@ -760,9 +759,11 @@ class OpenAIRealtimeLLMService(LLMService):
"""
logger.debug("Resetting conversation")
await self._disconnect()
if self._context:
self._context.llm_needs_settings_update = True
self._context.llm_needs_initial_messages = True
# Prepare to setup server-side conversation from local context again
self._llm_needs_conversation_setup = True
await self._process_completed_function_calls(send_new_results=False)
await self._connect()
@traced_openai_realtime(operation="llm_request")
@@ -771,19 +772,29 @@ class OpenAIRealtimeLLMService(LLMService):
self._run_llm_when_api_session_ready = True
return
if self._context.llm_needs_initial_messages:
messages = self._context.get_messages_for_initializing_history()
adapter: OpenAIRealtimeLLMAdapter = self.get_llm_adapter()
# Configure the LLM for this session if needed
if self._llm_needs_conversation_setup:
logger.debug(
f"Setting up conversation on OpenAI Realtime LLM service with initial messages: {adapter.get_messages_for_logging(self._context)}"
)
# Send initial messages
llm_invocation_params = adapter.get_llm_invocation_params(self._context)
messages = llm_invocation_params["messages"]
for item in messages:
evt = events.ConversationItemCreateEvent(item=item)
self._messages_added_manually[evt.item.id] = True
await self.send_client_event(evt)
self._context.llm_needs_initial_messages = False
if self._context.llm_needs_settings_update:
# Send new settings if needed
await self._update_settings()
self._context.llm_needs_settings_update = False
logger.debug(f"Creating response: {self._context.get_messages_for_logging()}")
# We're done configuring the LLM for this session
self._llm_needs_conversation_setup = False
logger.debug(f"Creating response")
await self.push_frame(LLMFullResponseStartFrame())
await self.start_processing_metrics()
@@ -794,19 +805,50 @@ class OpenAIRealtimeLLMService(LLMService):
)
)
async def _process_completed_function_calls(self, send_new_results: bool):
# Check for set of completed function calls in the context
sent_new_result = False
for message in self._context.get_messages():
if message.get("role") and message.get("content") != "IN_PROGRESS":
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
if send_new_results:
sent_new_result = True
await self._send_tool_result(tool_call_id, message.get("content"))
self._completed_tool_calls.add(tool_call_id)
# If we reported any new tool call results to the service, trigger
# another response
if sent_new_result:
await self._create_response()
async def _send_user_audio(self, frame):
payload = base64.b64encode(frame.audio).decode("utf-8")
await self.send_client_event(events.InputAudioBufferAppendEvent(audio=payload))
async def _send_tool_result(self, tool_call_id: str, result: str):
item = events.ConversationItem(
type="function_call_output",
call_id=tool_call_id,
output=json.dumps(result),
)
await self.send_client_event(events.ConversationItemCreateEvent(item=item))
def create_context_aggregator(
self,
context: OpenAILLMContext,
*,
user_params: LLMUserAggregatorParams = LLMUserAggregatorParams(),
assistant_params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(),
) -> OpenAIContextAggregatorPair:
) -> LLMContextAggregatorPair:
"""Create an instance of OpenAIContextAggregatorPair from an OpenAILLMContext.
NOTE: this method exists only for backward compatibility. New code
should instead do:
context = LLMContext(...)
context_aggregator = LLMContextAggregatorPair(context)
Constructor keyword arguments for both the user and assistant aggregators can be provided.
Args:
@@ -819,11 +861,41 @@ class OpenAIRealtimeLLMService(LLMService):
the user and one for the assistant, encapsulated in an
OpenAIContextAggregatorPair.
"""
context.set_llm_adapter(self.get_llm_adapter())
OpenAIRealtimeLLMContext.upgrade_to_realtime(context)
user = OpenAIRealtimeUserContextAggregator(context, params=user_params)
# Log warning about transcription frame direction change in 0.0.92.
# We're putting this warning here rather than in the constructor so
# that it shows up for folks who haven't updated their code at all
# since 0.0.92, gives them a way to acknowledge and dismiss the
# warning, and encourages adoption of a new preferred pattern.
logger.warning(
"As of version 0.0.92, TranscriptionFrames and InterimTranscriptionFrames "
"now go upstream from OpenAIRealtimeLLMService, so if you're using "
"TranscriptProcessor, say, you'll want to adjust accordingly:\n\n"
"pipeline = Pipeline(\n"
" [\n"
" transport.input(),\n"
" context_aggregator.user(),\n\n"
" # BEFORE\n"
" llm,\n"
" transcript.user(),\n\n"
" # AFTER\n"
" transcript.user(),\n"
" llm,\n\n"
" transport.output(),\n"
" transcript.assistant(),\n"
" context_aggregator.assistant(),\n"
" ]\n"
")\n\n"
"Also, LLMTextFrames are no longer pushed from "
"OpenAIRealtimeLLMService when it's configured with "
"output_modalities=['audio']. Listen for TTSTextFrames instead.\n\n"
"Once you've made the appropriate changes (if needed), you can "
"dismiss this warning by updating to the new context-setup pattern:\n\n"
" context = LLMContext(messages, tools)\n"
" context_aggregator = LLMContextAggregatorPair(context)\n"
)
context = LLMContext.from_openai_context(context)
assistant_params.expect_stripped_words = False
assistant = OpenAIRealtimeAssistantContextAggregator(context, params=assistant_params)
return OpenAIContextAggregatorPair(_user=user, _assistant=assistant)
return LLMContextAggregatorPair(
context, user_params=user_params, assistant_params=assistant_params
)

21
src/pipecat/services/openai_realtime/context.py
View File

@@ -4,18 +4,15 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""OpenAI Realtime LLM context and aggregator implementations."""
"""OpenAI Realtime LLM context and aggregator implementations.
import warnings
.. deprecated:: 0.0.91
OpenAI Realtime no longer uses types from this module under the hood.
It now uses `LLMContext` and `LLMContextAggregatorPair`.
Using the new patterns should allow you to not need types from this module.
See deprecation warning in pipecat.services.openai.realtime.context for
more details.
"""
from pipecat.services.openai.realtime.context import *
with warnings.catch_warnings():
warnings.simplefilter("always")
warnings.warn(
"Types in pipecat.services.openai_realtime.context are deprecated. "
"Please use the equivalent types from "
"pipecat.services.openai.realtime.context instead.",
DeprecationWarning,
stacklevel=2,
)

94
src/pipecat/services/simli/video.py
View File

@@ -7,9 +7,12 @@
"""Simli video service for real-time avatar generation."""
import asyncio
import warnings
from typing import Optional
import numpy as np
from loguru import logger
from pydantic import BaseModel
from pipecat.frames.frames import (
CancelFrame,
@@ -41,30 +44,103 @@ class SimliVideoService(FrameProcessor):
audio resampling, video frame processing, and connection management.
"""
class InputParams(BaseModel):
"""Input parameters for Simli video configuration.
Parameters:
max_session_length: Absolute maximum session duration in seconds.
Avatar will disconnect after this time even if it's speaking.
max_idle_time: Maximum duration in seconds the avatar is not speaking
before the avatar disconnects.
"""
max_session_length: Optional[int] = None
max_idle_time: Optional[int] = None
def __init__(
self,
simli_config: SimliConfig,
*,
api_key: Optional[str] = None,
face_id: Optional[str] = None,
simli_config: Optional[SimliConfig] = None,
use_turn_server: bool = False,
latency_interval: int = 0,
simli_url: str = "https://api.simli.ai",
is_trinity_avatar: bool = False,
params: Optional[InputParams] = None,
**kwargs,
):
"""Initialize the Simli video service.
Args:
api_key: Simli API key for authentication.
face_id: Simli Face ID. For Trinity avatars, specify "faceId/emotionId"
to use a different emotion than the default.
simli_config: Configuration object for Simli client settings.
use_turn_server: Whether to use TURN server for connection. Defaults to False.
latency_interval: Latency interval setting for sending health checks to check the latency to Simli Servers. Defaults to 0.
simli_url: URL of the simli servers. Can be changed for custom deployments of enterprise users.
is_trinity_avatar: boolean to tell simli client that this is a Trinity avatar which reduces latency when using Trinity.
Use api_key and face_id instead.
.. deprecated:: 0.0.92
The 'simli_config' parameter is deprecated and will be removed in a future version.
Please use 'api_key' and 'face_id' parameters instead.
use_turn_server: Whether to use TURN server for connection. Defaults to False.
latency_interval: Latency interval setting for sending health checks to check
the latency to Simli Servers. Defaults to 0.
simli_url: URL of the simli servers. Can be changed for custom deployments
of enterprise users.
is_trinity_avatar: Boolean to tell simli client that this is a Trinity avatar
which reduces latency when using Trinity.
params: Additional input parameters for session configuration.
**kwargs: Additional arguments passed to the parent FrameProcessor.
"""
super().__init__()
super().__init__(**kwargs)
params = params or SimliVideoService.InputParams()
# Handle deprecated simli_config parameter
if simli_config is not None:
if api_key is not None or face_id is not None:
raise ValueError(
"Cannot specify both simli_config and api_key/face_id. "
"Please use api_key and face_id (simli_config is deprecated)."
)
warnings.warn(
"The 'simli_config' parameter is deprecated and will be removed in a future version. "
"Please use 'api_key' and 'face_id' parameters instead, with optional 'params' for "
"max_session_length and max_idle_time configuration.",
DeprecationWarning,
stacklevel=2,
)
# Use the provided simli_config
config = simli_config
else:
# Validate new parameters
if api_key is None:
raise ValueError("api_key is required")
if face_id is None:
raise ValueError("face_id is required")
# Build SimliConfig from new parameters
# Only pass optional parameters if explicitly provided to use SimliConfig defaults
config_kwargs = {
"apiKey": api_key,
"faceId": face_id,
}
if params.max_session_length is not None:
config_kwargs["maxSessionLength"] = params.max_session_length
if params.max_idle_time is not None:
config_kwargs["maxIdleTime"] = params.max_idle_time
config = SimliConfig(**config_kwargs)
self._initialized = False
simli_config.maxIdleTime += 5
simli_config.maxSessionLength += 5
# Add buffer time to session limits
config.maxIdleTime += 5
config.maxSessionLength += 5
self._simli_client = SimliClient(
simli_config,
config,
use_turn_server,
latency_interval,
simliURL=simli_url,

386
src/pipecat/transports/daily/transport.py
View File

@@ -16,7 +16,7 @@ import time
from concurrent.futures import CancelledError as FuturesCancelledError
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass
from typing import Any, Awaitable, Callable, Dict, Mapping, Optional
from typing import Any, Awaitable, Callable, Dict, Mapping, Optional, Tuple
import aiohttp
from loguru import logger
@@ -419,6 +419,11 @@ class DailyAudioTrack:
track: CustomAudioTrack
# This is just a type alias for the errors returned by daily-python. Right now
# they are just a string.
CallClientError = str
class DailyTransportClient(EventHandler):
"""Core client for interacting with Daily's API.
@@ -553,14 +558,17 @@ class DailyTransportClient(EventHandler):
async def send_message(
self, frame: OutputTransportMessageFrame | OutputTransportMessageUrgentFrame
):
) -> Optional[CallClientError]:
"""Send an application message to participants.
Args:
frame: The message frame to send.
Returns:
error: An error description or None.
"""
if not self._joined:
return
return "Unable to send messages before joining."
participant_id = None
if isinstance(
@@ -572,7 +580,7 @@ class DailyTransportClient(EventHandler):
self._client.send_app_message(
frame.message, participant_id, completion=completion_callback(future)
)
await future
return await future
async def read_next_audio_frame(self) -> Optional[InputAudioRawFrame]:
"""Reads the next 20ms audio frame from the virtual speaker."""
@@ -754,9 +762,6 @@ class DailyTransportClient(EventHandler):
logger.info(f"Joined {self._room_url}")
if self._params.transcription_enabled:
await self.start_transcription(self._params.transcription_settings)
await self._callbacks.on_joined(data)
self._joined_event.set()
@@ -842,9 +847,6 @@ class DailyTransportClient(EventHandler):
# Call callback before leaving.
await self._callbacks.on_before_leave()
if self._params.transcription_enabled:
await self.stop_transcription()
# Remove any custom tracks, if any.
for track_name, _ in self._custom_audio_tracks.items():
await self.remove_custom_audio_track(track_name)
@@ -873,7 +875,7 @@ class DailyTransportClient(EventHandler):
self._client.release()
self._client = None
def participants(self):
def participants(self) -> Mapping[str, Any]:
"""Get current participants in the room.
Returns:
@@ -881,7 +883,7 @@ class DailyTransportClient(EventHandler):
"""
return self._client.participants()
def participant_counts(self):
def participant_counts(self) -> Mapping[str, Any]:
"""Get participant count information.
Returns:
@@ -889,165 +891,173 @@ class DailyTransportClient(EventHandler):
"""
return self._client.participant_counts()
async def start_dialout(self, settings):
async def start_dialout(self, settings) -> Tuple[str, Optional[CallClientError]]:
"""Start a dial-out call to a phone number.
Args:
settings: Dial-out configuration settings.
"""
logger.debug(f"Starting dialout: settings={settings}")
Returns:
session_id: Dail-out session ID.
error: An error description or None.
"""
future = self._get_event_loop().create_future()
self._client.start_dialout(settings, completion=completion_callback(future))
error = await future
if error:
logger.error(f"Unable to start dialout: {error}")
return await future
async def stop_dialout(self, participant_id):
async def stop_dialout(self, participant_id) -> Optional[CallClientError]:
"""Stop a dial-out call for a specific participant.
Args:
participant_id: ID of the participant to stop dial-out for.
"""
logger.debug(f"Stopping dialout: participant_id={participant_id}")
Returns:
error: An error description or None.
"""
future = self._get_event_loop().create_future()
self._client.stop_dialout(participant_id, completion=completion_callback(future))
error = await future
if error:
logger.error(f"Unable to stop dialout: {error}")
return await future
async def send_dtmf(self, settings):
async def send_dtmf(self, settings) -> Optional[CallClientError]:
"""Send DTMF tones during a call.
Args:
settings: DTMF settings including tones and target session.
Returns:
error: An error description or None.
"""
session_id = settings.get("sessionId") or self._dial_out_session_id
if not session_id:
logger.error("Unable to send DTMF: 'sessionId' is not set")
return
return "Can't send DTMF if 'sessionId' is not set"
# Update 'sessionId' field.
settings["sessionId"] = session_id
future = self._get_event_loop().create_future()
self._client.send_dtmf(settings, completion=completion_callback(future))
await future
return await future
async def sip_call_transfer(self, settings):
async def sip_call_transfer(self, settings) -> Optional[CallClientError]:
"""Transfer a SIP call to another destination.
Args:
settings: SIP call transfer settings.
Returns:
error: An error description or None.
"""
session_id = (
settings.get("sessionId") or self._dial_out_session_id or self._dial_in_session_id
)
if not session_id:
logger.error("Unable to transfer SIP call: 'sessionId' is not set")
return
return "Can't transfer SIP call if 'sessionId' is not set"
# Update 'sessionId' field.
settings["sessionId"] = session_id
future = self._get_event_loop().create_future()
self._client.sip_call_transfer(settings, completion=completion_callback(future))
await future
return await future
async def sip_refer(self, settings):
async def sip_refer(self, settings) -> Optional[CallClientError]:
"""Send a SIP REFER request.
Args:
settings: SIP REFER settings.
Returns:
error: An error description or None.
"""
future = self._get_event_loop().create_future()
self._client.sip_refer(settings, completion=completion_callback(future))
await future
return await future
async def start_recording(self, streaming_settings, stream_id, force_new):
async def start_recording(
self, streaming_settings, stream_id, force_new
) -> Tuple[str, Optional[CallClientError]]:
"""Start recording the call.
Args:
streaming_settings: Recording configuration settings.
stream_id: Unique identifier for the recording stream.
force_new: Whether to force a new recording session.
"""
logger.debug(
f"Starting recording: stream_id={stream_id} force_new={force_new} settings={streaming_settings}"
)
Returns:
stream_id: Unique identifier for the recording stream.
error: An error description or None.
"""
future = self._get_event_loop().create_future()
self._client.start_recording(
streaming_settings, stream_id, force_new, completion=completion_callback(future)
)
error = await future
if error:
logger.error(f"Unable to start recording: {error}")
return await future
async def stop_recording(self, stream_id):
async def stop_recording(self, stream_id) -> Optional[CallClientError]:
"""Stop recording the call.
Args:
stream_id: Unique identifier for the recording stream to stop.
"""
logger.debug(f"Stopping recording: stream_id={stream_id}")
Returns:
error: An error description or None.
"""
future = self._get_event_loop().create_future()
self._client.stop_recording(stream_id, completion=completion_callback(future))
error = await future
if error:
logger.error(f"Unable to stop recording: {error}")
return await future
async def start_transcription(self, settings):
async def start_transcription(self, settings) -> Optional[CallClientError]:
"""Start transcription for the call.
Args:
settings: Transcription configuration settings.
Returns:
error: An error description or None.
"""
if not self._token:
logger.warning("Transcription can't be started without a room token")
return
logger.debug(f"Starting transcription: settings={settings}")
return "Transcription can't be started without a room token"
future = self._get_event_loop().create_future()
self._client.start_transcription(
settings=self._params.transcription_settings.model_dump(exclude_none=True),
completion=completion_callback(future),
)
error = await future
if error:
logger.error(f"Unable to start transcription: {error}")
return await future
async def stop_transcription(self):
"""Stop transcription for the call."""
async def stop_transcription(self) -> Optional[CallClientError]:
"""Stop transcription for the call.
Returns:
error: An error description or None.
"""
if not self._token:
return
logger.debug(f"Stopping transcription")
return "Transcription can't be stopped without a room token"
future = self._get_event_loop().create_future()
self._client.stop_transcription(completion=completion_callback(future))
error = await future
if error:
logger.error(f"Unable to stop transcription: {error}")
return await future
async def send_prebuilt_chat_message(self, message: str, user_name: Optional[str] = None):
async def send_prebuilt_chat_message(
self, message: str, user_name: Optional[str] = None
) -> Optional[CallClientError]:
"""Send a chat message to Daily's Prebuilt main room.
Args:
message: The chat message to send.
user_name: Optional user name that will appear as sender of the message.
Returns:
error: An error description or None.
"""
if not self._joined:
return
return "Can't send message if not joined"
future = self._get_event_loop().create_future()
self._client.send_prebuilt_chat_message(
message, user_name=user_name, completion=completion_callback(future)
)
await future
return await future
async def capture_participant_transcription(self, participant_id: str):
"""Enable transcription capture for a specific participant.
@@ -1167,38 +1177,51 @@ class DailyTransportClient(EventHandler):
return track
async def remove_custom_audio_track(self, track_name: str):
async def remove_custom_audio_track(self, track_name: str) -> Optional[CallClientError]:
"""Remove a custom audio track.
Args:
track_name: Name of the custom audio track to remove.
Returns:
error: An error description or None.
"""
future = self._get_event_loop().create_future()
self._client.remove_custom_audio_track(
track_name=track_name,
completion=completion_callback(future),
)
await future
return await future
async def update_transcription(self, participants=None, instance_id=None):
async def update_transcription(
self, participants=None, instance_id=None
) -> Optional[CallClientError]:
"""Update transcription settings for specific participants.
Args:
participants: List of participant IDs to enable transcription for.
instance_id: Optional transcription instance ID.
Returns:
error: An error description or None.
"""
future = self._get_event_loop().create_future()
self._client.update_transcription(
participants, instance_id, completion=completion_callback(future)
)
await future
return await future
async def update_subscriptions(self, participant_settings=None, profile_settings=None):
async def update_subscriptions(
self, participant_settings=None, profile_settings=None
) -> Optional[CallClientError]:
"""Update media subscription settings.
Args:
participant_settings: Per-participant subscription settings.
profile_settings: Global subscription profile settings.
Returns:
error: An error description or None.
"""
future = self._get_event_loop().create_future()
self._client.update_subscriptions(
@@ -1206,32 +1229,42 @@ class DailyTransportClient(EventHandler):
profile_settings=profile_settings,
completion=completion_callback(future),
)
await future
return await future
async def update_publishing(self, publishing_settings: Mapping[str, Any]):
async def update_publishing(
self, publishing_settings: Mapping[str, Any]
) -> Optional[CallClientError]:
"""Update media publishing settings.
Args:
publishing_settings: Publishing configuration settings.
Returns:
error: An error description or None.
"""
future = self._get_event_loop().create_future()
self._client.update_publishing(
publishing_settings=publishing_settings,
completion=completion_callback(future),
)
await future
return await future
async def update_remote_participants(self, remote_participants: Mapping[str, Any]):
async def update_remote_participants(
self, remote_participants: Mapping[str, Any]
) -> Optional[CallClientError]:
"""Update settings for remote participants.
Args:
remote_participants: Remote participant configuration settings.
Returns:
error: An error description or None.
"""
future = self._get_event_loop().create_future()
self._client.update_remote_participants(
remote_participants=remote_participants, completion=completion_callback(future)
)
await future
return await future
#
#
@@ -1922,7 +1955,9 @@ class DailyOutputTransport(BaseOutputTransport):
Args:
frame: The transport message frame to send.
"""
await self._client.send_message(frame)
error = await self._client.send_message(frame)
if error:
logger.error(f"Unable to send message: {error}")
async def register_video_destination(self, destination: str):
"""Register a video output destination.
@@ -2166,7 +2201,7 @@ class DailyTransport(BaseTransport):
if self._output:
await self._output.queue_frame(frame, FrameDirection.DOWNSTREAM)
def participants(self):
def participants(self) -> Mapping[str, Any]:
"""Get current participants in the room.
Returns:
@@ -2174,7 +2209,7 @@ class DailyTransport(BaseTransport):
"""
return self._client.participants()
def participant_counts(self):
def participant_counts(self) -> Mapping[str, Any]:
"""Get participant count information.
Returns:
@@ -2182,76 +2217,155 @@ class DailyTransport(BaseTransport):
"""
return self._client.participant_counts()
async def start_dialout(self, settings=None):
async def start_dialout(self, settings=None) -> Tuple[str, Optional[CallClientError]]:
"""Start a dial-out call to a phone number.
Args:
settings: Dial-out configuration settings.
"""
await self._client.start_dialout(settings)
async def stop_dialout(self, participant_id):
Returns:
session_id: Dail-out session ID.
error: An error description or None.
"""
logger.debug(f"Starting dialout: settings={settings}")
session_id, error = await self._client.start_dialout(settings)
if error:
logger.error(f"Unable to start dialout: {error}")
return session_id, error
async def stop_dialout(self, participant_id) -> Optional[CallClientError]:
"""Stop a dial-out call for a specific participant.
Args:
participant_id: ID of the participant to stop dial-out for.
"""
await self._client.stop_dialout(participant_id)
async def sip_call_transfer(self, settings):
Returns:
error: An error description or None.
"""
logger.debug(f"Stopping dialout: participant_id={participant_id}")
error = await self._client.stop_dialout(participant_id)
if error:
logger.error(f"Unable to stop dialout: {error}")
return error
async def sip_call_transfer(self, settings) -> Optional[CallClientError]:
"""Transfer a SIP call to another destination.
Args:
settings: SIP call transfer settings.
"""
await self._client.sip_call_transfer(settings)
async def sip_refer(self, settings):
Returns:
error: An error description or None.
"""
logger.debug(f"Staring SIP call transfer: settings={settings}")
error = await self._client.sip_call_transfer(settings)
if error:
logger.error(f"Unable to transfer SIP call: {error}")
return error
async def sip_refer(self, settings) -> Optional[CallClientError]:
"""Send a SIP REFER request.
Args:
settings: SIP REFER settings.
"""
await self._client.sip_refer(settings)
async def start_recording(self, streaming_settings=None, stream_id=None, force_new=None):
Returns:
error: An error description or None.
"""
logger.debug(f"Staring SIP REFER: settings={settings}")
error = await self._client.sip_refer(settings)
if error:
logger.error(f"Unable to perform SIP REFER: {error}")
return error
async def start_recording(
self, streaming_settings=None, stream_id=None, force_new=None
) -> Tuple[str, Optional[CallClientError]]:
"""Start recording the call.
Args:
streaming_settings: Recording configuration settings.
stream_id: Unique identifier for the recording stream.
force_new: Whether to force a new recording session.
"""
await self._client.start_recording(streaming_settings, stream_id, force_new)
async def stop_recording(self, stream_id=None):
Returns:
stream_id: Unique identifier for the recording stream.
error: An error description or None.
"""
logger.debug(
f"Starting recording: stream_id={stream_id} force_new={force_new} settings={streaming_settings}"
)
r_id, error = await self._client.start_recording(streaming_settings, stream_id, force_new)
if error:
logger.error(f"Unable to start recording: {error}")
return r_id, error
async def stop_recording(self, stream_id=None) -> Optional[CallClientError]:
"""Stop recording the call.
Args:
stream_id: Unique identifier for the recording stream to stop.
"""
await self._client.stop_recording(stream_id)
async def start_transcription(self, settings=None):
Returns:
error: An error description or None.
"""
logger.debug(f"Stopping recording: stream_id={stream_id}")
error = await self._client.stop_recording(stream_id)
if error:
logger.error(f"Unable to stop recording: {error}")
return error
async def start_transcription(self, settings=None) -> Optional[CallClientError]:
"""Start transcription for the call.
Args:
settings: Transcription configuration settings.
Returns:
error: An error description or None.
"""
await self._client.start_transcription(settings)
logger.debug(f"Starting transcription: settings={settings}")
async def stop_transcription(self):
"""Stop transcription for the call."""
await self._client.stop_transcription()
error = await self._client.start_transcription(settings)
if error:
logger.error(f"Unable to start transcription: {error}")
return error
async def send_prebuilt_chat_message(self, message: str, user_name: Optional[str] = None):
async def stop_transcription(self) -> Optional[CallClientError]:
"""Stop transcription for the call.
Returns:
error: An error description or None.
"""
logger.debug(f"Stopping transcription")
error = await self._client.stop_transcription()
if error:
logger.error(f"Unable to stop transcription: {error}")
return error
async def send_prebuilt_chat_message(
self, message: str, user_name: Optional[str] = None
) -> Optional[CallClientError]:
"""Send a chat message to Daily's Prebuilt main room.
Args:
message: The chat message to send.
user_name: Optional user name that will appear as sender of the message.
Returns:
error: An error description or None.
"""
await self._client.send_prebuilt_chat_message(message, user_name)
error = await self._client.send_prebuilt_chat_message(message, user_name)
if error:
logger.error(f"Unable to send prebuilt chat message: {error}")
return error
async def capture_participant_transcription(self, participant_id: str):
"""Enable transcription capture for a specific participant.
@@ -2297,32 +2411,66 @@ class DailyTransport(BaseTransport):
participant_id, framerate, video_source, color_format
)
async def update_publishing(self, publishing_settings: Mapping[str, Any]):
async def update_publishing(
self, publishing_settings: Mapping[str, Any]
) -> Optional[CallClientError]:
"""Update media publishing settings.
Args:
publishing_settings: Publishing configuration settings.
"""
await self._client.update_publishing(publishing_settings=publishing_settings)
async def update_subscriptions(self, participant_settings=None, profile_settings=None):
Returns:
error: An error description or None.
"""
logger.debug(f"Updating publishing settings: settings={publishing_settings}")
error = await self._client.update_publishing(publishing_settings=publishing_settings)
if error:
logger.error(f"Unable to update publishing settings: {error}")
return error
async def update_subscriptions(
self, participant_settings=None, profile_settings=None
) -> Optional[CallClientError]:
"""Update media subscription settings.
Args:
participant_settings: Per-participant subscription settings.
profile_settings: Global subscription profile settings.
Returns:
error: An error description or None.
"""
await self._client.update_subscriptions(
participant_settings=participant_settings, profile_settings=profile_settings
logger.debug(
f"Updating subscriptions: participant_settings={participant_settings} profile_settings={profile_settings}"
)
async def update_remote_participants(self, remote_participants: Mapping[str, Any]):
error = await self._client.update_subscriptions(
participant_settings=participant_settings, profile_settings=profile_settings
)
if error:
logger.error(f"Unable to update subscription settings: {error}")
return error
async def update_remote_participants(
self, remote_participants: Mapping[str, Any]
) -> Optional[CallClientError]:
"""Update settings for remote participants.
Args:
remote_participants: Remote participant configuration settings.
Returns:
error: An error description or None.
"""
await self._client.update_remote_participants(remote_participants=remote_participants)
logger.debug(f"Updating remote participants: remote_participants={remote_participants}")
error = await self._client.update_remote_participants(
remote_participants=remote_participants
)
if error:
logger.error(f"Unable to update remote participants: {error}")
return error
async def _on_active_speaker_changed(self, participant: Any):
"""Handle active speaker change events."""
@@ -2330,6 +2478,12 @@ class DailyTransport(BaseTransport):
async def _on_joined(self, data):
"""Handle room joined events."""
if self._params.transcription_enabled:
# We report an error because we are starting transcription
# internally and if it fails we need to know.
error = await self.start_transcription(self._params.transcription_settings)
if error:
await self._on_error(f"Unable to start transcription: {error}")
await self._call_event_handler("on_joined", data)
async def _on_left(self):
@@ -2338,6 +2492,12 @@ class DailyTransport(BaseTransport):
async def _on_before_leave(self):
"""Handle before leave room events."""
if self._params.transcription_enabled:
# We report an error because we are stopping transcription
# internally and if it fails we need to know.
error = await self.stop_transcription()
if error:
await self._on_error(f"Unable to stop transcription: {error}")
await self._call_event_handler("on_before_leave")
async def _on_error(self, error):

4
src/pipecat/utils/tracing/service_decorators.py
View File

@@ -905,7 +905,9 @@ def traced_openai_realtime(operation: str) -> Callable:
# Capture context messages being sent
if hasattr(self, "_context") and self._context:
try:
messages = self._context.get_messages_for_logging()
messages = self.get_llm_adapter().get_messages_for_logging(
self._context
)
if messages:
operation_attrs["context_messages"] = json.dumps(messages)
except Exception as e:

12
uv.lock generated
View File

@@ -1282,13 +1282,13 @@ wheels = [
[[package]]
name = "daily-python"
version = "0.20.0"
version = "0.21.0"
source = { registry = "https://pypi.org/simple" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/9b/02/ce81ebf11a04cd133a5539e08f85060574711fff05a1d6ad29705f0755c1/daily_python-0.20.0-cp37-abi3-macosx_10_15_x86_64.whl", hash = "sha256:7da3f1df8cd9ef7f7fcc96ce688348dc903f62d82b6dd155a53bc64b7a74f3a7", size = 13259887, upload-time = "2025-10-16T22:14:12.262Z" },
{ url = "https://files.pythonhosted.org/packages/4a/1e/51f06f3486c978e1184af2271e800ce6a6e8a8f95d61ee6624bae88ae9cd/daily_python-0.20.0-cp37-abi3-macosx_11_0_arm64.whl", hash = "sha256:d02fd7b8c8079ceaa550ef23db052cdf70a8ffaf8ab6a8bc1a1e97bf0b939464", size = 11642453, upload-time = "2025-10-16T22:14:14.477Z" },
{ url = "https://files.pythonhosted.org/packages/71/c9/f767f0b479abd39330569ad61fb9db4661aae56cd74bb27c6f3483595463/daily_python-0.20.0-cp37-abi3-manylinux_2_28_aarch64.whl", hash = "sha256:a5c8718982c221dc18b41fb0692c9f8435f115f72e74994c94d3b9c6dad7c534", size = 13634216, upload-time = "2025-10-16T22:14:16.235Z" },
{ url = "https://files.pythonhosted.org/packages/e8/10/5c6d7b000bee36c2a0587a092a34c7486d2de831fc8e44ed42b16a6bd99f/daily_python-0.20.0-cp37-abi3-manylinux_2_28_x86_64.whl", hash = "sha256:ca9132aef1bdb5be663d1894b440dab1f998ebb3f45dfc31d44effabded4bc08", size = 14282189, upload-time = "2025-10-16T22:14:18.229Z" },
{ url = "https://files.pythonhosted.org/packages/ff/11/99590f8b7aad077f3f9b5b59d39b010aee0bd01b14dece8ae1e93d8080e7/daily_python-0.21.0-cp37-abi3-macosx_10_15_x86_64.whl", hash = "sha256:bdec96417825181559769bb2258ae688d1215949a1878336194e36fb452274a8", size = 13277066, upload-time = "2025-10-29T00:20:49.523Z" },
{ url = "https://files.pythonhosted.org/packages/e5/db/8c57f1a1b713ba3393584ac2be32d8074d3022a2c2c17c28eb4cd2aa3629/daily_python-0.21.0-cp37-abi3-macosx_11_0_arm64.whl", hash = "sha256:18677fa1415a0dc48b891cdf2fb8fe9dabc70e1b019d5aaa3d0699ccc8d187c9", size = 11644908, upload-time = "2025-10-29T00:20:52.106Z" },
{ url = "https://files.pythonhosted.org/packages/64/b6/b03f2f58a367d6ef4bb728715471542fdfa68afa8a177670139c3a2aadb7/daily_python-0.21.0-cp37-abi3-manylinux_2_28_aarch64.whl", hash = "sha256:97eb97352fe74227061b678e330b8befcfa4c694feb6eb2b09fe6eacec00ad6d", size = 13652356, upload-time = "2025-10-29T00:20:54.813Z" },
{ url = "https://files.pythonhosted.org/packages/f6/76/bde65f6f8d4c1679dc6c185fa37dae9223f6ddb4b7ced728ef46504956f7/daily_python-0.21.0-cp37-abi3-manylinux_2_28_x86_64.whl", hash = "sha256:68c3e36f609fc2fce79e4d17ecf1021eadd836506db6c5125f95c682bcf3612a", size = 14304643, upload-time = "2025-10-29T00:20:57.194Z" },
]
[[package]]
@@ -4637,7 +4637,7 @@ requires-dist = [
{ name = "azure-cognitiveservices-speech", marker = "extra == 'azure'", specifier = "~=1.42.0" },
{ name = "cartesia", marker = "extra == 'cartesia'", specifier = "~=2.0.3" },
{ name = "coremltools", marker = "extra == 'local-smart-turn'", specifier = ">=8.0" },
{ name = "daily-python", marker = "extra == 'daily'", specifier = "~=0.20.0" },
{ name = "daily-python", marker = "extra == 'daily'", specifier = "~=0.21.0" },
{ name = "deepgram-sdk", marker = "extra == 'deepgram'", specifier = "~=4.7.0" },
{ name = "docstring-parser", specifier = "~=0.16" },
{ name = "einops", marker = "extra == 'moondream'", specifier = "~=0.8.0" },