Save changes

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):
-        super().__init__()
-        self._llm1 = llm1
-        self._llm2 = llm2
-        self._race_counter = 0
-        self._active_races = {}  # race_id -> winner_name
-
-    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 = []
+        self._current_llm_name = None
 
-                        async def process_frame(self, frame, direction):
+    def set_llm_name(self, name: str):
-                            self.collected_frames.append(frame)
+        """Set the name of the LLM this processor instance is handling."""
+        self._current_llm_name = name
 
-                    collector = FrameCollector()
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
-
+        if isinstance(frame, LLMTextFrame):
-                    # Temporarily link LLM to collector
+            if not LLMRaceProcessor._response_started:
-                    llm.link(collector)
+                # First response wins the race
-
+                LLMRaceProcessor._winning_llm_name = self._current_llm_name
-                    # Process the frame
+                LLMRaceProcessor._response_started = True
-                    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!"
+                    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:
-                        logger.info(f"[LLM_RACE {race_id}] {name} lost")
+            # Always pass through non-LLM frames
-
-                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)
-
-        else:
-            # 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
+    # Make sure both LLMs share the same context - they should both process context frames
-    context_aggregator2 = llm2.create_context_aggregator(context)
+    # In a ParallelPipeline, the context frames will be duplicated to both branches
 
-    # Create race processor with both LLMs
+    # Create separate race processors for each LLM to track which one responds first
-    race_processor = LLMRaceProcessor(llm1, llm2)
+    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):