radical.asyncflow 0.2.0

A High Performance Asynchronous Workflow Scripting Library

RADICAL AsyncFlow (RAF) is a fast asynchronous scripting library built on top of asyncio for building powerful async/sync workflows on HPC, clusters, and local machines. It supports pluggable execution backends with intuitive task dependencies and workflow composition.

• ⚡ Powerful asynchronous workflows — Compose complex async and sync workflows easily, with intuitive task dependencies and campaign orchestration.

• 🌐 Portable across environments — Run seamlessly on HPC systems, clusters, and local machines with pluggable execution backends.

• 🧩 Flexible and extensible — Supports campaign management and advanced workflow patterns, built on Python’s asyncio and RADICAL Cybertools expertise.

Currently, RAF supports the following execution backends:

• Radical.Pilot
• Dask.Parallel
• Concurrent.Executor
• Noop with dry_run
• Custom implementations

⚙️ Installation

Radical Asyncflow package is available on PyPI.

pip install radical-asyncflow

For developers:

git clone https://github.com/radical-cybertools/radical.asyncflow
cd radical.asyncflow
pip install -e .[dev,lint,doc]

📚 Documentation

👉 AsyncFlow Documentation and API References

Basic Usage

import asyncio

from radical.asyncflow import WorkflowEngine
from radical.asyncflow import ConcurrentExecutionBackend

from concurrent.futures import ThreadPoolExecutor

async def main():
    # Create backend and workflow
    backend = await ConcurrentExecutionBackend(ThreadPoolExecutor())
    flow = await WorkflowEngine.create(backend=backend)

    @flow.executable_task
    async def task1():
        return "/bin/echo 5"

    @flow.function_task
    async def task2(t1_result):
        return int(t1_result.strip()) * 2 * 2

    # create the workflow
    t1_fut = task1()
    t2_result = await task2(t1_fut) # t2 depends on t1 (waits for it)
  
    print(t2_result)
    # shutdown the execution backend
    await flow.shutdown()

if __name__ == "__main__":
    asyncio.run(main())

What AsyncFlow Can Be Used For

• AI & LLM Workflows - Build complex AI agent systems and orchestrate multiple language model calls with automatic dependency resolution in parallel.
• Data Processing Pipelines - Create data science pipelines, and real-time analytics with async task coordination.
• High-Performance Computing - Execute scientific computing workflows and distributed simulations on HPC clusters with scaling.
• Cross-Platform Execution - Deploy the same workflows locally for development, or HPC infrastructure without code changes