nat.builder.workflow_builder

Attributes

logger

Classes

ConfiguredTelemetryExporter
ConfiguredFunction
ConfiguredFunctionGroup
ConfiguredLLM
ConfiguredEmbedder
ConfiguredMemory
ConfiguredObjectStore
ConfiguredRetriever
ConfiguredAuthProvider
ConfiguredTTCStrategy
ConfiguredMiddleware
ConfiguredTrainer
ConfiguredTrainerAdapter
ConfiguredTrajectoryBuilder
WorkflowBuilder	Helper class that provides a standard way to create an ABC using

Functions

_log_build_failure(→ None)	Common method to log comprehensive build failure information.
_build_function_impl(→ ConfiguredFunction)	Helper for core function building logic.
_build_function_group_impl(→ ConfiguredFunctionGroup)	Core function group building logic shared between WorkflowBuilder and PerUserWorkflowBuilder.

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 ConfiguredFunctionGroup

config: nat.data_models.function.FunctionGroupBaseConfig

instance: nat.builder.function.FunctionGroup

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 ConfiguredMiddleware

config: nat.data_models.middleware.MiddlewareBaseConfig

instance: nat.middleware.middleware.Middleware

class ConfiguredTrainer

config: nat.data_models.finetuning.TrainerConfig

instance: nat.finetuning.interfaces.finetuning_runner.Trainer

class ConfiguredTrainerAdapter

config: nat.data_models.finetuning.TrainerAdapterConfig

instance: nat.finetuning.interfaces.trainer_adapter.TrainerAdapter

class ConfiguredTrajectoryBuilder

config: nat.data_models.finetuning.TrajectoryBuilderConfig

instance: nat.finetuning.interfaces.trajectory_builder.TrajectoryBuilder

_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

async _build_function_impl(

*,

name: str,

config: nat.data_models.function.FunctionBaseConfig,

registry: nat.cli.type_registry.TypeRegistry,

exit_stack: contextlib.AsyncExitStack,

inner_builder: nat.builder.child_builder.ChildBuilder,

llms: dict[str, nat.builder.llm.LLMProviderInfo],

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

middleware_instances: list[nat.middleware.function_middleware.FunctionMiddleware],

) → ConfiguredFunction

Helper for core function building logic.

Args:

name: The function name config: The function configuration registry: Type registry to look up the function registration exit_stack: Async exit stack for context management inner_builder: ChildBuilder instance for dependency tracking llms: Dictionary of LLM instances dependencies: Dictionary to store function dependencies middleware_instances: Pre-resolved middleware instances

async _build_function_group_impl(

*,

name: str,

config: nat.data_models.function.FunctionGroupBaseConfig,

registry: nat.cli.type_registry.TypeRegistry,

exit_stack: contextlib.AsyncExitStack,

inner_builder: nat.builder.child_builder.ChildBuilder,

llms: dict[str, nat.builder.llm.LLMProviderInfo],

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

middleware_instances: list[nat.middleware.function_middleware.FunctionMiddleware],

) → ConfiguredFunctionGroup

Core function group building logic shared between WorkflowBuilder and PerUserWorkflowBuilder.

Args:

name: The function group name config: The function group configuration registry: Type registry to look up the function group registration exit_stack: Async exit stack for context management inner_builder: ChildBuilder instance for dependency tracking llms: Dictionary of LLM instances dependencies: Dictionary to store function group dependencies middleware_instances: Pre-resolved middleware instances

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]

_removed_root_handlers: list[tuple[logging.Handler, int]] = []

_telemetry_exporters: dict[str, ConfiguredTelemetryExporter]

_functions: dict[str, ConfiguredFunction]

_function_groups: dict[str, ConfiguredFunctionGroup]

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

_middleware: dict[str, ConfiguredMiddleware]

_trainers: dict[str, ConfiguredTrainer]

_trainer_adapters: dict[str, ConfiguredTrainerAdapter]

_trajectory_builders: dict[str, ConfiguredTrajectoryBuilder]

_context_state

_exit_stack: contextlib.AsyncExitStack | None = None

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

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

completed_components: list[tuple[str, str]] = []

remaining_components: list[tuple[str, str]] = []

property sync_builder: nat.builder.sync_builder.SyncBuilder

Get the synchronous version of the builder.

Returns:

The SyncBuilder object (synchronous wrapper).

async 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 _resolve_middleware_instances(

middleware_names: list[str],

component_name: str,

component_type: str,

) → list[nat.middleware.function_middleware.FunctionMiddleware]

