LLM-Orchestrator

Production-ready workflow orchestration engine for Large Language Models, written in Rust

Overview

LLM-Orchestrator is an enterprise-grade, production-ready workflow orchestration engine built in Rust for LLM-powered applications. It provides type-safe DAG-based execution with automatic parallelization, comprehensive fault tolerance, and seamless integration with multiple LLM providers.

Current Implementation Status

This project has successfully completed Phase 1-3 implementation with the following features:

✅ Fully Implemented:

Core workflow engine with DAG execution
OpenAI and Anthropic (Claude) provider integration
Exponential backoff retry logic with jitter
Async Tokio runtime for high-performance execution
Handlebars template engine for prompt rendering
CLI tools (validate & run commands)
Comprehensive test suite (56 tests passing)
Structured logging and observability
Type-safe error handling

Key Features

Multi-Provider Support: Seamless integration with OpenAI and Anthropic (Claude)
DAG-Based Execution: Automatic dependency resolution and parallel task execution
Fault Tolerance: Exponential backoff retries with configurable jitter
Template Engine: Handlebars-based prompt templating with workflow context
Type-Safe: Leverages Rust's type system for compile-time safety and zero-cost abstractions
Async Runtime: Built on Tokio for high-performance concurrent execution
Production-Ready: Comprehensive error handling, observability, and testing

Quick Start

Installation

Choose your preferred installation method:

Via npm (Recommended)

# Install globally
npm install -g @llm-dev-ops/llm-orchestrator

# Verify installation
llm-orchestrator --version

Via Docker

# Pull the latest image
docker pull ghcr.io/globalbusinessadvisors/llm-orchestrator:latest

# Run a workflow
docker run -v $(pwd):/workspace \
  -e OPENAI_API_KEY="$OPENAI_API_KEY" \
  ghcr.io/globalbusinessadvisors/llm-orchestrator:latest \
  run /workspace/workflow.yaml --input '{"name": "Alice"}'

Build from Source

# Clone and build from source
git clone https://github.com/globalbusinessadvisors/llm-orchestrator.git
cd llm-orchestrator
cargo build --release

# The binary will be available at target/release/llm-orchestrator

Environment Setup

Set up your API keys:

export OPENAI_API_KEY="sk-..."
export ANTHROPIC_API_KEY="sk-ant-..."

Create Your First Workflow

Create a file simple-workflow.yaml:

name: simple-workflow
version: "1.0"
description: A simple two-step workflow

steps:
  - id: greet
    type: llm
    provider: openai
    model: gpt-3.5-turbo
    prompt: "Say hello to {{inputs.name}}"
    temperature: 0.7
    max_tokens: 50
    output:
      - greeting

  - id: analyze
    type: llm
    depends_on:
      - greet
    provider: openai
    model: gpt-3.5-turbo
    prompt: "Analyze this greeting: {{steps.greet.greeting}}"
    temperature: 0.5
    max_tokens: 100
    output:
      - analysis

Validate and Run

# Validate the workflow
./target/release/llm-orchestrator validate simple-workflow.yaml

# Run the workflow with inputs
./target/release/llm-orchestrator run simple-workflow.yaml \
  --input '{"name": "Alice"}'

# Run with verbose logging
./target/release/llm-orchestrator run simple-workflow.yaml \
  --input '{"name": "Alice"}' \
  --verbose

Architecture

LLM-Orchestrator is built with a modular, type-safe architecture:

┌─────────────────────────────────────────────────────────┐
│                    LLM-ORCHESTRATOR                     │
├─────────────────────────────────────────────────────────┤
│                                                         │
│  Workflow YAML → Parser → Validator → DAG Builder      │
│                                ↓                        │
│  Execution Context ← Template Engine (Handlebars)      │
│         ↓                                               │
│  Workflow Executor → Step Scheduler → Task Router      │
│         ↓                     ↓                         │
│  Provider Registry → [OpenAI, Anthropic, ...]          │
│         ↓                                               │
│  Retry Executor ← Exponential Backoff + Jitter         │
│         ↓                                               │
│  Results (JSON) + Observability (Tracing)              │
│                                                         │
└─────────────────────────────────────────────────────────┘

Core Components

