PipeParallel

The PipeParallel controller executes multiple pipes simultaneously, then combines their results into a single output. This is highly effective for running independent tasks concurrently in isolated branches.

How it works

PipeParallel runs a list of sub-pipes in concurrent branches.

1. Isolation: Before execution, PipeParallel creates a deep copy of the current WorkingMemory for each branch. This means every parallel pipe starts with the exact same state, but they run in complete isolation—a change in one branch will not affect another.
2. Concurrent Execution: All specified pipes are executed at the same time using asyncio.gather.
3. Combination: After all branches have finished, their results are always combined into a single object of the pipe's declared output concept, keyed by each branch's result name. That combined object is the pipe's main output.
4. Optional branch exposure: If add_each_output is true, each branch's individual result is also added to the working memory under the name specified in its result key, so downstream pipes can consume branches directly.

The output concept

The declared output of a PipeParallel is the concept the branch results are combined into. It must be one of:

• Composite — the native untyped composition: the combined object simply holds each branch result under its result name. Use this when you don't want to declare a bespoke concept.
• A structured concept whose fields correspond to the branch result names: every required field of the structure must match a branch result name, and every result name must match a declared field. Branch output types are checked too: when a field is typed with a content class, the branch feeding it must produce that concept or one refining it (e.g. a concept that refines Text fits a TextContent field).

Anything else is rejected at validation time: scalar native concepts (Text, Image, ...), Dynamic, Anything, and multiplicity suffixes (Foo[], Foo[3]) are all invalid — a parallel combination is a named composite, never a scalar or a list (a list aggregation is PipeBatch's shape).

Combining under absence

A branch result may legitimately be absent at run time: the branch pipe declares an optional output (?), or the branch is lifted (skipped) because a plain input is fed by one of the parallel's own optional (?) inputs. The combine handles absence by output kind:

• Composite output: the absent component is simply omitted from the combined object (an absence note is kept in the ledger for observability).
• Structured output, non-required field: the absence is absorbed as the field's default (null unless the field declares another default).
• Structured output, required field: this is rejected statically at validation time (optional_branch_required_field) — a required field cannot be fed by a maybe-absent branch. Make the field non-required, or sink the absence upstream with a ? input on the branch path.

A branch and its field must agree on multiplicity

A branch that produces a list, such as a PipeBatch whose output is Idea[], fills a field that holds a list, and a branch that produces one item fills a field that holds one item. When they disagree, the combine refuses the branch results, in the dry run of pipelex validate as in a run, and the refusal names the branch, its result and the field, and says which of the two to change; when the branch's multiplicity comes from its own nb_output, multiple_output or batch_over in branches, the refusal names that setting instead of the pipe's output. The advice is given only when the branch's concept fits the field, and whatever else the combine refused is reported beside it. The refusal is a fault in your method, so a hosted run answers it with HTTP 422 and keeps its message under STRICT disclosure, with the change to make as its next step:

PipeParallel 'analyze_topics' cannot combine its branch results into its output 'TopicReview'. Branch 'draft_ideas' gives result 'ideas' as a list, 'Idea[]', but field 'ideas' of 'TopicReview' holds a single item. Declare the field as a list in the structure of 'TopicReview', with type 'list', item_type 'concept' and item_concept_ref 'Idea', or make branch 'draft_ideas' output a single 'Idea'.

Configuration

PipeParallel is configured in your pipeline's .mthds file.

MTHDS Parameters

Parallel Step Configuration

Each entry in the branches array is a table with the following keys:

A branch's result and batch_as are stored names, as on a sequence step: the branch stores a value under each for a pipe to read through an input, so each is a plain input name matching [a-z][a-z0-9_]*, and one in any other form, such as Sentiment or text.sentiment, is refused with invalid_input_name. Since the combined output holds each branch result under its result name, the fields of a structured output concept that the branches fill have plain names too.

A branch is always a pipe step. A binding step, { from = "invoice.total", result = "total_amount" }, is refused in branches with binding_step_invalid: the branches run at once, and a binding orders a value before the steps that read it, so bind the value in the calling sequence, in a step before the PipeParallel, and have the branch read the bound name. A branch may carry batch_over and batch_as, but its batch_over is a plain name: a dotted one, such as catalog.pages, binds before it batches (see Batching over a field), so it is refused with binding_step_invalid too, and the calling sequence binds the list before the PipeParallel.

Example: Extracting different details from a text

Imagine you have a product description and you want to extract the product features and the product sentiment at the same time.

[concept.ProductDescription]
description = "A product description"
refines = "Text"

[concept.ProductFeatures]
description = "The features of a product"
refines = "Text"

[concept.ProductSentiment]
description = "The sentiment expressed about a product"
refines = "Text"

[concept.ProductAnalysis]
description = "Combined product analysis"

[concept.ProductAnalysis.structure]
features  = { type = "concept", concept_ref = "ProductFeatures", description = "Product features", required = true }
sentiment = { type = "concept", concept_ref = "ProductSentiment", description = "Product sentiment", required = true }

[pipe.extract_features]
type = "PipeLLM"
description = "Extract features from text"
inputs = { description = "ProductDescription" }
output = "ProductFeatures"
prompt = "Extract the product features from this description:\n@description"

[pipe.analyze_sentiment]
type = "PipeLLM"
description = "Analyze sentiment of text"
inputs = { description = "ProductDescription" }
output = "ProductSentiment"
prompt = "Analyze the sentiment of this product description:\n@description"

# The PipeParallel definition
[pipe.analyze_product_in_parallel]
type = "PipeParallel"
description = "Extract features and sentiment at the same time"
inputs = { description = "ProductDescription" }
output = "ProductAnalysis" # The concept the branch results are combined into
add_each_output = true
branches = [
    { pipe = "extract_features", result = "features" },
    { pipe = "analyze_sentiment", result = "sentiment" },
]

How this works:

1. The analyze_product_in_parallel pipe starts. It receives a ProductDescription.
2. Two parallel branches are created, both with access to the ProductDescription.
3. The extract_features pipe runs in one branch, and the analyze_sentiment pipe runs in the other, simultaneously.
4. After both complete, PipeParallel collects the results. The output from extract_features is named features, and the output from analyze_sentiment is named sentiment.
5. A new structured object of type ProductAnalysis is created and populated with the results, like {"features": ..., "sentiment": ...}. This object becomes the main output of the analyze_product_in_parallel pipe.
6. Because add_each_output is true, features and sentiment are also available individually in the working memory for downstream pipes.

If you don't want to declare a ProductAnalysis concept, set output = "Composite" instead: the combined object then holds the same {"features": ..., "sentiment": ...} shape as an untyped composition. A later template reads each part by name, as it would read a field: {{ analysis.features }}.

Related Documentation

• Pipeline Orchestration - Overview of pipeline orchestration capabilities
• Native Concepts - Includes the Composite concept