aiq.builder.component_utils

Attributes

logger
_component_group_order

Classes

ComponentInstanceData

A data model to hold component runtime instance metadata to support generating build sequences.

Functions

iterate_leaf_to_root(...)	A recursive generator that yields leaf nodes from the bottom to the root of a directed graph.
group_from_component(...)	Determines the component group from a runtime instance configuration object.
recursive_componentref_discovery(...)	Discovers instances of ComponentRefs in a configuration object and updates the dependency graph.
update_dependency_graph(→ networkx.DiGraph)	Updates the hierarchical component instance dependency graph from a configuration runtime instance.
config_to_dependency_objects(→ tuple[dict[str, ...)	Generates a map of component runtime instance IDs to use when generating a build sequence.
build_dependency_sequence(→ list[ComponentInstanceData])	Generates the depencency sequence from an AgentIQ configuration object

Module Contents

logger

_component_group_order

class ComponentInstanceData(/, **data: Any)

Bases: pydantic.BaseModel

A data model to hold component runtime instance metadata to support generating build sequences.

Args:

component_group (ComponentGroup): The component group in an AgentIQ configuration object. name (ComponentRef): The name of the component runtime instance. config (TypedBaseModel): The runtime instance’s configuration object. instance_id (str): Unique identifier for each runtime instance. is_root (bool): A flag to indicate if the runtime instance is the root of the workflow.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

component_group: aiq.data_models.component.ComponentGroup

name: aiq.data_models.component_ref.ComponentRef

config: aiq.data_models.common.TypedBaseModel

instance_id: str

is_root: bool = False

iterate_leaf_to_root(

graph: networkx.DiGraph,

) → collections.abc.Generator[aiq.data_models.component_ref.ComponentRefNode]

A recursive generator that yields leaf nodes from the bottom to the root of a directed graph.

Args:

graph (nx.DiGraph): A networkx directed graph object.

Yields:

ComponentRefNode: An object contain a ComponentRef and its component group.

group_from_component(

component: aiq.data_models.common.TypedBaseModel,

) → aiq.data_models.component.ComponentGroup | None

Determines the component group from a runtime instance configuration object.

Args:

component (TypedBaseModel): A runtime instance configuration object.

Returns:

ComponentGroup | None: The component group of the runtime instance configuration object. If the

component is not a valid runtime instance, None is returned.

recursive_componentref_discovery(

cls: aiq.data_models.common.TypedBaseModel,

value: Any,

type_hint: type[Any],

) → collections.abc.Generator[tuple[str, aiq.data_models.component_ref.ComponentRefNode]]

Discovers instances of ComponentRefs in a configuration object and updates the dependency graph.

Args:

cls (TypedBaseModel): A configuration object for a runtime instance. value (typing.Any): The current traversed value from the configuration object. type_hint (type[typing.Any]): The type of the current traversed value from the configuration object.

update_dependency_graph(

config: aiq.data_models.config.AIQConfig,

instance_config: aiq.data_models.common.TypedBaseModel,

dependency_graph: networkx.DiGraph,

) → networkx.DiGraph

Updates the hierarchical component instance dependency graph from a configuration runtime instance.

Args:

config (AIQConfig): An AgentIQ configuration object with runtime instance details. instance_config (TypedBaseModel): A component’s runtime instance configuration object. dependency_graph (nx.DiGraph): A graph tracking runtime instance component dependencies.

Returns:

nx.DiGraph: An dependency graph that has been updated with the provided runtime instance.

config_to_dependency_objects(

config: aiq.data_models.config.AIQConfig,

) → tuple[dict[str, ComponentInstanceData], networkx.DiGraph]

Generates a map of component runtime instance IDs to use when generating a build sequence.

Args:

config (AIQConfig): The AgentIQ workflow configuration object.

Returns:

tuple[dict[str, ComponentInstanceData], nx.DiGraph]: A tuple containing a map of component runtime instance

IDs to a component object containing its metadata and a dependency graph of nested components.

build_dependency_sequence(

config: aiq.data_models.config.AIQConfig,

) → list[ComponentInstanceData]

Generates the depencency sequence from an AgentIQ configuration object

Args:

config (AIQConfig): An AgentIQ configuration object.

Returns:

list[ComponentInstanceData]: A list representing the instatiation sequence to ensure all valid

runtime instance references.