aiq.builder.function

Attributes

_InvokeFnT
_StreamFnT
_T
logger

Classes

Function	Abstract base class providing core functionality for AIQ Toolkit functions.
LambdaFunction	Abstract base class providing core functionality for AIQ Toolkit functions.

Module Contents

_InvokeFnT

_StreamFnT

_T

logger

class Function(

*,

config: aiq.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,

)

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

Abstract base class providing core functionality for AIQ Toolkit 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 AIQ Toolkit 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

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

abstractmethod _ainvoke(

value: aiq.builder.function_base.InputT,

) → aiq.builder.function_base.SingleOutputT

Async:

async ainvoke(

value: aiq.builder.function_base.InputT | Any,

) → aiq.builder.function_base.SingleOutputT

async ainvoke(

value: aiq.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.

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: aiq.builder.function_base.InputT,

) → collections.abc.AsyncGenerator[aiq.builder.function_base.StreamingOutputT]

Async:

async astream(

value: aiq.builder.function_base.InputT | Any,

) → collections.abc.AsyncGenerator[aiq.builder.function_base.SingleOutputT]

async astream(

value: aiq.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.

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: aiq.data_models.function.FunctionBaseConfig,

info: aiq.builder.function_info.FunctionInfo,

)

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

Abstract base class providing core functionality for AIQ Toolkit 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 AIQ Toolkit function system, providing: - Type handling via generics - Schema management for inputs and outputs - Type conversion capabilities - Abstract interface that concrete function classes must implement

_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: aiq.builder.function_base.InputT,

) → aiq.builder.function_base.SingleOutputT

async _astream(

value: aiq.builder.function_base.InputT,

) → collections.abc.AsyncGenerator[aiq.builder.function_base.StreamingOutputT]

static from_info(

*,

config: aiq.data_models.function.FunctionBaseConfig,

info: aiq.builder.function_info.FunctionInfo,

) → LambdaFunction[InputT, StreamingOutputT, SingleOutputT]