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

Merge pull request #596 from pipecat-ai/aleix/prepare-0.0.44

Browse Source 
prepare for pipecat 0.0.44
This commit is contained in:
Aleix Conchillo Flaqué
2024-10-15 18:13:07 -07:00
committed by GitHub
parent 0c4a513ca2 4eb2c95b63
commit d255b7d1b2
6 changed files with 97 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
CHANGELOG.md
View File

@@ -5,10 +5,14 @@ All notable changes to **Pipecat** will be documented in this file.
The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
## [Unreleased]
## [0.0.44] - 2024-10-15
### Added
- Added support for OpenAI Realtime API with the new
`OpenAILLMServiceRealtimeBeta` processor.
(see https://platform.openai.com/docs/guides/realtime/overview)
- Added `RTVIBotTranscriptionProcessor` which will send the RTVI
`bot-transcription` protocol message. These are TTS text aggregated (into
sentences) messages.
@@ -17,14 +21,44 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
`filter_code` to filter code from text and `filter_tables` to filter tables
from text.
- Added `CanonicalMetricsService`. This processor uses the new
`AudioBufferProcessor` to capture conversation audio and later send it to
Canonical AI.
(see https://canonical.chat/)
- Added `AudioBufferProcessor`. This processor can be used to buffer mixed user and
bot audio. This can later be saved into an audio file or processed by some
audio analyzer.
- Added `on_first_participant_joined` event to `LiveKitTransport`.
### Changed
- LLM text responses are now logged properly as unicode characters.
- `UserStartedSpeakingFrame`, `UserStoppedSpeakingFrame`,
`BotStartedSpeakingFrame`, `BotStoppedSpeakingFrame`, `BotSpeakingFrame` and
`UserImageRequestFrame` are now based from `SystemFrame`
### Fixed
- Merge `RTVIBotLLMProcessor`/`RTVIBotLLMTextProcessor` and
`RTVIBotTTSProcessor`/`RTVIBotTTSTextProcessor` to avoid out of order issues.
- Fixed an issue in RTVI protocol that could cause a `bot-llm-stopped` or
`bot-tts-stopped` message to be sent before a `bot-llm-text` or `bot-tts-text`
message.
- Fixed `DeepgramSTTService` constructor settings not being merged with default
ones.
- Fixed an issue in Daily transport that would cause tasks to be hanging if
urgent transport messages were being sent from a transport event handler.
- Fixed an issue in `BaseOutputTransport` that would cause `EndFrame` to be
pushed downed too early and call `FrameProcessor.cleanup()` before letting the
transport stop properly.
## [0.0.43] - 2024-10-10
### Added

19
src/pipecat/processors/audio/audio_buffer_processor.py
View File

@@ -1,18 +1,17 @@
#
# Copyright (c) 2024, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import wave
from io import BytesIO
from pipecat.frames.frames import (
AudioRawFrame,
BotInterruptionFrame,
BotStartedSpeakingFrame,
BotStoppedSpeakingFrame,
Frame,
InputAudioRawFrame,
OutputAudioRawFrame,
StartInterruptionFrame,
StopInterruptionFrame,
UserStartedSpeakingFrame,
UserStoppedSpeakingFrame,
)
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
@@ -39,18 +38,18 @@ class AudioBufferProcessor(FrameProcessor):
def _buffer_has_audio(self, buffer: bytearray):
return buffer is not None and len(buffer) > 0
def _has_audio(self):
def has_audio(self):
return (
self._buffer_has_audio(self._user_audio_buffer)
and self._buffer_has_audio(self._assistant_audio_buffer)
and self._sample_rate is not None
)
def _reset_audio_buffer(self):
def reset_audio_buffer(self):
self._user_audio_buffer = bytearray()
self._assistant_audio_buffer = bytearray()
def _merge_audio_buffers(self):
def merge_audio_buffers(self):
with BytesIO() as buffer:
with wave.open(buffer, "wb") as wf:
wf.setnchannels(2)

38
src/pipecat/services/canonical.py
View File

@@ -1,9 +1,21 @@
#
# Copyright (c) 2024, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import aiohttp
import os
import uuid
from datetime import datetime
from typing import Dict, List, Tuple
import aiohttp
from pipecat.frames.frames import CancelFrame, EndFrame, Frame
from pipecat.processors.audio.audio_buffer_processor import AudioBufferProcessor
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.ai_services import AIService
from loguru import logger
try:
@@ -18,27 +30,18 @@ except ModuleNotFoundError as e:
raise Exception(f"Missing module: {e}")
from pipecat.frames.frames import CancelFrame, EndFrame, Frame
from pipecat.processors.audio.audio_buffer_processor import AudioBufferProcessor
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.ai_services import AIService
# Multipart upload part size in bytes, cannot be smaller than 5MB
PART_SIZE = 1024 * 1024 * 5
"""
This class extends AudioBufferProcessor to handle audio processing and uploading
for the Canonical Voice API.
"""
class CanonicalMetricsService(AIService):
"""
Initialize a CanonicalAudioProcessor instance.
"""Initialize a CanonicalAudioProcessor instance.
This class extends AudioBufferProcessor to handle audio processing and uploading
for the Canonical Voice API.
This class uses an AudioBufferProcessor to get the conversation audio and
uploads it to Canonical Voice API for audio processing.
Args:
call_id (str): Your unique identifier for the call. This is used to match the call in the Canonical Voice system to the call in your system.
assistant (str): Identifier for the AI assistant. This can be whatever you want, it's intended for you convenience so you can distinguish
between different assistants and a grouping mechanism for calls.
@@ -52,7 +55,6 @@ class CanonicalMetricsService(AIService):
output_dir (str): Directory path for saving temporary audio files.
The constructor also ensures that the output directory exists.
This class requires a Canonical API key to be set in the CANONICAL_API_KEY environment variable.
"""
def __init__(
@@ -90,17 +92,17 @@ class CanonicalMetricsService(AIService):
async def _process_audio(self):
pipeline = self._audio_buffer_processor
if pipeline._has_audio():
if pipeline.has_audio():
os.makedirs(self._output_dir, exist_ok=True)
filename = self._get_output_filename()
wave_data = pipeline._merge_audio_buffers()
wave_data = pipeline.merge_audio_buffers()
async with aiofiles.open(filename, "wb") as file:
await file.write(wave_data)
try:
await self._multipart_upload(filename)
pipeline._reset_audio_buffer()
pipeline.reset_audio_buffer()
await aiofiles.os.remove(filename)
except FileNotFoundError:
pass

33
src/pipecat/transports/base_output.py
View File

@@ -95,15 +95,6 @@ class BaseOutputTransport(FrameProcessor):
self._audio_out_task = self.get_event_loop().create_task(self._audio_out_task_handler())
async def stop(self, frame: EndFrame):
# At this point we have enqueued an EndFrame and we need to wait for
# that EndFrame to be processed by the sink tasks. We also need to wait
# for these tasks before cancelling the camera and audio tasks below
# because they might be still rendering.
if self._sink_task:
await self._sink_task
if self._sink_clock_task:
await self._sink_clock_task
# Cancel and wait for the camera output task to finish.
if self._camera_out_task and self._params.camera_out_enabled:
self._camera_out_task.cancel()
@@ -191,9 +182,12 @@ class BaseOutputTransport(FrameProcessor):
await self.push_frame(frame, direction)
# Control frames.
elif isinstance(frame, EndFrame):
await self._sink_clock_queue.put((sys.maxsize, frame.id, frame))
await self._sink_queue.put(frame)
# Process sink tasks.
await self._stop_sink_tasks(frame)
# Now we can stop.
await self.stop(frame)
# We finally push EndFrame down so PipelineTask stops nicely.
await self.push_frame(frame, direction)
# Other frames.
elif isinstance(frame, OutputAudioRawFrame):
await self._handle_audio(frame)
@@ -205,6 +199,20 @@ class BaseOutputTransport(FrameProcessor):
else:
await self._sink_queue.put(frame)
async def _stop_sink_tasks(self, frame: EndFrame):
# Let the sink tasks process the queue until they reach this EndFrame.
await self._sink_clock_queue.put((sys.maxsize, frame.id, frame))
await self._sink_queue.put(frame)
# At this point we have enqueued an EndFrame and we need to wait for
# that EndFrame to be processed by the sink tasks. We also need to wait
# for these tasks before cancelling the camera and audio tasks below
# because they might be still rendering.
if self._sink_task:
await self._sink_task
if self._sink_clock_task:
await self._sink_clock_task
async def _handle_interruptions(self, frame: Frame):
if not self.interruptions_allowed:
return
@@ -278,7 +286,8 @@ class BaseOutputTransport(FrameProcessor):
elif isinstance(frame, TTSStoppedFrame):
await self._bot_stopped_speaking()
await self.push_frame(frame)
else:
# We will push EndFrame later.
elif not isinstance(frame, EndFrame):
await self.push_frame(frame)
async def _sink_task_handler(self):

8
src/pipecat/transports/services/daily.py
View File

@@ -241,7 +241,7 @@ class DailyTransportClient(EventHandler):
self._callbacks = callbacks
async def send_message(self, frame: TransportMessageFrame | TransportMessageUrgentFrame):
if not self._client:
if not self._joined or self._leaving:
return
participant_id = None
@@ -740,13 +740,17 @@ class DailyOutputTransport(BaseOutputTransport):
self._messages_task.cancel()
await self._messages_task
self._messages_task = None
self._messages_task = None
# Leave the room.
await self._client.leave()
async def cancel(self, frame: CancelFrame):
# Parent stop.
await super().cancel(frame)
# Cancel messages task
if self._messages_task:
self._messages_task.cancel()
await self._messages_task
self._messages_task = None
# Leave the room.
await self._client.leave()

6
src/pipecat/transports/services/livekit.py
View File

@@ -1,3 +1,9 @@
#
# Copyright (c) 2024, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import asyncio
from dataclasses import dataclass
from typing import Any, Awaitable, Callable, List