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

Merge pull request #3385 from pipecat-ai/aleix/context-aggregator-turn-stop-messages

Browse Source 
user and assistant aggregator turn events
This commit is contained in:
Aleix Conchillo Flaqué
2026-01-09 09:52:48 -08:00
committed by GitHub
parent 5a0a5c120b 25f6ba76d6
commit c973445ab7
13 changed files with 689 additions and 109 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
changelog/3385.added.md Normal file
View File

@@ -0,0 +1,4 @@
- `LLMAssistantAggregator` now exposes the following events:
- `on_assistant_turn_started`: triggered when the assistant turn starts
- `on_assistant_turn_stopped`: triggered when the assistant turn ends
- `on_assistant_thought`: triggered when there's an assistant thought available

1
changelog/3385.deprecated.md Normal file
View File

@@ -0,0 +1 @@
- `TranscriptProcessor` and related data classes and frames (`TranscriptionMessage`, `ThoughtTranscriptionMessage`, `TranscriptionUpdateFrame`) are deprecated. Use `LLMUserAggregator`'s and `LLMAssistantAggregator`'s new events (`on_user_turn_stopped` and `on_assistant_turn_stopped`) instead.

1
changelog/3385.other.md Normal file
View File

@@ -0,0 +1 @@
- Added a new foundational example `28-user-assistant-turns.py` that shows how to use the new `LLMUserAggregator` and `LLMAssistantAggregator` events to gather a conversation transcript.

215
examples/foundational/28-user-assistant-turns.py Normal file
View File

@@ -0,0 +1,215 @@
#
# Copyright (c) 2024-2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import os
from typing import Optional
from dotenv import load_dotenv
from loguru import logger
from pipecat.audio.turn.smart_turn.local_smart_turn_v3 import LocalSmartTurnAnalyzerV3
from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.audio.vad.vad_analyzer import VADParams
from pipecat.frames.frames import LLMRunFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response_universal import (
AssistantTurnStoppedMessage,
LLMContextAggregatorPair,
LLMUserAggregatorParams,
UserTurnStoppedMessage,
)
from pipecat.runner.types import RunnerArguments
from pipecat.runner.utils import create_transport
from pipecat.services.cartesia.tts import CartesiaTTSService
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.openai.llm import OpenAILLMService
from pipecat.transports.base_transport import BaseTransport, TransportParams
from pipecat.transports.daily.transport import DailyParams
from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams
from pipecat.turns.user_stop import TurnAnalyzerUserTurnStopStrategy
from pipecat.turns.user_turn_strategies import UserTurnStrategies
load_dotenv(override=True)
class TranscriptHandler:
"""Handles real-time transcript processing and output.
Maintains a list of conversation messages and outputs them either to a log
or to a file as they are received. Each message includes its timestamp and role.
Attributes:
messages: List of all processed transcript messages
output_file: Optional path to file where transcript is saved. If None, outputs to log only.
"""
def __init__(self, output_file: Optional[str] = None):
"""Initialize handler with optional file output.
Args:
output_file: Path to output file. If None, outputs to log only.
"""
self.output_file: Optional[str] = output_file
logger.debug(
f"TranscriptHandler initialized {'with output_file=' + output_file if output_file else 'with log output only'}"
)
async def save_message(self, role: str, content: str, timestamp: str):
"""Save a single transcript message.
Outputs the message to the log and optionally to a file.
Args:
role: Who generated this transcript
content: The transcript to save
"""
line = f"[{timestamp}] {role}: {content}"
# Always log the message
logger.info(f"Transcript: {line}")
# Optionally write to file
if self.output_file:
try:
with open(self.output_file, "a", encoding="utf-8") as f:
f.write(line + "\n\n")
except Exception as e:
logger.error(f"Error saving transcript message to file: {e}")
async def on_user_transcript(self, message: UserTurnStoppedMessage):
"""Handle new user transcript message.
Args:
message: The new user message
"""
logger.debug(f"Received user transcript update")
await self.save_message("user", message.content, message.timestamp)
async def on_assistant_transcript(self, message: AssistantTurnStoppedMessage):
"""Handle new assistant transcript message.
Args:
message: The new assistant message
"""
logger.debug(f"Received assistant transcript update")
await self.save_message("assistant", message.content, message.timestamp)
# We store functions so objects (e.g. SileroVADAnalyzer) don't get
# instantiated. The function will be called when the desired transport gets
# selected.
transport_params = {
"daily": lambda: DailyParams(
audio_in_enabled=True,
audio_out_enabled=True,
vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)),
),
"twilio": lambda: FastAPIWebsocketParams(
audio_in_enabled=True,
audio_out_enabled=True,
vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)),
),
"webrtc": lambda: TransportParams(
audio_in_enabled=True,
audio_out_enabled=True,
vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)),
),
}
async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
logger.info(f"Starting bot")
stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
tts = CartesiaTTSService(
api_key=os.getenv("CARTESIA_API_KEY"),
voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady
)
llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
messages = [
{
"role": "system",
"content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be spoken aloud, so avoid special characters that can't easily be spoken, such as emojis or bullet points. Respond to what the user said in a creative, helpful, and brief way. Say hello.",
},
]
context = LLMContext(messages)
context_aggregator = LLMContextAggregatorPair(
context,
user_params=LLMUserAggregatorParams(
user_turn_strategies=UserTurnStrategies(
stop=[TurnAnalyzerUserTurnStopStrategy(turn_analyzer=LocalSmartTurnAnalyzerV3())]
),
),
)
user_aggregator = context_aggregator.user()
assistant_aggregator = context_aggregator.assistant()
# Create transcript processor and handler
transcript_handler = TranscriptHandler() # Output to log only
# transcript_handler = TranscriptHandler(output_file="transcript.txt") # Output to file and log
pipeline = Pipeline(
[
transport.input(), # Transport user input
stt, # STT
user_aggregator, # User responses
llm, # LLM
tts, # TTS
transport.output(), # Transport bot output
assistant_aggregator, # Assistant spoken responses
]
)
task = PipelineTask(
pipeline,
params=PipelineParams(
enable_metrics=True,
enable_usage_metrics=True,
),
idle_timeout_secs=runner_args.pipeline_idle_timeout_secs,
)
@transport.event_handler("on_client_connected")
async def on_client_connected(transport, client):
logger.info(f"Client connected")
# Start conversation - empty prompt to let LLM follow system instructions
await task.queue_frames([LLMRunFrame()])
@transport.event_handler("on_client_disconnected")
async def on_client_disconnected(transport, client):
logger.info(f"Client disconnected")
await task.cancel()
@user_aggregator.event_handler("on_user_turn_stopped")
async def on_user_turn_stopped(aggregator, strategy, message: UserTurnStoppedMessage):
await transcript_handler.on_user_transcript(message)
@assistant_aggregator.event_handler("on_assistant_turn_stopped")
async def on_assistant_turn_stopped(aggregator, message: AssistantTurnStoppedMessage):
await transcript_handler.on_assistant_transcript(message)
runner = PipelineRunner(handle_sigint=runner_args.handle_sigint)
await runner.run(task)
async def bot(runner_args: RunnerArguments):
"""Main bot entry point compatible with Pipecat Cloud."""
transport = await create_transport(runner_args, transport_params)
await run_bot(transport, runner_args)
if __name__ == "__main__":
from pipecat.runner.run import main
main()

