Add UserTurnCompletionLLMServiceMixin (#3518)
* Added UserTurnCompletionLLMServiceMixin class * Added 22-filter-incomplete-turns.py foundational example * Removed old 22 natural conversation foundational examples * Added test_user_turn_completion_mixin.py
This commit is contained in:
Mark Backman
1
changelog/3518.added.md
Normal file
1
changelog/3518.added.md
Normal file
@@ -0,0 +1 @@
|
||||
- Added `UserTurnCompletionLLMServiceMixin` for LLM services to detect and filter incomplete user turns. When enabled via `filter_incomplete_user_turns` in `LLMUserAggregatorParams`, the LLM outputs a turn completion marker at the start of each response: ✓ (complete), ○ (incomplete short), or ◐ (incomplete long). Incomplete turns are suppressed, and configurable timeouts automatically re-prompt the user.
|
||||
180
examples/foundational/22-filter-incomplete-turns.py
Normal file
180
examples/foundational/22-filter-incomplete-turns.py
Normal file
@@ -0,0 +1,180 @@
|
||||
#
|
||||
# Copyright (c) 2024-2026, Daily
|
||||
#
|
||||
# SPDX-License-Identifier: BSD 2-Clause License
|
||||
#
|
||||
|
||||
"""Example 22: Filter Incomplete Turns
|
||||
|
||||
Demonstrates LLM-based turn completion detection to suppress bot responses when
|
||||
the user was cut off mid-thought. The LLM outputs one of three markers:
|
||||
- ✓ (complete): User finished their thought, respond normally
|
||||
- ○ (incomplete short): User was cut off, wait ~5s then prompt
|
||||
- ◐ (incomplete long): User needs time to think, wait ~15s then prompt
|
||||
|
||||
When incomplete is detected, the bot's response is suppressed. After the timeout
|
||||
expires, the LLM is automatically prompted to re-engage the user.
|
||||
"""
|
||||
|
||||
import os
|
||||
|
||||
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)
|
||||
|
||||
|
||||
# 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"))
|
||||
|
||||
llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
|
||||
|
||||
tts = CartesiaTTSService(
|
||||
api_key=os.getenv("CARTESIA_API_KEY"),
|
||||
voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady
|
||||
)
|
||||
|
||||
messages = [
|
||||
{
|
||||
"role": "system",
|
||||
"content": f"""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 and helpful way.""",
|
||||
}
|
||||
]
|
||||
|
||||
context = LLMContext(messages)
|
||||
user_aggregator, assistant_aggregator = LLMContextAggregatorPair(
|
||||
context,
|
||||
user_params=LLMUserAggregatorParams(
|
||||
user_turn_strategies=UserTurnStrategies(
|
||||
stop=[TurnAnalyzerUserTurnStopStrategy(turn_analyzer=LocalSmartTurnAnalyzerV3())]
|
||||
),
|
||||
# Enable turn completion filtering - the LLM will output:
|
||||
# ✓ = complete (respond normally)
|
||||
# ○ = incomplete short (wait 5s, then prompt)
|
||||
# ◐ = incomplete long (wait 15s, then prompt)
|
||||
filter_incomplete_user_turns=True,
|
||||
# Optional: customize turn completion behavior
|
||||
# turn_completion_config=TurnCompletionConfig(
|
||||
# incomplete_short_timeout=5.0,
|
||||
# incomplete_long_timeout=15.0,
|
||||
# incomplete_short_prompt="Custom prompt...",
|
||||
# incomplete_long_prompt="Custom prompt...",
|
||||
# instructions="Custom turn completion instructions...",
|
||||
# ),
|
||||
),
|
||||
)
|
||||
|
||||
pipeline = Pipeline(
|
||||
[
|
||||
transport.input(), # Transport user input
|
||||
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")
|
||||
# Kick off the conversation.
|
||||
messages.append(
|
||||
{
|
||||
"role": "system",
|
||||
"content": "Please introduce yourself to the user, asking them a question that will require a complete response. To start, say 'Let me start with a fun one. If you could travel anywhere in the world right now, where would you go and why?'",
|
||||
}
|
||||
)
|
||||
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):
|
||||
timestamp = f"[{message.timestamp}] " if message.timestamp else ""
|
||||
line = f"{timestamp}user: {message.content}"
|
||||
logger.info(f"Transcript: {line}")
|
||||
|
||||
@assistant_aggregator.event_handler("on_assistant_turn_stopped")
|
||||
async def on_assistant_turn_stopped(aggregator, message: AssistantTurnStoppedMessage):
|
||||
timestamp = f"[{message.timestamp}] " if message.timestamp else ""
|
||||
line = f"{timestamp}assistant: {message.content}"
|
||||
logger.info(f"Transcript: {line}")
|
||||
|
||||
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()
|
||||
@@ -1,212 +0,0 @@
|
||||
#
|
||||
# Copyright (c) 2024-2026, Daily
|
||||
#
|
||||
# SPDX-License-Identifier: BSD 2-Clause License
|
||||
#
|
||||
|
||||
|
||||
import os
|
||||
|
||||
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, TextFrame
|
||||
from pipecat.pipeline.parallel_pipeline import ParallelPipeline
|
||||
from pipecat.pipeline.pipeline import Pipeline
|
||||
from pipecat.pipeline.runner import PipelineRunner
|
||||
from pipecat.pipeline.task import PipelineParams, PipelineTask
|
||||
from pipecat.processors.aggregators.gated_llm_context import GatedLLMContextAggregator
|
||||
from pipecat.processors.aggregators.llm_context import LLMContext
|
||||
from pipecat.processors.aggregators.llm_response_universal import (
|
||||
LLMContextAggregatorPair,
|
||||
LLMUserAggregatorParams,
|
||||
)
|
||||
from pipecat.processors.filters.null_filter import NullFilter
|
||||
from pipecat.processors.filters.wake_notifier_filter import WakeNotifierFilter
|
||||
from pipecat.processors.user_idle_processor import UserIdleProcessor
|
||||
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.llm_service import LLMService
|
||||
from pipecat.services.openai.llm import OpenAIContextAggregatorPair, 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
|
||||
from pipecat.utils.sync.event_notifier import EventNotifier
|
||||
|
||||
load_dotenv(override=True)
|
||||
|
||||
|
||||
class TurnDetectionLLM(Pipeline):
|
||||
def __init__(self, llm: LLMService, context_aggregator: OpenAIContextAggregatorPair):
|
||||
# This is the LLM that will be used to detect if the user has finished a
|
||||
# statement. This doesn't really need to be an LLM, we could use NLP
|
||||
# libraries for that, but it was easier as an example because we
|
||||
# leverage the context aggregators.
|
||||
statement_llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
|
||||
|
||||
statement_messages = [
|
||||
{
|
||||
"role": "system",
|
||||
"content": "Determine if the user's statement is a complete sentence or question, ending in a natural pause or punctuation. Return 'YES' if it is complete and 'NO' if it seems to leave a thought unfinished.",
|
||||
},
|
||||
]
|
||||
|
||||
statement_context = LLMContext(statement_messages)
|
||||
statement_context_aggregator = LLMContextAggregatorPair(statement_context)
|
||||
|
||||
# We have instructed the LLM to return 'YES' if it thinks the user
|
||||
# completed a sentence. So, if it's 'YES' we will return true in this
|
||||
# predicate which will wake up the notifier.
|
||||
async def wake_check_filter(frame):
|
||||
logger.debug(f"Completeness check frame: {frame}")
|
||||
return frame.text == "YES"
|
||||
|
||||
# This is a notifier that we use to synchronize the two LLMs.
|
||||
notifier = EventNotifier()
|
||||
|
||||
# This a filter that will wake up the notifier if the given predicate
|
||||
# (wake_check_filter) returns true.
|
||||
completness_check = WakeNotifierFilter(
|
||||
notifier, types=(TextFrame,), filter=wake_check_filter
|
||||
)
|
||||
|
||||
# This processor keeps the last context and will let it through once the
|
||||
# notifier is woken up. We start with the gate open because we send an
|
||||
# initial context frame to start the conversation.
|
||||
gated_context_aggregator = GatedLLMContextAggregator(notifier=notifier, start_open=True)
|
||||
|
||||
# Notify if the user hasn't said anything.
|
||||
async def user_idle_notifier(frame):
|
||||
await notifier.notify()
|
||||
|
||||
# Sometimes the LLM will fail detecting if a user has completed a
|
||||
# sentence, this will wake up the notifier if that happens.
|
||||
user_idle = UserIdleProcessor(callback=user_idle_notifier, timeout=3.0)
|
||||
|
||||
# The ParallePipeline input are the user transcripts. We have two
|
||||
# contexts. The first one will be used to determine if the user finished
|
||||
# a statement and if so the notifier will be woken up. The second
|
||||
# context is simply the regular context but it's gated waiting for the
|
||||
# notifier to be woken up.
|
||||
super().__init__(
|
||||
[
|
||||
ParallelPipeline(
|
||||
[
|
||||
statement_context_aggregator.user(),
|
||||
statement_llm,
|
||||
completness_check,
|
||||
NullFilter(),
|
||||
],
|
||||
[context_aggregator.user(), gated_context_aggregator, llm],
|
||||
),
|
||||
user_idle,
|
||||
]
|
||||
)
|
||||
|
||||
|
||||
# We use lambdas to defer transport parameter creation until the transport
|
||||
# type is selected at runtime.
|
||||
transport_params = {
|
||||
"daily": lambda: DailyParams(
|
||||
audio_in_enabled=True,
|
||||
audio_out_enabled=True,
|
||||
),
|
||||
"webrtc": lambda: TransportParams(
|
||||
audio_in_enabled=True,
|
||||
audio_out_enabled=True,
|
||||
),
|
||||
"twilio": lambda: FastAPIWebsocketParams(
|
||||
audio_in_enabled=True,
|
||||
audio_out_enabled=True,
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
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
|
||||
)
|
||||
|
||||
# This is the regular LLM.
|
||||
llm_main = 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 and helpful way.",
|
||||
},
|
||||
]
|
||||
|
||||
context = LLMContext(messages)
|
||||
context_aggregator = LLMContextAggregatorPair(
|
||||
context,
|
||||
user_params=LLMUserAggregatorParams(
|
||||
user_turn_strategies=UserTurnStrategies(
|
||||
stop=[TurnAnalyzerUserTurnStopStrategy(turn_analyzer=LocalSmartTurnAnalyzerV3())]
|
||||
),
|
||||
vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)),
|
||||
),
|
||||
)
|
||||
|
||||
# LLM + turn detection (with an extra LLM as a judge)
|
||||
llm = TurnDetectionLLM(llm_main, context_aggregator)
|
||||
|
||||
pipeline = Pipeline(
|
||||
[
|
||||
transport.input(), # Transport user input
|
||||
stt, # STT
|
||||
llm, # LLM with turn detection
|
||||
tts, # TTS
|
||||
transport.output(), # Transport bot output
|
||||
context_aggregator.assistant(), # 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")
|
||||
# Kick off the conversation.
|
||||
messages.append({"role": "system", "content": "Please introduce yourself to the user."})
|
||||
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()
|
||||
|
||||
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()
|
||||
@@ -1,414 +0,0 @@
|
||||
#
|
||||
# Copyright (c) 2024-2026, Daily
|
||||
#
|
||||
# SPDX-License-Identifier: BSD 2-Clause License
|
||||
#
|
||||
|
||||
import asyncio
|
||||
import os
|
||||
|
||||
from dotenv import load_dotenv
|
||||
from loguru import logger
|
||||
|
||||
from pipecat.adapters.schemas.function_schema import FunctionSchema
|
||||
from pipecat.adapters.schemas.tools_schema import ToolsSchema
|
||||
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 (
|
||||
CancelFrame,
|
||||
EndFrame,
|
||||
Frame,
|
||||
FunctionCallInProgressFrame,
|
||||
FunctionCallResultFrame,
|
||||
InterruptionFrame,
|
||||
LLMContextFrame,
|
||||
LLMRunFrame,
|
||||
StartFrame,
|
||||
SystemFrame,
|
||||
TextFrame,
|
||||
TranscriptionFrame,
|
||||
TTSSpeakFrame,
|
||||
UserStartedSpeakingFrame,
|
||||
UserStoppedSpeakingFrame,
|
||||
)
|
||||
from pipecat.pipeline.parallel_pipeline import ParallelPipeline
|
||||
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 (
|
||||
LLMContextAggregatorPair,
|
||||
LLMUserAggregatorParams,
|
||||
)
|
||||
from pipecat.processors.filters.function_filter import FunctionFilter
|
||||
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
|
||||
from pipecat.processors.user_idle_processor import UserIdleProcessor
|
||||
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.llm_service import FunctionCallParams, LLMService
|
||||
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
|
||||
from pipecat.utils.sync.base_notifier import BaseNotifier
|
||||
from pipecat.utils.sync.event_notifier import EventNotifier
|
||||
from pipecat.utils.time import time_now_iso8601
|
||||
|
||||
load_dotenv(override=True)
|
||||
|
||||
|
||||
classifier_statement = "Determine if the user's statement ends with a complete thought and you should respond. The user text is transcribed speech. It may contain multiple fragments concatentated together. You are trying to determine only the completeness of the last user statement. The previous assistant statement is provided only for context. Categorize the text as either complete with the user now expecting a response, or incomplete. Return 'YES' if text is likely complete and the user is expecting a response. Return 'NO' if the text seems to be a partial expression or unfinished thought."
|
||||
|
||||
|
||||
class StatementJudgeContextFilter(FrameProcessor):
|
||||
async def process_frame(self, frame: Frame, direction: FrameDirection):
|
||||
await super().process_frame(frame, direction)
|
||||
# We must not block system frames.
|
||||
if isinstance(frame, SystemFrame):
|
||||
await self.push_frame(frame, direction)
|
||||
return
|
||||
|
||||
# We only want to handle LLMContextFrames, and only want to push through a simplified
|
||||
# context frame that contains a system prompt and the most recent user messages,
|
||||
# concatenated.
|
||||
if isinstance(frame, LLMContextFrame):
|
||||
logger.debug(f"Context Frame: {frame}")
|
||||
# Take text content from the most recent user messages.
|
||||
messages = frame.context.get_messages()
|
||||
user_text_messages = []
|
||||
last_assistant_message = None
|
||||
for message in reversed(messages):
|
||||
if message["role"] != "user":
|
||||
if message["role"] == "assistant":
|
||||
last_assistant_message = message
|
||||
break
|
||||
if isinstance(message["content"], str):
|
||||
user_text_messages.append(message["content"])
|
||||
elif isinstance(message["content"], list):
|
||||
for content in message["content"]:
|
||||
if content["type"] == "text":
|
||||
user_text_messages.insert(0, content["text"])
|
||||
# If we have any user text content, push a context frame with the simplified context.
|
||||
if user_text_messages:
|
||||
logger.debug(f"User text messages: {user_text_messages}")
|
||||
user_message = " ".join(reversed(user_text_messages))
|
||||
logger.debug(f"User message: {user_message}")
|
||||
messages = [
|
||||
{
|
||||
"role": "system",
|
||||
"content": classifier_statement,
|
||||
}
|
||||
]
|
||||
if last_assistant_message:
|
||||
messages.append(last_assistant_message)
|
||||
messages.append({"role": "user", "content": user_message})
|
||||
await self.push_frame(LLMContextFrame(LLMContext(messages)))
|
||||
|
||||
|
||||
class CompletenessCheck(FrameProcessor):
|
||||
def __init__(self, notifier: BaseNotifier):
|
||||
super().__init__()
|
||||
self._notifier = notifier
|
||||
|
||||
async def process_frame(self, frame: Frame, direction: FrameDirection):
|
||||
await super().process_frame(frame, direction)
|
||||
|
||||
if isinstance(frame, TextFrame) and frame.text == "YES":
|
||||
logger.debug("Completeness check YES")
|
||||
await self.broadcast_frame(UserStoppedSpeakingFrame)
|
||||
await self._notifier.notify()
|
||||
elif isinstance(frame, TextFrame) and frame.text == "NO":
|
||||
logger.debug("Completeness check NO")
|
||||
else:
|
||||
await self.push_frame(frame, direction)
|
||||
|
||||
|
||||
class OutputGate(FrameProcessor):
|
||||
def __init__(self, *, notifier: BaseNotifier, start_open: bool = False, **kwargs):
|
||||
super().__init__(**kwargs)
|
||||
self._gate_open = start_open
|
||||
self._frames_buffer = []
|
||||
self._notifier = notifier
|
||||
self._gate_task = None
|
||||
|
||||
def close_gate(self):
|
||||
self._gate_open = False
|
||||
|
||||
def open_gate(self):
|
||||
self._gate_open = True
|
||||
|
||||
async def process_frame(self, frame: Frame, direction: FrameDirection):
|
||||
await super().process_frame(frame, direction)
|
||||
|
||||
# We must not block system frames.
|
||||
if isinstance(frame, SystemFrame):
|
||||
if isinstance(frame, StartFrame):
|
||||
await self._start()
|
||||
if isinstance(frame, (EndFrame, CancelFrame)):
|
||||
await self._stop()
|
||||
if isinstance(frame, InterruptionFrame):
|
||||
self._frames_buffer = []
|
||||
self.close_gate()
|
||||
await self.push_frame(frame, direction)
|
||||
return
|
||||
|
||||
# Don't block function call frames
|
||||
if isinstance(frame, (FunctionCallInProgressFrame, FunctionCallResultFrame)):
|
||||
await self.push_frame(frame, direction)
|
||||
return
|
||||
|
||||
# Ignore frames that are not following the direction of this gate.
|
||||
if direction != FrameDirection.DOWNSTREAM:
|
||||
await self.push_frame(frame, direction)
|
||||
return
|
||||
|
||||
if self._gate_open:
|
||||
await self.push_frame(frame, direction)
|
||||
return
|
||||
|
||||
self._frames_buffer.append((frame, direction))
|
||||
|
||||
async def _start(self):
|
||||
self._frames_buffer = []
|
||||
if not self._gate_task:
|
||||
self._gate_task = self.create_task(self._gate_task_handler())
|
||||
|
||||
async def _stop(self):
|
||||
if self._gate_task:
|
||||
await self.cancel_task(self._gate_task)
|
||||
self._gate_task = None
|
||||
|
||||
async def _gate_task_handler(self):
|
||||
while True:
|
||||
try:
|
||||
await self._notifier.wait()
|
||||
self.open_gate()
|
||||
for frame, direction in self._frames_buffer:
|
||||
await self.push_frame(frame, direction)
|
||||
self._frames_buffer = []
|
||||
except asyncio.CancelledError:
|
||||
break
|
||||
|
||||
|
||||
async def fetch_weather_from_api(params: FunctionCallParams):
|
||||
await params.result_callback({"conditions": "nice", "temperature": "75"})
|
||||
|
||||
|
||||
class TurnDetectionLLM(Pipeline):
|
||||
def __init__(self, llm: LLMService):
|
||||
# This is the LLM that will be used to detect if the user has finished a
|
||||
# statement. This doesn't really need to be an LLM, we could use NLP
|
||||
# libraries for that, but we have the machinery to use an LLM, so we
|
||||
# might as well!
|
||||
statement_llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
|
||||
|
||||
# We have instructed the LLM to return 'YES' if it thinks the user
|
||||
# completed a sentence. So, if it's 'YES' we will return true in this
|
||||
# predicate which will wake up the notifier.
|
||||
async def wake_check_filter(frame):
|
||||
logger.debug(f"Completeness check frame: {frame}")
|
||||
return frame.text == "YES"
|
||||
|
||||
# This is a notifier that we use to synchronize the two LLMs.
|
||||
notifier = EventNotifier()
|
||||
|
||||
# This turns the LLM context into an inference request to classify the user's speech
|
||||
# as complete or incomplete.
|
||||
statement_judge_context_filter = StatementJudgeContextFilter()
|
||||
|
||||
# This sends a UserStoppedSpeakingFrame and triggers the notifier event
|
||||
completeness_check = CompletenessCheck(notifier=notifier)
|
||||
|
||||
# # Notify if the user hasn't said anything.
|
||||
async def user_idle_notifier(frame):
|
||||
await notifier.notify()
|
||||
|
||||
# Sometimes the LLM will fail detecting if a user has completed a
|
||||
# sentence, this will wake up the notifier if that happens.
|
||||
user_idle = UserIdleProcessor(callback=user_idle_notifier, timeout=5.0)
|
||||
|
||||
# We start with the gate open because we send an initial context frame
|
||||
# to start the conversation.
|
||||
bot_output_gate = OutputGate(notifier=notifier, start_open=True)
|
||||
|
||||
async def pass_only_llm_trigger_frames(frame):
|
||||
return (
|
||||
isinstance(frame, LLMContextFrame)
|
||||
or isinstance(frame, InterruptionFrame)
|
||||
or isinstance(frame, FunctionCallInProgressFrame)
|
||||
or isinstance(frame, FunctionCallResultFrame)
|
||||
)
|
||||
|
||||
async def filter_all(frame):
|
||||
return False
|
||||
|
||||
super().__init__(
|
||||
[
|
||||
ParallelPipeline(
|
||||
[
|
||||
# Ignore everything except an LLMContextFrame. Pass a specially constructed
|
||||
# simplified context frame to the statement classifier LLM. The only frame this
|
||||
# sub-pipeline will output is a UserStoppedSpeakingFrame.
|
||||
statement_judge_context_filter,
|
||||
statement_llm,
|
||||
completeness_check,
|
||||
FunctionFilter(filter=filter_all, direction=FrameDirection.UPSTREAM),
|
||||
],
|
||||
[
|
||||
# Block everything except frames that trigger LLM inference.
|
||||
FunctionFilter(filter=pass_only_llm_trigger_frames),
|
||||
llm,
|
||||
bot_output_gate, # Buffer all llm/tts output until notified.
|
||||
],
|
||||
),
|
||||
user_idle,
|
||||
]
|
||||
)
|
||||
|
||||
|
||||
# We use lambdas to defer transport parameter creation until the transport
|
||||
# type is selected at runtime.
|
||||
transport_params = {
|
||||
"daily": lambda: DailyParams(
|
||||
audio_in_enabled=True,
|
||||
audio_out_enabled=True,
|
||||
),
|
||||
"twilio": lambda: FastAPIWebsocketParams(
|
||||
audio_in_enabled=True,
|
||||
audio_out_enabled=True,
|
||||
),
|
||||
"webrtc": lambda: TransportParams(
|
||||
audio_in_enabled=True,
|
||||
audio_out_enabled=True,
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
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
|
||||
)
|
||||
|
||||
# This is the regular LLM.
|
||||
llm_main = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
|
||||
# You can also register a function_name of None to get all functions
|
||||
# sent to the same callback with an additional function_name parameter.
|
||||
llm_main.register_function("get_current_weather", fetch_weather_from_api)
|
||||
|
||||
@llm_main.event_handler("on_function_calls_started")
|
||||
async def on_function_calls_started(service, function_calls):
|
||||
await tts.queue_frame(TTSSpeakFrame("Let me check on that."))
|
||||
|
||||
weather_function = FunctionSchema(
|
||||
name="get_current_weather",
|
||||
description="Get the current weather",
|
||||
properties={
|
||||
"location": {
|
||||
"type": "string",
|
||||
"description": "The city and state, e.g. San Francisco, CA",
|
||||
},
|
||||
"format": {
|
||||
"type": "string",
|
||||
"enum": ["celsius", "fahrenheit"],
|
||||
"description": "The temperature unit to use. Infer this from the users location.",
|
||||
},
|
||||
},
|
||||
required=["location", "format"],
|
||||
)
|
||||
tools = ToolsSchema(standard_tools=[weather_function])
|
||||
|
||||
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 and helpful way.",
|
||||
},
|
||||
]
|
||||
|
||||
context = LLMContext(messages, tools)
|
||||
user_aggregator, assistant_aggregator = LLMContextAggregatorPair(
|
||||
context,
|
||||
user_params=LLMUserAggregatorParams(
|
||||
user_turn_strategies=UserTurnStrategies(
|
||||
stop=[TurnAnalyzerUserTurnStopStrategy(turn_analyzer=LocalSmartTurnAnalyzerV3())]
|
||||
),
|
||||
vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)),
|
||||
),
|
||||
)
|
||||
|
||||
# LLM + turn detection (with an extra LLM as a judge)
|
||||
llm = TurnDetectionLLM(llm_main)
|
||||
|
||||
pipeline = Pipeline(
|
||||
[
|
||||
transport.input(),
|
||||
stt,
|
||||
user_aggregator,
|
||||
llm,
|
||||
tts,
|
||||
transport.output(),
|
||||
assistant_aggregator,
|
||||
]
|
||||
)
|
||||
|
||||
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")
|
||||
# Kick off the conversation.
|
||||
messages.append({"role": "system", "content": "Please introduce yourself to the user."})
|
||||
await task.queue_frames([LLMRunFrame()])
|
||||
|
||||
@transport.event_handler("on_app_message")
|
||||
async def on_app_message(transport, message, sender):
|
||||
logger.debug(f"Received app message: {message}")
|
||||
if "message" not in message:
|
||||
return
|
||||
|
||||
await task.queue_frames(
|
||||
[
|
||||
UserStartedSpeakingFrame(),
|
||||
TranscriptionFrame(
|
||||
user_id="", timestamp=time_now_iso8601(), text=message["message"]
|
||||
),
|
||||
UserStoppedSpeakingFrame(),
|
||||
]
|
||||
)
|
||||
|
||||
@transport.event_handler("on_client_disconnected")
|
||||
async def on_client_disconnected(transport, client):
|
||||
logger.info(f"Client disconnected")
|
||||
await task.cancel()
|
||||
|
||||
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()
|
||||
@@ -1,622 +0,0 @@
|
||||
#
|
||||
# Copyright (c) 2024-2026, Daily
|
||||
#
|
||||
# SPDX-License-Identifier: BSD 2-Clause License
|
||||
#
|
||||
|
||||
import asyncio
|
||||
import os
|
||||
|
||||
from dotenv import load_dotenv
|
||||
from loguru import logger
|
||||
|
||||
from pipecat.adapters.schemas.function_schema import FunctionSchema
|
||||
from pipecat.adapters.schemas.tools_schema import ToolsSchema
|
||||
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 (
|
||||
CancelFrame,
|
||||
EndFrame,
|
||||
Frame,
|
||||
FunctionCallInProgressFrame,
|
||||
FunctionCallResultFrame,
|
||||
InterruptionFrame,
|
||||
LLMContextFrame,
|
||||
LLMRunFrame,
|
||||
StartFrame,
|
||||
SystemFrame,
|
||||
TextFrame,
|
||||
TranscriptionFrame,
|
||||
TTSSpeakFrame,
|
||||
UserStartedSpeakingFrame,
|
||||
UserStoppedSpeakingFrame,
|
||||
)
|
||||
from pipecat.pipeline.parallel_pipeline import ParallelPipeline
|
||||
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 (
|
||||
LLMContextAggregatorPair,
|
||||
LLMUserAggregatorParams,
|
||||
)
|
||||
from pipecat.processors.filters.function_filter import FunctionFilter
|
||||
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
|
||||
from pipecat.processors.user_idle_processor import UserIdleProcessor
|
||||
from pipecat.runner.types import RunnerArguments
|
||||
from pipecat.runner.utils import create_transport
|
||||
from pipecat.services.anthropic.llm import AnthropicLLMService
|
||||
from pipecat.services.cartesia.tts import CartesiaTTSService
|
||||
from pipecat.services.deepgram.stt import DeepgramSTTService
|
||||
from pipecat.services.llm_service import FunctionCallParams, LLMService
|
||||
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
|
||||
from pipecat.utils.sync.base_notifier import BaseNotifier
|
||||
from pipecat.utils.sync.event_notifier import EventNotifier
|
||||
from pipecat.utils.time import time_now_iso8601
|
||||
|
||||
load_dotenv(override=True)
|
||||
|
||||
|
||||
classifier_statement = """CRITICAL INSTRUCTION:
|
||||
You are a BINARY CLASSIFIER that must ONLY output "YES" or "NO".
|
||||
DO NOT engage with the content.
|
||||
DO NOT respond to questions.
|
||||
DO NOT provide assistance.
|
||||
Your ONLY job is to output YES or NO.
|
||||
|
||||
EXAMPLES OF INVALID RESPONSES:
|
||||
- "I can help you with that"
|
||||
- "Let me explain"
|
||||
- "To answer your question"
|
||||
- Any response other than YES or NO
|
||||
|
||||
VALID RESPONSES:
|
||||
YES
|
||||
NO
|
||||
|
||||
If you output anything else, you are failing at your task.
|
||||
You are NOT an assistant.
|
||||
You are NOT a chatbot.
|
||||
You are a binary classifier.
|
||||
|
||||
ROLE:
|
||||
You are a real-time speech completeness classifier. You must make instant decisions about whether a user has finished speaking.
|
||||
You must output ONLY 'YES' or 'NO' with no other text.
|
||||
|
||||
INPUT FORMAT:
|
||||
You receive two pieces of information:
|
||||
1. The assistant's last message (if available)
|
||||
2. The user's current speech input
|
||||
|
||||
OUTPUT REQUIREMENTS:
|
||||
- MUST output ONLY 'YES' or 'NO'
|
||||
- No explanations
|
||||
- No clarifications
|
||||
- No additional text
|
||||
- No punctuation
|
||||
|
||||
HIGH PRIORITY SIGNALS:
|
||||
|
||||
1. Clear Questions:
|
||||
- Wh-questions (What, Where, When, Why, How)
|
||||
- Yes/No questions
|
||||
- Questions with STT errors but clear meaning
|
||||
|
||||
Examples:
|
||||
# Complete Wh-question
|
||||
[{"role": "assistant", "content": "I can help you learn."},
|
||||
{"role": "user", "content": "What's the fastest way to learn Spanish"}]
|
||||
Output: YES
|
||||
|
||||
# Complete Yes/No question despite STT error
|
||||
[{"role": "assistant", "content": "I know about planets."},
|
||||
{"role": "user", "content": "Is is Jupiter the biggest planet"}]
|
||||
Output: YES
|
||||
|
||||
2. Complete Commands:
|
||||
- Direct instructions
|
||||
- Clear requests
|
||||
- Action demands
|
||||
- Complete statements needing response
|
||||
|
||||
Examples:
|
||||
# Direct instruction
|
||||
[{"role": "assistant", "content": "I can explain many topics."},
|
||||
{"role": "user", "content": "Tell me about black holes"}]
|
||||
Output: YES
|
||||
|
||||
# Action demand
|
||||
[{"role": "assistant", "content": "I can help with math."},
|
||||
{"role": "user", "content": "Solve this equation x plus 5 equals 12"}]
|
||||
Output: YES
|
||||
|
||||
3. Direct Responses:
|
||||
- Answers to specific questions
|
||||
- Option selections
|
||||
- Clear acknowledgments with completion
|
||||
|
||||
Examples:
|
||||
# Specific answer
|
||||
[{"role": "assistant", "content": "What's your favorite color?"},
|
||||
{"role": "user", "content": "I really like blue"}]
|
||||
Output: YES
|
||||
|
||||
# Option selection
|
||||
[{"role": "assistant", "content": "Would you prefer morning or evening?"},
|
||||
{"role": "user", "content": "Morning"}]
|
||||
Output: YES
|
||||
|
||||
MEDIUM PRIORITY SIGNALS:
|
||||
|
||||
1. Speech Pattern Completions:
|
||||
- Self-corrections reaching completion
|
||||
- False starts with clear ending
|
||||
- Topic changes with complete thought
|
||||
- Mid-sentence completions
|
||||
|
||||
Examples:
|
||||
# Self-correction reaching completion
|
||||
[{"role": "assistant", "content": "What would you like to know?"},
|
||||
{"role": "user", "content": "Tell me about... no wait, explain how rainbows form"}]
|
||||
Output: YES
|
||||
|
||||
# Topic change with complete thought
|
||||
[{"role": "assistant", "content": "The weather is nice today."},
|
||||
{"role": "user", "content": "Actually can you tell me who invented the telephone"}]
|
||||
Output: YES
|
||||
|
||||
# Mid-sentence completion
|
||||
[{"role": "assistant", "content": "Hello I'm ready."},
|
||||
{"role": "user", "content": "What's the capital of? France"}]
|
||||
Output: YES
|
||||
|
||||
2. Context-Dependent Brief Responses:
|
||||
- Acknowledgments (okay, sure, alright)
|
||||
- Agreements (yes, yeah)
|
||||
- Disagreements (no, nah)
|
||||
- Confirmations (correct, exactly)
|
||||
|
||||
Examples:
|
||||
# Acknowledgment
|
||||
[{"role": "assistant", "content": "Should we talk about history?"},
|
||||
{"role": "user", "content": "Sure"}]
|
||||
Output: YES
|
||||
|
||||
# Disagreement with completion
|
||||
[{"role": "assistant", "content": "Is that what you meant?"},
|
||||
{"role": "user", "content": "No not really"}]
|
||||
Output: YES
|
||||
|
||||
LOW PRIORITY SIGNALS:
|
||||
|
||||
1. STT Artifacts (Consider but don't over-weight):
|
||||
- Repeated words
|
||||
- Unusual punctuation
|
||||
- Capitalization errors
|
||||
- Word insertions/deletions
|
||||
|
||||
Examples:
|
||||
# Word repetition but complete
|
||||
[{"role": "assistant", "content": "I can help with that."},
|
||||
{"role": "user", "content": "What what is the time right now"}]
|
||||
Output: YES
|
||||
|
||||
# Missing punctuation but complete
|
||||
[{"role": "assistant", "content": "I can explain that."},
|
||||
{"role": "user", "content": "Please tell me how computers work"}]
|
||||
Output: YES
|
||||
|
||||
2. Speech Features:
|
||||
- Filler words (um, uh, like)
|
||||
- Thinking pauses
|
||||
- Word repetitions
|
||||
- Brief hesitations
|
||||
|
||||
Examples:
|
||||
# Filler words but complete
|
||||
[{"role": "assistant", "content": "What would you like to know?"},
|
||||
{"role": "user", "content": "Um uh how do airplanes fly"}]
|
||||
Output: YES
|
||||
|
||||
# Thinking pause but incomplete
|
||||
[{"role": "assistant", "content": "I can explain anything."},
|
||||
{"role": "user", "content": "Well um I want to know about the"}]
|
||||
Output: NO
|
||||
|
||||
DECISION RULES:
|
||||
|
||||
1. Return YES if:
|
||||
- ANY high priority signal shows clear completion
|
||||
- Medium priority signals combine to show completion
|
||||
- Meaning is clear despite low priority artifacts
|
||||
|
||||
2. Return NO if:
|
||||
- No high priority signals present
|
||||
- Thought clearly trails off
|
||||
- Multiple incomplete indicators
|
||||
- User appears mid-formulation
|
||||
|
||||
3. When uncertain:
|
||||
- If you can understand the intent → YES
|
||||
- If meaning is unclear → NO
|
||||
- Always make a binary decision
|
||||
- Never request clarification
|
||||
|
||||
Examples:
|
||||
# Incomplete despite corrections
|
||||
[{"role": "assistant", "content": "What would you like to know about?"},
|
||||
{"role": "user", "content": "Can you tell me about"}]
|
||||
Output: NO
|
||||
|
||||
# Complete despite multiple artifacts
|
||||
[{"role": "assistant", "content": "I can help you learn."},
|
||||
{"role": "user", "content": "How do you I mean what's the best way to learn programming"}]
|
||||
Output: YES
|
||||
|
||||
# Trailing off incomplete
|
||||
[{"role": "assistant", "content": "I can explain anything."},
|
||||
{"role": "user", "content": "I was wondering if you could tell me why"}]
|
||||
Output: NO
|
||||
"""
|
||||
|
||||
conversational_system_message = """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 and helpful way.
|
||||
|
||||
Please be very concise in your responses. Unless you are explicitly asked to do otherwise, give me the shortest complete answer possible without unnecessary elaboration. Generally you should answer with a single sentence.
|
||||
"""
|
||||
|
||||
|
||||
class StatementJudgeContextFilter(FrameProcessor):
|
||||
async def process_frame(self, frame: Frame, direction: FrameDirection):
|
||||
await super().process_frame(frame, direction)
|
||||
# We must not block system frames.
|
||||
if isinstance(frame, SystemFrame):
|
||||
await self.push_frame(frame, direction)
|
||||
return
|
||||
|
||||
# We only want to handle LLMContextFrames, and only want to push through a simplified
|
||||
# context frame that contains a system prompt and the most recent user messages,
|
||||
if isinstance(frame, LLMContextFrame):
|
||||
# Take text content from the most recent user messages.
|
||||
messages = frame.context.get_messages()
|
||||
user_text_messages = []
|
||||
last_assistant_message = None
|
||||
for message in reversed(messages):
|
||||
if message["role"] != "user":
|
||||
if message["role"] == "assistant":
|
||||
last_assistant_message = message
|
||||
break
|
||||
if isinstance(message["content"], str):
|
||||
user_text_messages.append(message["content"])
|
||||
elif isinstance(message["content"], list):
|
||||
for content in message["content"]:
|
||||
if content["type"] == "text":
|
||||
user_text_messages.insert(0, content["text"])
|
||||
# If we have any user text content, push a context frame with the simplified context.
|
||||
if user_text_messages:
|
||||
user_message = " ".join(reversed(user_text_messages))
|
||||
logger.debug(f"!!! {user_message}")
|
||||
messages = [
|
||||
{
|
||||
"role": "system",
|
||||
"content": classifier_statement,
|
||||
}
|
||||
]
|
||||
if last_assistant_message:
|
||||
messages.append(last_assistant_message)
|
||||
messages.append({"role": "user", "content": user_message})
|
||||
await self.push_frame(LLMContextFrame(LLMContext(messages)))
|
||||
|
||||
|
||||
class CompletenessCheck(FrameProcessor):
|
||||
def __init__(self, notifier: BaseNotifier):
|
||||
super().__init__()
|
||||
self._notifier = notifier
|
||||
|
||||
async def process_frame(self, frame: Frame, direction: FrameDirection):
|
||||
await super().process_frame(frame, direction)
|
||||
|
||||
if isinstance(frame, TextFrame) and frame.text == "YES":
|
||||
logger.debug("!!! Completeness check YES")
|
||||
await self.broadcast_frame(UserStoppedSpeakingFrame)
|
||||
await self._notifier.notify()
|
||||
elif isinstance(frame, TextFrame) and frame.text == "NO":
|
||||
logger.debug("!!! Completeness check NO")
|
||||
else:
|
||||
await self.push_frame(frame, direction)
|
||||
|
||||
|
||||
class OutputGate(FrameProcessor):
|
||||
def __init__(self, *, notifier: BaseNotifier, start_open: bool = False, **kwargs):
|
||||
super().__init__(**kwargs)
|
||||
self._gate_open = start_open
|
||||
self._frames_buffer = []
|
||||
self._notifier = notifier
|
||||
self._gate_task = None
|
||||
|
||||
def close_gate(self):
|
||||
self._gate_open = False
|
||||
|
||||
def open_gate(self):
|
||||
self._gate_open = True
|
||||
|
||||
async def process_frame(self, frame: Frame, direction: FrameDirection):
|
||||
await super().process_frame(frame, direction)
|
||||
|
||||
# We must not block system frames.
|
||||
if isinstance(frame, SystemFrame):
|
||||
if isinstance(frame, StartFrame):
|
||||
await self._start()
|
||||
if isinstance(frame, (EndFrame, CancelFrame)):
|
||||
await self._stop()
|
||||
if isinstance(frame, InterruptionFrame):
|
||||
self._frames_buffer = []
|
||||
self.close_gate()
|
||||
await self.push_frame(frame, direction)
|
||||
return
|
||||
|
||||
# Don't block function call frames
|
||||
if isinstance(frame, (FunctionCallInProgressFrame, FunctionCallResultFrame)):
|
||||
await self.push_frame(frame, direction)
|
||||
return
|
||||
|
||||
# Ignore frames that are not following the direction of this gate.
|
||||
if direction != FrameDirection.DOWNSTREAM:
|
||||
await self.push_frame(frame, direction)
|
||||
return
|
||||
|
||||
if self._gate_open:
|
||||
await self.push_frame(frame, direction)
|
||||
return
|
||||
|
||||
self._frames_buffer.append((frame, direction))
|
||||
|
||||
async def _start(self):
|
||||
self._frames_buffer = []
|
||||
if not self._gate_task:
|
||||
self._gate_task = self.create_task(self._gate_task_handler())
|
||||
|
||||
async def _stop(self):
|
||||
if self._gate_task:
|
||||
await self.cancel_task(self._gate_task)
|
||||
self._gate_task = None
|
||||
|
||||
async def _gate_task_handler(self):
|
||||
while True:
|
||||
try:
|
||||
await self._notifier.wait()
|
||||
self.open_gate()
|
||||
for frame, direction in self._frames_buffer:
|
||||
await self.push_frame(frame, direction)
|
||||
self._frames_buffer = []
|
||||
except asyncio.CancelledError:
|
||||
break
|
||||
|
||||
|
||||
class TurnDetectionLLM(Pipeline):
|
||||
def __init__(self, llm: LLMService):
|
||||
# This is the LLM that will be used to detect if the user has finished a
|
||||
# statement. This doesn't really need to be an LLM, we could use NLP
|
||||
# libraries for that, but we have the machinery to use an LLM, so we might as well!
|
||||
statement_llm = AnthropicLLMService(api_key=os.getenv("ANTHROPIC_API_KEY"))
|
||||
|
||||
# This is a notifier that we use to synchronize the two LLMs.
|
||||
notifier = EventNotifier()
|
||||
|
||||
# This turns the LLM context into an inference request to classify the user's speech
|
||||
# as complete or incomplete.
|
||||
statement_judge_context_filter = StatementJudgeContextFilter()
|
||||
|
||||
# This sends a UserStoppedSpeakingFrame and triggers the notifier event
|
||||
completeness_check = CompletenessCheck(notifier=notifier)
|
||||
|
||||
# # Notify if the user hasn't said anything.
|
||||
async def user_idle_notifier(frame):
|
||||
await notifier.notify()
|
||||
|
||||
# Sometimes the LLM will fail detecting if a user has completed a
|
||||
# sentence, this will wake up the notifier if that happens.
|
||||
user_idle = UserIdleProcessor(callback=user_idle_notifier, timeout=5.0)
|
||||
|
||||
# We start with the gate open because we send an initial context frame
|
||||
# to start the conversation.
|
||||
bot_output_gate = OutputGate(notifier=notifier, start_open=True)
|
||||
|
||||
async def block_user_stopped_speaking(frame):
|
||||
return not isinstance(frame, UserStoppedSpeakingFrame)
|
||||
|
||||
async def pass_only_llm_trigger_frames(frame):
|
||||
return (
|
||||
isinstance(frame, LLMContextFrame)
|
||||
or isinstance(frame, InterruptionFrame)
|
||||
or isinstance(frame, FunctionCallInProgressFrame)
|
||||
or isinstance(frame, FunctionCallResultFrame)
|
||||
)
|
||||
|
||||
async def filter_all(frame):
|
||||
return False
|
||||
|
||||
super().__init__(
|
||||
[
|
||||
ParallelPipeline(
|
||||
[
|
||||
# Pass everything except UserStoppedSpeaking to the elements after
|
||||
# this ParallelPipeline
|
||||
FunctionFilter(filter=block_user_stopped_speaking),
|
||||
],
|
||||
[
|
||||
# Ignore everything except an LLMContextFrame. Pass a specially constructed
|
||||
# simplified context frame to the statement classifier LLM. The only frame this
|
||||
# sub-pipeline will output is a UserStoppedSpeakingFrame.
|
||||
statement_judge_context_filter,
|
||||
statement_llm,
|
||||
completeness_check,
|
||||
FunctionFilter(filter=filter_all, direction=FrameDirection.UPSTREAM),
|
||||
],
|
||||
[
|
||||
# Block everything except frames that trigger LLM inference.
|
||||
FunctionFilter(filter=pass_only_llm_trigger_frames),
|
||||
llm,
|
||||
bot_output_gate, # Buffer all llm/tts output until notified.
|
||||
],
|
||||
),
|
||||
user_idle,
|
||||
]
|
||||
)
|
||||
|
||||
|
||||
async def fetch_weather_from_api(params: FunctionCallParams):
|
||||
await params.result_callback({"conditions": "nice", "temperature": "75"})
|
||||
|
||||
|
||||
# We use lambdas to defer transport parameter creation until the transport
|
||||
# type is selected at runtime.
|
||||
transport_params = {
|
||||
"daily": lambda: DailyParams(
|
||||
audio_in_enabled=True,
|
||||
audio_out_enabled=True,
|
||||
),
|
||||
"twilio": lambda: FastAPIWebsocketParams(
|
||||
audio_in_enabled=True,
|
||||
audio_out_enabled=True,
|
||||
),
|
||||
"webrtc": lambda: TransportParams(
|
||||
audio_in_enabled=True,
|
||||
audio_out_enabled=True,
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
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
|
||||
)
|
||||
|
||||
# This is the regular LLM.
|
||||
llm_main = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
|
||||
# Register a function_name of None to get all functions
|
||||
# sent to the same callback with an additional function_name parameter.
|
||||
llm_main.register_function("get_current_weather", fetch_weather_from_api)
|
||||
|
||||
@llm_main.event_handler("on_function_calls_started")
|
||||
async def on_function_calls_started(service, function_calls):
|
||||
await tts.queue_frame(TTSSpeakFrame("Let me check on that."))
|
||||
|
||||
weather_function = FunctionSchema(
|
||||
name="get_current_weather",
|
||||
description="Get the current weather",
|
||||
properties={
|
||||
"location": {
|
||||
"type": "string",
|
||||
"description": "The city and state, e.g. San Francisco, CA",
|
||||
},
|
||||
"format": {
|
||||
"type": "string",
|
||||
"enum": ["celsius", "fahrenheit"],
|
||||
"description": "The temperature unit to use. Infer this from the users location.",
|
||||
},
|
||||
},
|
||||
required=["location", "format"],
|
||||
)
|
||||
tools = ToolsSchema(standard_tools=[weather_function])
|
||||
|
||||
messages = [
|
||||
{
|
||||
"role": "system",
|
||||
"content": conversational_system_message,
|
||||
},
|
||||
]
|
||||
|
||||
context = LLMContext(messages, tools)
|
||||
user_aggregator, assistant_aggregator = LLMContextAggregatorPair(
|
||||
context,
|
||||
user_params=LLMUserAggregatorParams(
|
||||
user_turn_strategies=UserTurnStrategies(
|
||||
stop=[TurnAnalyzerUserTurnStopStrategy(turn_analyzer=LocalSmartTurnAnalyzerV3())]
|
||||
),
|
||||
vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)),
|
||||
),
|
||||
)
|
||||
|
||||
# LLM + turn detection (with an extra LLM as a judge)
|
||||
llm = TurnDetectionLLM(llm_main)
|
||||
|
||||
pipeline = Pipeline(
|
||||
[
|
||||
transport.input(),
|
||||
stt,
|
||||
user_aggregator,
|
||||
llm,
|
||||
tts,
|
||||
transport.output(),
|
||||
assistant_aggregator,
|
||||
]
|
||||
)
|
||||
|
||||
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")
|
||||
# Kick off the conversation.
|
||||
messages.append(
|
||||
{
|
||||
"role": "user",
|
||||
"content": "Start by just saying \"Hello I'm ready.\" Don't say anything else.",
|
||||
}
|
||||
)
|
||||
await task.queue_frames([LLMRunFrame()])
|
||||
|
||||
@transport.event_handler("on_app_message")
|
||||
async def on_app_message(transport, message, sender):
|
||||
logger.debug(f"Received app message: {message}")
|
||||
if "message" not in message:
|
||||
return
|
||||
|
||||
await task.queue_frames(
|
||||
[
|
||||
UserStartedSpeakingFrame(),
|
||||
TranscriptionFrame(
|
||||
user_id="", timestamp=time_now_iso8601(), text=message["message"]
|
||||
),
|
||||
UserStoppedSpeakingFrame(),
|
||||
]
|
||||
)
|
||||
|
||||
@transport.event_handler("on_client_disconnected")
|
||||
async def on_client_disconnected(transport, client):
|
||||
logger.info(f"Client disconnected")
|
||||
await task.cancel()
|
||||
|
||||
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()
|
||||
@@ -1,813 +0,0 @@
|
||||
#
|
||||
# Copyright (c) 2024-2026, Daily
|
||||
#
|
||||
# SPDX-License-Identifier: BSD 2-Clause License
|
||||
#
|
||||
|
||||
import asyncio
|
||||
import os
|
||||
import time
|
||||
|
||||
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 (
|
||||
CancelFrame,
|
||||
EndFrame,
|
||||
Frame,
|
||||
FunctionCallInProgressFrame,
|
||||
FunctionCallResultFrame,
|
||||
InputAudioRawFrame,
|
||||
InterruptionFrame,
|
||||
LLMContextFrame,
|
||||
LLMFullResponseStartFrame,
|
||||
StartFrame,
|
||||
SystemFrame,
|
||||
TextFrame,
|
||||
TranscriptionFrame,
|
||||
UserStartedSpeakingFrame,
|
||||
UserStoppedSpeakingFrame,
|
||||
)
|
||||
from pipecat.pipeline.parallel_pipeline import ParallelPipeline
|
||||
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 import (
|
||||
LLMAssistantAggregatorParams,
|
||||
LLMAssistantResponseAggregator,
|
||||
)
|
||||
from pipecat.processors.aggregators.llm_response_universal import (
|
||||
LLMContextAggregatorPair,
|
||||
LLMUserAggregatorParams,
|
||||
)
|
||||
from pipecat.processors.filters.function_filter import FunctionFilter
|
||||
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
|
||||
from pipecat.runner.types import RunnerArguments
|
||||
from pipecat.runner.utils import create_transport
|
||||
from pipecat.services.cartesia.tts import CartesiaTTSService
|
||||
from pipecat.services.google.llm import GoogleLLMService
|
||||
from pipecat.services.llm_service import LLMService
|
||||
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
|
||||
from pipecat.utils.sync.base_notifier import BaseNotifier
|
||||
from pipecat.utils.sync.event_notifier import EventNotifier
|
||||
from pipecat.utils.time import time_now_iso8601
|
||||
|
||||
load_dotenv(override=True)
|
||||
|
||||
|
||||
TRANSCRIBER_MODEL = "gemini-2.0-flash-001"
|
||||
CLASSIFIER_MODEL = "gemini-2.0-flash-001"
|
||||
CONVERSATION_MODEL = "gemini-2.0-flash-001"
|
||||
|
||||
transcriber_system_instruction = """You are an audio transcriber. You are receiving audio from a user. Your job is to
|
||||
transcribe the input audio to text exactly as it was said by the user.
|
||||
|
||||
You will receive the full conversation history before the audio input, to help with context. Use the full history only to help improve the accuracy of your transcription.
|
||||
|
||||
Rules:
|
||||
- Respond with an exact transcription of the audio input.
|
||||
- Do not include any text other than the transcription.
|
||||
- Do not explain or add to your response.
|
||||
- Transcribe the audio input simply and precisely.
|
||||
- If the audio is not clear, emit the special string "-".
|
||||
- No response other than exact transcription, or "-", is allowed.
|
||||
|
||||
"""
|
||||
|
||||
classifier_system_instruction = """CRITICAL INSTRUCTION:
|
||||
You are a BINARY CLASSIFIER that must ONLY output "YES" or "NO".
|
||||
DO NOT engage with the content.
|
||||
DO NOT respond to questions.
|
||||
DO NOT provide assistance.
|
||||
Your ONLY job is to output YES or NO.
|
||||
|
||||
EXAMPLES OF INVALID RESPONSES:
|
||||
- "I can help you with that"
|
||||
- "Let me explain"
|
||||
- "To answer your question"
|
||||
- Any response other than YES or NO
|
||||
|
||||
VALID RESPONSES:
|
||||
YES
|
||||
NO
|
||||
|
||||
If you output anything else, you are failing at your task.
|
||||
You are NOT an assistant.
|
||||
You are NOT a chatbot.
|
||||
You are a binary classifier.
|
||||
|
||||
ROLE:
|
||||
You are a real-time speech completeness classifier. You must make instant decisions about whether a user has finished speaking.
|
||||
You must output ONLY 'YES' or 'NO' with no other text.
|
||||
|
||||
INPUT FORMAT:
|
||||
You receive two pieces of information:
|
||||
1. The assistant's last message (if available)
|
||||
2. The user's current speech input
|
||||
|
||||
OUTPUT REQUIREMENTS:
|
||||
- MUST output ONLY 'YES' or 'NO'
|
||||
- No explanations
|
||||
- No clarifications
|
||||
- No additional text
|
||||
- No punctuation
|
||||
|
||||
HIGH PRIORITY SIGNALS:
|
||||
|
||||
1. Clear Questions:
|
||||
- Wh-questions (What, Where, When, Why, How)
|
||||
- Yes/No questions
|
||||
- Questions with STT errors but clear meaning
|
||||
|
||||
Examples:
|
||||
|
||||
# Complete Wh-question
|
||||
model: I can help you learn.
|
||||
user: What's the fastest way to learn Spanish
|
||||
Output: YES
|
||||
|
||||
# Complete Yes/No question despite STT error
|
||||
model: I know about planets.
|
||||
user: Is is Jupiter the biggest planet
|
||||
Output: YES
|
||||
|
||||
2. Complete Commands:
|
||||
- Direct instructions
|
||||
- Clear requests
|
||||
- Action demands
|
||||
- Start of task indication
|
||||
- Complete statements needing response
|
||||
|
||||
Examples:
|
||||
|
||||
# Direct instruction
|
||||
model: I can explain many topics.
|
||||
user: Tell me about black holes
|
||||
Output: YES
|
||||
|
||||
# Start of task indication
|
||||
user: Let's begin.
|
||||
Output: YES
|
||||
|
||||
# Start of task indication
|
||||
user: Let's get started.
|
||||
Output: YES
|
||||
|
||||
# Action demand
|
||||
model: I can help with math.
|
||||
user: Solve this equation x plus 5 equals 12
|
||||
Output: YES
|
||||
|
||||
3. Direct Responses:
|
||||
- Answers to specific questions
|
||||
- Option selections
|
||||
- Clear acknowledgments with completion
|
||||
- Providing information with a known format - mailing address
|
||||
- Providing information with a known format - phone number
|
||||
- Providing information with a known format - credit card number
|
||||
|
||||
Examples:
|
||||
|
||||
# Specific answer
|
||||
model: What's your favorite color?
|
||||
user: I really like blue
|
||||
Output: YES
|
||||
|
||||
# Option selection
|
||||
model: Would you prefer morning or evening?
|
||||
user: Morning
|
||||
Output: YES
|
||||
|
||||
# Providing information with a known format - mailing address
|
||||
model: What's your address?
|
||||
user: 1234 Main Street
|
||||
Output: NO
|
||||
|
||||
# Providing information with a known format - mailing address
|
||||
model: What's your address?
|
||||
user: 1234 Main Street Irving Texas 75063
|
||||
Output: Yes
|
||||
|
||||
# Providing information with a known format - phone number
|
||||
model: What's your phone number?
|
||||
user: 41086753
|
||||
Output: NO
|
||||
|
||||
# Providing information with a known format - phone number
|
||||
model: What's your phone number?
|
||||
user: 4108675309
|
||||
Output: Yes
|
||||
|
||||
# Providing information with a known format - phone number
|
||||
model: What's your phone number?
|
||||
user: 220
|
||||
Output: No
|
||||
|
||||
# Providing information with a known format - credit card number
|
||||
model: What's your credit card number?
|
||||
user: 5556
|
||||
Output: NO
|
||||
|
||||
# Providing information with a known format - phone number
|
||||
model: What's your credit card number?
|
||||
user: 5556710454680800
|
||||
Output: Yes
|
||||
|
||||
model: What's your credit card number?
|
||||
user: 414067
|
||||
Output: NO
|
||||
|
||||
|
||||
MEDIUM PRIORITY SIGNALS:
|
||||
|
||||
1. Speech Pattern Completions:
|
||||
- Self-corrections reaching completion
|
||||
- False starts with clear ending
|
||||
- Topic changes with complete thought
|
||||
- Mid-sentence completions
|
||||
|
||||
Examples:
|
||||
|
||||
# Self-correction reaching completion
|
||||
model: What would you like to know?
|
||||
user: Tell me about... no wait, explain how rainbows form
|
||||
Output: YES
|
||||
|
||||
# Topic change with complete thought
|
||||
model: The weather is nice today.
|
||||
user: Actually can you tell me who invented the telephone
|
||||
Output: YES
|
||||
|
||||
# Mid-sentence completion
|
||||
model: Hello I'm ready.
|
||||
user: What's the capital of? France
|
||||
Output: YES
|
||||
|
||||
2. Context-Dependent Brief Responses:
|
||||
- Acknowledgments (okay, sure, alright)
|
||||
- Agreements (yes, yeah)
|
||||
- Disagreements (no, nah)
|
||||
- Confirmations (correct, exactly)
|
||||
|
||||
Examples:
|
||||
|
||||
# Acknowledgment
|
||||
model: Should we talk about history?
|
||||
user: Sure
|
||||
Output: YES
|
||||
|
||||
# Disagreement with completion
|
||||
model: Is that what you meant?
|
||||
user: No not really
|
||||
Output: YES
|
||||
|
||||
LOW PRIORITY SIGNALS:
|
||||
|
||||
1. STT Artifacts (Consider but don't over-weight):
|
||||
- Repeated words
|
||||
- Unusual punctuation
|
||||
- Capitalization errors
|
||||
- Word insertions/deletions
|
||||
|
||||
Examples:
|
||||
|
||||
# Word repetition but complete
|
||||
model: I can help with that.
|
||||
user: What what is the time right now
|
||||
Output: YES
|
||||
|
||||
# Missing punctuation but complete
|
||||
model: I can explain that.
|
||||
user: Please tell me how computers work
|
||||
Output: YES
|
||||
|
||||
2. Speech Features:
|
||||
- Filler words (um, uh, like)
|
||||
- Thinking pauses
|
||||
- Word repetitions
|
||||
- Brief hesitations
|
||||
|
||||
Examples:
|
||||
|
||||
# Filler words but complete
|
||||
model: What would you like to know?
|
||||
user: Um uh how do airplanes fly
|
||||
Output: YES
|
||||
|
||||
# Thinking pause but incomplete
|
||||
model: I can explain anything.
|
||||
user: Well um I want to know about the
|
||||
Output: NO
|
||||
|
||||
DECISION RULES:
|
||||
|
||||
1. Return YES if:
|
||||
- ANY high priority signal shows clear completion
|
||||
- Medium priority signals combine to show completion
|
||||
- Meaning is clear despite low priority artifacts
|
||||
|
||||
2. Return NO if:
|
||||
- No high priority signals present
|
||||
- Thought clearly trails off
|
||||
- Multiple incomplete indicators
|
||||
- User appears mid-formulation
|
||||
|
||||
3. When uncertain:
|
||||
- If you can understand the intent → YES
|
||||
- If meaning is unclear → NO
|
||||
- Always make a binary decision
|
||||
- Never request clarification
|
||||
|
||||
Examples:
|
||||
|
||||
# Incomplete despite corrections
|
||||
model: What would you like to know about?
|
||||
user: Can you tell me about
|
||||
Output: NO
|
||||
|
||||
# Complete despite multiple artifacts
|
||||
model: I can help you learn.
|
||||
user: How do you I mean what's the best way to learn programming
|
||||
Output: YES
|
||||
|
||||
# Trailing off incomplete
|
||||
model: I can explain anything.
|
||||
user: I was wondering if you could tell me why
|
||||
Output: NO
|
||||
"""
|
||||
|
||||
conversation_system_instruction = """You are a helpful assistant participating in a voice converation.
|
||||
|
||||
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 and helpful way.
|
||||
|
||||
If you know that a number string is a phone number from the context of the conversation, write it as a phone number. For example 210-333-4567.
|
||||
|
||||
If you know that a number string is a credit card number, write it as a credit card number. For example 4111-1111-1111-1111.
|
||||
|
||||
Please be very concise in your responses. Unless you are explicitly asked to do otherwise, give me shortest complete answer possible without unnecessary elaboration. Generally you should answer with a single sentence.
|
||||
"""
|
||||
|
||||
|
||||
class AudioAccumulator(FrameProcessor):
|
||||
"""Buffers user audio until the user stops speaking.
|
||||
|
||||
Always pushes a fresh context with a single audio message.
|
||||
"""
|
||||
|
||||
def __init__(self, **kwargs):
|
||||
super().__init__(**kwargs)
|
||||
self._audio_frames = []
|
||||
self._start_secs = 0.2 # this should match VAD start_secs (hardcoding for now)
|
||||
self._max_buffer_size_secs = 30
|
||||
self._user_speaking_vad_state = False
|
||||
self._user_speaking_utterance_state = False
|
||||
|
||||
async def reset(self):
|
||||
self._audio_frames = []
|
||||
self._user_speaking_vad_state = False
|
||||
self._user_speaking_utterance_state = False
|
||||
|
||||
async def process_frame(self, frame: Frame, direction: FrameDirection):
|
||||
await super().process_frame(frame, direction)
|
||||
|
||||
# ignore context frame
|
||||
if isinstance(frame, LLMContextFrame):
|
||||
return
|
||||
|
||||
if isinstance(frame, TranscriptionFrame):
|
||||
# We could gracefully handle both audio input and text/transcription input ...
|
||||
# but let's leave that as an exercise to the reader. :-)
|
||||
return
|
||||
if isinstance(frame, UserStartedSpeakingFrame):
|
||||
self._user_speaking_vad_state = True
|
||||
self._user_speaking_utterance_state = True
|
||||
|
||||
elif isinstance(frame, UserStoppedSpeakingFrame):
|
||||
data = b"".join(frame.audio for frame in self._audio_frames)
|
||||
logger.debug(
|
||||
f"Processing audio buffer seconds: ({len(self._audio_frames)}) ({len(data)}) {len(data) / 2 / 16000}"
|
||||
)
|
||||
self._user_speaking = False
|
||||
context = LLMContext()
|
||||
await context.add_audio_frames_message(audio_frames=self._audio_frames)
|
||||
await self.push_frame(LLMContextFrame(context=context))
|
||||
elif isinstance(frame, InputAudioRawFrame):
|
||||
# Append the audio frame to our buffer. Treat the buffer as a ring buffer, dropping the oldest
|
||||
# frames as necessary.
|
||||
# Use a small buffer size when an utterance is not in progress. Just big enough to backfill the start_secs.
|
||||
# Use a larger buffer size when an utterance is in progress.
|
||||
# Assume all audio frames have the same duration.
|
||||
self._audio_frames.append(frame)
|
||||
frame_duration = len(frame.audio) / 2 * frame.num_channels / frame.sample_rate
|
||||
buffer_duration = frame_duration * len(self._audio_frames)
|
||||
# logger.debug(f"!!! Frame duration: {frame_duration}")
|
||||
if self._user_speaking_utterance_state:
|
||||
while buffer_duration > self._max_buffer_size_secs:
|
||||
self._audio_frames.pop(0)
|
||||
buffer_duration -= frame_duration
|
||||
else:
|
||||
while buffer_duration > self._start_secs:
|
||||
self._audio_frames.pop(0)
|
||||
buffer_duration -= frame_duration
|
||||
|
||||
await self.push_frame(frame, direction)
|
||||
|
||||
|
||||
class CompletenessCheck(FrameProcessor):
|
||||
"""Checks the result of the classifier LLM to determine if the user has finished speaking.
|
||||
|
||||
Triggers the notifier if the user has finished speaking. Also triggers the notifier if an
|
||||
idle timeout is reached.
|
||||
"""
|
||||
|
||||
wait_time = 5.0
|
||||
|
||||
def __init__(self, notifier: BaseNotifier, audio_accumulator: AudioAccumulator, **kwargs):
|
||||
super().__init__()
|
||||
self._notifier = notifier
|
||||
self._audio_accumulator = audio_accumulator
|
||||
self._idle_task = None
|
||||
self._wakeup_time = 0
|
||||
|
||||
async def process_frame(self, frame: Frame, direction: FrameDirection):
|
||||
await super().process_frame(frame, direction)
|
||||
|
||||
if isinstance(frame, (EndFrame, CancelFrame)):
|
||||
if self._idle_task:
|
||||
await self.cancel_task(self._idle_task)
|
||||
self._idle_task = None
|
||||
await self.push_frame(frame, direction)
|
||||
elif isinstance(frame, UserStartedSpeakingFrame):
|
||||
if self._idle_task:
|
||||
await self.cancel_task(self._idle_task)
|
||||
elif isinstance(frame, TextFrame) and frame.text.startswith("YES"):
|
||||
logger.debug("Completeness check YES")
|
||||
if self._idle_task:
|
||||
await self.cancel_task(self._idle_task)
|
||||
await self.broadcast_frame(UserStoppedSpeakingFrame)
|
||||
await self._audio_accumulator.reset()
|
||||
await self._notifier.notify()
|
||||
elif isinstance(frame, TextFrame):
|
||||
if frame.text.strip():
|
||||
logger.debug(f"Completeness check NO - '{frame.text}'")
|
||||
# start timer to wake up if necessary
|
||||
if self._wakeup_time:
|
||||
self._wakeup_time = time.time() + self.wait_time
|
||||
else:
|
||||
# logger.debug("!!! CompletenessCheck idle wait START")
|
||||
self._wakeup_time = time.time() + self.wait_time
|
||||
self._idle_task = self.create_task(self._idle_task_handler())
|
||||
else:
|
||||
await self.push_frame(frame, direction)
|
||||
|
||||
async def _idle_task_handler(self):
|
||||
try:
|
||||
while time.time() < self._wakeup_time:
|
||||
await asyncio.sleep(0.01)
|
||||
# logger.debug(f"!!! CompletenessCheck idle wait OVER")
|
||||
await self._audio_accumulator.reset()
|
||||
await self._notifier.notify()
|
||||
except asyncio.CancelledError:
|
||||
# logger.debug(f"!!! CompletenessCheck idle wait CANCEL")
|
||||
pass
|
||||
except Exception as e:
|
||||
logger.error(f"CompletenessCheck idle wait error: {e}")
|
||||
raise e
|
||||
finally:
|
||||
# logger.debug(f"!!! CompletenessCheck idle wait FINALLY")
|
||||
self._wakeup_time = 0
|
||||
self._idle_task = None
|
||||
|
||||
|
||||
class LLMAggregatorBuffer(LLMAssistantResponseAggregator):
|
||||
"""Buffers the output of the transcription LLM. Used by the bot output gate."""
|
||||
|
||||
def __init__(self, **kwargs):
|
||||
super().__init__(params=LLMAssistantAggregatorParams(expect_stripped_words=False))
|
||||
self._transcription = ""
|
||||
|
||||
async def process_frame(self, frame: Frame, direction: FrameDirection):
|
||||
await super().process_frame(frame, direction)
|
||||
# parent method pushes frames
|
||||
if isinstance(frame, UserStartedSpeakingFrame):
|
||||
self._transcription = ""
|
||||
|
||||
async def push_aggregation(self):
|
||||
if self._aggregation:
|
||||
self._transcription = self._aggregation
|
||||
self._aggregation = ""
|
||||
|
||||
logger.debug(f"[Transcription] {self._transcription}")
|
||||
|
||||
async def wait_for_transcription(self):
|
||||
while not self._transcription:
|
||||
await asyncio.sleep(0.01)
|
||||
tx = self._transcription
|
||||
self._transcription = ""
|
||||
return tx
|
||||
|
||||
|
||||
class ConversationAudioContextAssembler(FrameProcessor):
|
||||
"""Takes the single-message context generated by the AudioAccumulator and adds it to the conversation LLM's context."""
|
||||
|
||||
def __init__(self, context: LLMContext, **kwargs):
|
||||
super().__init__(**kwargs)
|
||||
self._context = context
|
||||
|
||||
async def process_frame(self, frame: Frame, direction: FrameDirection):
|
||||
await super().process_frame(frame, direction)
|
||||
|
||||
# We must not block system frames.
|
||||
if isinstance(frame, SystemFrame):
|
||||
await self.push_frame(frame, direction)
|
||||
return
|
||||
|
||||
if isinstance(frame, LLMContextFrame):
|
||||
last_message = frame.context.get_messages()[-1]
|
||||
self._context._messages.append(last_message)
|
||||
await self.push_frame(LLMContextFrame(context=self._context))
|
||||
|
||||
|
||||
class OutputGate(FrameProcessor):
|
||||
"""Buffers output frames until the notifier is triggered.
|
||||
|
||||
When the notifier fires, waits until a transcription is ready, then:
|
||||
1. Replaces the last user audio message with the transcription.
|
||||
2. Flushes the frames buffer.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
notifier: BaseNotifier,
|
||||
context: LLMContext,
|
||||
llm_transcription_buffer: LLMAggregatorBuffer,
|
||||
**kwargs,
|
||||
):
|
||||
super().__init__(**kwargs)
|
||||
self._gate_open = False
|
||||
self._frames_buffer = []
|
||||
self._notifier = notifier
|
||||
self._context = context
|
||||
self._transcription_buffer = llm_transcription_buffer
|
||||
self._gate_task = None
|
||||
|
||||
def close_gate(self):
|
||||
self._gate_open = False
|
||||
|
||||
def open_gate(self):
|
||||
self._gate_open = True
|
||||
|
||||
async def process_frame(self, frame: Frame, direction: FrameDirection):
|
||||
await super().process_frame(frame, direction)
|
||||
|
||||
# We must not block system frames.
|
||||
if isinstance(frame, SystemFrame):
|
||||
if isinstance(frame, StartFrame):
|
||||
await self._start()
|
||||
if isinstance(frame, (EndFrame, CancelFrame)):
|
||||
await self._stop()
|
||||
if isinstance(frame, InterruptionFrame):
|
||||
self._frames_buffer = []
|
||||
self.close_gate()
|
||||
await self.push_frame(frame, direction)
|
||||
return
|
||||
|
||||
# Don't block function call frames
|
||||
if isinstance(frame, (FunctionCallInProgressFrame, FunctionCallResultFrame)):
|
||||
await self.push_frame(frame, direction)
|
||||
return
|
||||
|
||||
# Ignore frames that are not following the direction of this gate.
|
||||
if direction != FrameDirection.DOWNSTREAM:
|
||||
await self.push_frame(frame, direction)
|
||||
return
|
||||
|
||||
if isinstance(frame, LLMFullResponseStartFrame):
|
||||
# Remove the audio message from the context. We will never need it again.
|
||||
# If the completeness check fails, a new audio message will be appended to the context.
|
||||
# If the completeness check succeeds, our notifier will fire and we will append the
|
||||
# transcription to the context.
|
||||
self._context._messages.pop()
|
||||
|
||||
if self._gate_open:
|
||||
await self.push_frame(frame, direction)
|
||||
return
|
||||
|
||||
self._frames_buffer.append((frame, direction))
|
||||
|
||||
async def _start(self):
|
||||
self._frames_buffer = []
|
||||
if not self._gate_task:
|
||||
self._gate_task = self.create_task(self._gate_task_handler())
|
||||
|
||||
async def _stop(self):
|
||||
if self._gate_task:
|
||||
await self.cancel_task(self._gate_task)
|
||||
self._gate_task = None
|
||||
|
||||
async def _gate_task_handler(self):
|
||||
while True:
|
||||
try:
|
||||
await self._notifier.wait()
|
||||
|
||||
transcription = await self._transcription_buffer.wait_for_transcription() or "-"
|
||||
self._context.add_message({"role": "user", "content": transcription})
|
||||
|
||||
self.open_gate()
|
||||
for frame, direction in self._frames_buffer:
|
||||
await self.push_frame(frame, direction)
|
||||
self._frames_buffer = []
|
||||
except asyncio.CancelledError:
|
||||
break
|
||||
|
||||
|
||||
class TurnDetectionLLM(Pipeline):
|
||||
def __init__(self, llm: LLMService, context: LLMContext):
|
||||
# This is the LLM that will transcribe user speech.
|
||||
tx_llm = GoogleLLMService(
|
||||
name="Transcriber",
|
||||
model=TRANSCRIBER_MODEL,
|
||||
api_key=os.getenv("GOOGLE_API_KEY"),
|
||||
temperature=0.0,
|
||||
system_instruction=transcriber_system_instruction,
|
||||
)
|
||||
|
||||
# This is the LLM that will classify user speech as complete or incomplete.
|
||||
classifier_llm = GoogleLLMService(
|
||||
name="Classifier",
|
||||
model=CLASSIFIER_MODEL,
|
||||
api_key=os.getenv("GOOGLE_API_KEY"),
|
||||
temperature=0.0,
|
||||
system_instruction=classifier_system_instruction,
|
||||
)
|
||||
|
||||
# This is a notifier that we use to synchronize the two LLMs.
|
||||
notifier = EventNotifier()
|
||||
|
||||
# This turns the LLM context into an inference request to classify the user's speech
|
||||
# as complete or incomplete.
|
||||
# statement_judge_context_filter = StatementJudgeAudioContextAccumulator(notifier=notifier)
|
||||
|
||||
audio_accumulator = AudioAccumulator()
|
||||
# This sends a UserStoppedSpeakingFrame and triggers the notifier event
|
||||
completeness_check = CompletenessCheck(
|
||||
notifier=notifier, audio_accumulator=audio_accumulator
|
||||
)
|
||||
|
||||
async def block_user_stopped_speaking(frame):
|
||||
return not isinstance(frame, UserStoppedSpeakingFrame)
|
||||
|
||||
conversation_audio_context_assembler = ConversationAudioContextAssembler(context=context)
|
||||
|
||||
llm_aggregator_buffer = LLMAggregatorBuffer()
|
||||
|
||||
bot_output_gate = OutputGate(
|
||||
notifier=notifier, context=context, llm_transcription_buffer=llm_aggregator_buffer
|
||||
)
|
||||
|
||||
super().__init__(
|
||||
[
|
||||
audio_accumulator,
|
||||
ParallelPipeline(
|
||||
[
|
||||
# Pass everything except UserStoppedSpeaking to the elements after
|
||||
# this ParallelPipeline
|
||||
FunctionFilter(filter=block_user_stopped_speaking),
|
||||
],
|
||||
[
|
||||
ParallelPipeline(
|
||||
[
|
||||
classifier_llm,
|
||||
completeness_check,
|
||||
],
|
||||
[
|
||||
tx_llm,
|
||||
llm_aggregator_buffer,
|
||||
],
|
||||
)
|
||||
],
|
||||
[
|
||||
conversation_audio_context_assembler,
|
||||
llm,
|
||||
bot_output_gate, # buffer output until notified, then flush frames and update context
|
||||
],
|
||||
),
|
||||
]
|
||||
)
|
||||
|
||||
|
||||
# We use lambdas to defer transport parameter creation until the transport
|
||||
# type is selected at runtime.
|
||||
transport_params = {
|
||||
"daily": lambda: DailyParams(
|
||||
audio_in_enabled=True,
|
||||
audio_out_enabled=True,
|
||||
),
|
||||
"twilio": lambda: FastAPIWebsocketParams(
|
||||
audio_in_enabled=True,
|
||||
audio_out_enabled=True,
|
||||
),
|
||||
"webrtc": lambda: TransportParams(
|
||||
audio_in_enabled=True,
|
||||
audio_out_enabled=True,
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
|
||||
logger.info(f"Starting bot")
|
||||
|
||||
tts = CartesiaTTSService(
|
||||
api_key=os.getenv("CARTESIA_API_KEY"),
|
||||
voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady
|
||||
)
|
||||
|
||||
# This is the regular LLM that responds conversationally.
|
||||
conversation_llm = GoogleLLMService(
|
||||
name="Conversation",
|
||||
model=CONVERSATION_MODEL,
|
||||
api_key=os.getenv("GOOGLE_API_KEY"),
|
||||
system_instruction=conversation_system_instruction,
|
||||
)
|
||||
|
||||
context = LLMContext()
|
||||
user_aggregator, assistant_aggregator = LLMContextAggregatorPair(
|
||||
context,
|
||||
user_params=LLMUserAggregatorParams(
|
||||
user_turn_strategies=UserTurnStrategies(
|
||||
stop=[TurnAnalyzerUserTurnStopStrategy(turn_analyzer=LocalSmartTurnAnalyzerV3())]
|
||||
),
|
||||
vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)),
|
||||
),
|
||||
)
|
||||
|
||||
llm = TurnDetectionLLM(conversation_llm, context)
|
||||
|
||||
pipeline = Pipeline(
|
||||
[
|
||||
transport.input(),
|
||||
user_aggregator,
|
||||
llm,
|
||||
tts,
|
||||
transport.output(),
|
||||
assistant_aggregator,
|
||||
],
|
||||
)
|
||||
|
||||
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")
|
||||
|
||||
@transport.event_handler("on_app_message")
|
||||
async def on_app_message(transport, message, sender):
|
||||
logger.debug(f"Received app message: {message}, sender: {sender}") # TODO: revert
|
||||
if "message" not in message:
|
||||
return
|
||||
|
||||
await task.queue_frames(
|
||||
[
|
||||
UserStartedSpeakingFrame(),
|
||||
TranscriptionFrame(
|
||||
user_id="", timestamp=time_now_iso8601(), text=message["message"]
|
||||
),
|
||||
UserStoppedSpeakingFrame(),
|
||||
]
|
||||
)
|
||||
|
||||
@transport.event_handler("on_client_disconnected")
|
||||
async def on_client_disconnected(transport, client):
|
||||
logger.info(f"Client disconnected")
|
||||
await task.cancel()
|
||||
|
||||
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()
|
||||
@@ -47,6 +47,7 @@ from pipecat.frames.frames import (
|
||||
LLMThoughtEndFrame,
|
||||
LLMThoughtStartFrame,
|
||||
LLMThoughtTextFrame,
|
||||
LLMUpdateSettingsFrame,
|
||||
SpeechControlParamsFrame,
|
||||
StartFrame,
|
||||
TextFrame,
|
||||
@@ -69,6 +70,7 @@ from pipecat.turns.user_idle_controller import UserIdleController
|
||||
from pipecat.turns.user_mute import BaseUserMuteStrategy
|
||||
from pipecat.turns.user_start import BaseUserTurnStartStrategy, UserTurnStartedParams
|
||||
from pipecat.turns.user_stop import BaseUserTurnStopStrategy, UserTurnStoppedParams
|
||||
from pipecat.turns.user_turn_completion_mixin import UserTurnCompletionConfig
|
||||
from pipecat.turns.user_turn_controller import UserTurnController
|
||||
from pipecat.turns.user_turn_strategies import ExternalUserTurnStrategies, UserTurnStrategies
|
||||
from pipecat.utils.string import TextPartForConcatenation, concatenate_aggregated_text
|
||||
@@ -89,6 +91,13 @@ class LLMUserAggregatorParams:
|
||||
has been idle (not speaking) for this duration. Set to None to disable
|
||||
idle detection.
|
||||
vad_analyzer: Voice Activity Detection analyzer instance.
|
||||
filter_incomplete_user_turns: Whether to filter out incomplete user turns.
|
||||
When enabled, the LLM outputs a turn completion marker at the start of
|
||||
each response: ✓ (complete), ○ (incomplete short), or ◐ (incomplete long).
|
||||
Incomplete responses are suppressed and timeouts trigger re-prompting.
|
||||
user_turn_completion_config: Configuration for turn completion behavior including
|
||||
custom instructions, timeouts, and prompts. Only used when
|
||||
filter_incomplete_user_turns is True.
|
||||
"""
|
||||
|
||||
user_turn_strategies: Optional[UserTurnStrategies] = None
|
||||
@@ -96,6 +105,8 @@ class LLMUserAggregatorParams:
|
||||
user_turn_stop_timeout: float = 5.0
|
||||
user_idle_timeout: Optional[float] = None
|
||||
vad_analyzer: Optional[VADAnalyzer] = None
|
||||
filter_incomplete_user_turns: bool = False
|
||||
user_turn_completion_config: Optional[UserTurnCompletionConfig] = None
|
||||
|
||||
|
||||
@dataclass
|
||||
@@ -488,6 +499,24 @@ class LLMUserAggregator(LLMContextAggregator):
|
||||
for s in self._params.user_mute_strategies:
|
||||
await s.setup(self.task_manager)
|
||||
|
||||
# Enable incomplete turn filtering on the LLM if configured
|
||||
if self._params.filter_incomplete_user_turns:
|
||||
# Get config or use defaults
|
||||
config = self._params.user_turn_completion_config or UserTurnCompletionConfig()
|
||||
|
||||
# Enable the feature on the LLM with config
|
||||
await self.push_frame(
|
||||
LLMUpdateSettingsFrame(
|
||||
settings={
|
||||
"filter_incomplete_user_turns": True,
|
||||
"user_turn_completion_config": config,
|
||||
}
|
||||
)
|
||||
)
|
||||
|
||||
# Auto-inject turn completion instructions into context
|
||||
self._context.add_message({"role": "system", "content": config.completion_instructions})
|
||||
|
||||
async def _stop(self, frame: EndFrame):
|
||||
await self._maybe_emit_user_turn_stopped(on_session_end=True)
|
||||
await self._cleanup()
|
||||
@@ -1145,13 +1174,40 @@ class LLMAssistantAggregator(LLMContextAggregator):
|
||||
async def _trigger_assistant_turn_stopped(self):
|
||||
aggregation = await self.push_aggregation()
|
||||
if aggregation:
|
||||
# Strip turn completion markers from the transcript
|
||||
content = self._maybe_strip_turn_completion_markers(aggregation)
|
||||
message = AssistantTurnStoppedMessage(
|
||||
content=aggregation, timestamp=self._assistant_turn_start_timestamp
|
||||
content=content, timestamp=self._assistant_turn_start_timestamp
|
||||
)
|
||||
await self._call_event_handler("on_assistant_turn_stopped", message)
|
||||
|
||||
self._assistant_turn_start_timestamp = ""
|
||||
|
||||
def _maybe_strip_turn_completion_markers(self, text: str) -> str:
|
||||
"""Strip turn completion markers from assistant transcript.
|
||||
|
||||
These markers (✓, ○, ◐) are used internally for turn completion
|
||||
detection and shouldn't appear in the final transcript.
|
||||
"""
|
||||
from pipecat.turns.user_turn_completion_mixin import (
|
||||
USER_TURN_COMPLETE_MARKER,
|
||||
USER_TURN_INCOMPLETE_LONG_MARKER,
|
||||
USER_TURN_INCOMPLETE_SHORT_MARKER,
|
||||
)
|
||||
|
||||
marker_found = False
|
||||
for marker in (
|
||||
USER_TURN_COMPLETE_MARKER,
|
||||
USER_TURN_INCOMPLETE_SHORT_MARKER,
|
||||
USER_TURN_INCOMPLETE_LONG_MARKER,
|
||||
):
|
||||
if marker in text:
|
||||
text = text.replace(marker, "")
|
||||
marker_found = True
|
||||
|
||||
# Only strip whitespace if we removed a marker
|
||||
return text.strip() if marker_found else text
|
||||
|
||||
|
||||
class LLMContextAggregatorPair:
|
||||
"""Pair of LLM context aggregators for updating context with user and assistant messages."""
|
||||
|
||||
@@ -439,7 +439,7 @@ class AnthropicLLMService(LLMService):
|
||||
|
||||
if event.type == "content_block_delta":
|
||||
if hasattr(event.delta, "text"):
|
||||
await self.push_frame(LLMTextFrame(event.delta.text))
|
||||
await self._push_llm_text(event.delta.text)
|
||||
completion_tokens_estimate += self._estimate_tokens(event.delta.text)
|
||||
elif hasattr(event.delta, "partial_json") and tool_use_block:
|
||||
json_accumulator += event.delta.partial_json
|
||||
|
||||
@@ -1107,7 +1107,7 @@ class AWSBedrockLLMService(LLMService):
|
||||
if "contentBlockDelta" in event:
|
||||
delta = event["contentBlockDelta"]["delta"]
|
||||
if "text" in delta:
|
||||
await self.push_frame(LLMTextFrame(delta["text"]))
|
||||
await self._push_llm_text(delta["text"])
|
||||
completion_tokens_estimate += self._estimate_tokens(delta["text"])
|
||||
elif "toolUse" in delta and "input" in delta["toolUse"]:
|
||||
# Handle partial JSON for tool use
|
||||
|
||||
@@ -1023,7 +1023,7 @@ class GoogleLLMService(LLMService):
|
||||
await self.push_frame(LLMThoughtEndFrame())
|
||||
else:
|
||||
accumulated_text += part.text
|
||||
await self.push_frame(LLMTextFrame(part.text))
|
||||
await self._push_llm_text(part.text)
|
||||
elif part.function_call:
|
||||
function_call = part.function_call
|
||||
function_call_id = function_call.id or str(uuid.uuid4())
|
||||
|
||||
@@ -56,6 +56,7 @@ from pipecat.processors.aggregators.llm_response import (
|
||||
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
|
||||
from pipecat.processors.frame_processor import FrameDirection
|
||||
from pipecat.services.ai_service import AIService
|
||||
from pipecat.turns.user_turn_completion_mixin import UserTurnCompletionLLMServiceMixin
|
||||
|
||||
# Type alias for a callable that handles LLM function calls.
|
||||
FunctionCallHandler = Callable[["FunctionCallParams"], Awaitable[None]]
|
||||
@@ -142,7 +143,7 @@ class FunctionCallRunnerItem:
|
||||
run_llm: Optional[bool] = None
|
||||
|
||||
|
||||
class LLMService(AIService):
|
||||
class LLMService(UserTurnCompletionLLMServiceMixin, AIService):
|
||||
"""Base class for all LLM services.
|
||||
|
||||
Handles function calling registration and execution with support for both
|
||||
@@ -186,6 +187,7 @@ class LLMService(AIService):
|
||||
super().__init__(**kwargs)
|
||||
self._run_in_parallel = run_in_parallel
|
||||
self._function_call_timeout_secs = function_call_timeout_secs
|
||||
self._filter_incomplete_user_turns: bool = False
|
||||
self._start_callbacks = {}
|
||||
self._adapter = self.adapter_class()
|
||||
self._functions: Dict[Optional[str], FunctionCallRegistryItem] = {}
|
||||
@@ -295,6 +297,35 @@ class LLMService(AIService):
|
||||
if not self._run_in_parallel:
|
||||
await self._cancel_sequential_runner_task()
|
||||
|
||||
async def _update_settings(self, settings: Mapping[str, Any]):
|
||||
"""Update LLM service settings.
|
||||
|
||||
Handles turn completion settings specially since they are not model
|
||||
parameters and should not be passed to the underlying LLM API.
|
||||
|
||||
Args:
|
||||
settings: Dictionary of settings to update.
|
||||
"""
|
||||
# Turn completion settings to extract (not model parameters)
|
||||
turn_completion_keys = {"filter_incomplete_user_turns", "user_turn_completion_config"}
|
||||
|
||||
# Handle turn completion settings
|
||||
if "filter_incomplete_user_turns" in settings:
|
||||
self._filter_incomplete_user_turns = settings["filter_incomplete_user_turns"]
|
||||
logger.info(
|
||||
f"{self}: Incomplete turn filtering {'enabled' if self._filter_incomplete_user_turns else 'disabled'}"
|
||||
)
|
||||
|
||||
# Configure the mixin with config object
|
||||
if self._filter_incomplete_user_turns and "user_turn_completion_config" in settings:
|
||||
self.set_user_turn_completion_config(settings["user_turn_completion_config"])
|
||||
|
||||
# Remove turn completion settings before passing to parent
|
||||
settings = {k: v for k, v in settings.items() if k not in turn_completion_keys}
|
||||
|
||||
# Let the parent handle remaining model parameters
|
||||
await super()._update_settings(settings)
|
||||
|
||||
async def process_frame(self, frame: Frame, direction: FrameDirection):
|
||||
"""Process a frame.
|
||||
|
||||
@@ -322,6 +353,20 @@ class LLMService(AIService):
|
||||
|
||||
await super().push_frame(frame, direction)
|
||||
|
||||
async def _push_llm_text(self, text: str):
|
||||
"""Push LLM text, using turn completion detection if enabled.
|
||||
|
||||
This helper method simplifies text pushing in LLM implementations by
|
||||
handling the conditional logic for turn completion internally.
|
||||
|
||||
Args:
|
||||
text: The text content from the LLM to push.
|
||||
"""
|
||||
if self._filter_incomplete_user_turns:
|
||||
await self._push_turn_text(text)
|
||||
else:
|
||||
await self.push_frame(LLMTextFrame(text))
|
||||
|
||||
async def _handle_interruptions(self, _: InterruptionFrame):
|
||||
for function_name, entry in self._functions.items():
|
||||
if entry.cancel_on_interruption:
|
||||
|
||||
@@ -422,7 +422,7 @@ class BaseOpenAILLMService(LLMService):
|
||||
# Keep iterating through the response to collect all the argument fragments
|
||||
arguments += tool_call.function.arguments
|
||||
elif chunk.choices[0].delta.content:
|
||||
await self.push_frame(LLMTextFrame(chunk.choices[0].delta.content))
|
||||
await self._push_llm_text(chunk.choices[0].delta.content)
|
||||
|
||||
# When gpt-4o-audio / gpt-4o-mini-audio is used for llm or stt+llm
|
||||
# we need to get LLMTextFrame for the transcript
|
||||
|
||||
428
src/pipecat/turns/user_turn_completion_mixin.py
Normal file
428
src/pipecat/turns/user_turn_completion_mixin.py
Normal file
@@ -0,0 +1,428 @@
|
||||
#
|
||||
# Copyright (c) 2024-2026, Daily
|
||||
#
|
||||
# SPDX-License-Identifier: BSD 2-Clause License
|
||||
#
|
||||
|
||||
"""Mixin for adding turn completion detection to LLM services.
|
||||
|
||||
This mixin enables LLM services to detect and process turn completion markers
|
||||
(COMPLETE/INCOMPLETE) in LLM responses, allowing for smarter conversation flow
|
||||
where the LLM can indicate whether the user's input was complete or if they
|
||||
were interrupted mid-thought.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
from dataclasses import dataclass
|
||||
from typing import Literal, Optional
|
||||
|
||||
from loguru import logger
|
||||
|
||||
from pipecat.frames.frames import (
|
||||
Frame,
|
||||
InterruptionFrame,
|
||||
LLMFullResponseEndFrame,
|
||||
LLMMessagesAppendFrame,
|
||||
LLMRunFrame,
|
||||
LLMTextFrame,
|
||||
)
|
||||
from pipecat.processors.frame_processor import FrameDirection
|
||||
|
||||
# Turn completion markers
|
||||
USER_TURN_COMPLETE_MARKER = "✓"
|
||||
USER_TURN_INCOMPLETE_SHORT_MARKER = "○" # Short wait - user likely continues soon
|
||||
USER_TURN_INCOMPLETE_LONG_MARKER = "◐" # Long wait - user needs more time
|
||||
|
||||
# Default prompts for incomplete timeouts
|
||||
DEFAULT_INCOMPLETE_SHORT_PROMPT = """The user paused briefly. Generate a brief, natural prompt to encourage them to continue.
|
||||
|
||||
IMPORTANT: You MUST respond with ✓ followed by your message. Do NOT output ○ or ◐ - the user has already been given time to continue.
|
||||
|
||||
Your response should:
|
||||
- Be contextually relevant to what was just discussed
|
||||
- Sound natural and conversational
|
||||
- Be very concise (1 sentence max)
|
||||
- Gently prompt them to continue
|
||||
|
||||
Example format: ✓ Go ahead, I'm listening.
|
||||
|
||||
Generate your ✓ response now."""
|
||||
|
||||
DEFAULT_INCOMPLETE_LONG_PROMPT = """The user has been quiet for a while. Generate a friendly check-in message.
|
||||
|
||||
IMPORTANT: You MUST respond with ✓ followed by your message. Do NOT output ○ or ◐ - the user has already been given plenty of time.
|
||||
|
||||
Your response should:
|
||||
- Acknowledge they might be thinking or busy
|
||||
- Offer to help or continue when ready
|
||||
- Be warm and understanding
|
||||
- Be brief (1 sentence)
|
||||
|
||||
Example format: ✓ No rush! Let me know when you're ready to continue.
|
||||
|
||||
Generate your ✓ response now."""
|
||||
|
||||
# System prompt instructions for turn completion that can be appended to any base prompt
|
||||
USER_TURN_COMPLETION_INSTRUCTIONS = """
|
||||
CRITICAL INSTRUCTION - MANDATORY RESPONSE FORMAT:
|
||||
Every single response MUST begin with a turn completion indicator. This is not optional.
|
||||
|
||||
TURN COMPLETION DECISION FRAMEWORK:
|
||||
Ask yourself: "Has the user provided enough information for me to give a meaningful, substantive response?"
|
||||
|
||||
Mark as COMPLETE (✓) when:
|
||||
- The user has answered your question with actual content
|
||||
- The user has made a complete request or statement
|
||||
- The user has provided all necessary information for you to respond meaningfully
|
||||
- The conversation can naturally progress to your substantive response
|
||||
|
||||
Mark as INCOMPLETE SHORT (○) when the user will likely continue soon:
|
||||
- The user was clearly cut off mid-sentence or mid-word
|
||||
- The user is in the middle of a thought that got interrupted
|
||||
- Brief technical interruption (they'll resume in a few seconds)
|
||||
|
||||
Mark as INCOMPLETE LONG (◐) when the user needs more time:
|
||||
- The user explicitly asks for time: "let me think", "give me a minute", "hold on"
|
||||
- The user is clearly pondering or deliberating: "hmm", "well...", "that's a good question"
|
||||
- The user acknowledged but hasn't answered yet: "That's interesting..."
|
||||
- The response feels like a preamble before the actual answer
|
||||
|
||||
RESPOND in one of these three formats:
|
||||
1. If COMPLETE: `✓` followed by a space and your full substantive response
|
||||
2. If INCOMPLETE SHORT: ONLY the character `○` (user will continue in a few seconds)
|
||||
3. If INCOMPLETE LONG: ONLY the character `◐` (user needs more time to think)
|
||||
|
||||
KEY INSIGHT: Grammatically complete ≠ conversationally complete
|
||||
- "That's a really good question." is grammatically complete but conversationally incomplete (use ◐)
|
||||
- "I'd go to Japan because I love" is mid-sentence (use ○)
|
||||
|
||||
EXAMPLES:
|
||||
|
||||
You ask: "Where would you travel?"
|
||||
User: "I'd go to Japan because I love"
|
||||
→ `○`
|
||||
(Cut off mid-sentence - they'll continue in seconds)
|
||||
|
||||
You ask: "Where would you travel?"
|
||||
User: "That's a good question. Let me think..."
|
||||
→ `◐`
|
||||
(User is deliberating - give them time)
|
||||
|
||||
You ask: "Where would you travel?"
|
||||
User: "Hmm, hold on a second."
|
||||
→ `◐`
|
||||
(User explicitly asked for time)
|
||||
|
||||
You ask: "Where would you travel?"
|
||||
User: "I'd go to Japan because I love the culture."
|
||||
→ `✓ Japan is a wonderful choice! The blend of ancient traditions and modern innovation is truly unique. Have you been before?`
|
||||
(Complete answer - give full response)
|
||||
|
||||
User: "I need help with"
|
||||
→ `○`
|
||||
(Cut off mid-request - they'll finish soon)
|
||||
|
||||
User: "Well, let me think about that for a moment."
|
||||
→ `◐`
|
||||
(User needs time to think)
|
||||
|
||||
User: "Can you help me book a flight to New York next week?"
|
||||
→ `✓ I'd be happy to help you with that! Let me gather some information...`
|
||||
(Complete request - provide full response)
|
||||
|
||||
User: "Give me a minute to gather my thoughts."
|
||||
→ `◐`
|
||||
(User explicitly asked for time)
|
||||
|
||||
FORMAT REQUIREMENTS:
|
||||
- ALWAYS use single-character indicators: `✓` (complete), `○` (short wait), or `◐` (long wait)
|
||||
- For COMPLETE: `✓` followed by a space and your full response
|
||||
- For INCOMPLETE: ONLY the single character (`○` or `◐`) with absolutely nothing else
|
||||
- Your turn indicator must be the very first character in your response
|
||||
|
||||
Remember: Focus on conversational completeness and how long the user might need. Was it a mid-sentence cutoff (○) or do they need time to think (◐)?"""
|
||||
|
||||
|
||||
@dataclass
|
||||
class UserTurnCompletionConfig:
|
||||
"""Configuration for turn completion behavior.
|
||||
|
||||
Attributes:
|
||||
instructions: Custom instructions for turn completion. If not provided,
|
||||
uses default USER_TURN_COMPLETION_INSTRUCTIONS.
|
||||
incomplete_short_timeout: Seconds to wait after short incomplete (○) before prompting.
|
||||
incomplete_long_timeout: Seconds to wait after long incomplete (◐) before prompting.
|
||||
incomplete_short_prompt: Custom prompt when short timeout expires.
|
||||
incomplete_long_prompt: Custom prompt when long timeout expires.
|
||||
"""
|
||||
|
||||
instructions: Optional[str] = None
|
||||
incomplete_short_timeout: float = 5.0
|
||||
incomplete_long_timeout: float = 10.0
|
||||
incomplete_short_prompt: Optional[str] = None
|
||||
incomplete_long_prompt: Optional[str] = None
|
||||
|
||||
@property
|
||||
def completion_instructions(self) -> str:
|
||||
"""Turn completion instructions, using default if not set."""
|
||||
return self.instructions or USER_TURN_COMPLETION_INSTRUCTIONS
|
||||
|
||||
@property
|
||||
def short_prompt(self) -> str:
|
||||
"""Short incomplete prompt, using default if not set."""
|
||||
return self.incomplete_short_prompt or DEFAULT_INCOMPLETE_SHORT_PROMPT
|
||||
|
||||
@property
|
||||
def long_prompt(self) -> str:
|
||||
"""Long incomplete prompt, using default if not set."""
|
||||
return self.incomplete_long_prompt or DEFAULT_INCOMPLETE_LONG_PROMPT
|
||||
|
||||
|
||||
class UserTurnCompletionLLMServiceMixin:
|
||||
"""Mixin that adds turn completion detection to LLM services.
|
||||
|
||||
This mixin provides methods to push LLM text with turn completion detection.
|
||||
It processes turn completion markers to enable smarter conversation flow:
|
||||
|
||||
- ✓ (COMPLETE): Push response normally
|
||||
- ○ (INCOMPLETE SHORT): Suppress response, wait ~5s, then prompt
|
||||
- ◐ (INCOMPLETE LONG): Suppress response, wait ~15s, then prompt
|
||||
|
||||
When incomplete timeouts expire, the mixin automatically prompts the LLM
|
||||
with a contextual follow-up message to re-engage the user.
|
||||
|
||||
Usage:
|
||||
The LLM service controls when to use turn completion by calling
|
||||
_push_turn_text instead of push_frame:
|
||||
|
||||
# With turn completion:
|
||||
if self._filter_incomplete_user_turns:
|
||||
await self._push_turn_text(chunk.text)
|
||||
else:
|
||||
await self.push_frame(LLMTextFrame(chunk.text))
|
||||
|
||||
The mixin requires that the base class has a `push_frame` method compatible
|
||||
with FrameProcessor's signature.
|
||||
"""
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
"""Initialize the turn completion mixin.
|
||||
|
||||
Args:
|
||||
*args: Positional arguments passed to parent class.
|
||||
**kwargs: Keyword arguments passed to parent class.
|
||||
"""
|
||||
super().__init__(*args, **kwargs)
|
||||
self._turn_text_buffer = ""
|
||||
# Safety mechanism: True when incomplete is detected. While the prompt
|
||||
# instructs the LLM to output ONLY the marker for incomplete turns, this flag
|
||||
# ensures graceful degradation if the LLM disobeys and outputs additional text.
|
||||
self._turn_suppressed = False
|
||||
self._turn_complete_found = False # True when ✓ (COMPLETE) is detected
|
||||
|
||||
# Timeout handling
|
||||
self._user_turn_completion_config = UserTurnCompletionConfig()
|
||||
self._incomplete_timeout_task: Optional[asyncio.Task] = None
|
||||
self._incomplete_type: Optional[Literal["short", "long"]] = None
|
||||
|
||||
def set_user_turn_completion_config(self, config: UserTurnCompletionConfig):
|
||||
"""Set the turn completion configuration.
|
||||
|
||||
Args:
|
||||
config: The turn completion configuration.
|
||||
"""
|
||||
self._user_turn_completion_config = config
|
||||
|
||||
async def _start_incomplete_timeout(self, incomplete_type: Literal["short", "long"]):
|
||||
"""Start a timeout task for incomplete turn handling.
|
||||
|
||||
Args:
|
||||
incomplete_type: Either "short" or "long" to determine timeout duration.
|
||||
"""
|
||||
# Cancel any existing timeout
|
||||
await self._cancel_incomplete_timeout()
|
||||
|
||||
self._incomplete_type = incomplete_type
|
||||
|
||||
if incomplete_type == "short":
|
||||
timeout = self._user_turn_completion_config.incomplete_short_timeout
|
||||
else:
|
||||
timeout = self._user_turn_completion_config.incomplete_long_timeout
|
||||
|
||||
logger.debug(f"Starting {incomplete_type} incomplete timeout ({timeout}s)")
|
||||
self._incomplete_timeout_task = self.create_task(
|
||||
self._incomplete_timeout_handler(incomplete_type, timeout),
|
||||
f"_incomplete_timeout_{incomplete_type}",
|
||||
)
|
||||
|
||||
async def _cancel_incomplete_timeout(self):
|
||||
"""Cancel any pending incomplete timeout task."""
|
||||
if self._incomplete_timeout_task and not self._incomplete_timeout_task.done():
|
||||
logger.debug("Cancelling incomplete timeout")
|
||||
await self.cancel_task(self._incomplete_timeout_task)
|
||||
self._incomplete_timeout_task = None
|
||||
self._incomplete_type = None
|
||||
|
||||
async def _incomplete_timeout_handler(
|
||||
self, incomplete_type: Literal["short", "long"], timeout: float
|
||||
):
|
||||
"""Handle incomplete timeout expiration.
|
||||
|
||||
Args:
|
||||
incomplete_type: Either "short" or "long".
|
||||
timeout: The timeout duration in seconds.
|
||||
"""
|
||||
try:
|
||||
await asyncio.sleep(timeout)
|
||||
|
||||
# Timeout expired - reset state before prompting LLM
|
||||
logger.info(f"Incomplete {incomplete_type} timeout expired, prompting LLM")
|
||||
await self._turn_reset()
|
||||
self._incomplete_timeout_task = None
|
||||
self._incomplete_type = None
|
||||
|
||||
# Get the appropriate prompt
|
||||
if incomplete_type == "short":
|
||||
prompt = self._user_turn_completion_config.short_prompt
|
||||
else:
|
||||
prompt = self._user_turn_completion_config.long_prompt
|
||||
|
||||
# Push through pipeline to trigger LLM response
|
||||
await self.push_frame(
|
||||
LLMMessagesAppendFrame(messages=[{"role": "system", "content": prompt}])
|
||||
)
|
||||
await self.push_frame(LLMRunFrame())
|
||||
|
||||
except asyncio.CancelledError:
|
||||
# Timeout was cancelled (user spoke or interruption)
|
||||
pass
|
||||
|
||||
async def _turn_reset(self):
|
||||
"""Reset turn completion state between responses.
|
||||
|
||||
Call this at the end of each LLM response to clear buffered text and reset state.
|
||||
If no marker was found, pushes the buffered text to avoid losing content.
|
||||
|
||||
Note: This does NOT cancel pending incomplete timeouts. Timeouts are only
|
||||
cancelled on InterruptionFrame (when user speaks).
|
||||
"""
|
||||
# Check if no marker was found in this response
|
||||
marker_found = self._turn_suppressed or self._turn_complete_found
|
||||
if not marker_found and self._turn_text_buffer:
|
||||
# Graceful degradation: push the buffered text so it's not lost
|
||||
logger.warning(
|
||||
f"{self}: filter_incomplete_user_turns is enabled but LLM response did not "
|
||||
f"contain turn completion markers (✓/○/◐). Pushing text anyway. "
|
||||
"The system prompt may be missing turn completion instructions."
|
||||
)
|
||||
await self.push_frame(LLMTextFrame(self._turn_text_buffer))
|
||||
|
||||
self._turn_text_buffer = ""
|
||||
self._turn_suppressed = False
|
||||
self._turn_complete_found = False
|
||||
|
||||
async def process_frame(self, frame: Frame, direction: FrameDirection):
|
||||
"""Process frames, handling turn completion state resets.
|
||||
|
||||
Args:
|
||||
frame: The frame to process.
|
||||
direction: The direction of frame processing.
|
||||
"""
|
||||
# Handle interruptions by cancelling timeout and resetting state
|
||||
if isinstance(frame, InterruptionFrame):
|
||||
await self._cancel_incomplete_timeout()
|
||||
await self._turn_reset()
|
||||
# Reset turn state at end of LLM response (but don't cancel timeout -
|
||||
# incomplete timeouts should continue running)
|
||||
elif isinstance(frame, LLMFullResponseEndFrame):
|
||||
await self._turn_reset()
|
||||
|
||||
# Pass frame to parent
|
||||
await super().process_frame(frame, direction)
|
||||
|
||||
async def _push_turn_text(self, text: str):
|
||||
"""Push LLM text with turn completion detection.
|
||||
|
||||
This method should be used instead of `push_frame(LLMTextFrame(text))` when
|
||||
turn completion is enabled. It will:
|
||||
1. Detect turn markers (✓, ○, or ◐)
|
||||
2. When ○ (SHORT) is found: suppress text, start short timeout
|
||||
3. When ◐ (LONG) is found: suppress text, start long timeout
|
||||
4. When ✓ (COMPLETE) is found: push all text with marker marked as skip_tts
|
||||
5. After marker detected: all subsequent text flows through immediately
|
||||
|
||||
Args:
|
||||
text: The text content from the LLM to push.
|
||||
"""
|
||||
# If we've already detected incomplete, suppress all remaining text.
|
||||
# This is a safety mechanism in case the LLM disobeys the prompt and outputs
|
||||
# additional text after the marker (e.g., "○ Please continue...").
|
||||
if self._turn_suppressed:
|
||||
return
|
||||
|
||||
# If ✓ (COMPLETE) was already found, push text immediately without buffering
|
||||
if self._turn_complete_found:
|
||||
await self.push_frame(LLMTextFrame(text))
|
||||
return
|
||||
|
||||
# Add text to buffer
|
||||
self._turn_text_buffer += text
|
||||
|
||||
# Check for incomplete markers (○ short, ◐ long)
|
||||
# These indicate the user was cut off or needs time - we suppress the bot's
|
||||
# response and start a timeout to re-prompt later.
|
||||
incomplete_type: Optional[Literal["short", "long"]] = None
|
||||
if USER_TURN_INCOMPLETE_SHORT_MARKER in self._turn_text_buffer:
|
||||
incomplete_type = "short"
|
||||
elif USER_TURN_INCOMPLETE_LONG_MARKER in self._turn_text_buffer:
|
||||
incomplete_type = "long"
|
||||
|
||||
if incomplete_type:
|
||||
marker = (
|
||||
USER_TURN_INCOMPLETE_SHORT_MARKER
|
||||
if incomplete_type == "short"
|
||||
else USER_TURN_INCOMPLETE_LONG_MARKER
|
||||
)
|
||||
logger.debug(
|
||||
f"INCOMPLETE {incomplete_type.upper()} ({marker}) detected, suppressing text"
|
||||
)
|
||||
self._turn_suppressed = True
|
||||
|
||||
# Push the marker with skip_tts=True so it's added to context (maintains
|
||||
# conversation continuity per prompt instructions) but not spoken by TTS
|
||||
frame = LLMTextFrame(self._turn_text_buffer)
|
||||
frame.skip_tts = True
|
||||
await self.push_frame(frame)
|
||||
|
||||
self._turn_text_buffer = ""
|
||||
await self._start_incomplete_timeout(incomplete_type)
|
||||
return
|
||||
|
||||
# Check for ✓ (COMPLETE) marker - user's turn was complete, respond normally
|
||||
if USER_TURN_COMPLETE_MARKER in self._turn_text_buffer:
|
||||
logger.debug(f"COMPLETE ({USER_TURN_COMPLETE_MARKER}) detected, pushing buffered text")
|
||||
|
||||
# Split buffer at the marker to handle cases where marker and text
|
||||
# arrive in the same chunk (e.g., "✓ Hello!" from some LLMs)
|
||||
marker_pos = self._turn_text_buffer.index(USER_TURN_COMPLETE_MARKER)
|
||||
marker_end = marker_pos + len(USER_TURN_COMPLETE_MARKER)
|
||||
|
||||
# Push the marker with skip_tts=True - adds to context but not spoken
|
||||
marker_text = self._turn_text_buffer[:marker_end]
|
||||
frame = LLMTextFrame(marker_text)
|
||||
frame.skip_tts = True
|
||||
await self.push_frame(frame)
|
||||
|
||||
# Push remaining text after marker as normal speech
|
||||
remaining_text = self._turn_text_buffer[marker_end:]
|
||||
if remaining_text:
|
||||
# Strip leading space after marker if present (✓ Hello -> Hello)
|
||||
if remaining_text.startswith(" "):
|
||||
remaining_text = remaining_text[1:]
|
||||
if remaining_text:
|
||||
await self.push_frame(LLMTextFrame(remaining_text))
|
||||
|
||||
# Mark complete - all subsequent text flows through immediately
|
||||
self._turn_text_buffer = ""
|
||||
self._turn_complete_found = True
|
||||
return
|
||||
@@ -567,6 +567,54 @@ class TestLLMAssistantAggregator(unittest.IsolatedAsyncioTestCase):
|
||||
self.assertEqual(len(stop_messages), 1)
|
||||
self.assertEqual(stop_messages[0].content, "Hello from Pipecat!")
|
||||
|
||||
async def test_turn_completion_markers_stripped_from_transcript(self):
|
||||
"""Turn completion markers should be stripped from assistant transcript."""
|
||||
from pipecat.turns.user_turn_completion_mixin import (
|
||||
USER_TURN_COMPLETE_MARKER,
|
||||
USER_TURN_INCOMPLETE_SHORT_MARKER,
|
||||
)
|
||||
|
||||
context = LLMContext()
|
||||
aggregator = LLMAssistantAggregator(context)
|
||||
|
||||
stop_messages = []
|
||||
|
||||
@aggregator.event_handler("on_assistant_turn_stopped")
|
||||
async def on_assistant_turn_stopped(aggregator, message: AssistantTurnStoppedMessage):
|
||||
stop_messages.append(message)
|
||||
|
||||
# Send text with a turn completion marker
|
||||
frames_to_send = [
|
||||
LLMFullResponseStartFrame(),
|
||||
LLMTextFrame(f"{USER_TURN_COMPLETE_MARKER} Hello from Pipecat!"),
|
||||
LLMFullResponseEndFrame(),
|
||||
]
|
||||
await run_test(aggregator, frames_to_send=frames_to_send)
|
||||
|
||||
# The marker should be stripped from the transcript
|
||||
self.assertEqual(len(stop_messages), 1)
|
||||
self.assertEqual(stop_messages[0].content, "Hello from Pipecat!")
|
||||
|
||||
# Test incomplete markers are also stripped
|
||||
stop_messages.clear()
|
||||
context2 = LLMContext()
|
||||
aggregator2 = LLMAssistantAggregator(context2)
|
||||
|
||||
@aggregator2.event_handler("on_assistant_turn_stopped")
|
||||
async def on_assistant_turn_stopped2(aggregator, message: AssistantTurnStoppedMessage):
|
||||
stop_messages.append(message)
|
||||
|
||||
frames_to_send = [
|
||||
LLMFullResponseStartFrame(),
|
||||
LLMTextFrame(USER_TURN_INCOMPLETE_SHORT_MARKER),
|
||||
LLMFullResponseEndFrame(),
|
||||
]
|
||||
await run_test(aggregator2, frames_to_send=frames_to_send)
|
||||
|
||||
# The incomplete marker should be stripped (resulting in empty content)
|
||||
self.assertEqual(len(stop_messages), 1)
|
||||
self.assertEqual(stop_messages[0].content, "")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
117
tests/test_user_turn_completion_mixin.py
Normal file
117
tests/test_user_turn_completion_mixin.py
Normal file
@@ -0,0 +1,117 @@
|
||||
#
|
||||
# Copyright (c) 2024-2026, Daily
|
||||
#
|
||||
# SPDX-License-Identifier: BSD 2-Clause License
|
||||
#
|
||||
|
||||
import unittest
|
||||
from unittest.mock import AsyncMock
|
||||
|
||||
from pipecat.frames.frames import LLMTextFrame
|
||||
from pipecat.processors.frame_processor import FrameProcessor
|
||||
from pipecat.turns.user_turn_completion_mixin import (
|
||||
USER_TURN_COMPLETE_MARKER,
|
||||
USER_TURN_INCOMPLETE_LONG_MARKER,
|
||||
USER_TURN_INCOMPLETE_SHORT_MARKER,
|
||||
UserTurnCompletionLLMServiceMixin,
|
||||
)
|
||||
|
||||
|
||||
class MockProcessor(UserTurnCompletionLLMServiceMixin, FrameProcessor):
|
||||
"""Simple mock processor using the turn completion mixin."""
|
||||
|
||||
pass
|
||||
|
||||
|
||||
class TestUserUserTurnCompletionLLMServiceMixin(unittest.IsolatedAsyncioTestCase):
|
||||
"""Tests for UserUserTurnCompletionLLMServiceMixin functionality."""
|
||||
|
||||
async def test_complete_marker_pushes_text(self):
|
||||
"""Test that ✓ marker is detected and text after it is pushed normally."""
|
||||
processor = MockProcessor()
|
||||
|
||||
# Capture frames that get pushed
|
||||
pushed_frames = []
|
||||
processor.push_frame = AsyncMock(
|
||||
side_effect=lambda f, *args, **kwargs: pushed_frames.append(f)
|
||||
)
|
||||
|
||||
# Simulate LLM generating: "✓ Hello there!"
|
||||
await processor._push_turn_text(f"{USER_TURN_COMPLETE_MARKER} Hello there!")
|
||||
|
||||
# Should have 2 text frames: marker (skip_tts) and content (normal)
|
||||
self.assertEqual(len(pushed_frames), 2)
|
||||
|
||||
# First frame should be the marker with skip_tts=True
|
||||
self.assertIsInstance(pushed_frames[0], LLMTextFrame)
|
||||
self.assertEqual(pushed_frames[0].text, USER_TURN_COMPLETE_MARKER)
|
||||
self.assertTrue(pushed_frames[0].skip_tts)
|
||||
|
||||
# Second frame should be the actual text without skip_tts
|
||||
self.assertIsInstance(pushed_frames[1], LLMTextFrame)
|
||||
self.assertEqual(pushed_frames[1].text, "Hello there!")
|
||||
self.assertFalse(pushed_frames[1].skip_tts)
|
||||
|
||||
async def test_incomplete_short_marker_suppresses_text(self):
|
||||
"""Test that ○ marker suppresses text with skip_tts."""
|
||||
processor = MockProcessor()
|
||||
|
||||
pushed_frames = []
|
||||
processor.push_frame = AsyncMock(
|
||||
side_effect=lambda f, *args, **kwargs: pushed_frames.append(f)
|
||||
)
|
||||
# Mock timeout to avoid needing task manager
|
||||
processor._start_incomplete_timeout = AsyncMock()
|
||||
|
||||
await processor._push_turn_text(USER_TURN_INCOMPLETE_SHORT_MARKER)
|
||||
|
||||
# Should have 1 text frame with skip_tts=True
|
||||
self.assertEqual(len(pushed_frames), 1)
|
||||
self.assertIsInstance(pushed_frames[0], LLMTextFrame)
|
||||
self.assertEqual(pushed_frames[0].text, USER_TURN_INCOMPLETE_SHORT_MARKER)
|
||||
self.assertTrue(pushed_frames[0].skip_tts)
|
||||
|
||||
async def test_incomplete_long_marker_suppresses_text(self):
|
||||
"""Test that ◐ marker suppresses text with skip_tts."""
|
||||
processor = MockProcessor()
|
||||
|
||||
pushed_frames = []
|
||||
processor.push_frame = AsyncMock(
|
||||
side_effect=lambda f, *args, **kwargs: pushed_frames.append(f)
|
||||
)
|
||||
# Mock timeout to avoid needing task manager
|
||||
processor._start_incomplete_timeout = AsyncMock()
|
||||
|
||||
await processor._push_turn_text(USER_TURN_INCOMPLETE_LONG_MARKER)
|
||||
|
||||
# Should have 1 text frame with skip_tts=True
|
||||
self.assertEqual(len(pushed_frames), 1)
|
||||
self.assertIsInstance(pushed_frames[0], LLMTextFrame)
|
||||
self.assertEqual(pushed_frames[0].text, USER_TURN_INCOMPLETE_LONG_MARKER)
|
||||
self.assertTrue(pushed_frames[0].skip_tts)
|
||||
|
||||
async def test_text_buffered_until_marker_found(self):
|
||||
"""Test that text is buffered until a marker is detected."""
|
||||
processor = MockProcessor()
|
||||
|
||||
pushed_frames = []
|
||||
processor.push_frame = AsyncMock(
|
||||
side_effect=lambda f, *args, **kwargs: pushed_frames.append(f)
|
||||
)
|
||||
|
||||
# Simulate token-by-token streaming without marker
|
||||
await processor._push_turn_text("Hello")
|
||||
await processor._push_turn_text(" there")
|
||||
|
||||
# No frames should be pushed yet (buffering)
|
||||
self.assertEqual(len(pushed_frames), 0)
|
||||
|
||||
# Now send the complete marker
|
||||
await processor._push_turn_text(f" {USER_TURN_COMPLETE_MARKER} How are you?")
|
||||
|
||||
# Now frames should be pushed
|
||||
self.assertEqual(len(pushed_frames), 2)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user