agi-node API

Path handling

Workers do not resolve dataset or workspace paths themselves. Path normalisation lives in agi_node.agi_dispatcher.base_worker_path_support and is reached through private BaseWorker helpers (_normalized_path, _share_root_path, _resolve_data_dir, _resolved_data_roots). There is no public normalisation entry point: BaseWorker applies it for you from setup_args, from_toml, and prepare_output_dir.

What that resolution guarantees:

• Relative values are resolved against the share root, so configuration files can ship entries such as <app>/dataset and still land on the correct worker share on every host.

• UNC-style shares on Windows (for example \\server\share) keep their double backslashes.

• Symlinks and bind mounts created by installers are preserved, and directories that only appear later during the run are still accepted.

The conventional argument fields are data_in and data_out. data_uri was the earlier name for data_in; it survives only as a legacy alias that app argument models migrate on load — see src/agilab/apps/builtin/minimal_app_project/src/minimal_app/app_args.py. Do not introduce it in new code.

Note

Earlier revisions of this page described a public BaseWorker.normalize_data_uri helper and a self.args.data_uri field. Neither exists. Call setup_args and let it resolve paths for you.

Argument helpers

Recent updates to BaseWorker standardise how workers load, merge, and persist their argument models. Every subclass can opt into the following hooks:

• default_settings_path and default_settings_section control the TOML source used by from_toml / to_toml.

• args_loader and args_merger are callables that fetch and combine raw settings with user overrides before instantiating the worker.

• args_ensure_defaults lets workers patch derived values (for example, normalising paths) after the merge but before instantiation.

• args_dumper and args_dump_mode define how to_toml emits the active configuration, enabling round-trips back into app_settings.toml.

If these helpers live in the worker module (for example load_args or dump_args defined alongside the class) or inside a sibling *_args/app_args module, BaseWorker auto-binds them during class creation. That lets most apps drop the explicit args_loader = … boilerplate while still allowing manual overrides for custom integrations.

Managed PC path remapping

• managed_pc_path_fields lists argument attributes that should be remapped to the managed-machine workspace (~/MyApp by default) when AgiEnv reports a managed PC.

• managed_pc_home_suffix customises the managed workspace folder name if your deployment uses something other than MyApp.

• BaseWorker.from_toml applies the remapping automatically; when instantiating a worker manually, use setup_args to apply defaults and remap paths in a single call.

• setup_args optionally accepts output_field (e.g. "data_uri") along with output_subdir, output_attr, output_clean and output_parents_up so managers can prepare their output directories without repeating boilerplate.

Output directory helpers

• prepare_output_dir centralises the setup of manager-side output folders (subdirectory dataframe by default). Hand it the base path you want to target and it resolves the path through the share resolver, clears old contents when clean is true (the default, and what setup_args passes as output_clean), creates the directory, and stores it on self.data_out unless you override attribute.

• It is also a validation boundary: it raises ValueError for a drive-relative root, for .. traversal in either the root or the subdirectory, and for a subdirectory that is absolute. Catch it if your manager accepts an operator-supplied output path.

With these attributes in place, BaseWorker.from_toml produces a configured instance and BaseWorker.to_toml writes the updated schema without each app copying boilerplate. BaseWorker.as_dict exposes a serialisable payload for Web pages and API consumers, while _extend_payload stays available for apps that need to enrich the exported structure.

Reference

This page documents the public worker foundation and the concrete worker types. Operational build and hook entry points such as build, pre_install, and post_install remain covered in the runbook because they are packaging tooling rather than the main API surface extended by app authors.

base_worker

node module

Auteur: Jean-Pierre Morard

class agi_node.agi_dispatcher.base_worker.ArgsNamespace(mapping_or_iterable=(), /, **kwargs)

Bases: SimpleNamespace

Namespace that supports both attribute and key-style access.

get(key, default=None)

to_dict()

class agi_node.agi_dispatcher.base_worker.BaseWorker

Bases: ArtifactContract, ABC