26
examples/foundational/49a-thinking-anthropic.py
View File

@@ -12,16 +12,16 @@ from loguru import logger
from pipecat.audio.turn.smart_turn.local_smart_turn_v3 import LocalSmartTurnAnalyzerV3
from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.audio.vad.vad_analyzer import VADParams
from pipecat.frames.frames import LLMRunFrame, ThoughtTranscriptionMessage, TranscriptionMessage
from pipecat.frames.frames import LLMRunFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response_universal import (
AssistantThoughtMessage,
LLMContextAggregatorPair,
LLMUserAggregatorParams,
)
from pipecat.processors.transcript_processor import TranscriptProcessor
from pipecat.runner.types import RunnerArguments
from pipecat.runner.utils import create_transport
from pipecat.services.anthropic.llm import AnthropicLLMService
@@ -74,8 +74,6 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
),
)
transcript = TranscriptProcessor(process_thoughts=True)
messages = [
{
"role": "system",
@@ -93,17 +91,18 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
),
)
user_aggregator = context_aggregator.user()
assistant_aggregator = context_aggregator.assistant()
pipeline = Pipeline(
[
transport.input(), # Transport user input
stt,
transcript.user(), # User transcripts
context_aggregator.user(), # User responses
user_aggregator, # User responses
llm, # LLM
tts, # TTS
transport.output(), # Transport bot output
transcript.assistant(), # Assistant transcripts (including thoughts)
context_aggregator.assistant(), # Assistant spoken responses
assistant_aggregator, # Assistant spoken responses
]
)
@@ -143,14 +142,9 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
logger.info(f"Client disconnected")
await task.cancel()
# Register event handler for transcript updates
@transcript.event_handler("on_transcript_update")
async def on_transcript_update(processor, frame):
for msg in frame.messages:
if isinstance(msg, (ThoughtTranscriptionMessage, TranscriptionMessage)):
timestamp = f"[{msg.timestamp}] " if msg.timestamp else ""
role = "THOUGHT" if isinstance(msg, ThoughtTranscriptionMessage) else msg.role
logger.info(f"Transcript: {timestamp}{role}: {msg.content}")
@assistant_aggregator.event_handler("on_assistant_thought")
async def on_assistant_thought(aggregator, message: AssistantThoughtMessage):
logger.info(f"Thought (timestamp: {message.timestamp}): {message.content}")
runner = PipelineRunner(handle_sigint=runner_args.handle_sigint)

27
examples/foundational/49b-thinking-google.py
View File

