tokio-prompt-orchestrator

Multi-core, Tokio-native orchestration for LLM inference pipelines. Built with bounded backpressure channels, circuit breakers, request deduplication, and an optional autonomous self-tuning control loop.

Architecture

The pipeline is a five-stage directed acyclic graph of bounded async channels. Each stage runs as an independent Tokio task. Backpressure propagates upstream when a downstream channel fills, and excess requests are shed gracefully to a dead-letter queue rather than blocking the pipeline.

[Prompt Submissions] -> [Dedup Stage] -> [Circuit Breaker] -> [Rate Limiter]
                                                                     |
                                                           [Worker Pool (Tokio)]
                                                                     |
                                                     [LLM Providers (Anthropic/OpenAI)]
                                                                     |
                                                 [Self-Improving Control Loop]
                                                                     |
                       [Prometheus Metrics] <- [OpenTelemetry] <- [Results]
                       [Web API (HTTP/WS)]

                        +------------------+
  PromptRequest ------> |  RAG Stage       | cap: 512
                        | (context fetch)  |
                        +--------+---------+
                                 |
                        +--------v---------+
                        |  Assemble Stage  | cap: 512
                        | (prompt builder) |
                        +--------+---------+
                                 |
                        +--------v---------+
                        |  Inference Stage | cap: 1024
                        | (model worker)   |
                        +--------+---------+
                                 |
                        +--------v---------+
                        |  Post Stage      | cap: 512
                        | (filter/format)  |
                        +--------+---------+
                                 |
                        +--------v---------+
                        |  Stream Stage    | cap: 256
                        | (output sink)    |
                        +------------------+

Resilience layers applied at the inference stage:

• Deduplication: in-flight requests with identical prompts are coalesced into a single inference call; all waiting callers receive the same result.
• Circuit breaker: opens on consecutive failures, enters half-open probe mode after a configurable timeout.
• Retry with exponential backoff and jitter.
• Rate limiter: token-bucket guard at the pipeline entry point.
• Dead-letter queue: shed requests are stored in a ring buffer for inspection and replay.

Quickstart

Prebuilt binary (no Rust required)

Download orchestrator.exe from the releases page and run it. The first launch runs an interactive setup wizard that saves your API key and provider preference to orchestrator.env.

Which AI provider do you want to use?

1) Anthropic  (Claude)
2) OpenAI     (GPT-4o)
3) llama.cpp  (local, no key)
4) echo       (offline test mode)

Enter 1, 2, 3, or 4 [1]:

After setup the orchestrator starts a terminal REPL and a web API on http://127.0.0.1:8080 simultaneously. You can type prompts in the terminal while agents and IDEs connect over HTTP in the background.

Library usage

Add the dependency:

[dependencies]
tokio-prompt-orchestrator = "0.1"
tokio = { version = "1", features = ["rt-multi-thread", "macros"] }

Minimal async example using the echo worker (no API key required):

```rust,no_run use std::collections::HashMap; use tokio_prompt_orchestrator::{ spawn_pipeline, EchoWorker, PromptRequest, SessionId, }; use std::sync::Arc;

[tokio::main]

async fn main() -> Result<(), Box> { // Spawn the five-stage pipeline with the echo worker. // Swap EchoWorker for OpenAiWorker, AnthropicWorker, LlamaCppWorker, or VllmWorker. let worker: Arc = Arc::new(EchoWorker::new()); let handles = spawn_pipeline(worker);

// Send a prompt into the pipeline.
handles
    .input_tx
    .send(PromptRequest {
        session: SessionId::new("demo"),
        request_id: "req-1".to_string(),
        input: "Hello, pipeline!".to_string(),
        meta: HashMap::new(),
        deadline: None,
    })
    .await?;

// Collect output from the stream stage.
let mut guard = handles.output_rx.lock().await;
if let Some(rx) = guard.as_mut() {
    if let Some(output) = rx.recv().await {
        println!("Response: {}", output.text);
    }
}

Ok(())

}

See [`examples/`](examples/) for REST, WebSocket, SSE streaming, OpenAI, Anthropic, llama.cpp, and multi-worker round-robin setups.

---

## API Overview

| Type | Module | Description |
|------|--------|-------------|
| `PromptRequest` | `lib` | Input message sent into the pipeline |
| `SessionId` | `lib` | Session identifier for affinity sharding |
| `OrchestratorError` | `lib` | Crate-level error enum |
| `ModelWorker` | `worker` | Async trait implemented by all inference backends |
| `EchoWorker` | `worker` | Returns prompt words as tokens; for testing |
| `OpenAiWorker` | `worker` | OpenAI chat completions API |
| `AnthropicWorker` | `worker` | Anthropic Messages API |
| `LlamaCppWorker` | `worker` | Local llama.cpp HTTP server |
| `VllmWorker` | `worker` | vLLM inference server |
| `LoadBalancedWorker` | `worker` | Round-robin or least-loaded pool of workers |
| `spawn_pipeline` | `stages` | Launch the five-stage pipeline, return channel handles |
| `spawn_pipeline_with_config` | `stages` | Same, with a full `PipelineConfig` |
| `PipelineConfig` | `config` | TOML-deserialisable root configuration type |
| `CircuitBreaker` | `enhanced` | Failure-rate circuit breaker |
| `Deduplicator` | `enhanced` | In-flight request coalescer |
| `RetryPolicy` | `enhanced` | Exponential backoff with jitter |
| `CacheLayer` | `enhanced` | TTL LRU cache for inference results |
| `PriorityQueue` | `enhanced` | Four-level priority scheduler |
| `DeadLetterQueue` | `lib` | Ring buffer of shed requests |
| `send_with_shed` | `lib` | Non-blocking channel send with graceful shedding |
| `shard_session` | `lib` | FNV-1a session affinity shard helper |

---

## Configuration Reference

The orchestrator is configured via a TOML file. Pass it with `--config pipeline.toml`.

```toml
[pipeline]
name        = "production"
version     = "1.0"
description = "Optional description"