class BaseWorker v1.0

args_dump_mode = 'json'

args_dumper = None

args_ensure_defaults = None

args_loader = None

args_merger = None

as_dict(mode=None)

Return type:

dict[str, Any]

break()

Signal the service loop to exit on this worker.

Return type:

bool

static break_loop()

Signal the service loop to exit on this worker.

Return type:

bool

default_settings_path = 'app_settings.toml'

default_settings_section = 'args'

distribution_cache_inputs()

Return additional filesystem inputs that determine the work plan.

WorkDispatcher automatically fingerprints conventional argument fields such as data_in and submission_inbox. Applications whose planner reads other filesystem locations can expose those roots here without changing the stable build_distribution(workers) signature. Set distribution_cache_inputs_mode to "replace" when these declared paths are the complete planner input set.

Return type:

Iterable[Path | str]

distribution_cache_inputs_mode = 'augment'

env = None

static expand(path, base_directory=None)

Expand a given path to an absolute path. :type path: str :param path: The path to expand. :type path: str :type base_directory: str, optional :param base_directory: The base directory to use for expanding the path. Defaults to None. :type base_directory: str, optional

Returns:

The expanded absolute path.

Return type:

str

Raises:

None –

Note

This method handles both Unix and Windows paths and expands ‘~’ notation to the user’s home directory.

static expand_and_join(path1, path2)

Join two paths after expanding the first path.

Parameters:

• path1 (str) – The first path to expand and join.

• path2 (str) – The second path to join with the expanded first path.

Returns:

The joined path.

Return type:

str

classmethod from_toml(env, settings_path=None, section=None, **overrides)

Return type:

BaseWorker

static loop(*, poll_interval=None)

Run a long-lived service loop on this worker until signalled to stop.

The derived worker can implement a loop method accepting either zero arguments or a single stop_event argument. When the method signature accepts stop_event (keyword stop_event or should_stop), the worker implementation is responsible for honouring the event. Otherwise the base implementation repeatedly invokes the method and sleeps for the configured poll interval between calls. Returning False from the worker method requests termination of the loop.

Return type:

Dict[str, Any]

managed_pc_home_suffix = 'MyApp'

managed_pc_path_fields = ()

static normalize_dataset_path(data_path)

Normalise any dataset directory input so workers rely on consistent paths.

Return type:

str

pool_init(worker_vars)

Per-pool-child initializer hook; default is a no-op.

Return type:

None

pool_vars = None

Shared state handed to pool_init in every pool child. Apps usually set self.pool_vars = {"args": self.args} in start(). For process-based worker families (pandas/fireducks) the worker instance and pool_vars must be picklable.

prepare_output_dir(root, *, subdir='dataframe', attribute='data_out', clean=True)

Create (and optionally reset) a deterministic output directory.

Return type:

Path

classmethod resolve_generated_artifact_path(data_in_root, data_out_root, artifact_path)

Return type:

Path

classmethod resolve_input_folder(env, dataset_root, relative_dir, *, descriptor, fallback_subdirs=(), dataset_namespace=None, min_files=1, patterns=None, required_label='data files')

Return type:

Path

setup_args(args, *, env=None, error=None, output_field=None, output_subdir='dataframe', output_attr='data_out', output_clean=True, output_parents_up=0)

Return type:

Any

setup_data_directories(*, source_path, target_path=None, target_subdir='dataframe', reset_target=False)

Prepare normalised input/output dataset paths without relying on worker args.

Returns a namespace with the resolved input path (input_path), the normalised string used by downstream readers (normalized_input), the output directory as a Path (output_path), and its normalised string representation (normalized_output). Optionally clears and recreates the output directory.

Return type:

SimpleNamespace

static start(worker_inst)

Invoke the concrete worker’s start hook once initialised.

stop()

Returns:

to_toml(settings_path=None, section=None, create_missing=True)

Return type:

None

verbose = 1

work_init()

Per-works()-call initialization hook; default is a no-op.

Return type:

None

dag_worker

class agi_node.dag_worker.dag_worker.DagWorker

Bases: BaseWorker

Minimal-change DAG worker:

• Keeps your existing structure

• Adds a tiny signature-aware _invoke() so custom methods can vary in signature

• Uses _invoke() at the single call site in ._exec_multi_process()

get_work(fn_name, args, prev_result)

Back-compat: delegate to the signature-aware invoker.

works(workers_plan, workers_plan_metadata)

Execute the DAG plan and return this call’s elapsed seconds.

DagWorker intentionally ignores the pool/dask mode bits: stages are always scheduled on the in-worker thread pool because dependency ordering (not the ORCHESTRATE pool toggle) drives concurrency here.

Return type:

float

pandas_worker

pandas_worker Framework Callback Functions Module

This module provides the PandasWorker class, which extends the foundational functionalities of BaseWorker for processing data using multiprocessing or single-threaded approaches with pandas.

Classes:

PandasWorker: Worker class for data processing tasks using pandas.

Internal Libraries:

os

External Libraries:

concurrent.futures.ProcessPoolExecutor pathlib.Path pandas as pd BaseWorker from node import BaseWorker.node

class agi_node.pandas_worker.pandas_worker.PandasWorker

Bases: BaseWorker

PandasWorker Class

Inherits from BaseWorker to provide extended data processing functionalities using pandas.

verbose

Verbosity level for logging.

Type:

int

data_out

Path to the output directory.

Type:

str

worker_id

Identifier for the worker instance.

Type:

int

args

Configuration arguments for the worker.

Type:

dict

work_done(df=None)

Handles the post-processing of the DataFrame after work_pool execution.

Parameters:

df (DataFrame) – The pandas DataFrame to process. Defaults to None.

Raises:

ValueError – If an unsupported output format is specified.

Return type:

None

work_pool(x=None)

Processes a single task.

Parameters:

x (any) – The task to process. Defaults to None.

Returns:

A pandas DataFrame with the processed results.

Return type:

DataFrame

works(workers_plan, workers_plan_metadata)

Executes worker tasks based on the distribution tree.

Parameters:

• workers_plan (any) – Distribution tree structure.

• workers_plan_metadata (any) – Additional information about the workers.

Returns:

Execution time of this works() call in seconds.

Return type:

float

polars_worker

data_worker Framework Callback Functions Module

This module provides the PolarsWorker class, which extends the foundational functionalities of BaseWorker for processing data using a thread pool or single-threaded approaches. The pool path deliberately uses threads rather than processes: polars releases the GIL in its native kernels, so threads parallelise IO/native work without process spawn and pickling costs.

Classes:

PolarsWorker: Worker class for data processing tasks.

External Libraries:

concurrent.futures.ThreadPoolExecutor pathlib.Path polars as pl BaseWorker from node import BaseWorker

class agi_node.polars_worker.polars_worker.PolarsWorker

Bases: BaseWorker

PolarsWorker Class

Inherits from BaseWorker to provide extended data processing functionalities.

verbose

Verbosity level for logging.

Type:

int

data_out

Path to the output directory.

Type:

str

worker_id

Identifier for the worker instance.

Type:

int

args

Configuration arguments for the worker.

Type:

dict

work_done(df=None)

Handles the post-processing of the DataFrame after work_pool execution.

Parameters:

df (DataFrame) – The Polars DataFrame to process. Defaults to None.

Raises:

ValueError – If an unsupported output format is specified.

Return type:

None

work_pool(x=None)

Processes a single task.

Parameters:

x (any) – The task to process. Defaults to None.

Returns:

A Polars DataFrame with the processed results.

Return type:

DataFrame

works(workers_plan, workers_plan_metadata)

Executes worker tasks based on the distribution tree.

Parameters:

• workers_plan (any) – Distribution tree structure.

• workers_plan_metadata (any) – Additional information about the workers.

Returns:

Execution time of this works() call in seconds.

Return type:

float