nat.builder.workflow_builder#

Attributes#

logger

Classes#

`ConfiguredTelemetryExporter`
`ConfiguredFunction`
`ConfiguredLLM`
`ConfiguredEmbedder`
`ConfiguredMemory`
`ConfiguredObjectStore`
`ConfiguredRetriever`
`ConfiguredAuthProvider`
`ConfiguredTTCStrategy`
`WorkflowBuilder`	Helper class that provides a standard way to create an ABC using
`ChildBuilder`	Helper class that provides a standard way to create an ABC using

Module Contents#

logger#

class ConfiguredTelemetryExporter#

config: nat.data_models.telemetry_exporter.TelemetryExporterBaseConfig#

instance: nat.observability.exporter.base_exporter.BaseExporter#

class ConfiguredFunction#

config: nat.data_models.function.FunctionBaseConfig#

instance: nat.builder.function.Function#

class ConfiguredLLM#

config: nat.data_models.llm.LLMBaseConfig#

instance: nat.builder.llm.LLMProviderInfo#

class ConfiguredEmbedder#

config: nat.data_models.embedder.EmbedderBaseConfig#

instance: nat.builder.embedder.EmbedderProviderInfo#

class ConfiguredMemory#

config: nat.data_models.memory.MemoryBaseConfig#

instance: nat.memory.interfaces.MemoryEditor#

class ConfiguredObjectStore#

config: nat.data_models.object_store.ObjectStoreBaseConfig#

instance: nat.object_store.interfaces.ObjectStore#

class ConfiguredRetriever#

config: nat.data_models.retriever.RetrieverBaseConfig#

instance: nat.builder.retriever.RetrieverProviderInfo#

class ConfiguredAuthProvider#

config: nat.data_models.authentication.AuthProviderBaseConfig#

instance: nat.authentication.interfaces.AuthProviderBase#

class ConfiguredTTCStrategy#

config: nat.data_models.ttc_strategy.TTCStrategyBaseConfig#

instance: nat.experimental.test_time_compute.models.strategy_base.StrategyBase#

class WorkflowBuilder( *, general_config: nat.data_models.config.GeneralConfig | None = None, registry: nat.cli.type_registry.TypeRegistry | None = None, )#

Bases: nat.builder.builder.Builder, contextlib.AbstractAsyncContextManager

Helper class that provides a standard way to create an ABC using inheritance.

general_config = None#

_registry = None#

_logging_handlers: dict[str, logging.Handler]#

_telemetry_exporters: dict[str, ConfiguredTelemetryExporter]#

_functions: dict[str, ConfiguredFunction]#

_workflow: ConfiguredFunction | None = None#

_llms: dict[str, ConfiguredLLM]#

_auth_providers: dict[str, ConfiguredAuthProvider]#

_embedders: dict[str, ConfiguredEmbedder]#

_memory_clients: dict[str, ConfiguredMemory]#

_object_stores: dict[str, ConfiguredObjectStore]#

_retrievers: dict[str, ConfiguredRetriever]#

_ttc_strategies: dict[str, ConfiguredTTCStrategy]#

_context_state#

_exit_stack: contextlib.AsyncExitStack | None = None#

function_dependencies: dict[str, nat.data_models.function_dependencies.FunctionDependencies]#

current_function_building: str | None = None#

build( entry_function: str | None = None, ) → nat.builder.workflow.Workflow#

Creates an instance of a workflow object using the added components and the desired entry function.

Parameters#

entry_functionstr | None, optional: The function name to use as the entry point for the created workflow. If None, the entry point will be the specified workflow function. By default None

Returns#

Workflow: A created workflow.

Raises#

ValueError: If the workflow has not been set before building.

_get_exit_stack() → contextlib.AsyncExitStack#

async _build_function( name: str, config: nat.data_models.function.FunctionBaseConfig, ) → ConfiguredFunction#

async add_function( name: str | nat.data_models.component_ref.FunctionRef, config: nat.data_models.function.FunctionBaseConfig, ) → nat.builder.function.Function#

get_function( name: str | nat.data_models.component_ref.FunctionRef, ) → nat.builder.function.Function#

get_function_config( name: str | nat.data_models.component_ref.FunctionRef, ) → nat.data_models.function.FunctionBaseConfig#

async set_workflow( config: nat.data_models.function.FunctionBaseConfig, ) → nat.builder.function.Function#

get_workflow() → nat.builder.function.Function#

get_workflow_config() → nat.data_models.function.FunctionBaseConfig#

get_function_dependencies( fn_name: str | nat.data_models.component_ref.FunctionRef, ) → nat.data_models.function_dependencies.FunctionDependencies#

get_tool( fn_name: str | nat.data_models.component_ref.FunctionRef, wrapper_type: nat.builder.framework_enum.LLMFrameworkEnum | str, )#

async add_llm( name: str | nat.data_models.component_ref.LLMRef, config: nat.data_models.llm.LLMBaseConfig, )#

async get_llm( llm_name: str | nat.data_models.component_ref.LLMRef, wrapper_type: nat.builder.framework_enum.LLMFrameworkEnum | str, )#