[stages.rag]
enabled           = true
timeout_ms        = 5000
max_context_tokens = 2048

[stages.assemble]
enabled          = true
channel_capacity = 512

[stages.inference]
worker      = "open_ai"   # open_ai | anthropic | llama_cpp | vllm | echo
model       = "gpt-4o"
max_tokens  = 1024
temperature = 0.7
timeout_ms  = 30000

[stages.post_process]
enabled = true

[stages.stream]
enabled = true

[resilience]
retry_attempts               = 3
retry_base_ms                = 100
retry_max_ms                 = 5000
circuit_breaker_threshold    = 5
circuit_breaker_timeout_s    = 60
circuit_breaker_success_rate = 0.8

[rate_limits]
enabled             = false
requests_per_second = 100
burst_capacity      = 20

[deduplication]
enabled    = true
window_s   = 300
max_entries = 10000

[observability]
log_format        = "json"    # pretty | json
metrics_port      = 9090      # Prometheus scrape endpoint; omit to disable
tracing_endpoint  = "http://jaeger:4318"  # OTLP endpoint; omit to disable

Environment variables:

Configuration

Timeouts

Each stage can have its own timeout, and individual requests can carry a per-request deadline. These are configured independently:

[stages.rag]
timeout_ms = 5000       # RAG stage timeout (ms)

[stages.inference]
timeout_ms = 30000      # Per-worker inference call timeout (ms)
                        # Overrides DEFAULT_INFERENCE_TIMEOUT_SECS (120 s)

[resilience]
retry_base_ms = 100     # Initial retry delay (ms)
retry_max_ms  = 5000    # Maximum retry delay after back-off (ms)
retry_attempts = 3      # Maximum number of retries per request

Set a per-request deadline in code:

```rust,no_run use tokio_prompt_orchestrator::{PromptRequest, SessionId}; use std::collections::HashMap; use std::time::{Duration, Instant};

let req = PromptRequest { session: SessionId::new("user-1"), request_id: "req-42".into(), input: "Summarise this document.".into(), meta: HashMap::new(), deadline: Some(Instant::now() + Duration::from_secs(10)), };

### Circuit breaker

The circuit breaker opens after `circuit_breaker_threshold` consecutive
failures, waits `circuit_breaker_timeout_s` seconds, then probes in
HALF-OPEN mode. It closes again when the probe success rate meets
`circuit_breaker_success_rate`.

```toml
[resilience]
circuit_breaker_threshold    = 5    # failures before opening
circuit_breaker_timeout_s    = 60   # seconds before half-open probe
circuit_breaker_success_rate = 0.8  # probe success rate to re-close

Rate limiting

Token-bucket rate limiting is opt-in. Enable with --features rate-limiting.

[rate_limits]
enabled             = true
requests_per_second = 100   # sustained rate (tokens/s)
burst_capacity      = 20    # burst allowance on top of the steady rate

Live Dashboard

Start the TUI dashboard with the tui feature:

cargo run --bin tui --features tui

The dashboard shows per-stage queue depths, circuit breaker state, deduplication hit rate, latency sparklines, and a scrolling log panel.

Web API

Enable with --features web-api. The REST and WebSocket API is documented in WEB_API.md.

# Single prompt over REST
curl -X POST http://localhost:8080/v1/prompt \
  -H "Content-Type: application/json" \
  -d '{"input": "What is backpressure?"}'

# Streaming over WebSocket
wscat -c ws://localhost:8080/v1/stream

MCP Integration

Enable the MCP server with --features mcp and point Claude Desktop at it:

{
  "mcpServers": {
    "orchestrator": {
      "url": "http://127.0.0.1:8080"
    }
  }
}

Feature Flags

All features are opt-in. The default build has no optional features.

Deployment Guide

Standalone Binary

cargo build --release --features full
./target/release/orchestrator --config pipeline.toml

The orchestrator exposes: - Web API on port 8080 (configurable) - Prometheus metrics on port 9090 (configurable) - TUI dashboard via cargo run --bin tui --features tui

Docker

```bash docker build -t tokio-prompt-orchestrator . docker run -p 8080:8080 -p 9090:9090 \ -e ANTHROPIC_API_