From cc637f4dea1b73f3020d3cc97bfa36e02469f683 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Tue, 1 Jul 2025 15:22:30 -0400 Subject: [PATCH] Clean up docstrings after DirectFunction merge (#2105) * Add missing import for FunctionCallParams * Update docstrings in direct_function * Docstring fixes for run.py * Remove unused imports in llm_service * Add missing docstrings to llm_service * Remove FunctionCallParams import * Wording improvements * Type checking for FunctionCallParams --- .../adapters/schemas/direct_function.py | 120 ++++++++++++++---- src/pipecat/examples/run.py | 29 +++++ src/pipecat/services/llm_service.py | 25 ++-- 3 files changed, 139 insertions(+), 35 deletions(-) diff --git a/src/pipecat/adapters/schemas/direct_function.py b/src/pipecat/adapters/schemas/direct_function.py index 54763c3d8..e300eff81 100644 --- a/src/pipecat/adapters/schemas/direct_function.py +++ b/src/pipecat/adapters/schemas/direct_function.py @@ -1,6 +1,22 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""Direct function wrapper utilities for LLM function calling. + +This module provides utilities for wrapping "direct" functions that handle LLM +function calls. Direct functions have their metadata automatically extracted +from function signatures and docstrings, allowing them to be used without +accompanying configurations (as FunctionSchemas or in provider-specific +formats). +""" + import inspect import types from typing import ( + TYPE_CHECKING, Any, Callable, Dict, @@ -19,6 +35,9 @@ import docstring_parser from pipecat.adapters.schemas.function_schema import FunctionSchema +if TYPE_CHECKING: + from pipecat.services.llm_service import FunctionCallParams + class DirectFunction(Protocol): """Protocol for a "direct" function that handles LLM function calls. @@ -28,30 +47,58 @@ class DirectFunction(Protocol): `FunctionSchema`s or in provider-specific formats). """ - async def __call__(self, params: "FunctionCallParams", **kwargs: Any) -> None: ... + async def __call__(self, params: "FunctionCallParams", **kwargs: Any) -> None: + """Execute the direct function. + + Args: + params: Function call parameters from the LLM service. + **kwargs: Additional keyword arguments passed to the function. + """ + ... class BaseDirectFunctionWrapper: - """ - Base class for a wrapper around a DirectFunction that: - - extracts metadata from the function signature and docstring - - using that metadata, generates a corresponding FunctionSchema - """ + """Base class for a wrapper around a DirectFunction. - @classmethod - def special_first_param_name(cls) -> str: - """The name of the "special" first function parameter that is ignored by the metadata - extraction, as it's not relevant to the LLM. - """ - raise NotImplementedError("Subclasses must define the special first parameter name.") + Provides functionality to: + + - extract metadata from the function signature and docstring + - use that metadata to generate a corresponding FunctionSchema + """ def __init__(self, function: Callable): + """Initialize the direct function wrapper. + + Args: + function: The function to wrap and extract metadata from. + """ self.__class__.validate_function(function) self.function = function self._initialize_metadata() + @classmethod + def special_first_param_name(cls) -> str: + """Get the name of the special first function parameter. + + The special first parameter is ignored by metadata extraction as it's + not relevant to the LLM (e.g., 'params' for FunctionCallParams). + + Returns: + The name of the special first parameter. + """ + raise NotImplementedError("Subclasses must define the special first parameter name.") + @classmethod def validate_function(cls, function: Callable) -> None: + """Validate that the function meets direct function requirements. + + Args: + function: The function to validate. + + Raises: + Exception: If function doesn't meet requirements (not async, missing + parameters, incorrect first parameter name). + """ if not inspect.iscoroutinefunction(function): raise Exception(f"Direct function {function.__name__} must be async") params = list(inspect.signature(function).parameters.items()) @@ -67,6 +114,11 @@ class BaseDirectFunctionWrapper: ) def to_function_schema(self) -> FunctionSchema: + """Convert the wrapped function to a FunctionSchema. + + Returns: + A FunctionSchema instance with extracted metadata. + """ return FunctionSchema( name=self.name, description=self.description, @@ -75,6 +127,7 @@ class BaseDirectFunctionWrapper: ) def _initialize_metadata(self): + """Initialize metadata from function signature and docstring.""" # Get function name self.name = self.function.__name__ @@ -93,20 +146,20 @@ class BaseDirectFunctionWrapper: def _get_parameters_as_jsonschema( self, func: Callable, docstring_params: List[docstring_parser.DocstringParam] ) -> Tuple[Dict[str, Any], List[str]]: - """ - Get function parameters as a dictionary of JSON schemas and a list of required parameters. + """Get function parameters as a dictionary of JSON schemas and a list of required parameters. + Ignore the first parameter, as it's expected to be the "special" one. Args: - func: Function to get parameters from - docstring_params: List of parameters extracted from the function's docstring + func: Function to get parameters from. + docstring_params: List of parameters extracted from the function's docstring. Returns: A tuple containing: - - A dictionary mapping each function parameter to its JSON schema - - A list of required parameter names - """ + - A dictionary mapping each function parameter to its JSON schema + - A list of required parameter names + """ sig = inspect.signature(func) hints = get_type_hints(func) properties = {} @@ -141,8 +194,7 @@ class BaseDirectFunctionWrapper: return properties, required def _typehint_to_jsonschema(self, type_hint: Any) -> Dict[str, Any]: - """ - Convert a Python type hint to a JSON Schema. + """Convert a Python type hint to a JSON Schema. Args: type_hint: A Python type hint @@ -213,16 +265,32 @@ class BaseDirectFunctionWrapper: class DirectFunctionWrapper(BaseDirectFunctionWrapper): - """ - Wrapper around a DirectFunction that: - - extracts metadata from the function signature and docstring - - generates a corresponding FunctionSchema - - helps with function invocation + """Wrapper around a DirectFunction for LLM function calling. + + This class: + + - Extracts metadata from the function signature and docstring + - Generates a corresponding FunctionSchema + - Helps with function invocation """ @classmethod def special_first_param_name(cls) -> str: + """Get the special first parameter name for direct functions. + + Returns: + The string "params" which is expected as the first parameter. + """ return "params" async def invoke(self, args: Mapping[str, Any], params: "FunctionCallParams"): + """Invoke the wrapped function with the provided arguments. + + Args: + args: Arguments to pass to the function. + params: Function call parameters from the LLM service. + + Returns: + The result of the function call. + """ return await self.function(params=params, **args) diff --git a/src/pipecat/examples/run.py b/src/pipecat/examples/run.py index 60cb3b5c6..be2834a28 100644 --- a/src/pipecat/examples/run.py +++ b/src/pipecat/examples/run.py @@ -93,6 +93,15 @@ async def maybe_capture_participant_screen( def smallwebrtc_sdp_cleanup_ice_candidates(text: str, pattern: str) -> str: + """Clean up ICE candidates in SDP text for SmallWebRTC. + + Args: + text: SDP text to clean up. + pattern: Pattern to match for candidate filtering. + + Returns: + Cleaned SDP text with filtered ICE candidates. + """ result = [] lines = text.splitlines() for line in lines: @@ -105,6 +114,14 @@ def smallwebrtc_sdp_cleanup_ice_candidates(text: str, pattern: str) -> str: def smallwebrtc_sdp_cleanup_fingerprints(text: str) -> str: + """Remove unsupported fingerprint algorithms from SDP text. + + Args: + text: SDP text to clean up. + + Returns: + SDP text with sha-384 and sha-512 fingerprints removed. + """ result = [] lines = text.splitlines() for line in lines: @@ -114,6 +131,15 @@ def smallwebrtc_sdp_cleanup_fingerprints(text: str) -> str: def smallwebrtc_sdp_munging(sdp: str, host: str) -> str: + """Apply SDP modifications for SmallWebRTC compatibility. + + Args: + sdp: Original SDP string. + host: Host address for ICE candidate filtering. + + Returns: + Modified SDP string with fingerprint and ICE candidate cleanup. + """ sdp = smallwebrtc_sdp_cleanup_fingerprints(sdp) sdp = smallwebrtc_sdp_cleanup_ice_candidates(sdp, host) return sdp @@ -232,6 +258,9 @@ def run_example_webrtc( Args: app: The FastAPI application instance. + + Yields: + Control to the FastAPI application runtime. """ yield # Run app coros = [pc.disconnect() for pc in pcs_map.values()] diff --git a/src/pipecat/services/llm_service.py b/src/pipecat/services/llm_service.py index 98c8483fe..51a5688f9 100644 --- a/src/pipecat/services/llm_service.py +++ b/src/pipecat/services/llm_service.py @@ -8,28 +8,19 @@ import asyncio import inspect -import types from dataclasses import dataclass from typing import ( Any, Awaitable, Callable, Dict, - List, Mapping, Optional, Protocol, Sequence, - Set, - Tuple, Type, - Union, - get_args, - get_origin, - get_type_hints, ) -import docstring_parser from loguru import logger from pipecat.adapters.base_llm_adapter import BaseLLMAdapter @@ -312,6 +303,17 @@ class LLMService(AIService): *, cancel_on_interruption: bool = True, ): + """Register a direct function handler for LLM function calls. + + Direct functions have their metadata automatically extracted from their + signature and docstring, eliminating the need for accompanying + configurations (as FunctionSchemas or in provider-specific formats). + + Args: + handler: The direct function to register. Must follow DirectFunction protocol. + cancel_on_interruption: Whether to cancel this function call when an + interruption occurs. Defaults to True. + """ wrapper = DirectFunctionWrapper(handler) self._functions[wrapper.name] = FunctionCallRegistryItem( function_name=wrapper.name, @@ -330,6 +332,11 @@ class LLMService(AIService): del self._start_callbacks[function_name] def unregister_direct_function(self, handler: Any): + """Remove a registered direct function handler. + + Args: + handler: The direct function handler to remove. + """ wrapper = DirectFunctionWrapper(handler) del self._functions[wrapper.name] # Note: no need to remove start callback here, as direct functions don't support start callbacks.