get_llm_config( llm_name: str | nat.data_models.component_ref.LLMRef, ) → nat.data_models.llm.LLMBaseConfig#

async add_auth_provider( name: str | nat.data_models.component_ref.AuthenticationRef, config: nat.data_models.authentication.AuthProviderBaseConfig, ) → nat.authentication.interfaces.AuthProviderBase#

Add an authentication provider to the workflow by constructing it from a configuration object.

Note: The Authentication Provider API is experimental and the API may change in future releases.

Parameters#

namestr | AuthenticationRef: The name of the authentication provider to add.
configAuthProviderBaseConfig: The configuration for the authentication provider.

Returns#

AuthProviderBase: The authentication provider instance.

Raises#

ValueError: If the authentication provider is already in the list of authentication providers.

async get_auth_provider( auth_provider_name: str, ) → nat.authentication.interfaces.AuthProviderBase#

Get the authentication provider instance for the given name.

Note: The Authentication Provider API is experimental and the API may change in future releases.

Parameters#

auth_provider_namestr: The name of the authentication provider to get.

Returns#

AuthProviderBase: The authentication provider instance.

Raises#

ValueError: If the authentication provider is not found.

async add_embedder( name: str | nat.data_models.component_ref.EmbedderRef, config: nat.data_models.embedder.EmbedderBaseConfig, )#

async get_embedder( embedder_name: str | nat.data_models.component_ref.EmbedderRef, wrapper_type: nat.builder.framework_enum.LLMFrameworkEnum | str, )#

get_embedder_config( embedder_name: str | nat.data_models.component_ref.EmbedderRef, ) → nat.data_models.embedder.EmbedderBaseConfig#

async add_memory_client( name: str | nat.data_models.component_ref.MemoryRef, config: nat.data_models.memory.MemoryBaseConfig, ) → nat.memory.interfaces.MemoryEditor#

get_memory_client( memory_name: str | nat.data_models.component_ref.MemoryRef, ) → nat.memory.interfaces.MemoryEditor#: Return the instantiated memory client for the given name.

get_memory_client_config( memory_name: str | nat.data_models.component_ref.MemoryRef, ) → nat.data_models.memory.MemoryBaseConfig#

async add_object_store( name: str | nat.data_models.component_ref.ObjectStoreRef, config: nat.data_models.object_store.ObjectStoreBaseConfig, ) → nat.object_store.interfaces.ObjectStore#

async get_object_store_client( object_store_name: str | nat.data_models.component_ref.ObjectStoreRef, ) → nat.object_store.interfaces.ObjectStore#

get_object_store_config( object_store_name: str | nat.data_models.component_ref.ObjectStoreRef, ) → nat.data_models.object_store.ObjectStoreBaseConfig#

async add_retriever( name: str | nat.data_models.component_ref.RetrieverRef, config: nat.data_models.retriever.RetrieverBaseConfig, )#

async get_retriever( retriever_name: str | nat.data_models.component_ref.RetrieverRef, wrapper_type: nat.builder.framework_enum.LLMFrameworkEnum | str | None = None, )#

async get_retriever_config( retriever_name: str | nat.data_models.component_ref.RetrieverRef, ) → nat.data_models.retriever.RetrieverBaseConfig#

async add_ttc_strategy( name: str | str, config: nat.data_models.ttc_strategy.TTCStrategyBaseConfig, )#

async get_ttc_strategy( strategy_name: str | nat.data_models.component_ref.TTCStrategyRef, pipeline_type: nat.experimental.test_time_compute.models.stage_enums.PipelineTypeEnum, stage_type: nat.experimental.test_time_compute.models.stage_enums.StageTypeEnum, ) → nat.experimental.test_time_compute.models.strategy_base.StrategyBase#

async get_ttc_strategy_config( strategy_name: str | nat.data_models.component_ref.TTCStrategyRef, pipeline_type: nat.experimental.test_time_compute.models.stage_enums.PipelineTypeEnum, stage_type: nat.experimental.test_time_compute.models.stage_enums.StageTypeEnum, ) → nat.data_models.ttc_strategy.TTCStrategyBaseConfig#

get_user_manager()#

async add_telemetry_exporter( name: str, config: nat.data_models.telemetry_exporter.TelemetryExporterBaseConfig, ) → None#

Add an configured telemetry exporter to the builder.

Args:: name (str): The name of the telemetry exporter config (TelemetryExporterBaseConfig): The configuration for the exporter

_log_build_failure( component_name: str, component_type: str, completed_components: list[tuple[str, str]], remaining_components: list[tuple[str, str]], original_error: Exception, ) → None#

Common method to log comprehensive build failure information.

Args:: component_name (str): The name of the component that failed to build component_type (str): The type of the component that failed to build completed_components (list[tuple[str, str]]): List of (name, type) tuples for successfully built components remaining_components (list[tuple[str, str]]): List of (name, type) tuples for components still to be built original_error (Exception): The original exception that caused the failure

