nat.builder.function#

Attributes#

`_InvokeFnT`
`_StreamFnT`
`_T`
`logger`

Classes#

`Function`	Abstract base class providing core functionality for NAT functions.
`LambdaFunction`	Abstract base class providing core functionality for NAT functions.
`FunctionGroup`	A group of functions that can be used together, sharing the same configuration, context, and resources.

Module Contents#

_InvokeFnT#

_StreamFnT#

_T#

logger#

class Function( *, config: nat.data_models.function.FunctionBaseConfig, description: str | None, input_schema: type[pydantic.BaseModel] | None = None, streaming_output_schema: type[pydantic.BaseModel] | type[None] | None = None, single_output_schema: type[pydantic.BaseModel] | type[None] | None = None, converters: list[collections.abc.Callable[[Any], Any]] | None = None, instance_name: str | None = None, )#

Bases: nat.builder.function_base.FunctionBase[nat.builder.function_base.InputT, nat.builder.function_base.StreamingOutputT, nat.builder.function_base.SingleOutputT], abc.ABC

Abstract base class providing core functionality for NAT functions.

This class provides type handling via generics, schema management for inputs and outputs, and type conversion capabilities.

Parameters#

InputTTypeVar: The input type for the function
StreamingOutputTTypeVar: The output type for streaming results
SingleOutputTTypeVar: The output type for single results

Notes#

FunctionBase is the foundation of the NAT function system, providing: - Type handling via generics - Schema management for inputs and outputs - Type conversion capabilities - Abstract interface that concrete function classes must implement

config#

description#

instance_name#

_context#

convert(value: Any, to_type: type[_T]) → _T#

Converts the given value to the specified type using the function’s converter.

Parameters#

valuetyping.Any: The value to convert.
to_typetype: The type to convert the value to.

Returns#

_T: The converted value.

Raises#

ValueError: If the value cannot be converted to the specified type (when to_type is specified).

try_convert(value: Any, to_type: type[_T]) → _T | Any#

Converts the given value to the specified type using graceful error handling. If conversion fails, returns the original value and continues processing.

Parameters#

valuetyping.Any: The value to convert.
to_typetype: The type to convert the value to.

Returns#

_T | typing.Any: The converted value, or original value if conversion fails.

abstractmethod _ainvoke( value: nat.builder.function_base.InputT, ) → nat.builder.function_base.SingleOutputT#

Async:

async ainvoke( value: nat.builder.function_base.InputT | Any, ) → nat.builder.function_base.SingleOutputT#

async ainvoke( value: nat.builder.function_base.InputT | Any, to_type: type[_T], ) → _T

Runs the function with the given input and returns a single output from the function. This is the main entry point for running a function.

Parameters#

valueInputT | typing.Any: The input to the function.
to_typetype | None, optional: The type to convert the output to using the function’s converter. When not specified, the output will match single_output_type.

Returns#

typing.Any: The output of the function optionally converted to the specified type.

Raises#

ValueError: If the output of the function cannot be converted to the specified type.

async acall_invoke(*args, **kwargs)#

A wrapper around ainvoke that allows for calling the function with arbitrary arguments and keyword arguments. This is useful in scenarios where the function might be called by an LLM or other system which gives varying inputs to the function. The function will attempt to convert the args and kwargs to the input schema of the function.

Returns#

SingleOutputT: The output of the function.

abstractmethod _astream( value: nat.builder.function_base.InputT, ) → collections.abc.AsyncGenerator[nat.builder.function_base.StreamingOutputT]#

Async:

async astream( value: nat.builder.function_base.InputT | Any, ) → collections.abc.AsyncGenerator[nat.builder.function_base.SingleOutputT]#

async astream( value: nat.builder.function_base.InputT | Any, to_type: type[_T], ) → collections.abc.AsyncGenerator[_T]

Runs the function with the given input and returns a stream of outputs from the function. This is the main entry point for running a function with streaming output.

Parameters#

valueInputT | typing.Any: The input to the function.
to_typetype | None, optional: The type to convert the output to using the function’s converter. When not specified, the output will match streaming_output_type.

Yields#

typing.Any: The output of the function optionally converted to the specified type.

Raises#

ValueError: If the output of the function cannot be converted to the specified type (when to_type is specified).

