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

Cleanup the last few badly-named Frame types

Browse Source
This commit is contained in:
Moishe Lettvin
2024-03-28 12:36:24 -04:00
parent 22bbedec93
commit 27322108b7
26 changed files with 64 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
src/dailyai/pipeline/aggregators.py
View File

@@ -7,11 +7,11 @@ from dailyai.pipeline.frames import (
EndFrame,
EndPipeFrame,
Frame,
LLMMessagesQueueFrame,
LLMMessagesFrame,
LLMResponseEndFrame,
LLMResponseStartFrame,
TextFrame,
TranscriptionQueueFrame,
TranscriptionFrame,
UserStartedSpeakingFrame,
UserStoppedSpeakingFrame,
)
@@ -57,7 +57,7 @@ class ResponseAggregator(FrameProcessor):
{"role": self._role, "content": self.aggregation})
self.aggregation = ""
yield self._end_frame()
yield LLMMessagesQueueFrame(self.messages)
yield LLMMessagesFrame(self.messages)
elif isinstance(frame, self._accumulator_frame) and self.aggregating:
self.aggregation += f" {frame.text}"
if self._pass_through:
@@ -84,7 +84,7 @@ class UserResponseAggregator(ResponseAggregator):
role="user",
start_frame=UserStartedSpeakingFrame,
end_frame=UserStoppedSpeakingFrame,
accumulator_frame=TranscriptionQueueFrame,
accumulator_frame=TranscriptionFrame,
pass_through=False,
)
@@ -114,7 +114,7 @@ class LLMContextAggregator(AIService):
return
# Ignore transcription frames from the bot
if isinstance(frame, TranscriptionQueueFrame):
if isinstance(frame, TranscriptionFrame):
if frame.participantId == self.bot_participant_id:
return
@@ -126,19 +126,19 @@ class LLMContextAggregator(AIService):
# TODO: split up transcription by participant
if self.complete_sentences:
# type: ignore -- the linter thinks this isn't a TextQueueFrame, even
# type: ignore -- the linter thinks this isn't a TextFrame, even
# though we check it above
self.sentence += frame.text
if self.sentence.endswith((".", "?", "!")):
self.messages.append(
{"role": self.role, "content": self.sentence})
self.sentence = ""
yield LLMMessagesQueueFrame(self.messages)
yield LLMMessagesFrame(self.messages)
else:
# type: ignore -- the linter thinks this isn't a TextQueueFrame, even
# type: ignore -- the linter thinks this isn't a TextFrame, even
# though we check it above
self.messages.append({"role": self.role, "content": frame.text})
yield LLMMessagesQueueFrame(self.messages)
yield LLMMessagesFrame(self.messages)
class LLMUserContextAggregator(LLMContextAggregator):
@@ -334,7 +334,7 @@ class ParallelPipeline(FrameProcessor):
continue
seen_ids.add(id(frame))
# Skip passing along EndParallelPipeQueueFrame, because we use them
# Skip passing along EndPipeFrame, because we use them
# for our own flow control.
if not isinstance(frame, EndPipeFrame):
yield frame

4
src/dailyai/pipeline/frame_processor.py
View File

@@ -12,9 +12,9 @@ class FrameProcessor:
By convention, FrameProcessors should immediately yield any frames they don't process.
Stateful FrameProcessors should watch for the EndStreamQueueFrame and finalize their
Stateful FrameProcessors should watch for the EndFrame and finalize their
output, eg. yielding an unfinished sentence if they're aggregating LLM output to full
sentences. EndStreamQueueFrame is also a chance to clean up any services that need to
sentences. EndFrame is also a chance to clean up any services that need to
be closed, del'd, etc.
"""

6
src/dailyai/pipeline/frames.py
View File

@@ -102,7 +102,7 @@ class TextFrame(Frame):
@dataclass()
class TranscriptionQueueFrame(TextFrame):
class TranscriptionFrame(TextFrame):
"""A text frame with transcription-specific data. Will be placed in the
transport's receive queue when a participant speaks."""
participantId: str
@@ -126,7 +126,7 @@ class TTSEndFrame(ControlFrame):
@dataclass()
class LLMMessagesQueueFrame(Frame):
class LLMMessagesFrame(Frame):
"""A frame containing a list of LLM messages. Used to signal that an LLM
service should run a chat completion and emit an LLMStartFrames, TextFrames
and an LLMEndFrame.
@@ -137,7 +137,7 @@ class LLMMessagesQueueFrame(Frame):
@dataclass()
class OpenAILLMContextFrame(Frame):
"""Like an LLMMessagesQueueFrame, but with extra context specific to the
"""Like an LLMMessagesFrame, but with extra context specific to the
OpenAI API. The context in this message is also mutable, and will be
changed by the OpenAIContextAggregator frame processor."""
context: OpenAILLMContext

4
src/dailyai/pipeline/opeanai_llm_aggregator.py
View File

@@ -6,7 +6,7 @@ from dailyai.pipeline.frames import (
LLMResponseStartFrame,
OpenAILLMContextFrame,
TextFrame,
TranscriptionQueueFrame,
TranscriptionFrame,
UserStartedSpeakingFrame,
UserStoppedSpeakingFrame,
)
@@ -90,7 +90,7 @@ class OpenAIUserContextAggregator(OpenAIContextAggregator):
role="user",
start_frame=UserStartedSpeakingFrame,
end_frame=UserStoppedSpeakingFrame,
accumulator_frame=TranscriptionQueueFrame,
accumulator_frame=TranscriptionFrame,
pass_through=False,
)

4
src/dailyai/pipeline/pipeline.py
View File