@@ -9,20 +9,19 @@ import os
from dotenv import load_dotenv
from loguru import logger
from pipecat.audio.turn.smart_turn.base_smart_turn import SmartTurnParams
from pipecat.audio.turn.smart_turn.local_smart_turn_v3 import LocalSmartTurnAnalyzerV3
from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.audio.vad.vad_analyzer import VADParams
from pipecat.frames.frames import LLMRunFrame, ThoughtTranscriptionMessage, TranscriptionMessage
from pipecat.frames.frames import LLMRunFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response_universal import (
AssistantThoughtMessage,
LLMContextAggregatorPair,
LLMUserAggregatorParams,
)
from pipecat.processors.transcript_processor import TranscriptProcessor
from pipecat.runner.types import RunnerArguments
from pipecat.runner.utils import create_transport
from pipecat.services.cartesia.tts import CartesiaTTSService
@@ -80,8 +79,6 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
),
)
transcript = TranscriptProcessor(process_thoughts=True)
messages = [
{
"role": "system",
@@ -99,17 +96,18 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
),
)
user_aggregator = context_aggregator.user()
assistant_aggregator = context_aggregator.assistant()
pipeline = Pipeline(
[
transport.input(), # Transport user input
stt,
transcript.user(), # User transcripts
context_aggregator.user(), # User responses
user_aggregator, # User responses
llm, # LLM
tts, # TTS
transport.output(), # Transport bot output
transcript.assistant(), # Assistant transcripts (including thoughts)
context_aggregator.assistant(), # Assistant spoken responses
assistant_aggregator, # Assistant spoken responses
]
)
@@ -150,14 +148,9 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
logger.info(f"Client disconnected")
await task.cancel()
# Register event handler for transcript updates
@transcript.event_handler("on_transcript_update")
async def on_transcript_update(processor, frame):
for msg in frame.messages:
if isinstance(msg, (ThoughtTranscriptionMessage, TranscriptionMessage)):
timestamp = f"[{msg.timestamp}] " if msg.timestamp else ""
role = "THOUGHT" if isinstance(msg, ThoughtTranscriptionMessage) else msg.role
logger.info(f"Transcript: {timestamp}{role}: {msg.content}")
@assistant_aggregator.event_handler("on_assistant_thought")
async def on_assistant_thought(aggregator, message: AssistantThoughtMessage):
logger.info(f"Thought (timestamp: {message.timestamp}): {message.content}")
runner = PipelineRunner(handle_sigint=runner_args.handle_sigint)

27
examples/foundational/49c-thinking-functions-anthropic.py
View File

@@ -10,20 +10,19 @@ from dotenv import load_dotenv
from loguru import logger
from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.audio.turn.smart_turn.base_smart_turn import SmartTurnParams
from pipecat.audio.turn.smart_turn.local_smart_turn_v3 import LocalSmartTurnAnalyzerV3
from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.audio.vad.vad_analyzer import VADParams
from pipecat.frames.frames import LLMRunFrame, ThoughtTranscriptionMessage, TranscriptionMessage
from pipecat.frames.frames import LLMRunFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response_universal import (
AssistantThoughtMessage,
LLMContextAggregatorPair,
LLMUserAggregatorParams,
)
from pipecat.processors.transcript_processor import TranscriptProcessor
from pipecat.runner.types import RunnerArguments
from pipecat.runner.utils import create_transport
from pipecat.services.anthropic.llm import AnthropicLLMService
@@ -101,8 +100,6 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
tools = ToolsSchema(standard_tools=[check_flight_status, book_taxi])
transcript = TranscriptProcessor(process_thoughts=True)
messages = [
{
"role": "system",
@@ -120,17 +117,17 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
),
)
user_aggregator = context_aggregator.user()
assistant_aggregator = context_aggregator.assistant()
pipeline = Pipeline(
[
transport.input(), # Transport user input
stt,
transcript.user(), # User transcripts
context_aggregator.user(), # User responses
user_aggregator, # User responses
llm, # LLM
tts, # TTS
transport.output(), # Transport bot output
transcript.assistant(), # Assistant transcripts (including thoughts)
context_aggregator.assistant(), # Assistant spoken responses
assistant_aggregator, # Assistant spoken responses
]
)
@@ -169,13 +166,9 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
logger.info(f"Client disconnected")
await task.cancel()
@transcript.event_handler("on_transcript_update")
async def on_transcript_update(processor, frame):
for msg in frame.messages:
if isinstance(msg, (ThoughtTranscriptionMessage, TranscriptionMessage)):
timestamp = f"[{msg.timestamp}] " if msg.timestamp else ""
role = "THOUGHT" if isinstance(msg, ThoughtTranscriptionMessage) else msg.role
logger.info(f"Transcript: {timestamp}{role}: {msg.content}")
@assistant_aggregator.event_handler("on_assistant_thought")
async def on_assistant_thought(aggregator, message: AssistantThoughtMessage):
logger.info(f"Thought (timestamp: {message.timestamp}): {message.content}")
runner = PipelineRunner(handle_sigint=runner_args.handle_sigint)

23
examples/foundational/49d-thinking-functions-google.py
View File

@@ -10,7 +10,6 @@ from dotenv import load_dotenv
from loguru import logger
from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.audio.turn.smart_turn.base_smart_turn import SmartTurnParams
from pipecat.audio.turn.smart_turn.local_smart_turn_v3 import LocalSmartTurnAnalyzerV3
from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.audio.vad.vad_analyzer import VADParams
@@ -20,6 +19,7 @@ from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response_universal import (
AssistantThoughtMessage,
LLMContextAggregatorPair,
LLMUserAggregatorParams,
)
@@ -106,8 +106,6 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
tools = ToolsSchema(standard_tools=[check_flight_status, book_taxi])
transcript = TranscriptProcessor(process_thoughts=True)
messages = [
{
"role": "system",
@@ -125,17 +123,18 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
),
)
user_aggregator = context_aggregator.user()
assistant_aggregator = context_aggregator.assistant()
pipeline = Pipeline(
[
transport.input(), # Transport user input
stt,
transcript.user(), # User transcripts
context_aggregator.user(), # User responses
user_aggregator, # User responses
llm, # LLM
tts, # TTS
transport.output(), # Transport bot output
transcript.assistant(), # Assistant transcripts (including thoughts)
context_aggregator.assistant(), # Assistant spoken responses
assistant_aggregator, # Assistant spoken responses
]
)
@@ -174,13 +173,9 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
logger.info(f"Client disconnected")
await task.cancel()
@transcript.event_handler("on_transcript_update")
async def on_transcript_update(processor, frame):
for msg in frame.messages:
if isinstance(msg, (ThoughtTranscriptionMessage, TranscriptionMessage)):
timestamp = f"[{msg.timestamp}] " if msg.timestamp else ""
role = "THOUGHT" if isinstance(msg, ThoughtTranscriptionMessage) else msg.role
logger.info(f"Transcript: {timestamp}{role}: {msg.content}")
@assistant_aggregator.event_handler("on_assistant_thought")
async def on_assistant_thought(aggregator, message: AssistantThoughtMessage):
logger.info(f"Thought (timestamp: {message.timestamp}): {message.content}")
runner = PipelineRunner(handle_sigint=runner_args.handle_sigint)

