scripts(evals): improve eval assertion on exit

examples/foundational/26a-gemini-live-transcription.py

@@ -17,7 +17,6 @@ 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
 from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
 from pipecat.processors.transcript_processor import TranscriptProcessor
 from pipecat.runner.types import RunnerArguments

examples/foundational/26b-gemini-live-function-calling.py

@@ -20,7 +20,6 @@ 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
 from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
 from pipecat.runner.types import RunnerArguments
 from pipecat.runner.utils import create_transport

examples/foundational/26c-gemini-live-video.py

@@ -18,7 +18,6 @@ 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
 from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
 from pipecat.runner.types import RunnerArguments
 from pipecat.runner.utils import (

scripts/evals/eval.py

@@ -31,7 +31,13 @@ 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.audio.vad.vad_analyzer import VADParams
-from pipecat.frames.frames import EndTaskFrame, LLMRunFrame, OutputImageRawFrame
+from pipecat.frames.frames import (
+    CancelFrame,
+    EndFrame,
+    EndTaskFrame,
+    LLMRunFrame,
+    OutputImageRawFrame,
+)
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
@@ -50,6 +56,7 @@ SCRIPT_DIR = Path(__file__).resolve().parent
 
 PIPELINE_IDLE_TIMEOUT_SECS = 60
 EVAL_TIMEOUT_SECS = 120
+EVAL_RESULT_TIMEOUT_SECS = 10
 
 EvalPrompt = str | Tuple[str, ImageFile]
 
@@ -88,16 +95,15 @@ class EvalRunner:
         os.makedirs(self._logs_dir, exist_ok=True)
         os.makedirs(self._recordings_dir, exist_ok=True)
 
-    async def assert_eval(self, params: FunctionCallParams):
+    async def function_assert_eval(self, params: FunctionCallParams):
         result = params.arguments["result"]
         reasoning = params.arguments["reasoning"]
         logger.debug(f"🧠 EVAL REASONING(result: {result}): {reasoning}")
-        await self._queue.put(result)
         await params.result_callback(None)
-        await params.llm.push_frame(EndTaskFrame(), FrameDirection.UPSTREAM)
+        await params.llm.push_frame(EndTaskFrame(reason=result), FrameDirection.UPSTREAM)
 
-    async def assert_eval_false(self):
-        await self._queue.put(False)
+    async def assert_eval(self, result: bool):
+        await self._queue.put(result)
 
     async def run_eval(
         self,
@@ -136,8 +142,9 @@ class EvalRunner:
             logger.error(f"ERROR: Unable to run {example_file}: {e}")
 
         try:
-            result = await asyncio.wait_for(self._queue.get(), timeout=1.0)
+            result = await asyncio.wait_for(self._queue.get(), timeout=EVAL_RESULT_TIMEOUT_SECS)
         except asyncio.TimeoutError:
+            logger.error(f"ERROR: Timeout waiting for eval result.")
             result = False
 
         if result:
@@ -244,7 +251,7 @@ async def run_eval_pipeline(
 
     llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
 
-    llm.register_function("eval_function", eval_runner.assert_eval)
+    llm.register_function("eval_function", eval_runner.function_assert_eval)
 
     eval_function = FunctionSchema(
         name="eval_function",
@@ -348,7 +355,14 @@ async def run_eval_pipeline(
 
     @task.event_handler("on_idle_timeout")
     async def on_pipeline_idle_timeout(task):
-        await eval_runner.assert_eval_false()
+        await eval_runner.assert_eval(False)
+
+    @task.event_handler("on_pipeline_finished")
+    async def on_pipeline_finished(task, frame):
+        if isinstance(frame, EndFrame):
+            await eval_runner.assert_eval(frame.reason)
+        elif isinstance(frame, CancelFrame):
+            await eval_runner.assert_eval(False)
 
     # TODO(aleix): We should handle SIGINT and SIGTERM so we can cancel both the
     # eval and the example.