@@ -81,8 +81,8 @@ class Pipeline:
The source and sink queues must be set before calling this method.
This method will exit when an EndStreamQueueFrame is placed on the sink queue.
No more frames will be placed on the sink queue after an EndStreamQueueFrame, even
This method will exit when an EndFrame is placed on the sink queue.
No more frames will be placed on the sink queue after an EndFrame, even
if it's not the last frame yielded by the last frame_processor in the pipeline..
"""

8
src/dailyai/serializers/protobuf_serializer.py
View File

@@ -1,6 +1,6 @@
import dataclasses
from typing import Text
from dailyai.pipeline.frames import AudioFrame, Frame, TextFrame, TranscriptionQueueFrame
from dailyai.pipeline.frames import AudioFrame, Frame, TextFrame, TranscriptionFrame
import dailyai.pipeline.protobufs.frames_pb2 as frame_protos
from dailyai.serializers.abstract_frame_serializer import FrameSerializer
@@ -9,7 +9,7 @@ class ProtobufFrameSerializer(FrameSerializer):
SERIALIZABLE_TYPES = {
TextFrame: "text",
AudioFrame: "audio",
TranscriptionQueueFrame: "transcription"
TranscriptionFrame: "transcription"
}
SERIALIZABLE_FIELDS = {v: k for k, v in SERIALIZABLE_TYPES.items()}
@@ -45,9 +45,9 @@ class ProtobufFrameSerializer(FrameSerializer):
... serializer.serialize(TextFrame(text='hello world')))
TextFrame(text='hello world')
>>> serializer.deserialize(serializer.serialize(TranscriptionQueueFrame(
>>> serializer.deserialize(serializer.serialize(TranscriptionFrame(
... text="Hello there!", participantId="123", timestamp="2021-01-01")))
TranscriptionQueueFrame(text='Hello there!', participantId='123', timestamp='2021-01-01')
TranscriptionFrame(text='Hello there!', participantId='123', timestamp='2021-01-01')
"""
proto = frame_protos.Frame.FromString(data)

4
src/dailyai/services/ai_services.py
View File

@@ -13,7 +13,7 @@ from dailyai.pipeline.frames import (
TTSEndFrame,
TTSStartFrame,
TextFrame,
TranscriptionQueueFrame,
TranscriptionFrame,
)
from abc import abstractmethod
@@ -128,7 +128,7 @@ class STTService(AIService):
ww.close()
content.seek(0)
text = await self.run_stt(content)
yield TranscriptionQueueFrame(text, "", str(time.time()))
yield TranscriptionFrame(text, "", str(time.time()))
class FrameLogger(AIService):

4
src/dailyai/services/anthropic_llm_service.py
View File

@@ -1,6 +1,6 @@
from typing import AsyncGenerator
from anthropic import AsyncAnthropic
from dailyai.pipeline.frames import Frame, LLMMessagesQueueFrame, TextFrame
from dailyai.pipeline.frames import Frame, LLMMessagesFrame, TextFrame
from dailyai.services.ai_services import LLMService
@@ -18,7 +18,7 @@ class AnthropicLLMService(LLMService):
self.max_tokens = max_tokens
async def process_frame(self, frame: Frame) -> AsyncGenerator[Frame, None]:
if not isinstance(frame, LLMMessagesQueueFrame):
if not isinstance(frame, LLMMessagesFrame):
yield frame
stream = await self.client.messages.create(

4
src/dailyai/services/local_stt_service.py
View File

@@ -4,7 +4,7 @@ import math
import time
from typing import AsyncGenerator
import wave
from dailyai.pipeline.frames import AudioFrame, Frame, TranscriptionQueueFrame
from dailyai.pipeline.frames import AudioFrame, Frame, TranscriptionFrame
from dailyai.services.ai_services import STTService
@@ -61,7 +61,7 @@ class LocalSTTService(STTService):
self._content.seek(0)
text = await self.run_stt(self._content)
self._new_wave()
yield TranscriptionQueueFrame(text, '', str(time.time()))
yield TranscriptionFrame(text, '', str(time.time()))
# If we get this far, this is a frame of silence
self._current_silence_frames += 1

4
src/dailyai/services/openai_api_llm_service.py
View File

@@ -6,7 +6,7 @@ from dailyai.pipeline.frames import (
Frame,
LLMFunctionCallFrame,
LLMFunctionStartFrame,
LLMMessagesQueueFrame,
LLMMessagesFrame,
LLMResponseEndFrame,
LLMResponseStartFrame,
OpenAILLMContextFrame,
@@ -75,7 +75,7 @@ class BaseOpenAILLMService(LLMService):
async def process_frame(self, frame: Frame) -> AsyncGenerator[Frame, None]:
if isinstance(frame, OpenAILLMContextFrame):
context: OpenAILLMContext = frame.context
elif isinstance(frame, LLMMessagesQueueFrame):
elif isinstance(frame, LLMMessagesFrame):
context = OpenAILLMContext.from_messages(frame.messages)
else:
yield frame

4
src/dailyai/transports/daily_transport.py
View File

@@ -10,7 +10,7 @@ from typing import Any
from dailyai.pipeline.frames import (
ReceivedAppMessageFrame,
TranscriptionQueueFrame,
TranscriptionFrame,
)
from threading import Event
@@ -269,7 +269,7 @@ class DailyTransport(ThreadedTransport, EventHandler):
elif "session_id" in message:
participantId = message["session_id"]
if self._my_participant_id and participantId != self._my_participant_id:
frame = TranscriptionQueueFrame(
frame = TranscriptionFrame(
message["text"], participantId, message["timestamp"])
asyncio.run_coroutine_threadsafe(
self.receive_queue.put(frame), self._loop)