From 6a1170760495da0a0ea97346aeee9d326d771142 Mon Sep 17 00:00:00 2001 From: James Hush Date: Fri, 5 Sep 2025 16:55:09 +0800 Subject: [PATCH] Save changes --- examples/foundational/45-llm-hedge.py | 144 ++++++++++++-------------- 1 file changed, 65 insertions(+), 79 deletions(-) diff --git a/examples/foundational/45-llm-hedge.py b/examples/foundational/45-llm-hedge.py index b292e64e7..fef587975 100644 --- a/examples/foundational/45-llm-hedge.py +++ b/examples/foundational/45-llm-hedge.py @@ -12,12 +12,17 @@ from loguru import logger from openai.types.chat import ChatCompletionMessageParam from pipecat.audio.vad.silero import SileroVADAnalyzer -from pipecat.frames.frames import Frame, LLMMessagesFrame +from pipecat.frames.frames import Frame, LLMMessagesFrame, LLMMessagesUpdateFrame, LLMTextFrame +from pipecat.observers.loggers.debug_log_observer import DebugLogObserver, FrameEndpoint +from pipecat.observers.loggers.llm_log_observer import LLMLogObserver 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 +from pipecat.processors.aggregators.openai_llm_context import ( + OpenAILLMContext, + OpenAILLMContextFrame, +) from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.runner.types import RunnerArguments from pipecat.runner.utils import create_transport @@ -53,83 +58,45 @@ transport_params = { class LLMRaceProcessor(FrameProcessor): - """Processor that sends frames to two LLMs in parallel and uses the first response.""" + """Manages racing between two LLMs - only allows frames from the first LLM to respond.""" - def __init__(self, llm1: OpenAILLMService, llm2: OpenAILLMService): + def __init__(self): super().__init__() - self._llm1 = llm1 - self._llm2 = llm2 - self._race_counter = 0 - self._active_races = {} # race_id -> winner_name + self._current_llm_name = None - async def process_frame(self, frame, direction): - await super().process_frame(frame, direction) - - # Check if this is a frame we should race through both LLMs - if isinstance(frame, LLMMessagesFrame): - race_id = self._race_counter - self._race_counter += 1 - - logger.info(f"[LLM_RACE {race_id}] Starting parallel processing") - - # Create a result collector for this race - race_result = asyncio.Event() - winning_frames = [] - - async def llm_runner(llm: OpenAILLMService, name: str): - """Run LLM and collect results.""" - try: - # Create a frame collector - class FrameCollector(FrameProcessor): - def __init__(self): - super().__init__() - self.collected_frames = [] - - async def process_frame(self, frame, direction): - self.collected_frames.append(frame) - - collector = FrameCollector() - - # Temporarily link LLM to collector - llm.link(collector) - - # Process the frame - await llm.process_frame(frame, FrameDirection.DOWNSTREAM) - - # Check if we won the race - if race_id not in self._active_races: - self._active_races[race_id] = name - winning_frames.extend(collector.collected_frames) - race_result.set() - logger.info( - f"[LLM_RACE {race_id}] {name} WON with {len(collector.collected_frames)} frames!" - ) - else: - logger.info(f"[LLM_RACE {race_id}] {name} lost") - - except Exception as e: - logger.error(f"[LLM_RACE {race_id}] Error in {name}: {e}") - - # Start both LLMs racing - task1 = asyncio.create_task(llm_runner(self._llm1, "LLM1")) - task2 = asyncio.create_task(llm_runner(self._llm2, "LLM2")) - - # Wait for the first one to complete - await race_result.wait() - - # Cancel the slower task - task1.cancel() - task2.cancel() - - # Push the winning frames - for winning_frame in winning_frames: - await self.push_frame(winning_frame, direction) + def set_llm_name(self, name: str): + """Set the name of the LLM this processor instance is handling.""" + self._current_llm_name = name + async def process_frame(self, frame: Frame, direction: FrameDirection): + if isinstance(frame, LLMTextFrame): + if not LLMRaceProcessor._response_started: + # First response wins the race + LLMRaceProcessor._winning_llm_name = self._current_llm_name + LLMRaceProcessor._response_started = True + logger.info( + f"🏆 [LLM_RACE] {self._current_llm_name} wins the race! Text: '{frame.text}'" + ) + await self.push_frame(frame, direction) + elif LLMRaceProcessor._winning_llm_name == self._current_llm_name: + # Continue allowing frames from winning LLM + logger.info(f"✅ [LLM_RACE] {self._current_llm_name} continuing: '{frame.text}'") + await self.push_frame(frame, direction) + else: + # Drop frames from losing LLM + logger.info( + f"❌ [LLM_RACE] Dropping '{frame.text}' from losing LLM: {self._current_llm_name}" + ) else: - # Pass through non-LLM frames + # Always pass through non-LLM frames await self.push_frame(frame, direction) +# Class variables to share state between instances +LLMRaceProcessor._winning_llm_name = None +LLMRaceProcessor._response_started = False + + async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): logger.info(f"Starting bot with parallel LLM racing") @@ -154,19 +121,35 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): context = OpenAILLMContext(messages) context_aggregator = llm1.create_context_aggregator(context) - # Create a second context aggregator that shares the same context - context_aggregator2 = llm2.create_context_aggregator(context) + # Make sure both LLMs share the same context - they should both process context frames + # In a ParallelPipeline, the context frames will be duplicated to both branches - # Create race processor with both LLMs - race_processor = LLMRaceProcessor(llm1, llm2) + # Create separate race processors for each LLM to track which one responds first + race_processor1 = LLMRaceProcessor() + race_processor1.set_llm_name("LLM1") - # Simple pipeline - the race processor handles the parallel LLM execution internally + race_processor2 = LLMRaceProcessor() + race_processor2.set_llm_name("LLM2") + + # Create parallel LLM branches using ParallelPipeline + parallel_llms = ParallelPipeline( + [llm1, race_processor1], # Branch 1: LLM1 -> race processor 1 + [llm2, race_processor2], # Branch 2: LLM2 -> race processor 2 + ) + + # Set up debug observers with filtering - only log LLM frames going to TTS + debug_observer = DebugLogObserver( + frame_types={LLMTextFrame: (CartesiaTTSService, FrameEndpoint.DESTINATION)} + ) + llm_observer = LLMLogObserver() + + # Simple pipeline with parallel LLM processing pipeline = Pipeline( [ transport.input(), # Transport user input stt, # Speech to text context_aggregator.user(), # User responses (creates context frames for LLMs) - race_processor, # Parallel LLM racing processor + parallel_llms, # Parallel LLM processing tts, # TTS transport.output(), # Transport bot output context_aggregator.assistant(), # Assistant spoken responses @@ -179,15 +162,18 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): enable_metrics=True, enable_usage_metrics=True, ), + # observers=[debug_observer, llm_observer], 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. + # Use a simpler approach - add message to context and push a context frame messages.append({"role": "system", "content": "Please introduce yourself to the user."}) - await task.queue_frames([context_aggregator.user().get_context_frame()]) + # Create a new context with the updated messages + updated_context = OpenAILLMContext(messages) + await task.queue_frames([OpenAILLMContextFrame(context=updated_context)]) @transport.event_handler("on_client_disconnected") async def on_client_disconnected(transport, client):