Resolve middleware names to FunctionMiddleware instances.

async _build_function(

name: str,

config: nat.data_models.function.FunctionBaseConfig,

) → ConfiguredFunction

async _build_function_group(

name: str,

config: nat.data_models.function.FunctionGroupBaseConfig,

) → ConfiguredFunctionGroup

Build a function group from the provided configuration.

Args:

name: The name of the function group config: The function group configuration

Returns:

ConfiguredFunctionGroup: The built function group

Raises:

ValueError: If the function group builder returns invalid results

async add_function(

name: str | nat.data_models.component_ref.FunctionRef,

config: nat.data_models.function.FunctionBaseConfig,

) → nat.builder.function.Function

Add a function to the builder.

Args:

name: The name or reference for the function config: The configuration for the function

Returns:

The built function instance

async add_function_group(

name: str | nat.data_models.component_ref.FunctionGroupRef,

config: nat.data_models.function.FunctionGroupBaseConfig,

) → nat.builder.function.FunctionGroup

Add a function group to the builder.

Args:

name: The name or reference for the function group config: The configuration for the function group

Returns:

The built function group instance

_check_backwards_compatibility_function_name(name: str) → str

async get_function(

name: str | nat.data_models.component_ref.FunctionRef,

) → nat.builder.function.Function

Get a function by name.

Args:

name: The name or reference of the function

Returns:

The built function instance

async get_function_group(

name: str | nat.data_models.component_ref.FunctionGroupRef,

) → nat.builder.function.FunctionGroup

Get a function group by name.

Args:

name: The name or reference of the function group

Returns:

The built function group instance

get_function_config(

name: str | nat.data_models.component_ref.FunctionRef,

) → nat.data_models.function.FunctionBaseConfig

Get the configuration for a function.

Args:

name: The name or reference of the function

Returns:

The configuration for the function

get_function_group_config(

name: str | nat.data_models.component_ref.FunctionGroupRef,

) → nat.data_models.function.FunctionGroupBaseConfig

Get the configuration for a function group.

Args:

name: The name or reference of the function group

Returns:

The configuration for the function group

async set_workflow(

config: nat.data_models.function.FunctionBaseConfig,

) → nat.builder.function.Function

Set the workflow function.

Args:

config: The configuration for the workflow function

Returns:

The built workflow function instance

get_workflow() → nat.builder.function.Function

Get the workflow function.

Returns:

The workflow function instance

get_workflow_config() → nat.data_models.function.FunctionBaseConfig

Get the configuration for the workflow.

Returns:

The configuration for the workflow function

get_function_dependencies(

fn_name: str | nat.data_models.component_ref.FunctionRef,

) → nat.data_models.function_dependencies.FunctionDependencies

Get the dependencies for a function.

Args:

fn_name: The name of the function

Returns:

The function dependencies

get_function_group_dependencies(

fn_name: str | nat.data_models.component_ref.FunctionGroupRef,

) → nat.data_models.function_dependencies.FunctionDependencies

Get the dependencies for a function group.

Args:

fn_name: The name of the function group

Returns:

The function group dependencies

async get_tools(

tool_names: collections.abc.Sequence[str | nat.data_models.component_ref.FunctionRef | nat.data_models.component_ref.FunctionGroupRef],

wrapper_type: nat.builder.framework_enum.LLMFrameworkEnum | str,

) → list[Any]

Get multiple tools by name wrapped in the specified framework type.

Args:

tool_names: The names or references of the tools (functions or function groups) wrapper_type: The LLM framework type to wrap the tools in

Returns:

List of tools wrapped in the specified framework type

async get_tool(

fn_name: str | nat.data_models.component_ref.FunctionRef,

wrapper_type: nat.builder.framework_enum.LLMFrameworkEnum | str,

) → Any

Get a tool by name wrapped in the specified framework type.

Args:

fn_name: The name or reference of the tool (function) wrapper_type: The LLM framework type to wrap the tool in

Returns:

The tool wrapped in the specified framework type

async add_llm(

name: str | nat.data_models.component_ref.LLMRef,

config: nat.data_models.llm.LLMBaseConfig,

) → None

Add an LLM to the builder.

Args:

name: The name or reference for the LLM config: The configuration for the LLM

Returns:

The built LLM instance

async get_llm(

llm_name: str | nat.data_models.component_ref.LLMRef,

wrapper_type: nat.builder.framework_enum.LLMFrameworkEnum | str,

) → Any

Get an LLM by name wrapped in the specified framework type.

Args:

