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

Merge pull request #111 from daily-co/user-transcription-aggregator

Browse Source 
pipeline: add UserTranscriptionAggregator
This commit is contained in:
Aleix Conchillo Flaqué
2024-04-09 23:34:52 +08:00
committed by GitHub
parent 4f04b10202 97b923e37e
commit 8d73cdb2ee
4 changed files with 87 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
examples/foundational/07-interruptible.py
View File

@@ -3,8 +3,8 @@ import aiohttp
import logging
import os
from dailyai.pipeline.aggregators import (
LLMResponseAggregator,
UserResponseAggregator,
LLMAssistantResponseAggregator,
LLMUserResponseAggregator,
)
from dailyai.pipeline.pipeline import Pipeline
@@ -63,8 +63,8 @@ async def main(room_url: str, token):
await transport.run_interruptible_pipeline(
pipeline,
post_processor=LLMResponseAggregator(messages),
pre_processor=UserResponseAggregator(messages),
post_processor=LLMAssistantResponseAggregator(messages),
pre_processor=LLMUserResponseAggregator(messages),
)
transport.transcription_settings["extra"]["punctuate"] = False

8
examples/starter-apps/chatbot.py
View File

@@ -6,8 +6,8 @@ from PIL import Image
from typing import AsyncGenerator
from dailyai.pipeline.aggregators import (
LLMResponseAggregator,
UserResponseAggregator,
LLMAssistantResponseAggregator,
LLMUserResponseAggregator,
)
from dailyai.pipeline.frames import (
ImageFrame,
@@ -135,8 +135,8 @@ async def main(room_url: str, token):
await transport.run_interruptible_pipeline(
pipeline,
post_processor=LLMResponseAggregator(messages),
pre_processor=UserResponseAggregator(messages),
post_processor=LLMAssistantResponseAggregator(messages),
pre_processor=LLMUserResponseAggregator(messages),
)
transport.transcription_settings["extra"]["endpointing"] = True

8
examples/starter-apps/storybot.py
View File

@@ -19,8 +19,8 @@ from dailyai.services.deepgram_ai_services import DeepgramTTSService
from dailyai.services.elevenlabs_ai_service import ElevenLabsTTSService
from dailyai.pipeline.aggregators import (
LLMAssistantContextAggregator,
UserResponseAggregator,
LLMResponseAggregator,
LLMAssistantResponseAggregator,
LLMUserResponseAggregator,
)
from dailyai.pipeline.frames import (
EndPipeFrame,
@@ -209,8 +209,8 @@ async def main(room_url: str, token):
key_id=os.getenv("FAL_KEY_ID"),
key_secret=os.getenv("FAL_KEY_SECRET"),
)
lra = LLMResponseAggregator(messages)
ura = UserResponseAggregator(messages)
lra = LLMAssistantResponseAggregator(messages)
ura = LLMUserResponseAggregator(messages)
sp = StoryProcessor(messages, story)
sig = StoryImageGenerator(story, llm, img)

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

@@ -22,6 +22,79 @@ from typing import AsyncGenerator, Coroutine, List
class ResponseAggregator(FrameProcessor):
"""This frame processor aggregates frames between a start and an end frame
into complete text frame sentences.
For example, frame input/output:
UserStartedSpeakingFrame() -> None
TranscriptionFrame("Hello,") -> None
TranscriptionFrame(" world.") -> None
UserStoppedSpeakingFrame() -> TextFrame("Hello world.")
Doctest:
>>> async def print_frames(aggregator, frame):
... async for frame in aggregator.process_frame(frame):
... if isinstance(frame, TextFrame):
... print(frame.text)
>>> aggregator = ResponseAggregator(start_frame = UserStartedSpeakingFrame,
... end_frame=UserStoppedSpeakingFrame,
... accumulator_frame=TranscriptionFrame,
... pass_through=False)
>>> asyncio.run(print_frames(aggregator, UserStartedSpeakingFrame()))
>>> asyncio.run(print_frames(aggregator, TranscriptionFrame("Hello,", 1, 1)))
>>> asyncio.run(print_frames(aggregator, TranscriptionFrame("world.", 1, 2)))
>>> asyncio.run(print_frames(aggregator, UserStoppedSpeakingFrame()))
Hello, world.
"""
def __init__(
self,
*,
start_frame,
end_frame,
accumulator_frame,
pass_through=True,
):
self.aggregation = ""
self.aggregating = False
self._start_frame = start_frame
self._end_frame = end_frame
self._accumulator_frame = accumulator_frame
self._pass_through = pass_through
async def process_frame(self, frame: Frame) -> AsyncGenerator[Frame, None]:
if isinstance(frame, self._start_frame):
self.aggregating = True
elif isinstance(frame, self._end_frame):
self.aggregating = False
# 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:
output = self.aggregation
self.aggregation = ""
yield self._end_frame()
yield TextFrame(output.strip())
elif isinstance(frame, self._accumulator_frame) and self.aggregating:
self.aggregation += f" {frame.text}"
if self._pass_through:
yield frame
else:
yield frame
class UserResponseAggregator(ResponseAggregator):
def __init__(self):
super().__init__(
start_frame=UserStartedSpeakingFrame,
end_frame=UserStoppedSpeakingFrame,
accumulator_frame=TranscriptionFrame,
pass_through=False,
)
class LLMResponseAggregator(FrameProcessor):
def __init__(
self,
@@ -66,7 +139,7 @@ class ResponseAggregator(FrameProcessor):
yield frame
class LLMResponseAggregator(ResponseAggregator):
class LLMAssistantResponseAggregator(LLMResponseAggregator):
def __init__(self, messages: list[dict]):
super().__init__(
messages=messages,
@@ -77,7 +150,7 @@ class LLMResponseAggregator(ResponseAggregator):
)
class UserResponseAggregator(ResponseAggregator):
class LLMUserResponseAggregator(LLMResponseAggregator):
def __init__(self, messages: list[dict]):
super().__init__(
messages=messages,