KO | EN

GitLite

Minimalist Open Source Repository Search Engine

gitlite — search

#dsh #dsh-plugin #developer-tools #ai-agent #claude-code #codex #typescript #ai-agents #deepseek-harness #mcp #cli #rust

routilux

★ 86 Open GitHub ↗

Routines-based, event-driven workflow orchestration for Python—compose complex data/AI pipelines and run concurrent workflows across distributed systems with durable state and robust error handling.

// repository documentation

Was this content helpful?

★ 0 (0 ratings)

# Routilux ⚡

[![PyPI version](https://img.shields.io/pypi/v/routilux.svg)](https://pypi.org/project/routilux/)
[![Python 3.7+](https://img.shields.io/badge/python-3.7+-blue.svg)](https://www.python.org/downloads/)
[![License](https://img.shields.io/badge/license-Apache%202.0-green.svg)](https://raw.githubusercontent.com/lzjever/routilux/main/LICENSE)
[![Documentation](https://readthedocs.org/projects/routilux/badge/?version=latest)](https://routilux.readthedocs.io)
[![CI](https://github.com/lzjever/routilux/workflows/CI/badge.svg)](https://github.com/lzjever/routilux/actions)
[![codecov](https://codecov.io/gh/lzjever/routilux/branch/main/graph/badge.svg)](https://codecov.io/gh/lzjever/routilux)

**[简体中文](README.zh-CN.md)**

**Routilux** — Event-driven workflow orchestration. Too many pipelines to tame? One event queue for orchestration, concurrency, and resume-from-checkpoint. Build in minutes, recover anytime.

## ✨ Why Routilux?

- 🚀 **Event queue**: Non-blocking, one model for sequential and concurrent execution
- 🔗 **Flexible wiring**: Many-to-many routines, smart routing
- 📊 **State built-in**: Execution state, metrics, history out of the box
- 🛡️ **Error policies**: STOP / CONTINUE / RETRY / SKIP with automatic recovery
- ⚡ **Concurrent execution**: I/O parallelized without blocking the main flow
- 💾 **Checkpoint & resume**: Save and restore at any node; survive interruptions
- 🎯 **Production-ready**: Error handling, tracing, and monitoring
- 🎨 **Simple API**: Flow auto-detection; fewer parameters in most cases

## 🎯 Perfect For

- **Data Pipelines**: ETL processes, data transformation workflows
- **API Orchestration**: Coordinating multiple API calls with complex dependencies
- **Event Processing**: Real-time event streams and reactive systems
- **Workflow Automation**: Business process automation and task scheduling
- **Microservices Coordination**: Managing interactions between services
- **LLM Agent Workflows**: Complex AI agent orchestration and chaining

## 📦 Installation

### ⚡ One-Line Install (Recommended)

```bash
# Mac / Linux - Auto-detects best method (uv > pipx > pip)
curl -fsSL https://raw.githubusercontent.com/lzjever/routilux/main/install.sh | bash

# Or with wget
wget -qO- https://raw.githubusercontent.com/lzjever/routilux/main/install.sh | bash
```

**Installation options:**

```bash
# Use pipx instead of uv
METHOD=pipx curl -fsSL https://raw.githubusercontent.com/lzjever/routilux/main/install.sh | bash

# Install specific version
VERSION=0.14.0 curl -fsSL https://raw.githubusercontent.com/lzjever/routilux/main/install.sh | bash
```

### Recommended: Isolated CLI Installation (pipx)

The best way to install Routilux CLI without affecting your system:

```bash
# Install CLI with isolated environment
pipx install "routilux[cli]"

# Use anywhere
routilux --help
routilux run workflow.yaml
```

**Why pipx?**
- ✅ Creates isolated virtual environment (no dependency conflicts)
- ✅ CLI available globally
- ✅ Easy to update: `pipx upgrade routilux`
- ✅ Works on Mac and Linux

### Alternative: Modern (uv tool)

Using [uv](https://github.com/astral-sh/uv) (faster than pipx):

```bash
# Install
uv tool install "routilux[cli]"

# Use
routilux --help
```

### macOS / Linux: Homebrew

```bash
# Add tap and install
brew tap lzjever/routilux
brew install routilux

# Or directly
brew install lzjever/routilux/routilux
```

### Standard pip Install

For library use or development:

```bash
# Library only
pip install routilux

# With CLI support
pip install "routilux[cli]"
```

### Development Setup with uv (Recommended)

This project uses [uv](https://github.com/astral-sh/uv) for fast dependency management. Install uv first:

```bash
curl -LsSf https://astral.sh/uv/install.sh | sh
```

Then set up the development environment:

**Recommended: For active development**

```bash
# Install package with all development dependencies (recommended)
make dev-install

# Or manually with uv (dev group is installed by default)
uv sync --group docs --all-extras
```

**Alternative: Dependencies only (for CI/CD or code review)**

```bash
# Create virtual environment and install dependencies only (without installing the package)
# Useful for: CI/CD pipelines, code review, or when you only need development tools
make setup-venv

# Later, if you need to install the package:
make install
```

**Understanding dependency groups vs extras:**

- **Dependency groups** (`dev`, `docs`): Development dependencies that are not published to PyPI. The `dev` group is installed by default with `uv sync`.
- **Extras**: Currently none, but may be added in the future.

All `make` commands will automatically use `uv` if available, otherwise fall back to `pip`.

### Development Install (Legacy - using pip)

For development with all dependencies using pip:

```bash
pip install -e ".[dev]"
# Or using Makefile
make dev-install
```

## 🖥️ CLI

Routilux includes a command-line interface for workflow management:

```bash
# Install with CLI support
pip install routilux[cli]

# Run a workflow
routilux run --workflow flow.yaml

# Start server
routilux server start

# See all commands
routilux --help
```

### CLI Commands

- **`routilux init`** - Initialize a new project with example files
- **`routilux run`** - Execute a workflow from a DSL file
- **`routilux server`** - Start the HTTP server for API access
- **`routilux job`** - Submit and manage jobs
- **`routilux list`** - List available routines or flows
- **`routilux validate`** - Validate a workflow DSL file

See [CLI Documentation](docs/cli/usage.md) for details.

### Server with Flow Loading

Start the HTTP server with flow auto-loading:

```bash
# Start server with flows directory
routilux server start --flows-dir ./flows --port 8080

# Built-in routines (Mapper, Filter, etc.) are automatically available
# Flows from ./flows/*.yaml are loaded at startup
# Hot reload enabled - flow files are watched for changes
```

### Job Management

Submit and manage jobs via CLI:

```bash
# Submit job locally
routilux job submit --flow myflow --routine processor --data '{"input": "value"}'

# Submit job to remote server
routilux job submit --server http://localhost:8080 --flow myflow --routine processor --data '{}'

# Check job status
routilux job status <job_id>

# List jobs
routilux job list --flow myflow
```

## 🚀 Quick Start

For development with all dependencies using pip:

```bash
pip install -e ".[dev]"
# Or using Makefile
make dev-install
```

## 🚀 Quick Start

### Create Your First Workflow in 3 Steps

**Step 1: Define a Routine**

```python
from routilux import Routine

class DataProcessor(Routine):
    def __init__(self):
        super().__init__()
        # Define input slot
        self.input_slot = self.define_slot("input", handler=self.process_data)
        # Define output event
        self.output_event = self.define_event("output", ["result"])
    
    def process_data(self, data=None, **kwargs):
        # Flow is automatically detected from routine context
        result = f"Processed: {data}"
        self._stats["processed_count"] = self._stats.get("processed_count", 0) + 1
        self.emit("output", result=result)  # No need to pass flow!
```

**Step 2: Create and Connect a Flow**

```python
from routilux import Flow

flow = Flow(flow_id="my_workflow")

processor1 = DataProcessor()
processor2 = DataProcessor()

id1 = flow.add_routine(processor1, "processor1")
id2 = flow.add_routine(processor2, "processor2")

# Connect: processor1's output → processor2's input
flow.connect(id1, "output", id2, "input")
```

**Step 3: Execute**

```python
job_state = flow.execute(id1, entry_params={"data": "Hello, Routilux!"})
print(job_state.status)  # "completed"
print(processor1.stats())  # {"processed_count": 1}
```

**🎉 Done!** You've created your first workflow.

## 💡 Key Features

### 🔄 Event Queue Architecture

Routines communicate through events and slots using a unified event queue pattern:

```python
# Multiple routines can listen to the same event
flow.connect(processor1, "output", processor2, "input")
flow.connect(processor1, "output", processor3, "input")  # Fan-out

# Multiple events can feed into the same slot
flow.connect(processor1, "output", aggregator, "input")
flow.connect(processor2, "output", aggregator, "input")  # Fan-in

# emit() is non-blocking - returns immediately after enqueuing tasks
# Flow is automatically detected from routine context
self.emit("output", data="value")  # No flow parameter needed!
```

### 🎛️ Flexible State Management

Track everything automatically:

```python
# Access routine state
stats = routine.stats()  # {"processed_count": 42, "errors": 0}

# Track execution history
history = job_state.get_execution_history()

# Performance metrics
perf = flow.execution_tracker.get_routine_performance("processor1")
```

### 🛡️ Built-in Error Handling

Choose the right strategy for your use case:

```python
from routilux import ErrorHandler, ErrorStrategy

# Stop on error (default)
flow.set_error_handler(ErrorHandler(ErrorStrategy.STOP))

# Continue and log errors
flow.set_error_handler(ErrorHandler(ErrorStrategy.CONTINUE))

# Retry with exponential backoff
flow.set_error_handler(ErrorHandler(
    ErrorStrategy.RETRY,
    max_retries=3,
    retry_delay=1.0,
    backoff_multiplier=2.0
))
```

### ⚡ Unified Execution Model

Both sequential and concurrent modes use the same event queue mechanism:

```python
# Sequential mode (default): max_workers=1
flow = Flow()  # Sequential by default

# Concurrent mode: max_workers>1
flow.set_execution_strategy("concurrent", max_workers=4)

# Tasks are processed fairly in queue order
# Long chains don't block shorter ones
job_state = flow.execute(entry_routine_id)
flow.wait_for_completion()  # Wait for async tasks
```

### 💾 Persistence & Recovery

Save and resume workflows:

```python
# Save workflow state
job_state.save("workflow_state.json")

# Later, resume from saved state
saved_state = JobState.load("workflow_state.json")
flow.resume(saved_state)
```

## 📚 Documentation

**📖 Full documentation available at: [routilux.readthedocs.io](https://routilux.readthedocs.io)**

### Documentation Highlights

- **📄 [Where flows come from](docs/flow_sources.md)**: DSL, registry, API, and “Sync from registry” in Overseer
- **📘 [User Guide](https://routilux.readthedocs.io/en/latest/user_guide/index.html)**: Comprehensive guide covering all features
- **🔧 [API Reference](https://routilux.readthedocs.io/en/latest/api_reference/index.html)**: Complete API documentation
- **💻 [Examples](https://routilux.readthedocs.io/en/latest/examples/index.html)**: Real-world code examples
- **🏗️ [Design](https://routilux.readthedocs.io/en/latest/design/index.html)**: Architecture and design principles

### Build Documentation Locally

```bash
pip install -e ".[docs]"
cd docs && make html
```

## 🎓 Examples

Check out the `examples/` directory for practical examples:

### Core Examples

- **`basic_example.py`** - Your first workflow
- **`data_processing.py`** - Multi-stage data pipeline
- **`concurrent_flow_demo.py`** - Parallel execution
- **`error_handling_example.py`** - Error handling strategies
- **`state_management_example.py`** - State tracking and recovery
- **`builtin_routines_demo.py`** - Using built-in routines

### Real-World Usage Patterns

- **`data_pipeline.py`** - Multi-stage data processing with validation, transformation, and quality checks
- **`async_orchestration.py`** - Concurrent task execution with result aggregation
- **`long_running_workflow.py`** - Pause/resume execution with state persistence and recovery
- **`error_handling.py`** - Retry patterns and fallback mechanisms

Run examples:

```bash
python examples/basic_example.py
python examples/data_pipeline.py
python examples/async_orchestration.py
python examples/long_running_workflow.py
python examples/error_handling.py
```

## 🧩 Built-in Routines

Routilux comes with a rich set of built-in routines ready to use:

- **Text Processing**: `TextClipper`, `TextRenderer`, `ResultExtractor`
- **Data Processing**: `DataTransformer`, `DataValidator`, `DataFlattener`
- **Control Flow**: `ConditionalRouter` for dynamic routing
- **Utilities**: `TimeProvider` for timestamps

```python
from routilux.builtin_routines import ConditionalRouter, DataTransformer

# Use built-in routines directly
router = ConditionalRouter()
transformer = DataTransformer()
```

## 🏗️ Project Structure

```
routilux/
├── routilux/              # Main package
│   ├── routine.py         # Routine base class
│   ├── flow.py            # Flow manager
│   ├── job_state.py       # State management
│   ├── connection.py      # Connection management
│   ├── event.py           # Event class
│   ├── slot.py            # Slot class
│   ├── error_handler.py   # Error handling
│   └── execution_tracker.py # Performance tracking
├── tests/                 # Comprehensive test suite
├── examples/              # Usage examples
└── docs/                  # Sphinx documentation
```

## 🧪 Testing

Routilux comes with comprehensive tests:

```bash
# Run all tests
make test-all

# Run with coverage
make test-cov

# Run specific test suite
pytest tests/                    # Core tests
pytest routilux/builtin_routines/  # Built-in routines tests
```

## 🤝 Contributing

We welcome contributions! Here's how you can help:

1. **Star the project** ⭐ - Show your support
2. **Report bugs** 🐛 - Help us improve
3. **Suggest features** 💡 - Share your ideas
4. **Submit PRs** 🔧 - Contribute code

## 🏢 About Agentsmith

**Routilux** is part of the **Agentsmith** open-source ecosystem. Agentsmith is a ToB AI agent and algorithm development platform, currently deployed in multiple highway management companies, securities firms, and regulatory agencies in China. The Agentsmith team is gradually open-sourcing the platform by removing proprietary code and algorithm modules, as well as enterprise-specific customizations, while decoupling the system for modular use by the open-source community.

### 🌟 Agentsmith Open-Source Projects

- **[Varlord](https://github.com/lzjever/varlord)** ⚙️ - Configuration management library with multi-source support
- **[Routilux](https://github.com/lzjever/routilux)** ⚡ - Event-driven workflow orchestration framework
- **[Serilux](https://github.com/lzjever/serilux)** 📦 - Flexible serialization framework for Python objects
- **[Lexilux](https://github.com/lzjever/lexilux)** 🚀 - Unified LLM API client library

These projects are modular components extracted from the Agentsmith platform, designed to be used independently or together to build powerful applications.

## 📄 License

Routilux is licensed under the **Apache License 2.0**. See [LICENSE](LICENSE) for details.

## 🔗 Links

- **📦 PyPI**: [pypi.org/project/routilux](https://pypi.org/project/routilux)
- **📚 Documentation**: [routilux.readthedocs.io](https://routilux.readthedocs.io)
- **🐙 GitHub**: [github.com/lzjever/routilux](https://github.com/lzjever/routilux)
- **📧 Issues**: [github.com/lzjever/routilux/issues](https://github.com/lzjever/routilux/issues)

## ⭐ Show Your Support

If Routilux helps you build amazing workflows, consider giving it a star on GitHub!

---

**Built with ❤️ by the Routilux Team**

*Making workflow orchestration simple, powerful, and fun.*

routilux

Explore Similar Repositories

Was this content helpful?

Recent Feedback

Download README