llm_name: The name or reference of the LLM wrapper_type: The LLM framework type to wrap the LLM in

Returns:

The LLM wrapped in the specified framework type

get_llm_config(

llm_name: str | nat.data_models.component_ref.LLMRef,

) → nat.data_models.llm.LLMBaseConfig

Get the configuration for an LLM.

Args:

llm_name: The name or reference of the LLM

Returns:

The configuration for the LLM

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,

) → None

Add an embedder to the builder.

Args:

name: The name or reference for the embedder config: The configuration for the embedder

async get_embedder(

embedder_name: str | nat.data_models.component_ref.EmbedderRef,

wrapper_type: nat.builder.framework_enum.LLMFrameworkEnum | str,

)

Get an embedder by name wrapped in the specified framework type.

Args:

embedder_name: The name or reference of the embedder wrapper_type: The LLM framework type to wrap the embedder in

Returns:

The embedder wrapped in the specified framework type

get_embedder_config(

embedder_name: str | nat.data_models.component_ref.EmbedderRef,

) → nat.data_models.embedder.EmbedderBaseConfig

Get the configuration for an embedder.

Args:

embedder_name: The name or reference of the embedder

Returns:

The configuration for the embedder

async add_memory_client(

name: str | nat.data_models.component_ref.MemoryRef,

config: nat.data_models.memory.MemoryBaseConfig,

) → nat.memory.interfaces.MemoryEditor

Add a memory client to the builder.

Args:

name: The name or reference for the memory client config: The configuration for the memory client

Returns:

The built memory client instance

async 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

Get the configuration for a memory client.

Args:

memory_name: The name or reference of the memory client

Returns:

The configuration for the memory client

async add_object_store(

name: str | nat.data_models.component_ref.ObjectStoreRef,

config: nat.data_models.object_store.ObjectStoreBaseConfig,

) → nat.object_store.interfaces.ObjectStore

Add an object store to the builder.

Args:

name: The name or reference for the object store config: The configuration for the object store

Returns:

The built object store instance

async get_object_store_client(

object_store_name: str | nat.data_models.component_ref.ObjectStoreRef,

) → nat.object_store.interfaces.ObjectStore

Get an object store client by name.

Args:

object_store_name: The name or reference of the object store

Returns:

The object store client instance

get_object_store_config(

object_store_name: str | nat.data_models.component_ref.ObjectStoreRef,

) → nat.data_models.object_store.ObjectStoreBaseConfig

Get the configuration for an object store.

Args:

object_store_name: The name or reference of the object store

Returns:

The configuration for the object store

async add_retriever(

name: str | nat.data_models.component_ref.RetrieverRef,

config: nat.data_models.retriever.RetrieverBaseConfig,

) → None

Add a retriever to the builder.

Args:

name: The name or reference for the retriever config: The configuration for the retriever

async get_retriever(

retriever_name: str | nat.data_models.component_ref.RetrieverRef,

wrapper_type: nat.builder.framework_enum.LLMFrameworkEnum | str | None = None,

)

Get a retriever by name.

Args:

retriever_name: The name or reference of the retriever wrapper_type: Optional LLM framework type to wrap the retriever in

Returns:

The retriever instance, optionally wrapped in the specified framework type

async get_retriever_config(

retriever_name: str | nat.data_models.component_ref.RetrieverRef,

) → nat.data_models.retriever.RetrieverBaseConfig

Get the configuration for a retriever.

Args:

retriever_name: The name or reference of the retriever

Returns:

The configuration for the retriever

async add_trainer(

name: str | nat.data_models.component_ref.TrainerRef,

config: nat.data_models.finetuning.TrainerConfig,

) → nat.finetuning.interfaces.finetuning_runner.Trainer

Add a trainer to the builder.

Args:

name: The name or reference for the trainer config: The configuration for the trainer

Returns:

The built trainer instance

async add_trainer_adapter(

name: str | nat.data_models.component_ref.TrainerAdapterRef,

config: nat.data_models.finetuning.TrainerAdapterConfig,

) → nat.finetuning.interfaces.trainer_adapter.TrainerAdapter

Add a trainer adapter to the builder.

Args:

name: The name or reference for the trainer adapter config: The configuration for the trainer adapter

Returns:

The built trainer adapter instance

async add_trajectory_builder(

name: str | nat.data_models.component_ref.TrajectoryBuilderRef,

config: nat.data_models.finetuning.TrajectoryBuilderConfig,

) → nat.finetuning.interfaces.trajectory_builder.TrajectoryBuilder

