From ef806163b2b46b64465eb4b52674796072d25c6b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 13 May 2026 19:15:33 -0700 Subject: [PATCH] Tighten the pipeline_task contract for processors and tools `FrameProcessorSetup.pipeline_task` is now mandatory and `FrameProcessor.pipeline_task` raises if accessed before setup instead of returning `None`. `FunctionCallParams` gains a required `pipeline_task` field and `LLMService._run_function_call` populates it (plus reads `app_resources` directly off the pipeline task). Tests that build a processor or `FunctionCallParams` outside a real pipeline stub it with a `SimpleNamespace`. --- src/pipecat/processors/frame_processor.py | 18 +++++------ src/pipecat/services/llm_service.py | 15 ++++++--- tests/test_app_resources.py | 38 +++++------------------ tests/test_audio_buffer_processor.py | 9 +++++- tests/test_llm_service.py | 3 ++ 5 files changed, 38 insertions(+), 45 deletions(-) diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index 527b19dd0..06c8e46f9 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -75,10 +75,10 @@ class FrameProcessorSetup: Parameters: clock: The clock instance for timing operations. task_manager: The task manager for handling async operations. - observer: Optional observer for monitoring frame processing events. pipeline_task: The :class:`PipelineTask` running this pipeline. Stored on each processor as ``self.pipeline_task`` so processors can reach task-scoped state (e.g. ``self.pipeline_task.app_resources``). + observer: Optional observer for monitoring frame processing events. tool_resources: Deprecated. :class:`PipelineTask` continues to populate this with ``app_resources`` so that custom :class:`FrameProcessor` subclasses whose ``setup()`` overrides read ``setup.tool_resources`` @@ -93,8 +93,8 @@ class FrameProcessorSetup: clock: BaseClock task_manager: BaseTaskManager + pipeline_task: PipelineTask observer: BaseObserver | None = None - pipeline_task: PipelineTask | None = None tool_resources: Any = None def __getattribute__(self, name: str) -> Any: @@ -220,8 +220,9 @@ class FrameProcessor(BaseObject): # Observer self._observer: BaseObserver | None = None - # Pipeline Task - self._pipeline_task: PipelineTask | None = None + # Pipeline Task. Populated by ``setup()``; accessing the + # ``pipeline_task`` property before setup raises. + self._pipeline_task: PipelineTask | None = None # set in setup() # Other properties self._enable_metrics = False @@ -366,7 +367,7 @@ class FrameProcessor(BaseObject): return self._report_only_initial_ttfb @property - def pipeline_task(self) -> PipelineTask | None: + def pipeline_task(self) -> PipelineTask: """Get the :class:`PipelineTask` this processor is running in. Provides access to task-scoped state from inside a processor — most @@ -374,11 +375,10 @@ class FrameProcessor(BaseObject): shared bag of resources (DB handles, clients, feature flags, etc.). Returns: - The :class:`PipelineTask` instance that set up this processor, - or ``None`` if the processor has not yet been set up by one - (for example, before the task has started, or when the processor - was instantiated in isolation). + The :class:`PipelineTask` instance that set up this processor. """ + if not self._pipeline_task: + raise Exception(f"{self} pipeline task is still not set.") return self._pipeline_task def processors_with_metrics(self): diff --git a/src/pipecat/services/llm_service.py b/src/pipecat/services/llm_service.py index 58ed54bd0..28070f33b 100644 --- a/src/pipecat/services/llm_service.py +++ b/src/pipecat/services/llm_service.py @@ -15,6 +15,7 @@ import warnings from collections.abc import Awaitable, Callable, Mapping, Sequence from dataclasses import dataclass from typing import ( + TYPE_CHECKING, Any, Generic, Protocol, @@ -70,6 +71,10 @@ from pipecat.utils.context.llm_context_summarization import ( LLMContextSummarizationUtil, ) +if TYPE_CHECKING: + from pipecat.pipeline.task import PipelineTask + + # Type alias for a callable that handles LLM function calls. FunctionCallHandler = Callable[["FunctionCallParams"], Awaitable[None]] @@ -126,6 +131,7 @@ class FunctionCallParams: # treat it invariantly, rejecting `LLMService[XAdapter]` at the call # sites that build FunctionCallParams. llm: LLMService[Any] + pipeline_task: PipelineTask context: LLMContext result_callback: FunctionCallResultCallback app_resources: Any = None @@ -947,9 +953,6 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService, Generic[TAdapter] # it starts would leave the coroutine in a "never awaited" state. await asyncio.sleep(0) - # _pipeline_task may be unset when the service is driven without a PipelineTask. - app_resources = self._pipeline_task.app_resources if self._pipeline_task else None - try: if isinstance(item.handler, DirectFunctionWrapper): # Handler is a DirectFunctionWrapper @@ -960,9 +963,10 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService, Generic[TAdapter] tool_call_id=runner_item.tool_call_id, arguments=runner_item.arguments, llm=self, + pipeline_task=self.pipeline_task, context=runner_item.context, result_callback=function_call_result_callback, - app_resources=app_resources, + app_resources=self.pipeline_task.app_resources, ), ) else: @@ -972,9 +976,10 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService, Generic[TAdapter] tool_call_id=runner_item.tool_call_id, arguments=runner_item.arguments, llm=self, + pipeline_task=self.pipeline_task, context=runner_item.context, result_callback=function_call_result_callback, - app_resources=app_resources, + app_resources=self.pipeline_task.app_resources, ) await item.handler(params) except Exception as e: diff --git a/tests/test_app_resources.py b/tests/test_app_resources.py index 646b8d397..7eb1676d2 100644 --- a/tests/test_app_resources.py +++ b/tests/test_app_resources.py @@ -14,9 +14,8 @@ from unittest.mock import AsyncMock from pipecat.adapters.schemas.direct_function import DirectFunctionWrapper from pipecat.clocks.system_clock import SystemClock from pipecat.frames.frames import EndFrame, Frame, StartFrame -from pipecat.pipeline.base_task import PipelineTaskParams from pipecat.pipeline.pipeline import Pipeline -from pipecat.pipeline.task import PipelineTask +from pipecat.pipeline.task import PipelineTask, PipelineTaskParams from pipecat.processors.aggregators.llm_context import LLMContext from pipecat.processors.frame_processor import FrameDirection, FrameProcessor, FrameProcessorSetup from pipecat.services.llm_service import ( @@ -65,6 +64,7 @@ class TestFunctionCallParamsAppResources(unittest.TestCase): tool_call_id="1", arguments={}, llm=None, # type: ignore[arg-type] + pipeline_task=None, # type: ignore[arg-type] context=LLMContext(), result_callback=AsyncMock(), ) @@ -77,6 +77,7 @@ class TestFunctionCallParamsAppResources(unittest.TestCase): tool_call_id="1", arguments={}, llm=None, # type: ignore[arg-type] + pipeline_task=None, # type: ignore[arg-type] context=LLMContext(), result_callback=AsyncMock(), app_resources=resources, @@ -90,6 +91,7 @@ class TestFunctionCallParamsAppResources(unittest.TestCase): tool_call_id="1", arguments={}, llm=None, # type: ignore[arg-type] + pipeline_task=None, # type: ignore[arg-type] context=LLMContext(), result_callback=AsyncMock(), app_resources=resources, @@ -160,32 +162,6 @@ class TestLLMServiceFunctionCallReadsAppResources(unittest.IsolatedAsyncioTestCa self.assertIs(captured["params"].app_resources, resources) - async def test_app_resources_none_when_pipeline_task_unset(self): - service = _MockLLMService() - captured: dict[str, Any] = {} - - async def handler(params: FunctionCallParams): - captured["params"] = params - await params.result_callback({"ok": True}) - - service._functions["lookup"] = FunctionCallRegistryItem( - function_name="lookup", - handler=handler, - cancel_on_interruption=True, - ) - service.broadcast_frame = AsyncMock() # type: ignore[method-assign] - - runner_item = FunctionCallRunnerItem( - registry_item=service._functions["lookup"], - function_name="lookup", - tool_call_id="call-1", - arguments={}, - context=LLMContext(), - ) - await service._run_function_call(runner_item) - - self.assertIsNone(captured["params"].app_resources) - async def test_frame_processor_setup_tool_resources_warns_on_read(self): # ``FrameProcessorSetup.tool_resources`` is retained for backwards # compatibility with custom FrameProcessors whose ``setup()`` overrides @@ -198,6 +174,7 @@ class TestLLMServiceFunctionCallReadsAppResources(unittest.IsolatedAsyncioTestCa setup = FrameProcessorSetup( clock=SystemClock(), task_manager=task_manager, + pipeline_task=SimpleNamespace(app_resources=None), # type: ignore[arg-type] tool_resources=resources, ) @@ -317,9 +294,10 @@ class TestFrameProcessorPipelineTaskAccess(unittest.IsolatedAsyncioTestCase): self.assertIs(recorder.observed_task, task) self.assertIs(recorder.observed_app_resources, resources) - def test_pipeline_task_returns_none_when_not_set_up(self): + def test_pipeline_task_raises_when_not_set_up(self): recorder = _RecordingProcessor() - self.assertIsNone(recorder.pipeline_task) + with self.assertRaises(Exception): + _ = recorder.pipeline_task if __name__ == "__main__": diff --git a/tests/test_audio_buffer_processor.py b/tests/test_audio_buffer_processor.py index 27403e6fc..5b5f9996d 100644 --- a/tests/test_audio_buffer_processor.py +++ b/tests/test_audio_buffer_processor.py @@ -7,6 +7,7 @@ import asyncio import struct import unittest +from types import SimpleNamespace from pipecat.clocks.system_clock import SystemClock from pipecat.frames.frames import ( @@ -44,7 +45,13 @@ async def _make_processor(*, buffer_size: int = 0) -> AudioBufferProcessor: loop = asyncio.get_event_loop() task_manager = TaskManager() task_manager.setup(TaskManagerParams(loop=loop)) - await processor.setup(FrameProcessorSetup(clock=SystemClock(), task_manager=task_manager)) + await processor.setup( + FrameProcessorSetup( + clock=SystemClock(), + task_manager=task_manager, + pipeline_task=SimpleNamespace(app_resources=None), # type: ignore[arg-type] + ) + ) await processor.process_frame( StartFrame(audio_out_sample_rate=16000), FrameDirection.DOWNSTREAM diff --git a/tests/test_llm_service.py b/tests/test_llm_service.py index 9476f42a1..f51c5e3c2 100644 --- a/tests/test_llm_service.py +++ b/tests/test_llm_service.py @@ -5,6 +5,7 @@ # import unittest +from types import SimpleNamespace from unittest.mock import AsyncMock, patch from pipecat.adapters.base_llm_adapter import BaseLLMAdapter @@ -45,6 +46,8 @@ class MockLLMService(LLMService): user_turn_completion_config=None, ) super().__init__(settings=settings, **kwargs) + # Stub the pipeline task so FunctionCallParams can be constructed. + self._pipeline_task = SimpleNamespace(app_resources=None) class TestUnparameterizedSubclass(unittest.TestCase):