Agentic Workflows
A modular async pipeline engine with a typed SDK where nodes, flows, and workflows are defined in single files, compiled to a DAG, and executed in parallel with human-in-the-loop checkpoints.
Overview
OrbitAIM Workflows is an async pipeline engine for orchestrating multi-step cold-email campaigns. It is built in two layers: an immutable core engine (a PipelineGraph → PipelineExecutor that runs nodes in topological layers with parallel asyncio.gather) and a developer-facing SDK that compiles down to it. Developers write a node as a single file — Pydantic input/output schemas, the API call, and auto-registration co-located — and compose nodes into flows and flows into workflows using a `>>` operator that auto-wires matching fields by name. From the user's perspective everything is just a node: they discover units via GET /nodes, ask what a node can connect to, and POST /run to execute one node or a connected pipeline. A single human-in-the-loop mechanism handles both pre-flight forms and mid-execution pauses (folder pickers, prospect selection) via checkpoints and a /resume endpoint.
Key features
Single-File Node SDK
The @node decorator reads a function's Pydantic input/output annotations, builds its input_mapping, wraps it in a core BaseNode subclass, and registers it — schema, API call, and registration all in one file.
Composable Flows & Workflows
@flow groups nodes into a composite node and @workflow chains flows, both wired with a `>>` operator that auto-matches fields by name, with explicit port wiring available for fine control.
Parallel DAG Executor
PipelineExecutor runs nodes in topological layers via asyncio.gather, propagates failures as SKIPPED downstream, and produces an ExecutionReport with completed/failed/skipped nodes.
Human-in-the-Loop Checkpoints
A single HITLField mechanism handles pre-flight forms and mid-run pauses (folder/prospect pickers). Runs return waiting_for_input + a checkpoint id; POST /resume continues from the saved state.
Connection Discovery API
GET /nodes, /nodes/{a}/connections, and /nodes/{a}/accepted_from expose developer-declared compatibility with copy-pasteable /run bodies — everything a canvas UI needs to draw and validate wires.
Boot-time Compatibility Validation
accepts_from / connects_to are declared on both sides of every connection; a startup validator refuses to boot if they disagree, turning wiring mistakes into hard errors instead of silent bugs.
Built-in Observability
Every /run and /resume response carries a per-node timeline (layer, start/end, duration) and the resolved_inputs each node saw, so slow steps and HITL pauses are traceable.
Built with
Cold-email campaign setup is inherently multi-step — create a campaign, assign sending accounts, pick folders and prospects, enrich company data, generate the emails — and each step depends on the last, with several requiring human decisions in the middle. Hard-coding that sequence is brittle, and exposing raw field-level wiring to users is unusable. The system needed a way to define steps once, compose them freely, and pause cleanly for human input without the caller managing execution state.
Turns multi-step cold-email campaign automations into composable, drag-to-connect nodes — letting non-authors assemble end-to-end outreach pipelines from a UI while developers add new capabilities in a single file with zero boilerplate.
Engineering challenges
Designing a DAG executor that runs independent nodes in parallel per topological layer while correctly marking downstream nodes SKIPPED when a dependency fails.
Suspending mid-pipeline for human input — a node raises HumanInputRequired, sibling nodes in the same layer are flushed first, a checkpoint is saved, and the run returns waiting_for_input with a resumable id.
Auto-wiring node-to-node connections without exposing field mapping — an expander name-matches every output field of the source to input fields of the target and patches each node's input_mapping at compile time.
Supporting both static pre-flight HITL (forms shown before the run) and dynamic mid-run HITL (options resolved from an upstream field) through one HITLField declaration.
Keeping the core engine untouched — the entire SDK ergonomics layer (decorators, the `>>` operator, expansion) compiles to plain core PipelineGraph/Edge objects so the runtime never changes.
Enforcing connection compatibility at boot — every node/flow declares accepts_from and connects_to by hand, and a startup validator fails the server if the two sides of any declared connection disagree.
Key decisions
Two-layer architecture (immutable core + SDK) — the execution engine is frozen and testable, while developer ergonomics evolve entirely in the SDK layer above it.
One node = one file — schema, API call, and registration co-located, with registration as a side-effect of import so there is no registry file or __init__.py to maintain.
`>>` operator with name-based auto-wiring as the default, plus explicit field-level wiring as an escape hatch when source/target field names differ or need fallback chains.
'Everything is a node' to users — flows (composite nodes) appear in GET /nodes identically to single nodes, so the UI has one uniform concept to render and connect.
In-memory checkpoint store by default with a swappable backing (e.g. Redis) so HITL pauses can survive restarts when needed.
Every run response includes a per-node timeline and the resolved_inputs each node actually received — HITL nodes included — making pipelines debuggable end to end.