weavegraph-0.1.0-alpha.5 has been yanked.

Weavegraph

NOTE: NodeKind::Start and NodeKind::End are virtual structural endpoints.
You never register them with add_node; attempts to do so are ignored with a warning.
Define only your executable (custom) nodes and connect them with edges from Start and to End.

Graph-driven, concurrent agent workflow framework with versioned state, deterministic barrier merges, and rich diagnostics.

Weavegraph is a modern Rust framework for building complex, stateful workflows using graph-based execution. It's designed for AI agents, data processing pipelines, and any application requiring sophisticated state management with concurrent node execution.

✨ Key Features

🔄 Concurrent Graph Execution: Bounded-concurrency scheduler with dependency resolution and versioned barrier synchronization
📝 Rich Message System: Type-safe message construction with role-based messaging for AI workflows
🎯 Versioned State Management: Channel-based state with snapshot isolation and deterministic merges
🚨 Comprehensive Error Handling: Structured error propagation with beautiful diagnostics via miette and thiserror
📊 Built-in Observability: Rich tracing spans and event streaming for monitoring and debugging
💾 Flexible Persistence: SQLite checkpointing with automatic schema management, plus in-memory options
🎭 Conditional Workflows: Dynamic routing and edge conditions based on runtime state
🔧 Developer Experience: Extensive examples, comprehensive docs, and ergonomic APIs

🚀 Quick Start

Add Weavegraph to your Cargo.toml:

[dependencies]
weavegraph = "0.1"

Basic Example

use weavegraph::{
    graph::GraphBuilder,
    message::Message,
    node::{Node, NodeContext, NodePartial},
    state::VersionedState,
};
use async_trait::async_trait;

// Define a simple greeting node
struct GreetingNode;

#[async_trait]
impl Node for GreetingNode {
    async fn run(
        &self,
        _snapshot: weavegraph::state::StateSnapshot,
        ctx: NodeContext,
    ) -> Result<NodePartial, weavegraph::node::NodeError> {
        ctx.emit("greeting", "Generating welcome message")?;

        let greeting = Message::assistant("Hello! How can I help you today?");

        Ok(NodePartial::new().with_messages(vec![greeting]))
    }
}

#[tokio::main]
async fn main() -> miette::Result<()> {
    // Build a simple graph with a virtual Start -> greet -> End topology.
    use weavegraph::types::NodeKind;
    let app = GraphBuilder::new()
        .add_node(NodeKind::Custom("greet".into()), GreetingNode)
        .add_edge(NodeKind::Start, NodeKind::Custom("greet".into()))
        .add_edge(NodeKind::Custom("greet".into()), NodeKind::End)
        .compile();

    // Create initial state and run
    let state = VersionedState::new_with_user_message("Hello, system!");
    let result = app.invoke(state).await?;

    // Access results
    for message in result.messages.snapshot() {
        println!("{}: {}", message.role, message.content);
    }

    Ok(())
}

📋 Core Concepts

Messages

Messages are the primary communication primitive with convenient constructors:

use weavegraph::message::Message;

// Use convenience constructors (recommended)
let user_msg = Message::user("What's the weather like?");
let assistant_msg = Message::assistant("It's sunny and 75°F!");
let system_msg = Message::system("You are a helpful assistant.");

// For custom roles
let function_msg = Message::new("function", "Processing complete");

// Complex cases with builder pattern
let complex_msg = Message::builder()
    .role("custom_agent")
    .content("Task completed successfully")
    .build();

State Management

Versioned state with channel isolation and snapshot consistency:

use weavegraph::state::VersionedState;

// Simple initialization
let state = VersionedState::new_with_user_message("Hello!");

// Rich initialization with builder
let state = VersionedState::builder()
    .with_user_message("What's the weather?")
    .with_system_message("You are a weather assistant")
    .with_extra("location", serde_json::json!("San Francisco"))
    .build();

Graph Building

Declarative workflow definition with conditional routing:

use weavegraph::graphs::GraphBuilder;
use weavegraph::types::NodeKind;
use std::sync::Arc;