async acall_stream(*args, **kwargs)#

A wrapper around astream that allows for calling the function with arbitrary arguments and keyword arguments. This is useful in scenarios where the function might be called by an LLM or other system which gives varying inputs to the function. The function will attempt to convert the args and kwargs to the input schema of the function.

Yields#

StreamingOutputT: The output of the function.

class LambdaFunction( *, config: nat.data_models.function.FunctionBaseConfig, info: nat.builder.function_info.FunctionInfo, instance_name: str | None = None, )#

Bases: Function[nat.builder.function_base.InputT, nat.builder.function_base.StreamingOutputT, nat.builder.function_base.SingleOutputT]

Abstract base class providing core functionality for NAT functions.

This class provides type handling via generics, schema management for inputs and outputs, and type conversion capabilities.

Parameters#

InputTTypeVar: The input type for the function
StreamingOutputTTypeVar: The output type for streaming results
SingleOutputTTypeVar: The output type for single results

Notes#

_info#

_ainvoke_fn: _InvokeFnT#

_astream_fn: _StreamFnT#

property has_streaming_output: bool#

Check if this function supports streaming output.

Returns#

bool: True if the function supports streaming output, False otherwise

property has_single_output: bool#

Check if this function supports single output.

Returns#

bool: True if the function supports single output, False otherwise

async _ainvoke( value: nat.builder.function_base.InputT, ) → nat.builder.function_base.SingleOutputT#

async _astream( value: nat.builder.function_base.InputT, ) → collections.abc.AsyncGenerator[nat.builder.function_base.StreamingOutputT]#

static from_info( *, config: nat.data_models.function.FunctionBaseConfig, info: nat.builder.function_info.FunctionInfo, instance_name: str | None = None, ) → LambdaFunction[InputT, StreamingOutputT, SingleOutputT]#

class FunctionGroup( *, config: nat.data_models.function.FunctionGroupBaseConfig, instance_name: str | None = None, filter_fn: collections.abc.Callable[[collections.abc.Sequence[str]], collections.abc.Awaitable[collections.abc.Sequence[str]]] | None = None, )#

A group of functions that can be used together, sharing the same configuration, context, and resources.

Creates a new function group.

Parameters#

configFunctionGroupBaseConfig: The configuration for the function group.
instance_namestr | None, optional: The name of the function group. If not provided, the type of the function group will be used.
filter_fnCallable[[Sequence[str]], Awaitable[Sequence[str]]] | None, optional: A callback function to additionally filter the functions in the function group dynamically when the functions are accessed via any accessor method.

_config#

_instance_name#

_functions: dict[str, Function]#

_filter_fn = None#

_per_function_filter_fn: dict[str, collections.abc.Callable[[str], collections.abc.Awaitable[bool]]]#

add_function( name: str, fn: collections.abc.Callable, *, input_schema: type[pydantic.BaseModel] | None = None, description: str | None = None, converters: list[collections.abc.Callable] | None = None, filter_fn: collections.abc.Callable[[str], collections.abc.Awaitable[bool]] | None = None, )#

Adds a function to the function group.

Parameters#

namestr: The name of the function.
fnCallable: The function to add to the function group.
input_schematype[BaseModel] | None, optional: The input schema for the function.
descriptionstr | None, optional: The description of the function.
converterslist[Callable] | None, optional: The converters to use for the function.
filter_fnCallable[[str], Awaitable[bool]] | None, optional: A callback to determine if the function should be included in the function group. The callback will be called with the function name. The callback is invoked dynamically when the functions are accessed via any accessor method such as get_accessible_functions, get_included_functions, get_excluded_functions, get_all_functions.

Raises#

ValueError: When the function name is empty or blank. When the function name contains invalid characters. When the function already exists in the function group.

get_config() → nat.data_models.function.FunctionGroupBaseConfig#

Returns the configuration for the function group.

Returns#

FunctionGroupBaseConfig: The configuration for the function group.

_get_fn_name(name: str) → str#

async _fn_should_be_included(name: str) → bool#

async _get_all_but_excluded_functions( filter_fn: collections.abc.Callable[[collections.abc.Sequence[str]], collections.abc.Awaitable[collections.abc.Sequence[str]]] | None = None, ) → dict[str, Function]#: Returns a dictionary of all functions in the function group except the excluded functions.