_log_build_failure_component( failing_component: nat.builder.component_utils.ComponentInstanceData, completed_components: list[tuple[str, str]], remaining_components: list[tuple[str, str]], original_error: Exception, ) → None#

Log comprehensive component build failure information.

Args:: failing_component (ComponentInstanceData): The ComponentInstanceData that failed to build completed_components (list[tuple[str, str]]): List of (name, type) tuples for successfully built components remaining_components (list[tuple[str, str]]): List of (name, type) tuples for components still to be built original_error (Exception): The original exception that caused the failure

_log_build_failure_workflow( completed_components: list[tuple[str, str]], remaining_components: list[tuple[str, str]], original_error: Exception, ) → None#

Log comprehensive workflow build failure information.

Args:: completed_components (list[tuple[str, str]]): List of (name, type) tuples for successfully built components remaining_components (list[tuple[str, str]]): List of (name, type) tuples for components still to be built original_error (Exception): The original exception that caused the failure

async populate_builder( config: nat.data_models.config.Config, skip_workflow: bool = False, )#

Populate the builder with components and optionally set up the workflow.

Args:: config (Config): The configuration object containing component definitions. skip_workflow (bool): If True, skips the workflow instantiation step. Defaults to False.

classmethod from_config(config: nat.data_models.config.Config)#

Async:

class ChildBuilder(workflow_builder: WorkflowBuilder)#

Bases: nat.builder.builder.Builder

Helper class that provides a standard way to create an ABC using inheritance.

_workflow_builder#

_dependencies#

property dependencies: nat.data_models.function_dependencies.FunctionDependencies#

async add_function( name: str, config: nat.data_models.function.FunctionBaseConfig, ) → nat.builder.function.Function#

get_function(name: str) → nat.builder.function.Function#

get_function_config( name: str, ) → nat.data_models.function.FunctionBaseConfig#

async set_workflow( config: nat.data_models.function.FunctionBaseConfig, ) → nat.builder.function.Function#

get_workflow() → nat.builder.function.Function#

get_workflow_config() → nat.data_models.function.FunctionBaseConfig#

get_tool( fn_name: str, wrapper_type: nat.builder.framework_enum.LLMFrameworkEnum | str, )#

async add_llm(name: str, config: nat.data_models.llm.LLMBaseConfig)#

async add_auth_provider( name: str, config: nat.data_models.authentication.AuthProviderBaseConfig, )#

async get_auth_provider(auth_provider_name: str)#

async get_llm( llm_name: str, wrapper_type: nat.builder.framework_enum.LLMFrameworkEnum | str, )#

get_llm_config(llm_name: str) → nat.data_models.llm.LLMBaseConfig#

async add_embedder( name: str, config: nat.data_models.embedder.EmbedderBaseConfig, )#

async get_embedder( embedder_name: str, wrapper_type: nat.builder.framework_enum.LLMFrameworkEnum | str, )#

get_embedder_config( embedder_name: str, ) → nat.data_models.embedder.EmbedderBaseConfig#

async add_memory_client( name: str, config: nat.data_models.memory.MemoryBaseConfig, ) → nat.memory.interfaces.MemoryEditor#

get_memory_client( memory_name: str, ) → nat.memory.interfaces.MemoryEditor#: Return the instantiated memory client for the given name.

get_memory_client_config( memory_name: str, ) → nat.data_models.memory.MemoryBaseConfig#

async add_object_store( name: str, config: nat.data_models.object_store.ObjectStoreBaseConfig, )#

async get_object_store_client( object_store_name: str, ) → nat.object_store.interfaces.ObjectStore#: Return the instantiated object store client for the given name.

get_object_store_config( object_store_name: str, ) → nat.data_models.object_store.ObjectStoreBaseConfig#

async add_ttc_strategy( name: str, config: nat.data_models.ttc_strategy.TTCStrategyBaseConfig, )#

async get_ttc_strategy( strategy_name: str | nat.data_models.component_ref.TTCStrategyRef, pipeline_type: nat.experimental.test_time_compute.models.stage_enums.PipelineTypeEnum, stage_type: nat.experimental.test_time_compute.models.stage_enums.StageTypeEnum, ) → nat.experimental.test_time_compute.models.strategy_base.StrategyBase#

async get_ttc_strategy_config( strategy_name: str | nat.data_models.component_ref.TTCStrategyRef, pipeline_type: nat.experimental.test_time_compute.models.stage_enums.PipelineTypeEnum, stage_type: nat.experimental.test_time_compute.models.stage_enums.StageTypeEnum, ) → nat.data_models.ttc_strategy.TTCStrategyBaseConfig#

async add_retriever( name: str, config: nat.data_models.retriever.RetrieverBaseConfig, )#

async get_retriever( retriever_name: str, wrapper_type: nat.builder.framework_enum.LLMFrameworkEnum | str | None = None, )#

async get_retriever_config( retriever_name: str, ) → nat.data_models.retriever.RetrieverBaseConfig#

get_user_manager() → nat.builder.builder.UserManagerHolder#

get_function_dependencies( fn_name: str, ) → nat.data_models.function_dependencies.FunctionDependencies#