Workflow Definition Layer (llm-orchestrator-core/src/workflow.rs)
- YAML parsing with serde_yaml
- Schema validation
- Step configuration types (LLM, Transform, Embed, etc.)
DAG Engine (llm-orchestrator-core/src/dag.rs)
- Dependency graph construction with petgraph
- Cycle detection
- Topological sorting for execution order
- Parallel step identification
Execution Engine (llm-orchestrator-core/src/executor.rs)
- Async execution with Tokio runtime
- Dynamic provider dispatch via trait objects
- Conditional step execution
- Parallel execution with configurable concurrency
Provider System (llm-orchestrator-providers/src/)
- Abstract LLMProvider trait
- OpenAI implementation (Chat Completions API)
- Anthropic implementation (Messages API)
- Thread-safe provider registry
Retry Logic (llm-orchestrator-core/src/retry.rs)
- Configurable retry policies
- Exponential backoff with jitter
- Retryable error detection
Context & Templating (llm-orchestrator-core/src/context.rs)
- Handlebars template engine
- Input/output variable management
- Step result aggregation
CLI Interface (llm-orchestrator-cli/src/main.rs)
- Workflow validation
- Workflow execution
- Structured logging with tracing

Technology Stack

Language: Rust 1.75+ (memory safety, zero-cost abstractions, fearless concurrency)
Async Runtime: Tokio (high-performance async I/O, work-stealing scheduler)
Graph Processing: petgraph (DAG algorithms, topological sort, cycle detection)
HTTP Client: reqwest (connection pooling, async requests)
Serialization: serde + serde_json + serde_yaml (type-safe de/serialization)
Template Engine: handlebars (prompt rendering with context variables)
CLI Framework: clap (argument parsing, subcommands)
Observability: tracing + tracing-subscriber (structured logging)
Concurrency: dashmap (concurrent hash maps), parking_lot (optimized locks)

Workflow Specification

Step Types

LLM Step

Execute an LLM completion:

- id: llm_step
  type: llm
  provider: openai  # or anthropic
  model: gpt-4
  prompt: "Your prompt here with {{inputs.variable_name}}"
  system: "Optional system message"
  temperature: 0.7
  max_tokens: 500
  output:
    - result_variable

Supported Providers:

openai: GPT-3.5, GPT-4, GPT-4 Turbo models
anthropic: Claude 3 (Haiku, Sonnet, Opus) models

Transform Step

Transform data between steps:

- id: transform_step
  type: transform
  function: merge  # Built-in functions: merge, filter, concat
  inputs:
    - var1
    - var2
  output:
    - merged_result

Dependencies

Steps can depend on other steps for sequential execution:

- id: step2
  type: llm
  depends_on:
    - step1
  provider: openai
  model: gpt-4
  prompt: "Use output from step1: {{steps.step1.output_var}}"
  output:
    - result

Conditional Execution

Steps can be conditionally executed based on inputs or previous outputs:

- id: conditional_step
  type: llm
  condition: "inputs.enable == true"
  provider: openai
  model: gpt-3.5-turbo
  prompt: "This only runs if enable is true"
  output:
    - result

Retry Configuration

Configure fault tolerance with exponential backoff:

- id: resilient_step
  type: llm
  provider: openai
  model: gpt-4
  prompt: "Retry me if I fail"
  retry:
    max_attempts: 3
    strategy: exponential
    initial_delay: 1000
    max_delay: 30000
    backoff_multiplier: 2
    jitter: true
  output:
    - result

Programmatic Usage

Using the SDK

use llm_orchestrator_core::{Workflow, WorkflowExecutor};
use llm_orchestrator_providers::{OpenAIProvider, LLMProvider};
use std::sync::Arc;
use std::collections::HashMap;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Load workflow from YAML
    let yaml = std::fs::read_to_string("workflow.yaml")?;
    let workflow = Workflow::from_yaml(&yaml)?;

    // Create inputs
    let mut inputs = HashMap::new();
    inputs.insert("name".to_string(), serde_json::json!("Alice"));

    // Create executor with provider
    let provider = OpenAIProvider::from_env()?;
    let executor = WorkflowExecutor::new(workflow, inputs)?
        .with_provider("openai", Arc::new(provider))
        .with_max_concurrency(4);

    // Execute workflow
    let results = executor.execute().await?;

    // Access results
    for (step_id, result) in results {
        println!("Step {}: {:?}", step_id, result.outputs);
    }

    Ok(())
}

Creating Workflows Programmatically

use llm_orchestrator_core::workflow::*;
use std::collections::HashMap;

let mut workflow = Workflow::new("my-workflow");
workflow.version = "1.0".to_string();