async get_accessible_functions( filter_fn: collections.abc.Callable[[collections.abc.Sequence[str]], collections.abc.Awaitable[collections.abc.Sequence[str]]] | None = None, ) → dict[str, Function]#

Returns a dictionary of all accessible functions in the function group.

First, the functions are filtered by the function group’s configuration. If the function group is configured to: - include some functions, this will return only the included functions. - not include or exclude any function, this will return all functions in the group. - exclude some functions, this will return all functions in the group except the excluded functions.

Then, the functions are filtered by filter function and per-function filter functions.

Parameters#

filter_fnCallable[[Sequence[str]], Awaitable[Sequence[str]]] | None, optional: A callback function to additionally filter the functions in the function group dynamically. If not provided then fall back to the function group’s filter function. If no filter function is set for the function group all functions will be returned.

Returns#

dict[str, Function]: A dictionary of all accessible functions in the function group.

Raises#

ValueError: When the function group is configured to include functions that are not found in the group.

async get_excluded_functions( filter_fn: collections.abc.Callable[[collections.abc.Sequence[str]], collections.abc.Awaitable[collections.abc.Sequence[str]]] | None = None, ) → dict[str, Function]#

Returns a dictionary of all functions in the function group which are configured to be excluded or filtered out by a filter function or per-function filter function.

Parameters#

filter_fnCallable[[Sequence[str]], Awaitable[Sequence[str]]] | None, optional: A callback function to additionally filter the functions in the function group dynamically. If not provided then fall back to the function group’s filter function. If no filter function is set for the function group then no functions will be added to the returned dictionary.

Returns#

dict[str, Function]: A dictionary of all excluded functions in the function group.

Raises#

ValueError: When the function group is configured to exclude functions that are not found in the group.

async get_included_functions( filter_fn: collections.abc.Callable[[collections.abc.Sequence[str]], collections.abc.Awaitable[collections.abc.Sequence[str]]] | None = None, ) → dict[str, Function]#

Returns a dictionary of all functions in the function group which are: - configured to be included and added to the global function registry - not configured to be excluded. - not filtered out by a filter function.

Parameters#

filter_fnCallable[[Sequence[str]], Awaitable[Sequence[str]]] | None, optional: A callback function to additionally filter the functions in the function group dynamically. If not provided then fall back to the function group’s filter function. If no filter function is set for the function group all functions will be returned.

Returns#

dict[str, Function]: A dictionary of all included functions in the function group.

Raises#

ValueError: When the function group is configured to include functions that are not found in the group.

async get_all_functions( filter_fn: collections.abc.Callable[[collections.abc.Sequence[str]], collections.abc.Awaitable[collections.abc.Sequence[str]]] | None = None, ) → dict[str, Function]#

Returns a dictionary of all functions in the function group, regardless if they are included or excluded.

If a filter function has been set, the returned functions will additionally be filtered by the callback.

Parameters#

filter_fnCallable[[Sequence[str]], Awaitable[Sequence[str]]] | None, optional: A callback function to additionally filter the functions in the function group dynamically. If not provided then fall back to the function group’s filter function. If no filter function is set for the function group all functions will be returned.

Returns#

dict[str, Function]: A dictionary of all functions in the function group.

set_filter_fn( filter_fn: collections.abc.Callable[[collections.abc.Sequence[str]], collections.abc.Awaitable[collections.abc.Sequence[str]]], )#

Sets the filter function for the function group.

Parameters#

filter_fnCallable[[Sequence[str]], Awaitable[Sequence[str]]]: The filter function to set for the function group.

set_per_function_filter_fn( name: str, filter_fn: collections.abc.Callable[[str], collections.abc.Awaitable[bool]], )#

Sets the a per-function filter function for the a function within the function group.

Parameters#

namestr: The name of the function.
filter_fnCallable[[str], Awaitable[bool]]: The per-function filter function to set for the function group.

Raises#

ValueError: When the function is not found in the function group.

set_instance_name(instance_name: str)#

Sets the instance name for the function group.

Parameters#

instance_namestr: The instance name to set for the function group.

property instance_name: str#: Returns the instance name for the function group.