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

Storybot and Chatbot examples (#58)

Browse Source 
* storybot

* storybot

* added pipeline.queue_frames

* fixup
This commit is contained in:
chadbailey59
2024-03-13 15:12:59 -05:00
committed by GitHub
parent e33820fe36
commit cf302fb765
40 changed files with 594 additions and 502 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -5,6 +5,7 @@ from dailyai.pipeline.frame_processor import FrameProcessor
from dailyai.pipeline.frames import (
EndFrame,
AudioFrame,
EndPipeFrame,
Frame,
ImageFrame,
@@ -14,7 +15,7 @@ from dailyai.pipeline.frames import (
TextFrame,
TranscriptionQueueFrame,
UserStartedSpeakingFrame,
UserStoppedSpeakingFrame
UserStoppedSpeakingFrame,
)
from dailyai.pipeline.pipeline import Pipeline
from dailyai.services.ai_services import AIService
@@ -23,6 +24,7 @@ from typing import AsyncGenerator, Callable, Coroutine, List
from dailyai.services.openai_llm_context import OpenAILLMContext
class ResponseAggregator(FrameProcessor):
def __init__(
@@ -44,9 +46,7 @@ class ResponseAggregator(FrameProcessor):
self._accumulator_frame = accumulator_frame
self._pass_through = pass_through
async def process_frame(
self, frame: Frame
) -> AsyncGenerator[Frame, None]:
async def process_frame(self, frame: Frame) -> AsyncGenerator[Frame, None]:
if not self.messages:
return
@@ -54,9 +54,13 @@ class ResponseAggregator(FrameProcessor):
self.aggregating = True
elif isinstance(frame, self._end_frame):
self.aggregating = False
self.messages.append({"role": self._role, "content": self.aggregation})
self.aggregation = ""
yield LLMMessagesQueueFrame(self.messages)
# Sometimes VAD triggers quickly on and off. If we don't get any transcription,
# it creates empty LLM message queue frames
if len(self.aggregation) > 0:
self.messages.append({"role": self._role, "content": self.aggregation})
self.aggregation = ""
yield self._end_frame()
yield LLMMessagesQueueFrame(self.messages)
elif isinstance(frame, self._accumulator_frame) and self.aggregating:
self.aggregation += f" {frame.text}"
if self._pass_through:
@@ -64,6 +68,7 @@ class ResponseAggregator(FrameProcessor):
else:
yield frame
class LLMResponseAggregator(ResponseAggregator):
def __init__(self, messages: list[dict]):
super().__init__(
@@ -71,9 +76,10 @@ class LLMResponseAggregator(ResponseAggregator):
role="assistant",
start_frame=LLMResponseStartFrame,
end_frame=LLMResponseEndFrame,
accumulator_frame=TextFrame
accumulator_frame=TextFrame,
)
class UserResponseAggregator(ResponseAggregator):
def __init__(self, messages: list[dict]):
super().__init__(
@@ -82,7 +88,7 @@ class UserResponseAggregator(ResponseAggregator):
start_frame=UserStartedSpeakingFrame,
end_frame=UserStoppedSpeakingFrame,
accumulator_frame=TranscriptionQueueFrame,
pass_through=False
pass_through=False,
)
@@ -103,9 +109,7 @@ class LLMContextAggregator(AIService):
self.complete_sentences = complete_sentences
self.pass_through = pass_through
async def process_frame(
self, frame: Frame
) -> AsyncGenerator[Frame, None]:
async def process_frame(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, TextFrame):
yield frame
@@ -137,6 +141,7 @@ class LLMContextAggregator(AIService):
self.messages.append({"role": self.role, "content": frame.text})
yield LLMMessagesQueueFrame(self.messages)
class LLMUserContextAggregator(LLMContextAggregator):
def __init__(
self, messages: list[dict], bot_participant_id=None, complete_sentences=True
@@ -176,12 +181,11 @@ class SentenceAggregator(FrameProcessor):
>>> asyncio.run(print_frames(aggregator, TextFrame(" world.")))
Hello, world.
"""
def __init__(self):
self.aggregation = ""
async def process_frame(
self, frame: Frame
) -> AsyncGenerator[Frame, None]:
async def process_frame(self, frame: Frame) -> AsyncGenerator[Frame, None]:
if isinstance(frame, TextFrame):
m = re.search("(.*[?.!])(.*)", frame.text)
if m:
@@ -233,12 +237,11 @@ class LLMFullResponseAggregator(FrameProcessor):
Hello, world. I am an LLM.
LLMResponseEndFrame
"""
def __init__(self):
self.aggregation = ""
async def process_frame(
self, frame: Frame
) -> AsyncGenerator[Frame, None]:
async def process_frame(self, frame: Frame) -> AsyncGenerator[Frame, None]:
if isinstance(frame, TextFrame):
self.aggregation += frame.text
elif isinstance(frame, LLMResponseEndFrame):
@@ -274,8 +277,9 @@ class StatelessTextTransformer(FrameProcessor):
else:
yield frame
class ParallelPipeline(FrameProcessor):
""" Run multiple pipelines in parallel.
"""Run multiple pipelines in parallel.
This class takes frames from its source queue and sends them to each
sub-pipeline. Each sub-pipeline emits its frames into this class's
@@ -292,6 +296,7 @@ class ParallelPipeline(FrameProcessor):
Since frame handlers pass through unhandled frames by convention, this
class de-dupes frames in its sink before yielding them.
"""
def __init__(self, pipeline_definitions: List[List[FrameProcessor]]):
self.sources = [asyncio.Queue() for _ in pipeline_definitions]
self.sink: asyncio.Queue[Frame] = asyncio.Queue()
@@ -327,6 +332,7 @@ class ParallelPipeline(FrameProcessor):
if not isinstance(frame, EndPipeFrame):
yield frame
class GatedAggregator(FrameProcessor):
"""Accumulate frames, with custom functions to start and stop accumulation.
Yields gate-opening frame before any accumulated frames, then ensuing frames
@@ -352,6 +358,7 @@ class GatedAggregator(FrameProcessor):
>>> asyncio.run(print_frames(aggregator, TextFrame("Goodbye.")))
Goodbye.
"""
def __init__(self, gate_open_fn, gate_close_fn, start_open):
self.gate_open_fn = gate_open_fn
self.gate_close_fn = gate_close_fn

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

@@ -5,7 +5,9 @@ from dailyai.services.openai_llm_context import OpenAILLMContext
class Frame:
pass
def __str__(self):
return f"{self.__class__.__name__}"
class ControlFrame(Frame):
# Control frames should contain no instance data, so
@@ -21,10 +23,21 @@ class StartFrame(ControlFrame):
class EndFrame(ControlFrame):
pass
class EndPipeFrame(ControlFrame):
pass
class PipelineStartedFrame(ControlFrame):
"""
Used by the transport to indicate that execution of a pipeline is starting
(or restarting). It should be the first frame your app receives when it
starts, or when an interruptible pipeline has been interrupted.
"""
pass
class LLMResponseStartFrame(ControlFrame):
pass
@@ -37,22 +50,34 @@ class LLMResponseEndFrame(ControlFrame):
class AudioFrame(Frame):
data: bytes
def __str__(self):
return f"{self.__class__.__name__}, size: {len(self.data)} B"
@dataclass()
class ImageFrame(Frame):
url: str | None
image: bytes
def __str__(self):
return f"{self.__class__.__name__}, url: {self.url}, image size: {len(self.image)} B"
@dataclass()
class SpriteFrame(Frame):
images: list[bytes]
def __str__(self):
return f"{self.__class__.name__}, list size: {len(self.images)}"
@dataclass()
class TextFrame(Frame):
text: str
def __str__(self):
return f'{self.__class__.__name__}: "{self.text}"'
@dataclass()
class TranscriptionQueueFrame(TextFrame):
@@ -74,15 +99,28 @@ class AppMessageQueueFrame(Frame):
message: Any
participantId: str
class UserStartedSpeakingFrame(Frame):
pass
class UserStoppedSpeakingFrame(Frame):
pass
class BotStartedSpeakingFrame(Frame):
pass
class BotStoppedSpeakingFrame(Frame):
pass
@dataclass()
class LLMFunctionStartFrame(Frame):
function_name: str
@dataclass()
class LLMFunctionCallFrame(Frame):
function_name: str

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

@@ -19,7 +19,7 @@ class Pipeline:
source: asyncio.Queue | None = None,
sink: asyncio.Queue[Frame] | None = None,
):
""" Create a new pipeline. By default neither the source nor sink
"""Create a new pipeline. By default neither the source nor sink
queues are set, so you'll need to pass them to this constructor or
call set_source and set_sink before using the pipeline. Note that
the transport's run_*_pipeline methods will set the source and sink
@@ -30,18 +30,18 @@ class Pipeline:
self.sink: asyncio.Queue[Frame] | None = sink
def set_source(self, source: asyncio.Queue[Frame]):
""" Set the source queue for this pipeline. Frames from this queue
"""Set the source queue for this pipeline. Frames from this queue
will be processed by each frame_processor in the pipeline, or order
from first to last. """
from first to last."""
self.source = source
def set_sink(self, sink: asyncio.Queue[Frame]):
""" Set the sink queue for this pipeline. After the last frame_processor
"""Set the sink queue for this pipeline. After the last frame_processor
has processed a frame, its output will be placed on this queue."""
self.sink = sink
async def get_next_source_frame(self) -> AsyncGenerator[Frame, None]:
""" Convenience function to get the next frame from the source queue. This
"""Convenience function to get the next frame from the source queue. This
lets us consistently have an AsyncGenerator yield frames, from either the
source queue or a frame_processor."""
if self.source is None:
@@ -53,13 +53,15 @@ class Pipeline:
) -> AsyncGenerator[Frame, None]:
if processors:
async for frame in processors[0].process_frame(initial_frame):
async for final_frame in self.run_pipeline_recursively(frame, processors[1:]):
async for final_frame in self.run_pipeline_recursively(
frame, processors[1:]
):
yield final_frame
else:
yield initial_frame
async def run_pipeline(self):
""" Run the pipeline. Take each frame from the source queue, pass it to
"""Run the pipeline. Take each frame from the source queue, pass it to
the first frame_processor, pass the output of that frame_processor to the
next in the list, etc. until the last frame_processor has processed the
resulting frames, then place those frames in the sink queue.
@@ -68,7 +70,8 @@ class Pipeline:
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
if it's not the last frame yielded by the last frame_processor in the pipeline.."""
if it's not the last frame yielded by the last frame_processor in the pipeline..
"""
if self.source is None or self.sink is None:
raise ValueError("Source or sink queue not set")
@@ -76,13 +79,26 @@ class Pipeline:
try:
while True:
initial_frame = await self.source.get()
async for frame in self.run_pipeline_recursively(initial_frame, self.processors):
async for frame in self.run_pipeline_recursively(
initial_frame, self.processors
):
await self.sink.put(frame)
if isinstance(initial_frame, EndFrame) or isinstance(initial_frame, EndPipeFrame):
if isinstance(initial_frame, EndFrame) or isinstance(
initial_frame, EndPipeFrame
):
break
except asyncio.CancelledError:
# this means there's been an interruption, do any cleanup necessary here.
for processor in self.processors:
await processor.interrupted()
pass
async def queue_frames(self, frames: Frame | List[Frame]):
"""Insert frames directly into a pipeline. This is typically used inside a transport
participant_joined callback to prompt a bot to start a conversation, for example.
"""
if not isinstance(frames, List):
frames = [frames]
for f in frames:
await self.source.put(f)