workflow.steps.push(Step {
    id: "greet".to_string(),
    step_type: StepType::Llm,
    depends_on: vec![],
    condition: None,
    config: StepConfig::Llm(LlmStepConfig {
        provider: "openai".to_string(),
        model: "gpt-4".to_string(),
        prompt: "Hello {{inputs.name}}".to_string(),
        temperature: Some(0.7),
        max_tokens: Some(100),
        system: None,
        stream: false,
        extra: HashMap::new(),
    }),
    output: vec!["greeting".to_string()],
    timeout_seconds: None,
    retry: None,
});

Testing

Run the full test suite:

# Run all tests (unit + integration)
cargo test --workspace

# Run with output
cargo test --workspace -- --nocapture

# Run specific test
cargo test test_simple_workflow_execution

# Run integration tests only
cargo test --test integration_test

Test Summary:

Unit tests: 52 tests (41 core + 11 providers)
Integration tests: 4 comprehensive workflow execution tests
Doc tests: 4 documentation examples
Total: 60 tests passing

Observability

The orchestrator includes comprehensive structured logging using the tracing crate:

# Set log level via environment
export RUST_LOG=llm_orchestrator=debug

# Or use the --verbose flag
./target/release/llm-orchestrator run workflow.yaml --verbose

Log Events Include:

Workflow start/completion with execution ID
Step execution status (pending → running → completed/failed)
Retry attempts with backoff duration
Performance metrics (duration, tokens used)
Error details with full context
Provider API call traces

Example Log Output:

2025-11-14T12:00:00Z INFO  llm_orchestrator: Starting workflow execution workflow_id="simple-workflow"
2025-11-14T12:00:00Z INFO  llm_orchestrator: Executing step step_id="greet" step_type=Llm
2025-11-14T12:00:01Z INFO  llm_orchestrator: Step completed successfully step_id="greet" duration_ms=850
2025-11-14T12:00:01Z INFO  llm_orchestrator: Workflow completed successfully

Examples

Complete examples are available in the examples/ directory:

simple.yaml: Two-step workflow with dependencies
transform-only.yaml: Data transformation pipeline

Example: Simple Two-Step Workflow

name: simple-workflow
version: "1.0"
description: A simple two-step workflow

steps:
  - id: greet
    type: llm
    provider: openai
    model: gpt-3.5-turbo
    prompt: "Say hello to {{inputs.name}}"
    temperature: 0.7
    max_tokens: 50
    output:
      - greeting

  - id: analyze
    type: llm
    depends_on:
      - greet
    provider: openai
    model: gpt-3.5-turbo
    prompt: "Analyze this greeting: {{steps.greet.greeting}}"
    temperature: 0.5
    max_tokens: 100
    output:
      - analysis

Run it:

./target/release/llm-orchestrator run examples/simple.yaml \
  --input '{"name": "Alice"}'

Development

Building from Source

# Clone repository
git clone https://github.com/llm-devops/llm-orchestrator.git
cd llm-orchestrator

# Build debug version
cargo build

# Build optimized release version
cargo build --release

# Run tests
cargo test --workspace

# Run with logging
RUST_LOG=debug cargo run --bin llm-orchestrator -- validate examples/simple.yaml

Project Structure

llm-orchestrator/
├── crates/
│   ├── llm-orchestrator-core/      # Core workflow engine
│   │   ├── src/
│   │   │   ├── workflow.rs         # Workflow definitions & parsing
│   │   │   ├── dag.rs              # DAG construction & analysis
│   │   │   ├── executor.rs         # Async execution engine
│   │   │   ├── context.rs          # Execution context & templating
│   │   │   ├── retry.rs            # Retry logic with backoff
│   │   │   ├── error.rs            # Error types & handling
│   │   │   └── providers.rs        # Provider trait definitions
│   │   └── tests/
│   │       └── integration_test.rs # Integration tests
│   ├── llm-orchestrator-providers/ # LLM provider implementations
│   │   └── src/
│   │       ├── openai.rs           # OpenAI integration
│   │       └── anthropic.rs        # Anthropic/Claude integration
│   ├── llm-orchestrator-sdk/       # High-level SDK (future)
│   └── llm-orchestrator-cli/       # Command-line interface
│       └── src/
│           └── main.rs             # CLI implementation
├── examples/                        # Example workflows
│   ├── simple.yaml
│   └── transform-only.yaml
└── Cargo.toml                      # Workspace configuration