79
src/pipecat/frames/frames.py
View File

@@ -536,6 +536,10 @@ class TranscriptionMessage:
content: The message content/text.
user_id: Optional identifier for the user.
timestamp: Optional timestamp when the message was created.
.. deprecated:: 0.0.99
`TranscriptionMessage` is deprecated and will be removed in a future version.
Use `LLMUserAggregator`'s and `LLMAssistantAggregator`'s new events instead.
"""
role: Literal["user", "assistant"]
@@ -543,15 +547,44 @@ class TranscriptionMessage:
user_id: Optional[str] = None
timestamp: Optional[str] = None
def __post_init__(self):
import warnings
with warnings.catch_warnings():
warnings.simplefilter("always")
warnings.warn(
"TranscriptionMessage is deprecated and will be removed in a future version. "
"Use `LLMUserAggregator`'s and `LLMAssistantAggregator`'s new events instead.",
DeprecationWarning,
stacklevel=2,
)
@dataclass
class ThoughtTranscriptionMessage:
"""An LLM thought message in a conversation transcript."""
"""An LLM thought message in a conversation transcript.
.. deprecated:: 0.0.99
`ThoughtTranscriptionMessage` is deprecated and will be removed in a future version.
Use `LLMAssistantAggregator`'s new events instead.
"""
role: Literal["assistant"] = field(default="assistant", init=False)
content: str
timestamp: Optional[str] = None
def __post_init__(self):
import warnings
with warnings.catch_warnings():
warnings.simplefilter("always")
warnings.warn(
"ThoughtTranscriptionMessage is deprecated and will be removed in a future version. "
"Use `LLMAssistantAggregator`'s new events instead.",
DeprecationWarning,
stacklevel=2,
)
@dataclass
class TranscriptionUpdateFrame(DataFrame):
@@ -595,10 +628,28 @@ class TranscriptionUpdateFrame(DataFrame):
Parameters:
messages: List of new transcript messages that were added.
.. deprecated:: 0.0.99
`TranscriptionUpdateFrame` is deprecated and will be removed in a future version.
Use `LLMUserAggregator`'s and `LLMAssistantAggregator`'s new events instead.
"""
messages: List[TranscriptionMessage | ThoughtTranscriptionMessage]
def __post_init__(self):
super().__post_init__()
import warnings
with warnings.catch_warnings():
warnings.simplefilter("always")
warnings.warn(
"TranscriptionUpdateFrame is deprecated and will be removed in a future version. "
"Use `LLMUserAggregator`'s and `LLMAssistantAggregator`'s new events instead.",
DeprecationWarning,
stacklevel=2,
)
def __str__(self):
pts = format_pts(self.pts)
return f"{self.name}(pts: {pts}, messages: {len(self.messages)})"
@@ -1160,7 +1211,18 @@ class EmulateUserStartedSpeakingFrame(SystemFrame):
This frame is deprecated and will be removed in a future version.
"""
pass
def __post_init__(self):
super().__post_init__()
import warnings
with warnings.catch_warnings():
warnings.simplefilter("always")
warnings.warn(
"EmulateUserStartedSpeakingFrame is deprecated and will be removed in a future version.",
DeprecationWarning,
stacklevel=2,
)
@dataclass
@@ -1174,7 +1236,18 @@ class EmulateUserStoppedSpeakingFrame(SystemFrame):
This frame is deprecated and will be removed in a future version.
"""
pass
def __post_init__(self):
super().__post_init__()
import warnings
with warnings.catch_warnings():
warnings.simplefilter("always")
warnings.warn(
"EmulateUserStoppedSpeakingFrame is deprecated and will be removed in a future version.",
DeprecationWarning,
stacklevel=2,
)
@dataclass

178
src/pipecat/processors/aggregators/llm_response_universal.py
View File

