nat.builder.per_user_workflow_builder

Attributes

logger

Classes

PerUserWorkflowBuilder

Builder for per-user components that are lazily instantiated.

Module Contents

logger

class PerUserWorkflowBuilder(

user_id: str,

shared_builder: nat.builder.workflow_builder.WorkflowBuilder,

registry: nat.cli.type_registry.TypeRegistry | None = None,

)

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

Builder for per-user components that are lazily instantiated.

This builder is created per-user and only builds functions/function_groups that are marked as per-user. It delegates to a shared WorkflowBuilder for all shared components (LLMs, embedders, memory, etc.).

Lifecycle: - Created when a user first makes a request - Kept alive while the user is active - Cleaned up after user inactivity timeout

_user_id

_shared_builder

_workflow: nat.builder.workflow_builder.ConfiguredFunction | None = None

_registry = None

_per_user_functions: dict[str, nat.builder.workflow_builder.ConfiguredFunction]

_per_user_function_groups: dict[str, nat.builder.workflow_builder.ConfiguredFunctionGroup]

_exit_stack: contextlib.AsyncExitStack | None = None

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

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

completed_components: list[tuple[str, str]]

remaining_components: list[tuple[str, str]]

_get_exit_stack() → contextlib.AsyncExitStack

property sync_builder: nat.builder.sync_builder.SyncBuilder

Get the synchronous version of the builder.

Returns:

The SyncBuilder object (synchronous wrapper).

property user_id: str

async _resolve_middleware_instances_from_shared_builder(

middleware_names: collections.abc.Sequence[str],

component_type: str = 'function',

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

Resolve middleware names to FunctionMiddleware instances from the shared builder.

async _build_per_user_function(

name: str,

config: nat.data_models.function.FunctionBaseConfig,

) → nat.builder.workflow_builder.ConfiguredFunction

async _build_per_user_function_group(

name: str,

config: nat.data_models.function.FunctionGroupBaseConfig,

) → nat.builder.workflow_builder.ConfiguredFunctionGroup

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

_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

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

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

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

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

config: nat.data_models.authentication.AuthProviderBaseConfig,

) → nat.authentication.interfaces.AuthProviderBase

Add an authentication provider to the builder.

Args:

name: The name or reference for the authentication provider config: The configuration for the authentication provider

Returns:

The built authentication provider instance

async get_auth_provider(

auth_provider_name: str,

) → nat.authentication.interfaces.AuthProviderBase

Get an authentication provider by name.

Args:

auth_provider_name: The name or reference of the authentication provider

Returns:

The authentication provider instance

async add_embedder(

name: str,

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,

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

) → Any

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

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.memory.interfaces.MemoryEditor

Return the instantiated memory client for the given name.

get_memory_client_config(

memory_name: str,

) → 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,

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

) → nat.retriever.interface.Retriever

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

get_user_manager() → nat.builder.builder.UserManagerHolder

Get the user manager holder.

Returns:

The user manager holder instance

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

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

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

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_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 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 populate_builder(

config: nat.data_models.config.Config,

skip_workflow: bool = False,

)

Populate the per-user builder with per-user components from config.

Only builds components that are marked as per-user. Builds in dependency order to handle per-user functions depending on other per-user functions.

Args:

config: The full configuration object skip_workflow: If True, skips the workflow instantiation step. Defaults to False.

Raises:

ValueError: If a per-user component has invalid dependencies

async build(

entry_function: str | None = None,

) → nat.builder.workflow.Workflow

Creates a workflow instance for this specific user.

Combines per-user functions with shared components from the shared builder.

Parameters

entry_functionstr | None, optional

The function name to use as the entry point. If None, uses the workflow. By default None

Returns

Workflow

A per-user workflow instance

Raises

ValueError

If no workflow is set (neither per-user nor shared)

classmethod from_config(

user_id: str,

config: nat.data_models.config.Config,

shared_builder: nat.builder.workflow_builder.WorkflowBuilder,

)

Async:

Create and populate a PerUserWorkflowBuilder from config.

This is the primary entry point for creating per-user builders.

Args:

user_id: Unique identifier for the user config: Full configuration object shared_builder: The shared WorkflowBuilder instance

Yields:

PerUserWorkflowBuilder: Populated per-user builder instance