Running Tests

# All tests
cargo test --workspace

# Core tests only
cargo test -p llm-orchestrator-core

# Provider tests only
cargo test -p llm-orchestrator-providers

# Integration tests
cargo test --test integration_test

# With verbose output
cargo test --workspace -- --nocapture

Contributing

Contributions are welcome! To contribute:

Fork the repository
Create a feature branch (git checkout -b feature/amazing-feature)
Make your changes
Run tests (cargo test --workspace)
Run formatting (cargo fmt)
Run clippy (cargo clippy)
Commit your changes (git commit -m 'Add amazing feature')
Push to the branch (git push origin feature/amazing-feature)
Open a Pull Request

Performance

Built for production workloads:

Async execution: Non-blocking I/O with Tokio work-stealing scheduler
Parallel execution: Automatic parallelization of independent steps
Efficient memory: Zero-copy where possible, Arc for shared state
Connection pooling: HTTP client reuse across requests
Type safety: Compile-time guarantees eliminate entire classes of runtime errors

Error Handling

Comprehensive error handling with detailed error types:

use llm_orchestrator_core::error::OrchestratorError;

match executor.execute().await {
    Ok(results) => println!("Success: {:?}", results),
    Err(OrchestratorError::ValidationError(e)) => {
        eprintln!("Workflow validation failed: {}", e);
    },
    Err(OrchestratorError::ExecutionError(e)) => {
        eprintln!("Execution error: {}", e);
    },
    Err(e) => {
        eprintln!("Other error: {}", e);
    }
}

Error Categories:

ValidationError: Workflow schema validation failures
ExecutionError: Step execution failures
DagError: Dependency graph errors (cycles, missing steps)
TemplateError: Template rendering errors
Provider-specific errors (rate limits, auth, timeouts)

Roadmap

✅ Phase 1: Core Foundation (Completed)

Workflow parser with YAML support
DAG builder with cycle detection
Execution engine with Tokio
Execution context & templating
Error handling framework

✅ Phase 2: Provider Integration (Completed)

Provider trait abstraction
OpenAI provider (Chat Completions API)
Anthropic provider (Messages API)
Provider registry system
Error conversion & handling

✅ Phase 3: Fault Tolerance (Completed)

Retry executor with policies
Exponential backoff with jitter
Configurable retry strategies
Retryable error detection

✅ Phase 4: CLI & Tooling (Completed)

✅ Phase 5: Testing & Quality (Completed)

📋 Phase 6: Advanced Features (Planned)

Additional providers (Cohere, Google AI, Azure OpenAI)
Embedding step type
Vector search step type
Streaming response support
Workflow templates library
State persistence (SQLite, PostgreSQL)
Metrics & monitoring (Prometheus)
REST API server
Web UI for workflow design

📋 Phase 7: Production Hardening (Future)

License

This project is licensed under the Apache License 2.0.

See LICENSE for the full license text.

Support

Issues: GitHub Issues
Discussions: GitHub Discussions
Documentation: This README and inline code documentation

Acknowledgments

Built with excellent open-source libraries:

Rust - Memory-safe systems programming language
Tokio - Async runtime with work-stealing scheduler
petgraph - Graph data structures and algorithms
reqwest - HTTP client with connection pooling
serde - Serialization framework
handlebars - Template engine
clap - Command-line argument parser
tracing - Structured logging and diagnostics
dashmap - Concurrent hash map

LLM-Orchestrator - Production-ready workflow orchestration for Large Language Models, built with Rust.

Name		Name	Last commit message	Last commit date
Latest commit History 51 Commits
.github/workflows		.github/workflows
benchmarks/output		benchmarks/output
crates		crates
deploy/gcloud		deploy/gcloud
docs		docs
examples		examples
functions		functions
helm		helm
npm		npm
plans		plans
scripts		scripts
tests		tests
.dockerignore		.dockerignore
.gcloudignore		.gcloudignore
.gitignore		.gitignore
Cargo.lock		Cargo.lock
Cargo.toml		Cargo.toml
Dockerfile		Dockerfile
LICENSE		LICENSE
README.md		README.md
cloudbuild.yaml		cloudbuild.yaml
docker-compose.yml		docker-compose.yml
index.js		index.js
package.json		package.json
phase-2b-integration.yaml		phase-2b-integration.yaml

Uh oh!

Folders and files

Latest commit

History

Repository files navigation