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

Clean up docstrings after DirectFunction merge (#2105)

Browse Source 
* 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
This commit is contained in:
Mark Backman
2025-07-01 15:22:30 -04:00
committed by GitHub
parent 7f76a14c54
commit cc637f4dea
3 changed files with 139 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

120
src/pipecat/adapters/schemas/direct_function.py
View File

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

29
src/pipecat/examples/run.py
View File

@@ -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()]

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

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