let graph = GraphBuilder::new()
    .add_node(NodeKind::Custom("input".into()), InputProcessorNode)
    .add_node(NodeKind::Custom("analyze".into()), AnalyzerNode)
    .add_node(NodeKind::Custom("respond".into()), ResponseNode)
    .add_node(NodeKind::Custom("escalate".into()), EscalateNode)
    // Virtual Start/End: connect from Start and into End explicitly
    .add_edge(NodeKind::Start, NodeKind::Custom("input".into()))
    .add_edge(NodeKind::Custom("input".into()), NodeKind::Custom("analyze".into()))
  .add_conditional_edge(
    NodeKind::Custom("analyze".into()),
    Arc::new(|state| {
      if state.extra.contains_key("needs_escalation") {
        "escalate".to_string()
      } else {
        "respond".to_string()
      }
    })
  )
    .add_edge(NodeKind::Custom("respond".into()), NodeKind::End)
    .add_edge(NodeKind::Custom("escalate".into()), NodeKind::End)
    .compile();

Note: Conditional predicates must return the name of a valid next node or a virtual endpoint. The runtime accepts:

Custom nodes by name (e.g., "respond", "escalate") that were registered via add_node
The virtual endpoints "Start" and "End" If a predicate returns an unknown target, the route is skipped and a warning is logged.

Conditional Edges

Use conditional edges to route dynamically based on runtime state. Predicates return target node names (Vec), allowing flexible routing to single or multiple nodes.

Compact example:

use std::sync::Arc;
use weavegraph::graphs::{GraphBuilder, EdgePredicate};
use weavegraph::types::NodeKind;

let route: EdgePredicate = Arc::new(|snap| {
  if snap.extra.contains_key("needs_escalation") {
    vec!["escalate".to_string()]
  } else {
    vec!["respond".to_string()]
  }
});

let app = GraphBuilder::new()
  .add_node(NodeKind::Custom("analyze".into()), AnalyzeNode)
  .add_node(NodeKind::Custom("respond".into()), RespondNode)
  .add_node(NodeKind::Custom("escalate".into()), EscalateNode)
  .add_edge(NodeKind::Start, NodeKind::Custom("analyze".into()))
  .add_conditional_edge(NodeKind::Custom("analyze".into()), route)
  .add_edge(NodeKind::Custom("respond".into()), NodeKind::End)
  .add_edge(NodeKind::Custom("escalate".into()), NodeKind::End)
  .compile();

Troubleshooting:

If nothing happens after a node with a conditional edge, ensure the predicate returns valid target names matching registered nodes, or the virtual endpoints "Start"/"End".
For readability, use small helper predicates (EdgePredicate) and unit test them with sample StateSnapshots.

🔧 Examples

The repository includes comprehensive examples demonstrating different patterns:

# Basic node patterns and message handling
cargo run --example basic_nodes

# Advanced patterns: error handling, conditional routing, data transformation
cargo run --example advanced_patterns

# Error handling and pretty diagnostics
cargo run --example errors_pretty

Demo Applications

Historical demo applications showcase evolution of capabilities:

# Basic graph execution patterns (examples/demo1.rs)
cargo run --example demo1

# Direct scheduler usage and barrier synchronization (examples/demo2.rs)
cargo run --example demo2

# LLM workflows with Ollama integration (examples/demo3.rs)
cargo run --example demo3

# Advanced multi-step workflows (examples/demo4.rs)
cargo run --example demo4

Note: Demo3 requires Ollama running at http://localhost:11434 with models like gemma3. Use the provided docker-compose.yml to set up Ollama:

docker-compose up -d ollama

🏗️ Architecture

Weavegraph is built around several core modules:

[message] - Type-safe message construction and role-based messaging
[state] - Versioned state management with channel isolation
[node] - Node execution primitives and async trait definitions
[graph] - Workflow graph definition and conditional routing
[schedulers] - Concurrent execution with dependency resolution
[runtimes] - High-level execution runtime and checkpointing
[channels] - Channel-based state storage and versioning
[reducers] - State merge strategies and conflict resolution
[event_bus] - Event streaming and observability infrastructure

🔍 Observability & Debugging

Tracing

Rich tracing integration with configurable log levels:

# Debug level for weavegraph modules
RUST_LOG=debug cargo run --example basic_nodes

# Error level globally, debug for weavegraph
RUST_LOG=error,weavegraph=debug cargo run --example advanced_patterns

Event Streaming

Built-in event bus for monitoring workflow execution:

