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

Merge pull request #2724 from pipecat-ai/pk/update-natural-conversation-examples-with-universal-context

Browse Source 
Update natural conversation examples with universal context
This commit is contained in:
Aleix Conchillo Flaqué
2025-09-24 11:07:50 -07:00
committed by GitHub
parent f899c15b03 817c77f3fe
commit 6c2c43ade0
4 changed files with 105 additions and 112 deletions
Download Patch File Download Diff File Expand all files Collapse all files

71
examples/foundational/22b-natural-conversation-proposal.py
View File

@@ -9,8 +9,9 @@ import os
from dotenv import load_dotenv
from loguru import logger
from openai.types.chat import ChatCompletionToolParam
from pipecat.adapters.schemas.function_schema import FunctionSchema
from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.frames.frames import (
CancelFrame,
@@ -19,6 +20,7 @@ from pipecat.frames.frames import (
FunctionCallInProgressFrame,
FunctionCallResultFrame,
InterruptionFrame,
LLMContextFrame,
LLMRunFrame,
StartFrame,
SystemFrame,
@@ -32,10 +34,8 @@ 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.openai_llm_context import (
OpenAILLMContext,
OpenAILLMContextFrame,
)
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
from pipecat.processors.filters.function_filter import FunctionFilter
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.processors.user_idle_processor import UserIdleProcessor
@@ -66,13 +66,13 @@ class StatementJudgeContextFilter(FrameProcessor):
await self.push_frame(frame, direction)
return
# We only want to handle OpenAILLMContextFrames, and only want to push through a simplified
# 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, OpenAILLMContextFrame):
if isinstance(frame, LLMContextFrame):
logger.debug(f"Context Frame: {frame}")
# Take text content from the most recent user messages.
messages = frame.context.messages
messages = frame.context.get_messages()
user_text_messages = []
last_assistant_message = None
for message in reversed(messages):
@@ -100,7 +100,7 @@ class StatementJudgeContextFilter(FrameProcessor):
if last_assistant_message:
messages.append(last_assistant_message)
messages.append({"role": "user", "content": user_message})
await self.push_frame(OpenAILLMContextFrame(OpenAILLMContext(messages)))
await self.push_frame(LLMContextFrame(LLMContext(messages)))
class CompletenessCheck(FrameProcessor):
@@ -231,22 +231,26 @@ class TurnDetectionLLM(Pipeline):
async def pass_only_llm_trigger_frames(frame):
return (
isinstance(frame, OpenAILLMContextFrame)
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 OpenAILLMContextFrame. Pass a specially constructed
# 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.
@@ -302,30 +306,23 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
async def on_function_calls_started(service, function_calls):
await tts.queue_frame(TTSSpeakFrame("Let me check on that."))
tools = [
ChatCompletionToolParam(
type="function",
function={
"name": "get_current_weather",
"description": "Get the current weather",
"parameters": {
"type": "object",
"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"],
},
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 = [
{
@@ -334,8 +331,8 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
},
]
context = OpenAILLMContext(messages, tools)
context_aggregator = llm_main.create_context_aggregator(context)
context = LLMContext(messages, tools)
context_aggregator = LLMContextAggregatorPair(context)
# LLM + turn detection (with an extra LLM as a judge)
llm = TurnDetectionLLM(llm_main)
@@ -369,7 +366,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
await task.queue_frames([LLMRunFrame()])
@transport.event_handler("on_app_message")
async def on_app_message(transport, message):
async def on_app_message(transport, message, sender):
logger.debug(f"Received app message: {message}")
if "message" not in message:
return

71
examples/foundational/22c-natural-conversation-mixed-llms.py
View File

@@ -9,8 +9,9 @@ import os
from dotenv import load_dotenv
from loguru import logger
from openai.types.chat import ChatCompletionToolParam
from pipecat.adapters.schemas.function_schema import FunctionSchema
from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.frames.frames import (
CancelFrame,
@@ -19,6 +20,7 @@ from pipecat.frames.frames import (
FunctionCallInProgressFrame,
FunctionCallResultFrame,
InterruptionFrame,
LLMContextFrame,
LLMRunFrame,
StartFrame,
SystemFrame,
@@ -32,10 +34,8 @@ 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.openai_llm_context import (
OpenAILLMContext,
OpenAILLMContextFrame,
)
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
from pipecat.processors.filters.function_filter import FunctionFilter
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.processors.user_idle_processor import UserIdleProcessor
@@ -272,11 +272,11 @@ class StatementJudgeContextFilter(FrameProcessor):
await self.push_frame(frame, direction)
return
# We only want to handle OpenAILLMContextFrames, and only want to push through a simplified
# 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, OpenAILLMContextFrame):
if isinstance(frame, LLMContextFrame):
# Take text content from the most recent user messages.
messages = frame.context.messages
messages = frame.context.get_messages()
user_text_messages = []
last_assistant_message = None
for message in reversed(messages):
@@ -303,7 +303,7 @@ class StatementJudgeContextFilter(FrameProcessor):
if last_assistant_message:
messages.append(last_assistant_message)
messages.append({"role": "user", "content": user_message})
await self.push_frame(OpenAILLMContextFrame(OpenAILLMContext(messages)))
await self.push_frame(LLMContextFrame(LLMContext(messages)))
class CompletenessCheck(FrameProcessor):
@@ -425,12 +425,15 @@ class TurnDetectionLLM(Pipeline):
async def pass_only_llm_trigger_frames(frame):
return (
isinstance(frame, OpenAILLMContextFrame)
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(
@@ -440,12 +443,13 @@ class TurnDetectionLLM(Pipeline):
FunctionFilter(filter=block_user_stopped_speaking),
],
[
# Ignore everything except an OpenAILLMContextFrame. Pass a specially constructed
# 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.
@@ -505,30 +509,23 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
async def on_function_calls_started(service, function_calls):
await tts.queue_frame(TTSSpeakFrame("Let me check on that."))
tools = [
ChatCompletionToolParam(
type="function",
function={
"name": "get_current_weather",
"description": "Get the current weather",
"parameters": {
"type": "object",
"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"],
},
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 = [
{
@@ -537,8 +534,8 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
},
]
context = OpenAILLMContext(messages, tools)
context_aggregator = llm_main.create_context_aggregator(context)
context = LLMContext(messages, tools)
context_aggregator = LLMContextAggregatorPair(context)
# LLM + turn detection (with an extra LLM as a judge)
llm = TurnDetectionLLM(llm_main)
@@ -577,7 +574,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
await task.queue_frames([LLMRunFrame()])
@transport.event_handler("on_app_message")
async def on_app_message(transport, message):
async def on_app_message(transport, message, sender):
logger.debug(f"Received app message: {message}")
if "message" not in message:
return

63
examples/foundational/22d-natural-conversation-gemini-audio.py
View File

@@ -9,7 +9,6 @@ import os
import time
from dotenv import load_dotenv
from google.genai.types import Content, Part
from loguru import logger
from pipecat.audio.vad.silero import SileroVADAnalyzer
@@ -21,6 +20,7 @@ from pipecat.frames.frames import (
FunctionCallResultFrame,
InputAudioRawFrame,
InterruptionFrame,
LLMContextFrame,
LLMFullResponseStartFrame,
LLMRunFrame,
StartFrame,
@@ -34,20 +34,18 @@ 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.openai_llm_context import (
OpenAILLMContext,
OpenAILLMContextFrame,
)
from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
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 GoogleLLMContext, GoogleLLMService
from pipecat.services.google.llm import GoogleLLMService
from pipecat.services.llm_service import LLMService
from pipecat.sync.base_notifier import BaseNotifier
from pipecat.sync.event_notifier import EventNotifier
@@ -375,7 +373,7 @@ class AudioAccumulator(FrameProcessor):
await super().process_frame(frame, direction)
# ignore context frame
if isinstance(frame, OpenAILLMContextFrame):
if isinstance(frame, LLMContextFrame):
return
if isinstance(frame, TranscriptionFrame):
@@ -392,9 +390,9 @@ class AudioAccumulator(FrameProcessor):
f"Processing audio buffer seconds: ({len(self._audio_frames)}) ({len(data)}) {len(data) / 2 / 16000}"
)
self._user_speaking = False
context = GoogleLLMContext()
context = LLMContext()
context.add_audio_frames_message(audio_frames=self._audio_frames)
await self.push_frame(OpenAILLMContextFrame(context=context))
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.
@@ -513,7 +511,7 @@ class LLMAggregatorBuffer(LLMAssistantResponseAggregator):
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: OpenAILLMContext, **kwargs):
def __init__(self, context: LLMContext, **kwargs):
super().__init__(**kwargs)
self._context = context
@@ -525,11 +523,10 @@ class ConversationAudioContextAssembler(FrameProcessor):
await self.push_frame(frame, direction)
return
if isinstance(frame, OpenAILLMContextFrame):
GoogleLLMContext.upgrade_to_google(self._context)
last_message = frame.context.messages[-1]
if isinstance(frame, LLMContextFrame):
last_message = frame.context.get_messages()[-1]
self._context._messages.append(last_message)
await self.push_frame(OpenAILLMContextFrame(context=self._context))
await self.push_frame(LLMContextFrame(context=self._context))
class OutputGate(FrameProcessor):
@@ -543,7 +540,7 @@ class OutputGate(FrameProcessor):
def __init__(
self,
notifier: BaseNotifier,
context: OpenAILLMContext,
context: LLMContext,
llm_transcription_buffer: LLMAggregatorBuffer,
**kwargs,
):
@@ -610,19 +607,23 @@ class OutputGate(FrameProcessor):
self._gate_task = None
async def _gate_task_handler(self):
await self._notifier.wait()
while True:
try:
await self._notifier.wait()
transcription = await self._transcription_buffer.wait_for_transcription() or "-"
self._context.add_message(Content(role="user", parts=[Part(text=transcription)]))
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 = []
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: OpenAILLMContext):
def __init__(self, llm: LLMService, context: LLMContext):
# This is the LLM that will transcribe user speech.
tx_llm = GoogleLLMService(
name="Transcriber",
@@ -648,10 +649,10 @@ class TurnDetectionLLM(Pipeline):
# as complete or incomplete.
# statement_judge_context_filter = StatementJudgeAudioContextAccumulator(notifier=notifier)
audio_accumulater = AudioAccumulator()
audio_accumulator = AudioAccumulator()
# This sends a UserStoppedSpeakingFrame and triggers the notifier event
completeness_check = CompletenessCheck(
notifier=notifier, audio_accumulator=audio_accumulater
notifier=notifier, audio_accumulator=audio_accumulator
)
async def block_user_stopped_speaking(frame):
@@ -667,7 +668,7 @@ class TurnDetectionLLM(Pipeline):
super().__init__(
[
audio_accumulater,
audio_accumulator,
ParallelPipeline(
[
# Pass everything except UserStoppedSpeaking to the elements after
@@ -734,8 +735,8 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
system_instruction=conversation_system_instruction,
)
context = OpenAILLMContext()
context_aggregator = conversation_llm.create_context_aggregator(context)
context = LLMContext()
context_aggregator = LLMContextAggregatorPair(context)
llm = TurnDetectionLLM(conversation_llm, context)
@@ -761,12 +762,10 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
@transport.event_handler("on_client_connected")
async def on_client_connected(transport, client):
logger.info(f"Client connected")
# Kick off the conversation.
await task.queue_frames([LLMRunFrame()])
@transport.event_handler("on_app_message")
async def on_app_message(transport, message):
logger.debug(f"Received app message: {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

12
src/pipecat/transports/smallwebrtc/transport.py
View File

@@ -66,7 +66,7 @@ class SmallWebRTCCallbacks(BaseModel):
on_client_disconnected: Called when a client disconnects.
"""
on_app_message: Callable[[Any], Awaitable[None]]
on_app_message: Callable[[Any, str], Awaitable[None]]
on_client_connected: Callable[[SmallWebRTCConnection], Awaitable[None]]
on_client_disconnected: Callable[[SmallWebRTCConnection], Awaitable[None]]
@@ -254,7 +254,7 @@ class SmallWebRTCClient:
@self._webrtc_connection.event_handler("app-message")
async def on_app_message(connection: SmallWebRTCConnection, message: Any):
await self._handle_app_message(message)
await self._handle_app_message(message, connection.pc_id)
def _convert_frame(self, frame_array: np.ndarray, format_name: str) -> np.ndarray:
"""Convert a video frame to RGB format based on the input format.
@@ -512,9 +512,9 @@ class SmallWebRTCClient:
if not self._closing:
await self._callbacks.on_client_disconnected(self._webrtc_connection)
async def _handle_app_message(self, message: Any):
async def _handle_app_message(self, message: Any, sender: str):
"""Handle incoming application messages."""
await self._callbacks.on_app_message(message)
await self._callbacks.on_app_message(message, sender)
def _can_send(self):
"""Check if the connection is ready for sending data."""
@@ -935,11 +935,11 @@ class SmallWebRTCTransport(BaseTransport):
if self._output:
await self._output.queue_frame(frame, FrameDirection.DOWNSTREAM)
async def _on_app_message(self, message: Any):
async def _on_app_message(self, message: Any, sender: str):
"""Handle incoming application messages."""
if self._input:
await self._input.push_app_message(message)
await self._call_event_handler("on_app_message", message)
await self._call_event_handler("on_app_message", message, sender)
async def _on_client_connected(self, webrtc_connection):
"""Handle client connection events."""