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

Remove Queue in frame names

Browse Source
This commit is contained in:
Moishe Lettvin
2024-03-06 14:09:06 -05:00
parent b9556716dd
commit 62fd371b97
25 changed files with 241 additions and 240 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -5,13 +5,13 @@ from tblib import Frame
from dailyai.pipeline.frame_processor import FrameProcessor
from dailyai.pipeline.frames import (
ControlQueueFrame,
EndParallelPipeQueueFrame,
EndStreamQueueFrame,
ControlFrame,
EndPipeFrame,
EndFrame,
LLMMessagesQueueFrame,
LLMResponseEndQueueFrame,
QueueFrame,
TextQueueFrame,
LLMResponseEndFrame,
Frame,
TextFrame,
TranscriptionQueueFrame,
)
from dailyai.pipeline.pipeline import Pipeline
@@ -38,10 +38,10 @@ class LLMContextAggregator(AIService):
self.pass_through = pass_through
async def process_frame(
self, frame: QueueFrame
) -> AsyncGenerator[QueueFrame, None]:
self, frame: Frame
) -> AsyncGenerator[Frame, None]:
# We don't do anything with non-text frames, pass it along to next in the pipeline.
if not isinstance(frame, TextQueueFrame):
if not isinstance(frame, TextFrame):
yield frame
return
@@ -71,7 +71,7 @@ class LLMContextAggregator(AIService):
self.messages.append({"role": self.role, "content": frame.text})
yield LLMMessagesQueueFrame(self.messages)
async def finalize(self) -> AsyncGenerator[QueueFrame, None]:
async def finalize(self) -> AsyncGenerator[Frame, None]:
# Send any dangling words that weren't finished with punctuation.
if self.complete_sentences and self.sentence:
self.messages.append({"role": self.role, "content": self.sentence})
@@ -106,18 +106,18 @@ class SentenceAggregator(FrameProcessor):
self.aggregation = ""
async def process_frame(
self, frame: QueueFrame
) -> AsyncGenerator[QueueFrame, None]:
if isinstance(frame, TextQueueFrame):
self, frame: Frame
) -> AsyncGenerator[Frame, None]:
if isinstance(frame, TextFrame):
m = re.search("(.*[?.!])(.*)", frame.text)
if m:
yield TextQueueFrame(self.aggregation + m.group(1))
yield TextFrame(self.aggregation + m.group(1))
self.aggregation = m.group(2)
else:
self.aggregation += frame.text
elif isinstance(frame, EndStreamQueueFrame):
elif isinstance(frame, EndFrame):
if self.aggregation:
yield TextQueueFrame(self.aggregation)
yield TextFrame(self.aggregation)
yield frame
else:
yield frame
@@ -128,12 +128,12 @@ class LLMFullResponseAggregator(FrameProcessor):
self.aggregation = ""
async def process_frame(
self, frame: QueueFrame
) -> AsyncGenerator[QueueFrame, None]:
if isinstance(frame, TextQueueFrame):
self, frame: Frame
) -> AsyncGenerator[Frame, None]:
if isinstance(frame, TextFrame):
self.aggregation += frame.text
elif isinstance(frame, LLMResponseEndQueueFrame):
yield TextQueueFrame(self.aggregation)
elif isinstance(frame, LLMResponseEndFrame):
yield TextFrame(self.aggregation)
self.aggregation = ""
else:
yield frame
@@ -143,20 +143,20 @@ class StatelessTextTransformer(FrameProcessor):
def __init__(self, transform_fn):
self.transform_fn = transform_fn
async def process_frame(self, frame: QueueFrame) -> AsyncGenerator[QueueFrame, None]:
if isinstance(frame, TextQueueFrame):
async def process_frame(self, frame: Frame) -> AsyncGenerator[Frame, None]:
if isinstance(frame, TextFrame):
result = self.transform_fn(frame.text)
if isinstance(result, Coroutine):
result = await result
yield TextQueueFrame(result)
yield TextFrame(result)
else:
yield frame
class ParallelPipeline(FrameProcessor):
def __init__(self, pipeline_definitions: List[List[FrameProcessor]]):
self.sources = [asyncio.Queue() for _ in pipeline_definitions]
self.sink: asyncio.Queue[QueueFrame] = asyncio.Queue()
self.sink: asyncio.Queue[Frame] = asyncio.Queue()
self.pipelines: list[Pipeline] = [
Pipeline(
pipeline_definition,
@@ -166,10 +166,10 @@ class ParallelPipeline(FrameProcessor):
for source, pipeline_definition in zip(self.sources, pipeline_definitions)
]
async def process_frame(self, frame: QueueFrame) -> AsyncGenerator[QueueFrame, None]:
async def process_frame(self, frame: Frame) -> AsyncGenerator[Frame, None]:
for source in self.sources:
await source.put(frame)
await source.put(EndParallelPipeQueueFrame())
await source.put(EndPipeFrame())
await asyncio.gather(*[pipeline.run_pipeline() for pipeline in self.pipelines])
@@ -186,7 +186,7 @@ class ParallelPipeline(FrameProcessor):
seen_ids.add(id(frame))
# Skip passing along EndParallelPipeQueueFrame, because we use them for our own flow control.
if not isinstance(frame, EndParallelPipeQueueFrame):
if not isinstance(frame, EndPipeFrame):
yield frame
class GatedAggregator(FrameProcessor):
@@ -194,9 +194,9 @@ class GatedAggregator(FrameProcessor):
self.gate_open_fn = gate_open_fn
self.gate_close_fn = gate_close_fn
self.gate_open = start_open
self.accumulator: List[QueueFrame] = []
self.accumulator: List[Frame] = []
async def process_frame(self, frame: QueueFrame) -> AsyncGenerator[QueueFrame, None]:
async def process_frame(self, frame: Frame) -> AsyncGenerator[Frame, None]:
if self.gate_open:
if self.gate_close_fn(frame):
self.gate_open = False

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

@@ -1,7 +1,7 @@
from abc import abstractmethod
from typing import AsyncGenerator
from dailyai.pipeline.frames import ControlQueueFrame, QueueFrame
from dailyai.pipeline.frames import ControlFrame, Frame
"""
This is the base class for all frame processors. Frame processors consume a frame
@@ -20,16 +20,16 @@ be closed, del'd, etc.
class FrameProcessor:
@abstractmethod
async def process_frame(
self, frame: QueueFrame
) -> AsyncGenerator[QueueFrame, None]:
if isinstance(frame, ControlQueueFrame):
self, frame: Frame
) -> AsyncGenerator[Frame, None]:
if isinstance(frame, ControlFrame):
yield frame
@abstractmethod
async def finalize(self) -> AsyncGenerator[QueueFrame, None]:
async def finalize(self) -> AsyncGenerator[Frame, None]:
# This is a trick for the interpreter (and linter) to know that this is a generator.
if False:
yield QueueFrame()
yield Frame()
@abstractmethod
async def interrupted(self) -> None:

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

@@ -2,72 +2,73 @@ from dataclasses import dataclass
from typing import Any
class QueueFrame:
class Frame:
pass
class ControlFrame(Frame):
# Control frames should contain no instance data, so
# equality is based solely on the class.
def __eq__(self, other):
return isinstance(other, self.__class__)
return type(other) == self.__class__
class ControlQueueFrame(QueueFrame):
class StartFrame(ControlFrame):
pass
class StartStreamQueueFrame(ControlQueueFrame):
class EndFrame(ControlFrame):
pass
class EndPipeFrame(ControlFrame):
pass
class EndStreamQueueFrame(ControlQueueFrame):
pass
class EndParallelPipeQueueFrame(ControlQueueFrame):
class LLMResponseStartFrame(ControlFrame):
pass
class LLMResponseStartQueueFrame(QueueFrame):
pass
class LLMResponseEndQueueFrame(QueueFrame):
class LLMResponseEndFrame(ControlFrame):
pass
@dataclass()
class AudioQueueFrame(QueueFrame):
class AudioFrame(Frame):
data: bytes
@dataclass()
class ImageQueueFrame(QueueFrame):
class ImageFrame(Frame):
url: str | None
image: bytes
@dataclass()
class SpriteQueueFrame(QueueFrame):
class SpriteFrame(Frame):
images: list[bytes]
@dataclass()
class TextQueueFrame(QueueFrame):
class TextFrame(Frame):
text: str
@dataclass()
class TranscriptionQueueFrame(TextQueueFrame):
class TranscriptionQueueFrame(TextFrame):
participantId: str
timestamp: str
@dataclass()
class LLMMessagesQueueFrame(QueueFrame):
class LLMMessagesQueueFrame(Frame):
messages: list[dict[str, str]] # TODO: define this more concretely!
class AppMessageQueueFrame(QueueFrame):
class AppMessageQueueFrame(Frame):
message: Any
participantId: str
class UserStartedSpeakingFrame(QueueFrame):
class UserStartedSpeakingFrame(Frame):
pass
class UserStoppedSpeakingFrame(QueueFrame):
class UserStoppedSpeakingFrame(Frame):
pass

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

@@ -2,7 +2,7 @@ import asyncio
from typing import AsyncGenerator, List
from dailyai.pipeline.frame_processor import FrameProcessor
from dailyai.pipeline.frames import EndParallelPipeQueueFrame, EndStreamQueueFrame, QueueFrame
from dailyai.pipeline.frames import EndPipeFrame, EndFrame, Frame
"""
This class manages a pipe of FrameProcessors, and runs them in sequence. The "source"
@@ -17,19 +17,19 @@ class Pipeline:
self,
processors: List[FrameProcessor],
source: asyncio.Queue | None = None,
sink: asyncio.Queue[QueueFrame] | None = None,
sink: asyncio.Queue[Frame] | None = None,
):
self.processors = processors
self.source: asyncio.Queue[QueueFrame] | None = source
self.sink: asyncio.Queue[QueueFrame] | None = sink
self.source: asyncio.Queue[Frame] | None = source
self.sink: asyncio.Queue[Frame] | None = sink
def set_source(self, source: asyncio.Queue[QueueFrame]):
def set_source(self, source: asyncio.Queue[Frame]):
self.source = source
def set_sink(self, sink: asyncio.Queue[QueueFrame]):
def set_sink(self, sink: asyncio.Queue[Frame]):
self.sink = sink
async def get_next_source_frame(self) -> AsyncGenerator[QueueFrame, None]:
async def get_next_source_frame(self) -> AsyncGenerator[Frame, None]:
if self.source is None:
raise ValueError("Source queue not set")
yield await self.source.get()
@@ -52,9 +52,9 @@ class Pipeline:
async for frame in frame_generator:
await self.sink.put(frame)
if isinstance(
frame, EndStreamQueueFrame
frame, EndFrame
) or isinstance(
frame, EndParallelPipeQueueFrame
frame, EndPipeFrame
):
return
except asyncio.CancelledError: