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

Merge pull request #4395 from pipecat-ai/pk/app-resources-api-updates

Browse Source 
Broaden tool_resources to app_resources
This commit is contained in:
kompfner
2026-04-30 21:19:05 -04:00
committed by GitHub
parent a27d9fc30b c3ef199efa
commit 6d66bbceeb
7 changed files with 554 additions and 191 deletions
Download Patch File Download Diff File Expand all files Collapse all files

56
src/pipecat/pipeline/task.py
View File

@@ -14,6 +14,7 @@ including heartbeats, idle detection, and observer integration.
import asyncio
import importlib.util
import os
import warnings
from collections.abc import AsyncIterable, Iterable
from pathlib import Path
from typing import Any, TypeVar
@@ -193,6 +194,7 @@ class PipelineTask(BasePipelineTask):
*,
params: PipelineParams | None = None,
additional_span_attributes: dict | None = None,
app_resources: Any = None,
cancel_on_idle_timeout: bool = True,
cancel_timeout_secs: float = CANCEL_TIMEOUT_SECS,
check_dangling_tasks: bool = True,
@@ -216,6 +218,14 @@ class PipelineTask(BasePipelineTask):
params: Configuration parameters for the pipeline.
additional_span_attributes: Optional dictionary of attributes to propagate as
OpenTelemetry conversation span attributes.
app_resources: Optional application-defined bag of anything your
application code may want to share across this session (DB
handles, HTTP clients, etc.), passed by reference. Pipecat
passes it through untouched and exposes it on the task itself
as ``task.app_resources`` and passes it to tool handlers as
``FunctionCallParams.app_resources``. The framework never
copies or clears this object; the caller retains their handle
and can read any mutations after the task finishes.
cancel_on_idle_timeout: Whether the pipeline task should be cancelled if
the idle timeout is reached.
cancel_timeout_secs: Timeout (in seconds) to wait for cancellation to happen
@@ -235,13 +245,24 @@ class PipelineTask(BasePipelineTask):
rtvi_observer_params: The RTVI observer parameter to use if RTVI is enabled.
rtvi_processor: The RTVI processor to add if RTVI is enabled.
task_manager: Optional task manager for handling asyncio tasks.
tool_resources: Optional application-defined bag of resources (DB handles,
clients, state, etc.) passed by reference to every tool handler via
``FunctionCallParams.tool_resources``. The framework never copies or
clears this object; the caller retains their handle and can read any
mutations after the task finishes.
tool_resources: Deprecated alias for ``app_resources``.
.. deprecated:: 1.2.0
Use ``app_resources`` instead. ``tool_resources`` will be
removed in a future version.
"""
super().__init__()
if tool_resources is not None:
with warnings.catch_warnings():
warnings.simplefilter("always")
warnings.warn(
"`PipelineTask(tool_resources=...)` is deprecated since 1.2.0, "
"use `app_resources` instead.",
DeprecationWarning,
stacklevel=2,
)
if app_resources is None:
app_resources = tool_resources
self._params = params or PipelineParams()
self._additional_span_attributes = additional_span_attributes or {}
self._cancel_on_idle_timeout = cancel_on_idle_timeout
@@ -252,7 +273,7 @@ class PipelineTask(BasePipelineTask):
self._enable_tracing = enable_tracing and is_tracing_available()
self._enable_turn_tracking = enable_turn_tracking
self._idle_timeout_secs = idle_timeout_secs
self._tool_resources = tool_resources
self._app_resources = app_resources
observers = observers or []
self._turn_tracking_observer: TurnTrackingObserver | None = None
self._user_bot_latency_observer: UserBotLatencyObserver | None = None
@@ -391,6 +412,21 @@ class PipelineTask(BasePipelineTask):
"""
return self._params
@property
def app_resources(self) -> Any:
"""Get the application-defined resources passed to this task.
This is the same object passed to the constructor as
``app_resources``. Tool handlers can also access it via
``FunctionCallParams.app_resources``. The framework returns the
original reference; mutations are visible to all callers.
Returns:
The application-defined resources, or ``None`` if none were
passed.
"""
return self._app_resources
@property
def pipeline(self) -> BasePipeline:
"""Get the full pipeline managed by this pipeline task.
@@ -730,7 +766,13 @@ class PipelineTask(BasePipelineTask):
clock=self._clock,
task_manager=self._task_manager,
observer=self._observer,
tool_resources=self._tool_resources,
pipeline_task=self,
# Populate the deprecated `tool_resources` field for backwards
# compatibility with custom FrameProcessor subclasses whose
# ``setup()`` overrides still read it. Reading the field emits a
# DeprecationWarning; new code should read
# ``setup.pipeline_task.app_resources`` instead.
tool_resources=self._app_resources,
)
await self._pipeline.setup(setup)

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

@@ -16,10 +16,12 @@ from __future__ import annotations
import asyncio
import dataclasses
import traceback
import warnings
from collections.abc import Awaitable, Callable, Coroutine
from dataclasses import dataclass
from enum import Enum
from typing import (
TYPE_CHECKING,
Any,
Optional,
)
@@ -47,6 +49,9 @@ from pipecat.utils.asyncio.task_manager import BaseTaskManager
from pipecat.utils.base_object import BaseObject
from pipecat.utils.frame_queue import FrameQueue
if TYPE_CHECKING:
from pipecat.pipeline.task import PipelineTask
class FrameDirection(Enum):
"""Direction of frame flow in the processing pipeline.
@@ -71,15 +76,45 @@ class FrameProcessorSetup:
clock: The clock instance for timing operations.
task_manager: The task manager for handling async operations.
observer: Optional observer for monitoring frame processing events.
tool_resources: Application-defined resources shared with processors
for this pipeline run.
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``).
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``
keep working. New code should read
``setup.pipeline_task.app_resources`` instead.
.. deprecated:: 1.2.0
Reading this attribute emits a ``DeprecationWarning``. Read
``setup.pipeline_task.app_resources`` instead.
``tool_resources`` will be removed in a future version.
"""
clock: BaseClock
task_manager: BaseTaskManager
observer: BaseObserver | None = None
pipeline_task: PipelineTask | None = None
tool_resources: Any = None
def __getattribute__(self, name: str) -> Any:
# Warn when user code reads the deprecated ``tool_resources`` field.
# Set is unaffected (goes through ``__setattr__``), so PipelineTask can
# populate it for backwards compat without tripping the warning.
if name == "tool_resources":
value = object.__getattribute__(self, "tool_resources")
if value is not None:
with warnings.catch_warnings():
warnings.simplefilter("always")
warnings.warn(
"`FrameProcessorSetup.tool_resources` is deprecated since 1.2.0; "
"read `setup.pipeline_task.app_resources` instead.",
DeprecationWarning,
stacklevel=2,
)
return value
return object.__getattribute__(self, name)
class FrameProcessorQueue(asyncio.PriorityQueue):
"""A priority queue for systems frames and other frames.
@@ -188,6 +223,9 @@ class FrameProcessor(BaseObject):
# Observer
self._observer: BaseObserver | None = None
# Pipeline Task
self._pipeline_task: PipelineTask | None = None
# Other properties
self._enable_metrics = False
self._enable_usage_metrics = False
@@ -344,6 +382,22 @@ class FrameProcessor(BaseObject):
raise Exception(f"{self} TaskManager is still not initialized.")
return self._task_manager
@property
def pipeline_task(self) -> PipelineTask | None:
"""Get the :class:`PipelineTask` this processor is running in.
Provides access to task-scoped state from inside a processor — most
notably ``self.pipeline_task.app_resources`` for the application's
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).
"""
return self._pipeline_task
def processors_with_metrics(self):
"""Return processors that can generate metrics.
@@ -495,6 +549,7 @@ class FrameProcessor(BaseObject):
self._clock = setup.clock
self._task_manager = setup.task_manager
self._observer = setup.observer
self._pipeline_task = setup.pipeline_task
# Create processing tasks.
self.__create_input_task()

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

@@ -51,7 +51,7 @@ from pipecat.processors.aggregators.llm_context import (
LLMContext,
LLMSpecificMessage,
)
from pipecat.processors.frame_processor import FrameDirection, FrameProcessorSetup
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.ai_service import AIService
from pipecat.services.settings import LLMSettings, assert_given
from pipecat.services.websocket_service import WebsocketService
@@ -107,9 +107,10 @@ class FunctionCallParams:
For async function calls (``cancel_on_interruption=False``), call
it with ``properties=FunctionCallResultProperties(is_final=False)``
to push intermediate updates before the final result.
tool_resources: Application-defined bag of resources (DB handles, clients,
state, etc.) shared across tool calls for the pipeline session. Set
via ``PipelineTask(..., tool_resources=...)`` and passed by reference.
app_resources: The application-defined resources passed to
``PipelineTask(..., app_resources=...)``. Same object — passed by
reference, not a copy. Use it to share DB handles, clients, state,
feature flags, etc. across all of a session's tool handlers.
"""
function_name: str
@@ -118,7 +119,25 @@ class FunctionCallParams:
llm: LLMService
context: LLMContext
result_callback: FunctionCallResultCallback
tool_resources: Any = None
app_resources: Any = None
@property
def tool_resources(self) -> Any:
"""Deprecated alias for :attr:`app_resources`.
.. deprecated:: 1.2.0
Use :attr:`app_resources` instead. ``tool_resources`` will be
removed in a future version.
"""
with warnings.catch_warnings():
warnings.simplefilter("always")
warnings.warn(
"`FunctionCallParams.tool_resources` is deprecated since 1.2.0, "
"use `app_resources` instead.",
DeprecationWarning,
stacklevel=2,
)
return self.app_resources
@dataclass
@@ -256,7 +275,6 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService):
self._sequential_runner_task: asyncio.Task | None = None
self._skip_tts: bool | None = None
self._summary_task: asyncio.Task | None = None
self._tool_resources: Any = None
self._register_event_handler("on_function_calls_started")
self._register_event_handler("on_function_calls_cancelled")
@@ -303,15 +321,6 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService):
"""
raise NotImplementedError(f"run_inference() not supported by {self.__class__.__name__}")
async def setup(self, setup: FrameProcessorSetup):
"""Set up the LLM service.
Args:
setup: The frame processor setup data.
"""
await super().setup(setup)
self._tool_resources = setup.tool_resources
async def start(self, frame: StartFrame):
"""Start the LLM service.
@@ -882,6 +891,9 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService):
# 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
@@ -894,7 +906,7 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService):
llm=self,
context=runner_item.context,
result_callback=function_call_result_callback,
tool_resources=self._tool_resources,
app_resources=app_resources,
),
)
else:
@@ -906,7 +918,7 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService):
llm=self,
context=runner_item.context,
result_callback=function_call_result_callback,
tool_resources=self._tool_resources,
app_resources=app_resources,
)
await item.handler(params)
except Exception as e: