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

Tighten the pipeline_task contract for processors and tools

Browse Source 
`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`.
This commit is contained in:
Aleix Conchillo Flaqué
2026-05-13 19:15:33 -07:00
parent 7d28c46a5d
commit ef806163b2
5 changed files with 38 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
src/pipecat/processors/frame_processor.py
View File

@@ -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):

15
src/pipecat/services/llm_service.py
View File

@@ -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:

38
tests/test_app_resources.py
View File

@@ -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__":

9
tests/test_audio_buffer_processor.py
View File

@@ -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

3
tests/test_llm_service.py
View File

@@ -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):