aiq.builder.function_base#

Base class for AgentIQ functions providing type handling and schema management.

This module contains the FunctionBase abstract base class which provides core functionality for AgentIQ functions including type handling via generics, schema management for inputs and outputs, and type conversion capabilities.

Attributes#

Classes#

FunctionBase

Abstract base class providing core functionality for AgentIQ functions.

Module Contents#

InputT#
StreamingOutputT#
SingleOutputT#
logger#
class FunctionBase(
*,
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: Generic[InputT, StreamingOutputT, SingleOutputT], abc.ABC

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

_converter_list = []#
_input_schema#
_converter: aiq.utils.type_converter.TypeConverter#
property input_type: type[InputT]#

Get the input type of the function. The input type is determined by the generic parameters of the class.

For example, if a function is defined as def my_function(input: list[int]) -> str, the input_type is list[int].

Returns#

type[InputT]

The input type specified in the generic parameters

Raises#

ValueError

If the input type cannot be determined from the class definition

property input_class: type#

Get the python class of the input type. This is the class that can be used to check if a value is an instance of the input type. It removes any generic or annotation information from the input type.

For example, if a function is defined as def my_function(input: list[int]) -> str, the input_class is list.

Returns#

type

The python type of the input type

property input_schema: type[pydantic.BaseModel]#

Get the Pydantic model schema for validating inputs. The schema must be pydantic models. This allows for type validation and coercion, and documenting schema properties of the input value. If the input type is already a pydantic model, it will be returned as is.

For example, if a function is defined as def my_function(input: list[int]) -> str, the input_schema is:

class InputSchema(BaseModel):
    input: list[int]

Returns#

type[BaseModel]

The Pydantic model class for input validation

property converter_list: list[collections.abc.Callable[[Any], Any]]#

Get the list of type converters used by this function.

Returns#

list[Callable[[typing.Any], typing.Any]]

List of converter functions that transform input types

property streaming_output_type: type[StreamingOutputT]#

Get the streaming output type of the function. The streaming output type is determined by the generic parameters of the class.

For example, if a function is defined as def my_function(input: int) -> AsyncGenerator[dict[str, Any]], the streaming_output_type is dict[str, Any].

Returns#

type[StreamingOutputT]

The streaming output type specified in the generic parameters

Raises#

ValueError

If the streaming output type cannot be determined from the class definition

property streaming_output_class: type#

Get the python class of the output type. This is the class that can be used to check if a value is an instance of the output type. It removes any generic or annotation information from the output type.

For example, if a function is defined as def my_function(input: int) -> AsyncGenerator[dict[str, Any]], the streaming_output_class is dict.

Returns#

type

The python type of the output type

property streaming_output_schema: type[pydantic.BaseModel] | type[None]#

Get the Pydantic model schema for validating streaming outputs. The schema must be pydantic models. This allows for type validation and coercion, and documenting schema properties of the output value. If the output type is already a pydantic model, it will be returned as is.

For example, if a function is defined as def my_function(input: int) -> AsyncGenerator[dict[str, Any]], the streaming_output_schema is:

class StreamingOutputSchema(BaseModel):
    value: dict[str, Any]

Returns#

type[BaseModel] | type[None]

The Pydantic model class for streaming output validation, or NoneType if no streaming output.

property single_output_type: type[SingleOutputT]#

Get the single output type of the function. The single output type is determined by the generic parameters of the class. Returns NoneType if no single output is supported.

For example, if a function is defined as def my_function(input: int) -> list[str], the single_output_type is list[str].

Returns#

type[SingleOutputT]

The single output type specified in the generic parameters

Raises#

ValueError

If the single output type cannot be determined from the class definition

property single_output_class: type#

Get the python class of the output type. This is the class that can be used to check if a value is an instance of the output type. It removes any generic or annotation information from the output type.

For example, if a function is defined as def my_function(input: int) -> list[str], the single_output_class is list.

Returns#

type

The python type of the output type

property single_output_schema: type[pydantic.BaseModel] | type[None]#

Get the Pydantic model schema for validating single outputs. The schema must be pydantic models. This allows for type validation and coercion, and documenting schema properties of the output value. If the output type is already a pydantic model, it will be returned as is.

For example, if a function is defined as def my_function(input: int) -> list[str], the single_output_schema is:

class SingleOutputSchema(BaseModel):
    value: list[str]

Returns#

type[BaseModel] | type[None]

The Pydantic model class for single output validation, or None if no single output

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

_convert_input(value: Any)#