@@ -23,7 +23,6 @@ from loguru import logger
from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.frames.frames import (
AssistantImageRawFrame,
BotStoppedSpeakingFrame,
CancelFrame,
EndFrame,
Frame,
@@ -102,6 +101,59 @@ class LLMAssistantAggregatorParams:
expect_stripped_words: bool = True
@dataclass
class UserTurnStoppedMessage:
"""A user turn stopped message containing a user transcript update.
A message in a conversation transcript containing the user content. This is
the aggregated transcript that is then used in the context.
Parameters:
content: The message content/text.
timestamp: When the user turn started.
user_id: Optional identifier for the user.
"""
content: str
timestamp: str
user_id: Optional[str] = None
@dataclass
class AssistantTurnStoppedMessage:
"""An assistant turn stopped message containing an assistant transcript update.
A message in a conversation transcript containing the assistant
content. This is the aggregated transcript that is then used in the context.
Parameters:
content: The message content/text.
timestamp: When the assistant turn started.
"""
content: str
timestamp: str
@dataclass
class AssistantThoughtMessage:
"""An assistant thought message containing an assistant thought update.
A message in a conversation transcript containing the assistant thought
content.
Parameters:
content: The message content/text.
timestamp: When the thought started.
"""
content: str
timestamp: str
class LLMContextAggregator(FrameProcessor):
"""Base LLM aggregator that uses an LLMContext for conversation storage.
@@ -205,8 +257,12 @@ class LLMContextAggregator(FrameProcessor):
self._aggregation = []
@abstractmethod
async def push_aggregation(self):
"""Push the current aggregation downstream."""
async def push_aggregation(self) -> str:
"""Push the current aggregation downstream.
Returns:
The pushed aggregation.
"""
pass
def aggregation_string(self) -> str:
@@ -243,7 +299,7 @@ class LLMUserAggregator(LLMContextAggregator):
...
@aggregator.event_handler("on_user_turn_stopped")
async def on_user_turn_stopped(aggregator, strategy: BaseUserTurnStopStrategy):
async def on_user_turn_stopped(aggregator, strategy: BaseUserTurnStopStrategy, message: UserTurnStoppedMessage):
...
@aggregator.event_handler("on_user_turn_stop_timeout")
@@ -276,6 +332,7 @@ class LLMUserAggregator(LLMContextAggregator):
user_turn_strategies = self._params.user_turn_strategies or UserTurnStrategies()
self._user_is_muted = False
self._user_turn_start_timestamp = ""
self._user_turn_controller = UserTurnController(
user_turn_strategies=user_turn_strategies,
@@ -348,16 +405,18 @@ class LLMUserAggregator(LLMContextAggregator):
await self._user_turn_controller.process_frame(frame)
async def push_aggregation(self):
async def push_aggregation(self) -> str:
"""Push the current aggregation."""
if len(self._aggregation) == 0:
return
return ""
aggregation = self.aggregation_string()
await self.reset()
self._context.add_message({"role": self.role, "content": aggregation})
await self.push_context_frame()
return aggregation
async def _start(self, frame: StartFrame):
await self._user_turn_controller.setup(self.task_manager)
@@ -473,6 +532,8 @@ class LLMUserAggregator(LLMContextAggregator):
):
logger.debug(f"{self}: User started speaking (user turn start strategy: {strategy})")
self._user_turn_start_timestamp = time_now_iso8601()
if params.enable_user_speaking_frames:
await self.broadcast_frame(UserStartedSpeakingFrame)
@@ -493,9 +554,13 @@ class LLMUserAggregator(LLMContextAggregator):
await self.broadcast_frame(UserStoppedSpeakingFrame)
# Always push context frame.
await self.push_aggregation()
aggregation = await self.push_aggregation()
await self._call_event_handler("on_user_turn_stopped", strategy)
message = UserTurnStoppedMessage(
content=aggregation, timestamp=self._user_turn_start_timestamp
)
await self._call_event_handler("on_user_turn_stopped", strategy, message)
self._user_turn_start_timestamp = ""
async def _on_user_turn_stop_timeout(self, controller):
await self._call_event_handler("on_user_turn_stop_timeout")
@@ -514,6 +579,27 @@ class LLMAssistantAggregator(LLMContextAggregator):
The aggregator manages function calls in progress and coordinates between
text generation and tool execution phases of LLM responses.
Event handlers available:
- on_assistant_turn_started: Called when the assistant turn starts
- on_assistant_turn_stopped: Called when the assistant turn ends
- on_assistant_thought: Called when an assistant thought is available
Example::
@aggregator.event_handler("on_assistant_turn_started")
async def on_assistant_turn_started(aggregator):
...
@aggregator.event_handler("on_assistant_turn_stopped")
async def on_assistant_turn_stopped(aggregator, message: AssistantTurnStoppedMessage):
...
@aggregator.event_handler("on_assistant_thought")
async def on_assistant_thought(aggregator, message: AssistantThoughtMessage):
...
"""
def __init__(
@@ -557,9 +643,16 @@ class LLMAssistantAggregator(LLMContextAggregator):
self._function_calls_in_progress: Dict[str, Optional[FunctionCallInProgressFrame]] = {}
self._context_updated_tasks: Set[asyncio.Task] = set()
self._thought_aggregation_enabled = False
self._assistant_turn_start_timestamp = ""
self._thought_append_to_context = False
self._thought_llm: str = ""
self._thought_aggregation: List[TextPartForConcatenation] = []
self._thought_start_time: str = ""
self._register_event_handler("on_assistant_turn_started")
self._register_event_handler("on_assistant_turn_stopped")
self._register_event_handler("on_assistant_thought")
@property
def has_function_calls_in_progress(self) -> bool:
@@ -577,7 +670,7 @@ class LLMAssistantAggregator(LLMContextAggregator):
async def _reset_thought_aggregation(self):
"""Reset the thought aggregation state."""
self._thought_aggregation_enabled = False
self._thought_append_to_context = False
self._thought_llm = ""
self._thought_aggregation = []
@@ -627,22 +720,18 @@ class LLMAssistantAggregator(LLMContextAggregator):
await self._handle_user_image_frame(frame)
elif isinstance(frame, AssistantImageRawFrame):
await self._handle_assistant_image_frame(frame)
elif isinstance(frame, BotStoppedSpeakingFrame):
await self.push_aggregation()
await self.push_frame(frame, direction)
else:
await self.push_frame(frame, direction)
async def push_aggregation(self):
async def push_aggregation(self) -> str:
"""Push the current assistant aggregation with timestamp."""
if not self._aggregation:
return
return ""
aggregation = self.aggregation_string()
await self.reset()
if aggregation:
self._context.add_message({"role": "assistant", "content": aggregation})
self._context.add_message({"role": "assistant", "content": aggregation})
# Push context frame
await self.push_context_frame()
@@ -651,6 +740,8 @@ class LLMAssistantAggregator(LLMContextAggregator):
timestamp_frame = LLMContextAssistantTimestampFrame(timestamp=time_now_iso8601())
await self.push_frame(timestamp_frame)
return aggregation
async def _handle_llm_run(self, frame: LLMRunFrame):
await self.push_context_frame(FrameDirection.UPSTREAM)
@@ -665,7 +756,7 @@ class LLMAssistantAggregator(LLMContextAggregator):
await self.push_context_frame(FrameDirection.UPSTREAM)
async def _handle_interruptions(self, frame: InterruptionFrame):
await self.push_aggregation()
await self._trigger_assistant_turn_stopped()
self._started = 0
await self.reset()
@@ -788,7 +879,7 @@ class LLMAssistantAggregator(LLMContextAggregator):
text=frame.text,
)
await self.push_aggregation()
await self._trigger_assistant_turn_stopped()
await self.push_context_frame(FrameDirection.UPSTREAM)
async def _handle_assistant_image_frame(self, frame: AssistantImageRawFrame):
@@ -811,10 +902,11 @@ class LLMAssistantAggregator(LLMContextAggregator):
async def _handle_llm_start(self, _: LLMFullResponseStartFrame):
self._started += 1
await self._trigger_assistant_turn_started()
async def _handle_llm_end(self, _: LLMFullResponseEndFrame):
self._started -= 1
await self.push_aggregation()
await self._trigger_assistant_turn_stopped()
async def _handle_text(self, frame: TextFrame):
if not self._started or not frame.append_to_context:
@@ -835,11 +927,12 @@ class LLMAssistantAggregator(LLMContextAggregator):
return
await self._reset_thought_aggregation()
self._thought_aggregation_enabled = frame.append_to_context
self._thought_append_to_context = frame.append_to_context
self._thought_llm = frame.llm
self._thought_start_time = time_now_iso8601()
async def _handle_thought_text(self, frame: LLMThoughtTextFrame):
if not self._started or not self._thought_aggregation_enabled:
if not self._started:
return
# Make sure we really have text (spaces count, too!)
@@ -853,27 +946,46 @@ class LLMAssistantAggregator(LLMContextAggregator):
)
async def _handle_thought_end(self, frame: LLMThoughtEndFrame):
if not self._started or not self._thought_aggregation_enabled:
if not self._started:
return
thought = concatenate_aggregated_text(self._thought_aggregation)
llm = self._thought_llm
await self._reset_thought_aggregation()
self._context.add_message(
LLMSpecificMessage(
llm=llm,
message={
"type": "thought",
"text": thought,
"signature": frame.signature,
},
if self._thought_append_to_context:
llm = self._thought_llm
self._context.add_message(
LLMSpecificMessage(
llm=llm,
message={
"type": "thought",
"text": thought,
"signature": frame.signature,
},
)
)
)
message = AssistantThoughtMessage(content=thought, timestamp=self._thought_start_time)
await self._call_event_handler("on_assistant_thought", message)
def _context_updated_task_finished(self, task: asyncio.Task):
self._context_updated_tasks.discard(task)
async def _trigger_assistant_turn_started(self):
self._assistant_turn_start_timestamp = time_now_iso8601()
await self._call_event_handler("on_assistant_turn_started")
async def _trigger_assistant_turn_stopped(self):
aggregation = await self.push_aggregation()
if aggregation:
message = AssistantTurnStoppedMessage(
content=aggregation, timestamp=self._assistant_turn_start_timestamp
)
await self._call_event_handler("on_assistant_turn_stopped", message)
self._assistant_turn_start_timestamp = ""
class LLMContextAggregatorPair:
"""Pair of LLM context aggregators for updating context with user and assistant messages."""

14
src/pipecat/processors/transcript_processor.py
View File

@@ -269,6 +269,10 @@ class TranscriptProcessor:
@transcript.event_handler("on_transcript_update")
async def handle_update(processor, frame):
print(f"New messages: {frame.messages}")
.. deprecated:: 0.0.99
`TranscriptProcessor` is deprecated and will be removed in a future version.
Use `LLMUserAggregator`'s and `LLMAssistantAggregator`'s new events instead.
"""
def __init__(self, *, process_thoughts: bool = False):
@@ -283,6 +287,16 @@ class TranscriptProcessor:
self._assistant_processor = None
self._event_handlers = {}
import warnings
with warnings.catch_warnings():
warnings.simplefilter("always")
warnings.warn(
"`TranscriptProcessor` is deprecated and will be removed in a future version. "
"Use `LLMUserAggregator`'s and `LLMAssistantAggregator`'s new events instead.",
DeprecationWarning,
)
def user(self, **kwargs) -> UserTranscriptProcessor:
"""Get the user transcript processor.

8
src/pipecat/utils/base_object.py
View File

@@ -96,7 +96,7 @@ class BaseObject(ABC):
"""
if self._event_tasks:
event_names, tasks = zip(*self._event_tasks)
logger.debug(f"{self} waiting on event handlers to finish {list(event_names)}...")
logger.debug(f"{self}: waiting on event handlers to finish {list(event_names)}...")
await asyncio.wait(tasks)
def event_handler(self, event_name: str):
@@ -126,7 +126,7 @@ class BaseObject(ABC):
if event_name in self._event_handlers:
self._event_handlers[event_name].handlers.append(handler)
else:
logger.warning(f"Event handler {event_name} not registered")
logger.warning(f"{self}: event handler {event_name} not registered")
def _register_event_handler(self, event_name: str, sync: bool = False):
"""Register an event handler type.
@@ -140,7 +140,7 @@ class BaseObject(ABC):
name=event_name, handlers=[], is_sync=sync
)
else:
logger.warning(f"Event handler {event_name} already registered")
logger.warning(f"{self}: event handler {event_name} already registered")
async def _call_event_handler(self, event_name: str, *args, **kwargs):
"""Call all registered handlers for the specified event.
@@ -191,7 +191,7 @@ class BaseObject(ABC):
tb = traceback.extract_tb(e.__traceback__)
last = tb[-1]
logger.error(
f"Uncaught exception in event handler '{event_name}' ({last.filename}:{last.lineno}): {e}"
f"{self}: uncaught exception in event handler '{event_name}' ({last.filename}:{last.lineno}): {e}"
)
def _event_task_finished(self, task: asyncio.Task):

195
tests/test_context_aggregators_universal.py
View File

@@ -13,10 +13,17 @@ from pipecat.frames.frames import (
FunctionCallResultFrame,
FunctionCallsStartedFrame,
InterruptionFrame,
LLMContextAssistantTimestampFrame,
LLMContextFrame,
LLMFullResponseEndFrame,
LLMFullResponseStartFrame,
LLMMessagesAppendFrame,
LLMMessagesUpdateFrame,
LLMRunFrame,
LLMTextFrame,
LLMThoughtEndFrame,
LLMThoughtStartFrame,
LLMThoughtTextFrame,
TranscriptionFrame,
UserStartedSpeakingFrame,
UserStoppedSpeakingFrame,
@@ -26,6 +33,9 @@ from pipecat.frames.frames import (
from pipecat.pipeline.pipeline import Pipeline
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response_universal import (
AssistantThoughtMessage,
AssistantTurnStoppedMessage,
LLMAssistantAggregator,
LLMUserAggregator,
LLMUserAggregatorParams,
)
@@ -143,6 +153,7 @@ class TestLLMUserAggregator(unittest.IsolatedAsyncioTestCase):
should_start = None
should_stop = None
stop_message = None
@user_aggregator.event_handler("on_user_turn_started")
async def on_user_turn_started(aggregator, strategy):
@@ -150,9 +161,10 @@ class TestLLMUserAggregator(unittest.IsolatedAsyncioTestCase):
should_start = True
@user_aggregator.event_handler("on_user_turn_stopped")
async def on_user_turn_stopped(aggregator, strategy):
nonlocal should_stop
async def on_user_turn_stopped(aggregator, strategy, message):
nonlocal should_stop, stop_message
should_stop = True
stop_message = message
pipeline = Pipeline([user_aggregator])
@@ -177,6 +189,7 @@ class TestLLMUserAggregator(unittest.IsolatedAsyncioTestCase):
)
self.assertTrue(should_start)
self.assertTrue(should_stop)
self.assertEqual(stop_message.content, "Hello!")
async def test_user_turn_stop_timeout_no_transcription(self):
context = LLMContext()
@@ -196,7 +209,7 @@ class TestLLMUserAggregator(unittest.IsolatedAsyncioTestCase):
should_start = True
@user_aggregator.event_handler("on_user_turn_stopped")
async def on_user_turn_stopped(aggregator, strategy):
async def on_user_turn_stopped(aggregator, strategy, message):
nonlocal should_stop
should_stop = True
@@ -236,6 +249,7 @@ class TestLLMUserAggregator(unittest.IsolatedAsyncioTestCase):
should_start = None
should_stop = None
stop_message = None
timeout = None
@user_aggregator.event_handler("on_user_turn_started")
@@ -244,9 +258,10 @@ class TestLLMUserAggregator(unittest.IsolatedAsyncioTestCase):
should_start = True
@user_aggregator.event_handler("on_user_turn_stopped")
async def on_user_turn_stopped(aggregator, strategy):
nonlocal should_stop
async def on_user_turn_stopped(aggregator, strategy, message):
nonlocal should_stop, stop_message
should_stop = True
stop_message = message
@user_aggregator.event_handler("on_user_turn_stop_timeout")
async def on_user_turn_stop_timeout(aggregator):
@@ -271,6 +286,7 @@ class TestLLMUserAggregator(unittest.IsolatedAsyncioTestCase):
# The transcription strategy should kick-in before the user turn end timeout.
self.assertTrue(should_start)
self.assertTrue(should_stop)
self.assertEqual(stop_message.content, "Hello!")
self.assertFalse(timeout)
async def test_user_mute_strategies(self):
@@ -327,3 +343,172 @@ class TestLLMUserAggregator(unittest.IsolatedAsyncioTestCase):
# The user mute strategies should have muted the user.
self.assertFalse(user_turn)
class TestLLMAssistantAggregator(unittest.IsolatedAsyncioTestCase):
async def test_empty(self):
context = LLMContext()
aggregator = LLMAssistantAggregator(context)
should_start = None
should_stop = None
stop_message = None
@aggregator.event_handler("on_assistant_turn_started")
async def on_assistant_turn_started(aggregator):
nonlocal should_start
should_start = True
@aggregator.event_handler("on_assistant_turn_stopped")
async def on_assistant_turn_stopped(aggregator, message: AssistantTurnStoppedMessage):
nonlocal should_stop, stop_message
should_stop = True
stop_message = message
frames_to_send = [LLMFullResponseStartFrame(), LLMFullResponseEndFrame()]
await run_test(aggregator, frames_to_send=frames_to_send)
self.assertTrue(should_start)
self.assertIsNone(should_stop)
self.assertIsNone(stop_message)
async def test_simple(self):
context = LLMContext()
aggregator = LLMAssistantAggregator(context)
should_start = None
should_stop = None
stop_message = None
@aggregator.event_handler("on_assistant_turn_started")
async def on_assistant_turn_started(aggregator):
nonlocal should_start
should_start = True
@aggregator.event_handler("on_assistant_turn_stopped")
async def on_assistant_turn_stopped(aggregator, message: AssistantTurnStoppedMessage):
nonlocal should_stop, stop_message
should_stop = True
stop_message = message
frames_to_send = [
LLMFullResponseStartFrame(),
LLMTextFrame("Hello from Pipecat!"),
LLMFullResponseEndFrame(),
]
expected_down_frames = [LLMContextFrame, LLMContextAssistantTimestampFrame]
await run_test(
aggregator,
frames_to_send=frames_to_send,
expected_down_frames=expected_down_frames,
)
self.assertTrue(should_start)
self.assertTrue(should_stop)
self.assertEqual(stop_message.content, "Hello from Pipecat!")
async def test_multiple(self):
context = LLMContext()
aggregator = LLMAssistantAggregator(context)
should_start = None
should_stop = None
stop_message = None
@aggregator.event_handler("on_assistant_turn_started")
async def on_assistant_turn_started(aggregator):
nonlocal should_start
should_start = True
@aggregator.event_handler("on_assistant_turn_stopped")
async def on_assistant_turn_stopped(aggregator, message: AssistantTurnStoppedMessage):
nonlocal should_stop, stop_message
should_stop = True
stop_message = message
frames_to_send = [
LLMFullResponseStartFrame(),
LLMTextFrame("Hello "),
LLMTextFrame("from "),
LLMTextFrame("Pipecat!"),
LLMFullResponseEndFrame(),
]
expected_down_frames = [LLMContextFrame, LLMContextAssistantTimestampFrame]
await run_test(
aggregator,
frames_to_send=frames_to_send,
expected_down_frames=expected_down_frames,
)
self.assertTrue(should_start)
self.assertTrue(should_stop)
self.assertEqual(stop_message.content, "Hello from Pipecat!")
async def test_interruption(self):
context = LLMContext()
aggregator = LLMAssistantAggregator(context)
should_start = 0
should_stop = 0
stop_messages = []
@aggregator.event_handler("on_assistant_turn_started")
async def on_assistant_turn_started(aggregator):
nonlocal should_start
should_start += 1
@aggregator.event_handler("on_assistant_turn_stopped")
async def on_assistant_turn_stopped(aggregator, message: AssistantTurnStoppedMessage):
nonlocal should_stop, stop_messages
should_stop += 1
stop_messages.append(message)
frames_to_send = [
LLMFullResponseStartFrame(),
LLMTextFrame("Hello "),
SleepFrame(),
InterruptionFrame(),
LLMFullResponseStartFrame(),
LLMTextFrame("Hello "),
LLMTextFrame("there!"),
LLMFullResponseEndFrame(),
]
expected_down_frames = [
LLMContextFrame,
LLMContextAssistantTimestampFrame,
InterruptionFrame,
LLMContextFrame,
LLMContextAssistantTimestampFrame,
]
await run_test(
aggregator,
frames_to_send=frames_to_send,
expected_down_frames=expected_down_frames,
)
self.assertEqual(should_start, 2)
self.assertEqual(should_stop, 2)
self.assertEqual(stop_messages[0].content, "Hello")
self.assertEqual(stop_messages[1].content, "Hello there!")
async def test_thought(self):
context = LLMContext()
aggregator = LLMAssistantAggregator(context)
thought_message = None
@aggregator.event_handler("on_assistant_thought")
async def on_assistant_thought(aggregator, message: AssistantThoughtMessage):
nonlocal thought_message
thought_message = message
frames_to_send = [
LLMFullResponseStartFrame(),
LLMThoughtStartFrame(),
LLMThoughtTextFrame(text="I'm thinking!"),
LLMThoughtEndFrame(),
LLMFullResponseEndFrame(),
]
await run_test(aggregator, frames_to_send=frames_to_send)
self.assertEqual(thought_message.content, "I'm thinking!")