pivex

pivex helps you orchestrate pipelines and workflows that consist of complex asynchronous interactions. This is especially helpful in multi-agent AI pipelines, RAG, ETL, or anything where you need to orchestrate asyncronous (and synchronous) calls.

What does pivex do?

$3

1. Create complicated pipelines in code
1. Handle coordination of asynchronous calls
1. Use steps across pipelines
1. Debug complicated pipelines
- NOTE: Use 'inline breakpoints' for lambdas

$3

1. Declarative syntax for creating pipelines
1. Native async interface that also supports synchronous actions
1. Some type safety for pipelines and steps when using TypeScript

Examples

The syntax for creating pipelines leverages your IDE to help you understand how your pipeline works.

All pipelines expose an asynchronous start function but pipelines handle both synchronous and asynchronous steps correctly.

See the Important considerations section for more information.

$3

``typescript
const testPipeline = pipeline()
.sequential(
run((x: number) => x + 1),
run((x: number) => x + 2)
)
const result = await testPipeline.start(1)
expect(result).eq(4)
`

`mermaid
flowchart TB
start(("Input=2"))
subgraph Step 1
add1["AsyncAction

---

Multiply by 3"]
add2["AsyncAction

---

Multiply by 5"]
end
stop(("Output=30"))
start-->add1-->add2-->stop

`




$3


`typescript
const testPipeline = pipeline()
.parallel(
defer((x: number) => Promise.resolve(x * 3)),
defer((x: number) => Promise.resolve(x * 5))
)
const result = await testPipeline.start(2)
expect(result[0]).eq(6)
expect(result[1]).eq(10)
`

`mermaid
flowchart TB
start(("Input=2"))
subgraph step1["Step 1 (Actions in Parallel)"]
mult1["AsyncAction

---

Multiply by 3"]
mult2["AsyncAction

---

Multiply by 5"]
end
step1Output["--pivex coalesces output--"]
stop(("Output=[6, 10]"))
start-->step1
mult1-->step1Output
mult2-->step1Output
step1Output-->stop
style step1Output fill:#FFFFFF
`

$3

`typescript
const testPipeline = pipeline()
.sequential(
parallel(
defer((x: number) => Promise.resolve(x * 2)),
defer((x: number) => Promise.resolve(x * 3))
),
sequential(
run((nums: number[]) => nums.reduce((p, c) => p + c)),
run((x: number) => x / 2)
)
)
const result = await testPipeline.start(2)
expect(result).eq(5)
`

`mermaid
flowchart TB
start(("Input=2"))
subgraph step1["Step 1 (Actions in Parallel)"]
mult1["AsyncAction

---

Multiply by 3"]
mult2["AsyncAction

---

Multiply by 5"]
end
subgraph step2["Step 2 (Sequential actions)"]
sum["Sum the numbers"]
div["Divide the sum by 2"]
end
step1Output["--pivex coalesces output--"]
stop(("Output=5"))
start-->step1
mult1-->step1Output
mult2-->step1Output
step1Output-->sum
sum-->div
div-->stop
style step1Output fill:#FFFFFF
`

$3

You can provide names and enable logging for any step in the pipeline

`
const testPipeline = pipeline()
.sequential(
name('Step 1')
.log(true)
.run((x: number) => x + 1),
name('Step 2')
.log(true)
.run((x: number) => x + 2)
)
const result = await testPipeline.start(1)
expect(result).eq(4)
``

Design overview

$3

#### pipeline
- Takes initial input
- Contains steps
- Will execute child steps in sequential order
- Output from the first step is passed to the second, whose output is passed to the third, etc.
- Returns output from the last step

#### step
- Contains a single action or an array of child steps
- An action can be either synchronous or asynchronous. Either way the step will not complete until the action has completed
- Child steps can be executed in parallel or sequentially (in order)
- If parallel, the step starts all its children and awaits all before coalescing the results and returning an array of the results
- If sequential, executes all child steps in order, awaiting the results from each child step before passing results from previous step to the next step. The parent step returns the value returned from the last step.

#### action
- Contains a single function
- Can be synchronous or asynchronous

Important considerations

> WARNING: pivex does not check for infinite loops. We recommmend you use the constructor-builder syntax (see code examples above) to ensure no infinite loops are created.

- The pipeline start function is natively async but handles tasks with synchronous or asynchronous actions. If you have pipelines that are entirely synchronous this imposes a very small performance penalty (~65ms for 1M steps)

TODO

1. Improve typings between steps
1. Update typings for step.sync so they don't allow passing asynchronous calls
1. Add a stop function to cancel the pipeline