use weavegraph::event_bus::{EventBus, MemorySink};

// Custom event handling for testing
let event_bus = EventBus::with_sink(MemorySink::new());
let runner = AppRunner::with_bus(graph, event_bus);

// Events include node starts/completions, state changes, errors

Error Diagnostics

Beautiful error reporting with context and suggestions:

// Automatic error context and pretty printing
fn main() -> miette::Result<()> {
    // Your workflow code here
    Ok(())
}

💾 Persistence

SQLite Checkpointing

Automatic state persistence with configurable database location:

use weavegraph::runtimes::SqliteCheckpointer;

let checkpointer = SqliteCheckpointer::new("sqlite://workflow.db").await?;
let runner = AppRunner::with_checkpointer(graph, checkpointer);

Database URL resolution order:

WEAVEGRAPH_SQLITE_URL environment variable
Explicit URL in code
SQLITE_DB_NAME environment variable (filename only)
Default: sqlite://weavegraph.db

In-Memory Mode

For testing and ephemeral workflows:

let runner = AppRunner::new(graph); // Uses in-memory state

🧪 Testing

Run the comprehensive test suite:

# All tests with output
cargo test --all -- --nocapture

# Specific test categories
cargo test schedulers:: -- --nocapture
cargo test channels:: -- --nocapture
cargo test integration:: -- --nocapture

Property-based testing with proptest ensures correctness across edge cases.

Overview mermain flowchart of the app


flowchart TB

subgraph Client
  user[Client App or UI]
end

subgraph Build
  gb[GraphBuilder]
end

subgraph Graph
  cg[CompiledGraph]
end

subgraph Runtime
  app[App]
  sched[Scheduler]
  router[Router Edges and Commands]
  barrier[Barrier Applier]
end

subgraph Nodes
  usernode[User Nodes]
  llmnode[LLM Node]
  toolnode[Tool Node]
end

subgraph State
  vstate[Versioned State]
  snap[State Snapshot]
end

subgraph Reducers
  redreg[Reducer Registry]
end

subgraph Checkpoint
  cpif[Checkpointer]
end

subgraph Rig
  rigad[Rig Adapter]
  llmprov[LLM Provider]
end

subgraph Tools
  toolreg[Tool Registry]
  exttools[External Tools]
end

subgraph Stream
  stream[Stream Controller]
end

subgraph Viz
  viz[Visualizer]
end


user --> gb
gb --> cg

user --> app
cg --> app

app --> sched
sched --> snap
vstate --> snap

sched --> usernode
sched --> llmnode
sched --> toolnode

usernode --> barrier
llmnode --> barrier
toolnode --> barrier
redreg --> barrier
barrier --> vstate

snap --> router
app --> router
router --> sched

llmnode --> rigad
rigad --> llmprov
llmprov --> rigad
rigad --> llmnode

toolnode --> toolreg
toolnode --> exttools
exttools --> toolnode

barrier --> cpif

app --> stream
stream --> user

cg --> viz

🚀 Production Considerations

Performance

Bounded concurrency prevents resource exhaustion
Snapshot isolation eliminates state races
Channel-based architecture enables efficient partial updates
SQLite checkpointing supports failure recovery

Monitoring

Structured event streaming for observability platforms
Rich tracing spans for distributed tracing
Error aggregation and pretty diagnostics
Custom event sinks for metrics collection

Deployment

Docker-ready with provided compose configuration
Environment-based configuration
Graceful shutdown handling
Migration support for schema evolution

🎓 Project Background

Weavegraph originated as a capstone project for a Rust online course, developed by contributors with Python backgrounds and experience with LangGraph and LangChain. The goal was to bring similar graph-based workflow capabilities to Rust while leveraging its performance, safety, and concurrency advantages.

While rooted in educational exploration, Weavegraph has evolved into a production-ready framework with continued active development well beyond the classroom setting.

🤝 Contributing

We welcome contributions! Please see our contributing guidelines for details.

Areas we're particularly interested in:

Additional persistence backends (PostgreSQL, Redis)
Enhanced AI/LLM integration patterns
Performance optimizations
Documentation improvements
Example applications

📄 License

This project is licensed under the MIT License - see the LICENSE file for details.

weavegraph 0.1.0-alpha.5