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#
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#
_check_backwards_compatibility_function_name(name: str) → str#
async 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 add_function_group(
name: str | nat.data_models.component_ref.FunctionGroupRef,
config: nat.data_models.function.FunctionGroupBaseConfig,
) → nat.builder.function.FunctionGroup#
async get_function_group(
name: str | nat.data_models.component_ref.FunctionGroupRef,
) → nat.builder.function.FunctionGroup#
get_function_group_config(
name: str | nat.data_models.component_ref.FunctionGroupRef,
) → nat.data_models.function.FunctionGroupBaseConfig#
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_function_group_dependencies(
fn_name: str | nat.data_models.component_ref.FunctionGroupRef,
) → nat.data_models.function_dependencies.FunctionDependencies#
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]#
async get_tool(
fn_name: str | nat.data_models.component_ref.FunctionRef,
wrapper_type: nat.builder.framework_enum.LLMFrameworkEnum | str,
) → Any#
async add_llm(name: str, config: nat.data_models.llm.LLMBaseConfig) → None#
async get_llm(
llm_name: str,
wrapper_type: nat.builder.framework_enum.LLMFrameworkEnum | str,
) → Any#
get_llm_config(llm_name: str) → nat.data_models.llm.LLMBaseConfig#
async add_auth_provider(
name: str,
config: nat.data_models.authentication.AuthProviderBaseConfig,
) → nat.authentication.interfaces.AuthProviderBase#
async get_auth_provider(
auth_provider_name: str,
) → nat.authentication.interfaces.AuthProviderBase#
async add_embedder(
name: str,
config: nat.data_models.embedder.EmbedderBaseConfig,
) → None#
async get_embedder(
embedder_name: str,
wrapper_type: nat.builder.framework_enum.LLMFrameworkEnum | str,
) → Any#
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#
async get_memory_client(
memory_name: str,
) → nat.memory.interfaces.MemoryEditor#
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,
) → nat.object_store.interfaces.ObjectStore#
async get_object_store_client(
object_store_name: str,
) → nat.object_store.interfaces.ObjectStore#
get_object_store_config(
object_store_name: str,
) → nat.data_models.object_store.ObjectStoreBaseConfig#
async add_retriever(
name: str | nat.data_models.component_ref.RetrieverRef,
config: nat.data_models.retriever.RetrieverBaseConfig,
) → None#
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#
async get_retriever_config(
retriever_name: str | nat.data_models.component_ref.RetrieverRef,
) → nat.data_models.retriever.RetrieverBaseConfig#
async add_ttc_strategy(
name: str | nat.data_models.component_ref.TTCStrategyRef,
config: nat.data_models.ttc_strategy.TTCStrategyBaseConfig,
) → None#
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() → nat.builder.builder.UserManagerHolder#
async add_middleware(
name: str | nat.data_models.component_ref.MiddlewareRef,
config: nat.data_models.middleware.MiddlewareBaseConfig,
) → nat.middleware.middleware.Middleware#
async get_middleware(
middleware_name: str | nat.data_models.component_ref.MiddlewareRef,
) → nat.middleware.middleware.Middleware#
get_middleware_config(
middleware_name: str | nat.data_models.component_ref.MiddlewareRef,
) → nat.data_models.middleware.MiddlewareBaseConfig#
async add_trainer(
name: str | nat.data_models.component_ref.TrainerRef,
config: nat.data_models.finetuning.TrainerConfig,
) → nat.finetuning.interfaces.finetuning_runner.Trainer#
async add_trainer_adapter(
name: str | nat.data_models.component_ref.TrainerAdapterRef,
config: nat.data_models.finetuning.TrainerAdapterConfig,
) → nat.finetuning.interfaces.trainer_adapter.TrainerAdapter#
async add_trajectory_builder(
name: str | nat.data_models.component_ref.TrajectoryBuilderRef,
config: nat.data_models.finetuning.TrajectoryBuilderConfig,
) → nat.finetuning.interfaces.trajectory_builder.TrajectoryBuilder#
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#
async get_trainer_adapter(
trainer_adapter_name: str | nat.data_models.component_ref.TrainerAdapterRef,
) → nat.finetuning.interfaces.trainer_adapter.TrainerAdapter#
async get_trajectory_builder(
trajectory_builder_name: str | nat.data_models.component_ref.TrajectoryBuilderRef,
) → nat.finetuning.interfaces.trajectory_builder.TrajectoryBuilder#
async get_trainer_config(
trainer_name: str | nat.data_models.component_ref.TrainerRef,
) → nat.data_models.finetuning.TrainerConfig#
async get_trainer_adapter_config(
trainer_adapter_name: str | nat.data_models.component_ref.TrainerAdapterRef,
) → nat.data_models.finetuning.TrainerAdapterConfig#
async get_trajectory_builder_config(
trajectory_builder_name: str | nat.data_models.component_ref.TrajectoryBuilderRef,
) → nat.data_models.finetuning.TrajectoryBuilderConfig#
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