Fault-Tolerant Distributed Agentic Workflow Engine

A production-grade distributed workflow execution engine designed to orchestrate complex agentic workflows with built-in fault tolerance, retries, and scalability.

Overview

This project provides a robust framework for executing workflows composed of interdependent nodes (agents) in a distributed manner. It includes:

• Workflow DAG Execution: Define workflows as directed acyclic graphs (DAGs) with dependencies
• Fault Tolerance: Automatic retry mechanisms and graceful failure handling
• Distributed Task Processing: Async task queue integration using Redis
• Persistent State: PostgreSQL-backed workflow and execution tracking
• RESTful API: FastAPI-based HTTP endpoints for workflow management
• Database Migrations: Alembic for schema versioning and migrations

Architecture

The engine consists of the following components:

• Engine: Core orchestration logic for executing workflows
• Worker: Distributed task processor for executing individual nodes
• API: FastAPI application for workflow management and monitoring
• Database: SQLAlchemy ORM with async PostgreSQL support
• Queue: Redis-backed distributed task queue (arq)

Prerequisites

• Python 3.11 or higher
• PostgreSQL 12+
• Redis 6.0+
• Poetry (for dependency management)

Installation

1. Clone the repository

git clone https://github.com/yasaman138/agentic-workflow-engine.git
cd "agentic-workflow-engine"

2. Install dependencies using Poetry

poetry install

This will install all dependencies specified in pyproject.toml, including:

• FastAPI and Uvicorn
• SQLAlchemy with async PostgreSQL support
• Arq for distributed task processing
• Alembic for database migrations
• Development tools (pytest, mypy, ruff)

3. Set up environment variables

Create a .env file in the project root:

DATABASE_URL=postgresql+asyncpg://postgres:supersecretpassword@localhost:5432/workflow_engine
REDIS_URL=redis://localhost:6379
LOG_LEVEL=INFO

4. Start services using Docker Compose

docker-compose up -d

This will start:

• PostgreSQL database (port 5432)
• Redis server (port 6379)

5. Run database migrations

poetry run alembic upgrade head

Usage

Starting the API Server

poetry run uvicorn app.api:app --reload

The API will be available at http://localhost:8000

Starting the Worker

poetry run python -m app.worker

The worker will process tasks from the Redis queue.

Creating and Executing Workflows

Workflows are defined as JSON DAGs. Use the API endpoints to:

• Create workflows
• Trigger executions
• Monitor progress
• Retrieve results

Example DAG structure:

{
  "nodes": [
    {"id": "node1", "type": "agent"},
    {"id": "node2", "type": "agent"}
  ],
  "edges": [
    {"source": "node1", "target": "node2"}
  ]
}

Testing

Run the test suite using pytest:

poetry run pytest

With coverage:

poetry run pytest --cov=app

Test files include:

• tests/test_engine.py: Engine logic tests
• tests/test_worker.py: Worker process tests

Development

Code Quality

The project uses:

• Ruff: Fast Python linter and formatter
• MyPy: Static type checking with strict mode enabled
• Pytest: Testing framework with async support

Run linting and type checking:

poetry run ruff check .
poetry run mypy app

Format code:

poetry run ruff format .

Configuration Files

• pyproject.toml: Project metadata and dependencies
• pytest.ini: Pytest configuration
• alembic.ini: Database migration configuration
• docker-compose.yml: Local development services
• Dockerfile: Container image definition

Database

Migrations

Database schemas are managed using Alembic. To create a new migration:

poetry run alembic revision --autogenerate -m "description"

Apply migrations:

poetry run alembic upgrade head

Troubleshooting

Connection Issues

Ensure PostgreSQL and Redis are running:

docker-compose ps

Database Errors

Reset the database:

docker-compose down -v
docker-compose up -d
poetry run alembic upgrade head

Import Errors

Ensure the project root is in your Python path:

poetry run python -c "import app"

Project Structure

.
├── alembic/                 # Database migrations
│   ├── versions/           # Migration files
│   ├── env.py             # Migration environment
│   └── script.py.mako     # Migration template
├── app/                     # Main application
│   ├── __init__.py        # Package initialization
│   ├── api.py             # FastAPI application
│   ├── database.py        # Database configuration
│   ├── engine.py          # Workflow execution engine
│   ├── models.py          # SQLAlchemy models
│   ├── schemas.py         # Pydantic schemas
│   └── worker.py          # Task worker
├── scripts/                # Utility scripts
│   └── entrypoint.sh      # Docker entrypoint
├── tests/                  # Test suite
│   ├── conftest.py        # Pytest configuration
│   ├── test_engine.py     # Engine tests
│   └── test_worker.py     # Worker tests
├── docker-compose.yml      # Docker services configuration
├── Dockerfile             # Container image
├── pyproject.toml         # Poetry project configuration
├── pytest.ini             # Pytest configuration
└── alembic.ini           # Alembic configuration

Contributing

1. Create a feature branch from main
2. Make changes and ensure tests pass
3. Run linting and type checks
4. Submit a pull request