Add a trajectory builder to the builder.

Args:

name: The name or reference for the trajectory builder config: The configuration for the trajectory builder

Returns:

The built trajectory builder instance

async get_trainer(

trainer_name: str | nat.data_models.component_ref.TrainerRef,

trajectory_builder: nat.finetuning.interfaces.trajectory_builder.TrajectoryBuilder,

trainer_adapter: nat.finetuning.interfaces.trainer_adapter.TrainerAdapter,

) → nat.finetuning.interfaces.finetuning_runner.Trainer

Get a trainer by name with the specified trajectory builder and trainer adapter.

Args:

trainer_name: The name or reference of the trainer trajectory_builder: The trajectory builder instance trainer_adapter: The trainer adapter instance

Returns:

The trainer instance

async get_trainer_config(

trainer_name: str | nat.data_models.component_ref.TrainerRef,

) → nat.data_models.finetuning.TrainerConfig

Get the configuration for a trainer.

Args:

trainer_name: The name or reference of the trainer

Returns:

The configuration for the trainer

async get_trainer_adapter_config(

trainer_adapter_name: str | nat.data_models.component_ref.TrainerAdapterRef,

) → nat.data_models.finetuning.TrainerAdapterConfig

Get the configuration for a trainer adapter.

Args:

trainer_adapter_name: The name or reference of the trainer adapter

Returns:

The configuration for the trainer adapter

async get_trajectory_builder_config(

trajectory_builder_name: str | nat.data_models.component_ref.TrajectoryBuilderRef,

) → nat.data_models.finetuning.TrajectoryBuilderConfig

Get the configuration for a trajectory builder.

Args:

trajectory_builder_name: The name or reference of the trajectory builder

Returns:

The configuration for the trajectory builder

async get_trainer_adapter(

trainer_adapter_name: str | nat.data_models.component_ref.TrainerAdapterRef,

) → nat.finetuning.interfaces.trainer_adapter.TrainerAdapter

Get a trainer adapter by name.

Args:

trainer_adapter_name: The name or reference of the trainer adapter

Returns:

The trainer adapter instance

async get_trajectory_builder(

trajectory_builder_name: str | nat.data_models.component_ref.TrajectoryBuilderRef,

) → nat.finetuning.interfaces.trajectory_builder.TrajectoryBuilder

Get a trajectory builder by name.

Args:

trajectory_builder_name: The name or reference of the trajectory builder

Returns:

The trajectory builder instance

async add_ttc_strategy(

name: str | nat.data_models.component_ref.TTCStrategyRef,

config: nat.data_models.ttc_strategy.TTCStrategyBaseConfig,

) → None

Add a test-time compute strategy to the builder.

Args:

name: The name or reference for the TTC strategy config: The configuration for the TTC strategy

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

Get a test-time compute strategy by name.

Args:

strategy_name: The name or reference of the TTC strategy pipeline_type: The pipeline type for the strategy stage_type: The stage type for the strategy

Returns:

The TTC strategy instance

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 the configuration for a test-time compute strategy.

Args:

strategy_name: The name or reference of the TTC strategy pipeline_type: The pipeline type for the strategy stage_type: The stage type for the strategy

Returns:

The configuration for the TTC strategy

async add_middleware(

name: str | nat.data_models.component_ref.MiddlewareRef,

config: nat.data_models.middleware.MiddlewareBaseConfig,

) → nat.middleware.middleware.Middleware

Add middleware to the builder.

Args:

name: The name or reference for the middleware config: The configuration for the middleware

Returns:

The built middleware instance

Raises:

ValueError: If the middleware already exists

async get_middleware(

middleware_name: str | nat.data_models.component_ref.MiddlewareRef,

) → nat.middleware.middleware.Middleware

Get built middleware by name.

Args:

middleware_name: The name or reference of the middleware

Returns:

The built middleware instance

Raises:

ValueError: If the middleware is not found

get_middleware_config(

middleware_name: str | nat.data_models.component_ref.MiddlewareRef,

) → nat.data_models.middleware.MiddlewareBaseConfig

Get the configuration for middleware.

Args:

middleware_name: The name or reference of the middleware

Returns:

The configuration for the middleware

Raises:

ValueError: If the middleware is not found

get_user_manager()

Get the user manager holder.

Returns:

The user manager holder instance

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

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.

_validate_dependencies(config: nat.data_models.config.Config)

Validate no shared component has dependencies on any per-user components.

This prevents invalid configurations where shared components try to use per-user functions that do not exist at shared builder initialization time.

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

Async: