Skip to content

Meta Endpoint

Meta Endpoints

A MetaEndpoint is an Endpoint backed by a DAG of other endpoints. See Meta Endpoints for the concept overview.

MetaEndpoint subclasses Endpoint, so wrapping an existing deployed meta endpoint by name works identically. Use create() to build and deploy a new meta endpoint from a DAG.

MetaEndpoint: An Endpoint backed by a directed acyclic graph (DAG) of child endpoints and aggregation nodes.

A MetaEndpoint behaves identically to a regular Endpoint at runtime — callers do endpoint.inference(df) and get a DataFrame back. The DAG machinery is server-side: the deployed container loads the serialized DAG, dispatches each child invocation to fast_inference (sync) or async_inference (async), and runs aggregation nodes locally.

Common usage::

from workbench.api import MetaEndpoint
from workbench.utils.meta_endpoint_dag import MetaEndpointDAG
from workbench.utils.aggregation_nodes import Concat

dag = MetaEndpointDAG()
dag.add_endpoint("smiles-to-2d-v1")
dag.add_endpoint("smiles-to-3d-v1")
dag.add_aggregation(Concat(name="combine"))
dag.add_edge("smiles-to-2d-v1", "combine")
dag.add_edge("smiles-to-3d-v1", "combine")
dag.set_input_node("smiles-to-2d-v1", "smiles-to-3d-v1")
dag.set_output_node("combine")

end = MetaEndpoint.create(name="my-features-meta", dag=dag)
# Input does not need any id column — the DAG handles row alignment internally.
df = end.inference(input_df)

If any child endpoint in the DAG is async (e.g. smiles-to-3d-v1), the MetaEndpoint is automatically deployed as async too — its invocation budget needs to accommodate the slowest child.

MetaEndpoint

Bases: Endpoint

Endpoint backed by a :class:MetaEndpointDAG.

Constructor wraps an existing deployed MetaEndpoint by name, identical to :class:Endpoint. Use :meth:create to build and deploy a new one from a DAG.

Source code in src/workbench/api/meta_endpoint.py 
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
class MetaEndpoint(Endpoint):
    """Endpoint backed by a :class:`MetaEndpointDAG`.

    Constructor wraps an existing deployed MetaEndpoint by name, identical
    to :class:`Endpoint`. Use :meth:`create` to build and deploy a new one
    from a DAG.
    """

    @classmethod
    def create(
        cls,
        name: str,
        dag: MetaEndpointDAG,
        description: str | None = None,
        tags: list[str] | None = None,
        min_instances: int = 0,
        max_instances: int = 1,
    ) -> "MetaEndpoint":
        """Build, register, and deploy a MetaEndpoint from a DAG.

        Steps:
          1. Validate the DAG; populate per-endpoint async flags.
          2. Backtrace lineage from a primary endpoint to satisfy
             Workbench's Model machinery (FeatureSet, target, features).
          3. Run the standard ``FeatureSet.to_model()`` flow, passing the
             DAG / region / bucket as ``custom_args`` so the meta-endpoint
             template fills them in at training time.
          4. Deploy the endpoint (async if any DAG child is async) and
             stash the serialized DAG on it.

        Args:
            name: Endpoint / Model name.
            dag: A :class:`MetaEndpointDAG` describing the data flow.
            description: Optional description for the registered model.
            tags: Optional list of Workbench tags.
            min_instances: Autoscaler floor for async deployments (default: 0 —
                scale to zero when idle). Set to 1 in production to keep the
                meta warm. Ignored for sync deployments.
            max_instances: Autoscaler ceiling for async deployments (default: 1).
                The meta is a thin orchestrator that fans out to children, so
                it rarely needs more than one instance. Ignored for sync
                deployments.

        Returns:
            The deployed MetaEndpoint, ready for ``.inference()``.
        """
        if not Artifact.is_name_valid(name, delimiter="-", lower_case=False):
            raise ValueError(f"Invalid MetaEndpoint name: '{name}' (use only alphanumerics and '-')")

        log.important(f"Validating DAG for MetaEndpoint '{name}'...")
        dag.validate()
        dag.populate_child_metadata()
        is_async = dag.has_async_endpoint()
        log.important(
            f"DAG: {len(dag.endpoints)} endpoints, {len(dag.aggregations)} aggregation nodes "
            f"({'async' if is_async else 'sync'} deployment)"
        )

        # An async meta sizes its batch from the child fleet (see the batch
        # computation below), which requires every async child to have advertised
        # its `inference_batch_size` + `max_instances` — both stamped on the
        # child's `workbench_meta` when its deploy script completes. Fail fast,
        # before deploying anything, if a child isn't fully deployed yet: the
        # children must come up first.
        if is_async:
            unstamped = [
                ep
                for ep, child_is_async in dag.endpoint_async_flags.items()
                if child_is_async and (ep not in dag.endpoint_batch_sizes or ep not in dag.endpoint_max_instances)
            ]
            if unstamped:
                raise RuntimeError(
                    f"MetaEndpoint '{name}': async child endpoint(s) {unstamped} have not advertised "
                    f"'inference_batch_size' / 'max_instances' — they aren't fully deployed yet. Deploy "
                    f"the child endpoint(s) first (their deploy script stamps these on completion), then "
                    f"create this meta. The meta sizes its async batch from the child fleet, so it cannot "
                    f"be built before its children."
                )

        # Backtrace lineage from a primary endpoint to satisfy Workbench Model
        # machinery (every Model needs a FeatureSet to hang off of).
        feature_list, feature_set_name, target_column = cls._derive_lineage(dag)

        # Build the model via the standard FeatureSet → Model flow. The
        # meta-endpoint template's `{{dag}}`, `{{aws_region}}`, `{{s3_bucket}}`
        # placeholders are filled from custom_args.
        aws_clamp = AWSAccountClamp()
        sm_session = aws_clamp.sagemaker_session()
        workbench_bucket = ConfigManager().get_config("WORKBENCH_BUCKET")

        feature_set = FeatureSet(feature_set_name)
        feature_set.to_model(
            name=name,
            model_type=cls._derive_model_type(dag),
            model_framework=ModelFramework.META,
            tags=tags or [name],
            description=description or f"MetaEndpoint DAG over: {', '.join(dag.endpoints.values())}",
            target_column=target_column,
            feature_list=feature_list,
            custom_args={
                "dag": dag.to_dict(),
                "aws_region": sm_session.boto_region_name,
                "s3_bucket": workbench_bucket,
            },
        )

        # MetaEndpoint containers are thin orchestrators — they receive the
        # full input as one invocation and let each child's own async_inference
        # do the row-level fan-out against the child fleet.
        log.important(f"Deploying MetaEndpoint '{name}' ({'async' if is_async else 'sync'})...")
        model = Model(name)
        endpoint = model.to_endpoint(
            tags=tags or [name],
            async_endpoint=is_async,
            min_instances=min_instances if is_async else 0,
            max_instances=max_instances if is_async else None,
        )

        # The meta delegates row-level batching to its children, who saturate
        # their own fleets; size the async meta batch to the slowest child's
        # fleet: smallest async-child batch × that fleet's max_instances × 4
        # (≈4 rounds through the fleet). Smaller batches → shorter invocations
        # (less likely to hit the 60-min async cap) and finer-grained failure/
        # retry. All-sync metas stay large (rows are cheap). Child stamps are
        # guaranteed present here — validated right after populate_child_metadata.
        meta_batch = 500
        if is_async:
            async_children = [ep for ep, child_is_async in dag.endpoint_async_flags.items() if child_is_async]
            async_batches = [dag.endpoint_batch_sizes[ep] for ep in async_children]
            async_fleets = [dag.endpoint_max_instances[ep] for ep in async_children]
            meta_batch = min(async_batches) * max(async_fleets) * 4
        endpoint.upsert_workbench_meta(
            {
                "inference_batch_size": meta_batch,
                "meta_endpoint_dag": dag.to_dict(),
            }
        )

        log.important(f"MetaEndpoint '{name}' created successfully!")
        return cls(name)

    # ------------------------------------------------------------------
    # Internal helpers
    # ------------------------------------------------------------------

    @classmethod
    def _derive_model_type(cls, dag: MetaEndpointDAG) -> ModelType:
        """Pick the most accurate :class:`ModelType` for the DAG's output.

        - Output node is a terminal endpoint → borrow that endpoint's declared
          type (e.g., a downstream predictor endpoint contributes its own type).
        - Output node is :class:`~workbench.utils.aggregation_nodes.Concat` →
          ``TRANSFORMER`` (column-union of feature outputs).
        - Output node is :class:`~workbench.utils.aggregation_nodes.Vote` →
          ``CLASSIFIER`` (majority vote of class labels).
        - Output node is any other prediction aggregator → ``REGRESSOR``.
        """
        from workbench.utils.aggregation_nodes import Concat, Vote

        output_name = dag.output_node
        if output_name in dag.endpoints:
            return Model(dag.endpoints[output_name]).model_type

        agg = dag.aggregations[output_name]
        if isinstance(agg, Concat):
            return ModelType.TRANSFORMER
        if isinstance(agg, Vote):
            return ModelType.CLASSIFIER
        return ModelType.REGRESSOR

    @classmethod
    def _derive_lineage(cls, dag: MetaEndpointDAG) -> tuple[list[str], str, str | None]:
        """Derive a (feature_list, feature_set_name, target_column) tuple for the meta.

        Workbench Models need to trace back to a FeatureSet, and downstream
        tooling (``test_inference``, ``register_input_columns`` /
        ``register_output_columns``) all anchor on
        ``model.features()`` + ``model.get_input()``. For DAG-based
        MetaEndpoints we satisfy that contract by:

          - ``feature_list`` / ``feature_set_name`` — borrowed from the first
            input endpoint, since the meta consumes that endpoint's input
            columns and that endpoint's FeatureSet is guaranteed to contain
            usable smoke-test data.
          - ``target_column`` — :meth:`MetaEndpointDAG.terminal_target`,
            which walks back from the output node to the closest predictor
            endpoint(s) so mixed DAGs (smiles → features → predictor) report
            what the meta actually predicts, not what the input endpoint
            happens to produce.
        """
        if not dag.input_nodes:
            raise ValueError("DAG has no input nodes — cannot derive lineage")
        primary_endpoint_name = dag.endpoints[dag.input_nodes[0]]

        ep = Endpoint(primary_endpoint_name)
        if not ep.exists():
            raise ValueError(f"Primary endpoint '{primary_endpoint_name}' does not exist")

        primary_model = Model(ep.get_input())
        feature_list = primary_model.features() or list(dag.input_columns())
        feature_set_name = primary_model.get_input()
        # Target reflects what the DAG ultimately predicts (walks back from
        # the output node), not what the input endpoint happens to produce.
        # For mixed DAGs (smiles → features → predictor), this surfaces the
        # downstream predictor's target instead of the feature endpoint's None.
        target_column = dag.terminal_target()

        log.info(
            f"Lineage anchor: {primary_endpoint_name} -> {primary_model.name} -> {feature_set_name} "
            f"(target: {target_column})"
        )
        return feature_list, feature_set_name, target_column

    def get_dag(self) -> MetaEndpointDAG:
        """Reconstruct the MetaEndpointDAG from this endpoint's stored metadata."""
        meta = self.workbench_meta() or {}
        dag_dict = meta.get("meta_endpoint_dag")
        if not dag_dict:
            raise ValueError(
                f"MetaEndpoint '{self.name}' has no DAG in workbench_meta. Recreate via MetaEndpoint.create()."
            )
        return MetaEndpointDAG.from_dict(dag_dict)

    def run_dag_test(self, input_df: pd.DataFrame) -> pd.DataFrame:
        """Execute the DAG client-side against ``input_df``.

        Bypasses the deployed meta endpoint entirely: each child endpoint is
        invoked directly via the regular ``Endpoint(name).inference()`` API
        from this process. Useful for debugging the DAG topology, isolating
        which child is misbehaving, or running the DAG when the deployed
        meta endpoint is unavailable.

        Result is identical to ``self.inference(input_df)`` modulo transport
        and any container-only side effects (data capture, etc.).
        """
        return self.get_dag().run(input_df)

    def purge_async_queue(self) -> int:
        """Cancel queued async invocations for the meta and every async child.

        When a meta-level batch job is killed, orphaned work can be sitting
        at two layers: invocations queued at the meta endpoint, and child
        invocations queued by in-flight meta predict_fns. This override
        purges both so the entire DAG drains.

        Returns:
            int: Total number of staged input objects deleted across the
            meta and its async children.
        """
        total = super().purge_async_queue()
        dag = self.get_dag()
        for child_name in set(dag.endpoints.values()):
            if dag.endpoint_async_flags.get(child_name):
                total += Endpoint(child_name).purge_async_queue()
        return total

create(name, dag, description=None, tags=None, min_instances=0, max_instances=1) classmethod

Build, register, and deploy a MetaEndpoint from a DAG.

Steps
  1. Validate the DAG; populate per-endpoint async flags.
  2. Backtrace lineage from a primary endpoint to satisfy Workbench's Model machinery (FeatureSet, target, features).
  3. Run the standard FeatureSet.to_model() flow, passing the DAG / region / bucket as custom_args so the meta-endpoint template fills them in at training time.
  4. Deploy the endpoint (async if any DAG child is async) and stash the serialized DAG on it.

Parameters:

Name Type Description Default
name str

Endpoint / Model name.

 required
dag MetaEndpointDAG

A :class:MetaEndpointDAG describing the data flow.

 required
description str | None

Optional description for the registered model.

 None
tags list[str] | None

Optional list of Workbench tags.

 None
min_instances int

Autoscaler floor for async deployments (default: 0 — scale to zero when idle). Set to 1 in production to keep the meta warm. Ignored for sync deployments.

 0
max_instances int

Autoscaler ceiling for async deployments (default: 1). The meta is a thin orchestrator that fans out to children, so it rarely needs more than one instance. Ignored for sync deployments.

 1

Returns:

Type Description
'MetaEndpoint'

The deployed MetaEndpoint, ready for .inference().
Source code in src/workbench/api/meta_endpoint.py 
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
@classmethod
def create(
    cls,
    name: str,
    dag: MetaEndpointDAG,
    description: str | None = None,
    tags: list[str] | None = None,
    min_instances: int = 0,
    max_instances: int = 1,
) -> "MetaEndpoint":
    """Build, register, and deploy a MetaEndpoint from a DAG.

    Steps:
      1. Validate the DAG; populate per-endpoint async flags.
      2. Backtrace lineage from a primary endpoint to satisfy
         Workbench's Model machinery (FeatureSet, target, features).
      3. Run the standard ``FeatureSet.to_model()`` flow, passing the
         DAG / region / bucket as ``custom_args`` so the meta-endpoint
         template fills them in at training time.
      4. Deploy the endpoint (async if any DAG child is async) and
         stash the serialized DAG on it.

    Args:
        name: Endpoint / Model name.
        dag: A :class:`MetaEndpointDAG` describing the data flow.
        description: Optional description for the registered model.
        tags: Optional list of Workbench tags.
        min_instances: Autoscaler floor for async deployments (default: 0 —
            scale to zero when idle). Set to 1 in production to keep the
            meta warm. Ignored for sync deployments.
        max_instances: Autoscaler ceiling for async deployments (default: 1).
            The meta is a thin orchestrator that fans out to children, so
            it rarely needs more than one instance. Ignored for sync
            deployments.

    Returns:
        The deployed MetaEndpoint, ready for ``.inference()``.
    """
    if not Artifact.is_name_valid(name, delimiter="-", lower_case=False):
        raise ValueError(f"Invalid MetaEndpoint name: '{name}' (use only alphanumerics and '-')")

    log.important(f"Validating DAG for MetaEndpoint '{name}'...")
    dag.validate()
    dag.populate_child_metadata()
    is_async = dag.has_async_endpoint()
    log.important(
        f"DAG: {len(dag.endpoints)} endpoints, {len(dag.aggregations)} aggregation nodes "
        f"({'async' if is_async else 'sync'} deployment)"
    )

    # An async meta sizes its batch from the child fleet (see the batch
    # computation below), which requires every async child to have advertised
    # its `inference_batch_size` + `max_instances` — both stamped on the
    # child's `workbench_meta` when its deploy script completes. Fail fast,
    # before deploying anything, if a child isn't fully deployed yet: the
    # children must come up first.
    if is_async:
        unstamped = [
            ep
            for ep, child_is_async in dag.endpoint_async_flags.items()
            if child_is_async and (ep not in dag.endpoint_batch_sizes or ep not in dag.endpoint_max_instances)
        ]
        if unstamped:
            raise RuntimeError(
                f"MetaEndpoint '{name}': async child endpoint(s) {unstamped} have not advertised "
                f"'inference_batch_size' / 'max_instances' — they aren't fully deployed yet. Deploy "
                f"the child endpoint(s) first (their deploy script stamps these on completion), then "
                f"create this meta. The meta sizes its async batch from the child fleet, so it cannot "
                f"be built before its children."
            )

    # Backtrace lineage from a primary endpoint to satisfy Workbench Model
    # machinery (every Model needs a FeatureSet to hang off of).
    feature_list, feature_set_name, target_column = cls._derive_lineage(dag)

    # Build the model via the standard FeatureSet → Model flow. The
    # meta-endpoint template's `{{dag}}`, `{{aws_region}}`, `{{s3_bucket}}`
    # placeholders are filled from custom_args.
    aws_clamp = AWSAccountClamp()
    sm_session = aws_clamp.sagemaker_session()
    workbench_bucket = ConfigManager().get_config("WORKBENCH_BUCKET")

    feature_set = FeatureSet(feature_set_name)
    feature_set.to_model(
        name=name,
        model_type=cls._derive_model_type(dag),
        model_framework=ModelFramework.META,
        tags=tags or [name],
        description=description or f"MetaEndpoint DAG over: {', '.join(dag.endpoints.values())}",
        target_column=target_column,
        feature_list=feature_list,
        custom_args={
            "dag": dag.to_dict(),
            "aws_region": sm_session.boto_region_name,
            "s3_bucket": workbench_bucket,
        },
    )

    # MetaEndpoint containers are thin orchestrators — they receive the
    # full input as one invocation and let each child's own async_inference
    # do the row-level fan-out against the child fleet.
    log.important(f"Deploying MetaEndpoint '{name}' ({'async' if is_async else 'sync'})...")
    model = Model(name)
    endpoint = model.to_endpoint(
        tags=tags or [name],
        async_endpoint=is_async,
        min_instances=min_instances if is_async else 0,
        max_instances=max_instances if is_async else None,
    )

    # The meta delegates row-level batching to its children, who saturate
    # their own fleets; size the async meta batch to the slowest child's
    # fleet: smallest async-child batch × that fleet's max_instances × 4
    # (≈4 rounds through the fleet). Smaller batches → shorter invocations
    # (less likely to hit the 60-min async cap) and finer-grained failure/
    # retry. All-sync metas stay large (rows are cheap). Child stamps are
    # guaranteed present here — validated right after populate_child_metadata.
    meta_batch = 500
    if is_async:
        async_children = [ep for ep, child_is_async in dag.endpoint_async_flags.items() if child_is_async]
        async_batches = [dag.endpoint_batch_sizes[ep] for ep in async_children]
        async_fleets = [dag.endpoint_max_instances[ep] for ep in async_children]
        meta_batch = min(async_batches) * max(async_fleets) * 4
    endpoint.upsert_workbench_meta(
        {
            "inference_batch_size": meta_batch,
            "meta_endpoint_dag": dag.to_dict(),
        }
    )

    log.important(f"MetaEndpoint '{name}' created successfully!")
    return cls(name)

get_dag()

Reconstruct the MetaEndpointDAG from this endpoint's stored metadata.

Source code in src/workbench/api/meta_endpoint.py 
265
266
267
268
269
270
271
272
273
def get_dag(self) -> MetaEndpointDAG:
    """Reconstruct the MetaEndpointDAG from this endpoint's stored metadata."""
    meta = self.workbench_meta() or {}
    dag_dict = meta.get("meta_endpoint_dag")
    if not dag_dict:
        raise ValueError(
            f"MetaEndpoint '{self.name}' has no DAG in workbench_meta. Recreate via MetaEndpoint.create()."
        )
    return MetaEndpointDAG.from_dict(dag_dict)

purge_async_queue()

Cancel queued async invocations for the meta and every async child.

When a meta-level batch job is killed, orphaned work can be sitting at two layers: invocations queued at the meta endpoint, and child invocations queued by in-flight meta predict_fns. This override purges both so the entire DAG drains.

Returns:

Name Type Description
int int

Total number of staged input objects deleted across the
int

meta and its async children.
Source code in src/workbench/api/meta_endpoint.py 
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
def purge_async_queue(self) -> int:
    """Cancel queued async invocations for the meta and every async child.

    When a meta-level batch job is killed, orphaned work can be sitting
    at two layers: invocations queued at the meta endpoint, and child
    invocations queued by in-flight meta predict_fns. This override
    purges both so the entire DAG drains.

    Returns:
        int: Total number of staged input objects deleted across the
        meta and its async children.
    """
    total = super().purge_async_queue()
    dag = self.get_dag()
    for child_name in set(dag.endpoints.values()):
        if dag.endpoint_async_flags.get(child_name):
            total += Endpoint(child_name).purge_async_queue()
    return total

run_dag_test(input_df)

Execute the DAG client-side against input_df.

Bypasses the deployed meta endpoint entirely: each child endpoint is invoked directly via the regular Endpoint(name).inference() API from this process. Useful for debugging the DAG topology, isolating which child is misbehaving, or running the DAG when the deployed meta endpoint is unavailable.

Result is identical to self.inference(input_df) modulo transport and any container-only side effects (data capture, etc.).

Source code in src/workbench/api/meta_endpoint.py 
275
276
277
278
279
280
281
282
283
284
285
286
287
def run_dag_test(self, input_df: pd.DataFrame) -> pd.DataFrame:
    """Execute the DAG client-side against ``input_df``.

    Bypasses the deployed meta endpoint entirely: each child endpoint is
    invoked directly via the regular ``Endpoint(name).inference()`` API
    from this process. Useful for debugging the DAG topology, isolating
    which child is misbehaving, or running the DAG when the deployed
    meta endpoint is unavailable.

    Result is identical to ``self.inference(input_df)`` modulo transport
    and any container-only side effects (data capture, etc.).
    """
    return self.get_dag().run(input_df)