SymFlow

Neuro-Symbolic Workflow Orchestration with Phi-3 Mini LAM

SymFlow is a declarative workflow engine that transforms RDF workflow definitions into executable bash scripts with intelligent validation powered by Phi-3 Mini.

🎯 Core Features

1. RDF-Based Workflow Definition

Define workflows declaratively using RDF/Turtle syntax:

:LoadData a :Task ;
    :agent "agents/data_loader.sh" ;
    :agentConvention "args" ;
    :input "data/raw.csv" ;
    :output "output/loaded.csv" .

:CleanData a :Task ;
    :dependsOn :LoadData ;
    :agent "agents/data_cleaner.sh" ;
    :input "output/loaded.csv" ;
    :output "output/clean.csv" .

2. Hybrid Language Action Model (LAM)

Two-stage intelligent validation:

Stage 1: Rule Guards (< 1ms)

Validates required fields (agent, name)
Checks dependency existence
Ensures root tasks exist
Blocks execution on critical errors

Stage 2: Phi-3 Mini Analysis (~2-5s)

Context-aware semantic validation
Optimization suggestions
Best practice recommendations
Natural language explanations
Confidence scoring (0-100%)

3. Agent Convention Support

Flexible agent argument handling:

args (default): Pass inputs/outputs as command-line arguments
stdout: Redirect stdout/stderr to output files
hybrid: Combine both approaches

4. DAG-Based Execution

Automatic dependency resolution
Cycle detection
Parallelization potential analysis
Topological sorting for execution order

5. Bash Script Generation

ACID-compliant execution (atomic file operations)
Automatic error handling
Progress tracking
Rollback on failure

6. C3AN Platform Integration

Visual Builder: Design workflows using drag-and-drop interface at https://c3anplatform.vercel.app/
Automatic Conversion: SymFlow automatically detects C3AN JSON exports and converts them to RDF
Seamless Execution: Run exported JSON files directly (python3 symflow.py export.json)

🚀 Quick Start

Installation

# Clone repository
git clone <repository-url>
cd SymFlow

# Create virtual environment
python3 -m venv venv
source venv/bin/activate

# Install dependencies
pip install -r requirements.txt

# Install Ollama and pull Phi-3 Mini
curl -fsSL https://ollama.com/install.sh | sh
ollama pull phi3:mini

Basic Usage

# Generate and execute workflow (RDF)
python3 symflow.py workflow.ttl -o script.sh
./script.sh

# Run from C3AN Platform export (JSON)
python3 symflow.py workflow.json -o script.sh
./script.sh

# Validate only (no script generation)
python3 symflow.py workflow.ttl --validate-only

# Fast mode (skip Phi-3 analysis)
python3 symflow.py workflow.ttl -o script.sh --fast-mode

# Skip LAM entirely
python3 symflow.py workflow.ttl -o script.sh --skip-lam

📖 SymFlow Pipeline Architecture

SymFlow operates as a multi-stage compiler that transforms abstract workflow definitions into robust, executable code.

┌─────────────────────────────────────────────────────────────────────────────┐
│                       SymFlow Orchestration Pipeline                        │
├──────────────────────┬─────────────┬──────────────┬────────────┬────────────┤
│   1. INGESTION       │ 2. LAM      │ 3. GRAPH     │ 4. OPTIMIZ │ 5. CODEGEN │
│                      │ VALIDATION  │ CONSTRUCTION │  -ATION    │            │
├──────────────────────┼─────────────┼──────────────┼────────────┼────────────┤
│                      │             │              │            │            │
│  [ .ttl / .json ]    │  [ Neuro-   │   [ DAG ]    │ [ Parallel │ [ .sh ]    │
│         │            │  Symbolic ] │      │       │   Groups ] │      │     │
│         ▼            │      │      │      ▼       │      │     │      ▼     │
│  ┌──────────────┐    │  ┌───────┐  │  ┌───────┐   │ ┌────────┐ │ ┌────────┐ │
│  │  RDF Parser  │    │  │ Rules │  │  │ Nodes │   │ │ Grp A  │ │ │ Header │ │
│  └──────┬───────┘    │  └───┬───┘  │  └───┬───┘   │ └────────┘ │ ├───┼────┤ │
│         │            │      │      │      │       │      │     │ │ Atomic │ │
│         ▼            │      ▼      │      ▼       │      ▼     │ │  Ops   │ │
│  ┌──────────────┐    │  ┌───────┐  │  ┌───────┐   │ ┌────────┐ │ ├───┼────┤ │
│  │ C3AN Adapter │───▶│  │ Phi-3 │  │  │ Edges │   │ │ Grp B  │ │ │ Error  │ │
│  └──────────────┘    │  └───────┘  │  └───┬───┘   │ └────────┘ │ │Handler │ │
│                      │             │      │       │            │ └───┬────┘ │
│                      │             │      ▼       │            │     │      │
│                      │             │  ┌───────┐   │            │     ▼      │
│                      │             │  │ Order │   │            │ [ Script ] │
│                      │             │  └───────┘   │            │            │
│                      │             │              │            │            │
└──────────────────────┴─────────────┴──────────────┴────────────┴────────────┘

Pipeline Stages Breakdown

1. Ingestion & Adapters

Universal Input: Accepts both RDF Turtle files (.ttl) and C3AN JSON exports (.json).
Normalization: The C3ANAdapter converts JSON topologies into standard RDF triples on-the-fly.
Parsing: RDFParser extracts Tasks, Agents, Inputs, Outputs, and Dependencies into internal objects.

2. Neuro-Symbolic Validation (LAM)

Symbolic Guardrails: Rigid checks for cycle detection, missing dependencies, and file path validity. (Speed: <1ms)
Neural Analysis: Phi-3 Mini (3.8B LLM) analyzes the semantics of the workflow. It checks:
- Does TrainModel actually consume data from CleanData?
- Are the agent names descriptive enough?
- Are there security risks in the defined paths?
Result: A confidence score (0-100%) and a list of actionable optimization suggestions.

3. Graph Construction (DAG)

Dependency Resolution: Builds a Directed Acyclic Graph (DAG) using NetworkX.
Topological Sort: Determines the mathematically correct linear execution order (e.g., A must run before B).
Cycle Detection: Guaranteed mathematical proof that the workflow will not hang in an infinite loop.

4. Optimization Engine

Parallelization Detection: The DAG analyzer identifies independent sub-graphs.
Stage Grouping: Tasks that can run simultaneously are grouped together (e.g., ExtractAPI and ExtractDB can run in parallel).
This ensures maximum resource utilization on the target machine.

5. ACID Code Generation

BashGenerator: The final compilation step.
ACID Properties:
- Atomicity: Agents output to temporary files first (.tmp), then purely move them on success.
- Consistency: The script fails fast if any step fails.
- Isolation: Each task runs in its own process space.
- Durability: Results are persisted to disk immediately.
Outcome: A standalone, dependency-free Bash script (.sh) ready for deployment.

C3AN Integration Module

core/c3an_adapter.py

Adapts C3AN Platform JSON exports to SymFlow RDF
Maps visual agents to executable bash scripts
Handles agent ID sanitization and dependency mapping

🌐 C3AN Platform Integration

SymFlow is fully integrated with the C3AN Platform, a visual workflow builder.

How to Use

Design: Create your pipeline visually at c3anplatform.vercel.app
- Drag agents (RAGRetriever, VisionInspector, etc.) onto the canvas
- Connect them to form a pipeline
- Add tools (CNN Model, Vector DB) as needed
Export: Click "Download Triples" to get a .json file
Run: Execute directly with SymFlow

# SymFlow automatically detects JSON and converts it
python3 symflow.py my_visual_workflow.json -o pipeline.sh

Supported Agents

The C3AN adapter (core/c3an_adapter.py) maps visual blocks to these SymFlow agents:

RAGRetriever → tests/agents/rag/retriever.sh
VisionInspector → tests/agents/vision/inspector.sh
TimeSeriesAnalyzer → tests/agents/timeseries/analyzer.sh

🧠 LAM (Language Action Model)

What is LAM?

SymFlow's LAM is a hybrid neuro-symbolic validation system that combines:

Symbolic: Fast, deterministic rule-based checks
Neural: Phi-3 Mini (3.8B parameters) for semantic understanding

How It Works

# Stage 1: Rule Guards (always runs)
guard_results = RuleGuards().validate(tasks)
if not guard_results['valid']:
    print("❌ Critical errors found")
    exit(1)  # BLOCK EXECUTION

# Stage 2: Phi-3 Analysis (optional)
phi3_results = Phi3LAM().analyze(tasks, guard_results)
print(f"💡 Suggestions: {phi3_results['suggestions']}")
print(f"✓ Confidence: {phi3_results['confidence']}%")

Example Output

🧠 Running LAM analysis...
   🤖 Full mode: Rule guards + Phi-3 semantic analysis
   
   ⚠ [WARNING] CleanData: No explicit error checking before EngineerFeatures
      → Add guard clause to verify integrity of 'loaded_data.csv'
   
   💡 Suggestions:
      BEST_PRACTICE: Include data validation steps within CleanData
         Benefit: Reduces potential errors in feature generation
      OPTIMIZATION: Consider parallelizing CleanData and EngineerFeatures
         Benefit: Decreases overall workflow time
   
   📝 Workflow follows logical dependencies but could benefit from
      explicit guard clauses to ensure data integrity.
   
   ✓ Workflow validated (confidence: 70%)
   ℹ Analysis source: phi3_lam

LAM Modes

Mode	Speed	Validation	Use Case
Full (default)	~2-5s	Rule guards + Phi-3	Production workflows
Fast (`--fast-mode`)	< 1ms	Rule guards only	CI/CD, quick validation
Skip (`--skip-lam`)	0ms	None	Trusted workflows

📝 RDF Workflow Syntax

Task Definition

@prefix : <http://symflow.org/workflow#> .

:TaskName a :Task ;
    rdfs:label "Human-readable name" ;
    :agent "path/to/agent.sh" ;
    :agentConvention "args" ;  # or "stdout", "hybrid"
    :dependsOn :PreviousTask ;
    :input "input/file.csv" ;
    :output "output/file.csv" ;
    :description "What this task does" .

Agent Conventions

Args Convention (default):

# Agent receives: agent.sh INPUT OUTPUT
./data_loader.sh "input.csv" "output.csv"

Stdout Convention:

# Agent writes to stdout, redirected to file
./data_loader.sh "input.csv" > "output.csv"

Hybrid Convention:

# Combination: args for inputs, stdout for last output
./processor.sh "input.csv" "temp.log" > "final_output.csv"

🔍 Example Workflows

Data Science Pipeline

:LoadData → :CleanData → :EngineerFeatures → :TrainModel → :EvaluateModel

Features:

Loads CSV data
Cleans and validates
Engineers features
Trains ML model
Evaluates performance

LAM Insights:

Suggests data validation steps
Recommends parallelization
Identifies missing error handling

Document Processing Pipeline

:ExtractPDF → :ChunkText → :GenerateEmbeddings → :StoreVectors → :QuerySystem

Features:

Extracts text from PDFs
Chunks into segments
Generates embeddings
Stores in vector database
Enables semantic search

LAM Insights:

Validates chunk integrity
Suggests parallel processing
Recommends validation steps

ETL Pipeline

:ExtractAPI → :TransformAPI ┐
:ExtractDB  → :TransformDB  ├→ :MergeData → :LoadToDB
:ExtractFiles → :TransformFiles ┘

Features:

Parallel extraction from multiple sources
Data transformation
Merging and loading

LAM Insights:

Identifies race conditions
Suggests input validation
Recommends error handling

🛡️ Error Handling

LAM Blocking Example

Broken Workflow (invalid dependencies):

:CleanData a :Task ;
    :dependsOn :LoadData, :ValidateData ;  # ValidateData doesn't exist!
    ...

LAM Output:

🧠 Running LAM analysis...
   ✗ [ERROR] CleanData: Dependency "ValidateData" does not exist
   ✗ [ERROR] ProcessData: Dependency "NormalizeData" does not exist

❌ Workflow validation failed

Result: No bash script generated, execution blocked

🎓 Advanced Features

Custom Agents

Create bash agents that follow conventions:

#!/bin/bash
# data_loader.sh - Args convention

INPUT=$1
OUTPUT=$2

# Load and process data
python3 << EOF
import pandas as pd
df = pd.read_csv('$INPUT')
# ... processing ...
df.to_csv('$OUTPUT', index=False)
EOF

echo "✓ Loaded data: $OUTPUT"

Workflow Visualization

python3 symflow.py workflow.ttl --visualize

Generates DAG visualization showing:

Task dependencies
Execution order
Parallelization opportunities

Validation Only

python3 symflow.py workflow.ttl --validate-only

Runs LAM analysis without generating scripts:

Checks structural validity
Provides Phi-3 insights
Shows confidence score

📊 Performance

Metric	Value
Rule Guards	< 1ms
Phi-3 Analysis	~2-5 seconds
DAG Building	< 10ms
Script Generation	< 50ms
Total (Full Mode)	~2-5 seconds
Total (Fast Mode)	< 100ms

🔧 Configuration

Environment Variables

# Ollama API endpoint (default: http://localhost:11434)
export OLLAMA_HOST=http://localhost:11434

# Phi-3 model (default: phi3:mini)
export PHI3_MODEL=phi3:mini

Requirements

Python 3.8+
Ollama (for Phi-3 Mini)
Bash 4.0+
Dependencies: rdflib, networkx

📚 Documentation

Command-Line Options

usage: symflow.py [-h] [-o OUTPUT] [--skip-lam] [--fast-mode] 
                  [--visualize] [--validate-only]
                  workflow

positional arguments:
  workflow              Path to RDF workflow file (.ttl)

optional arguments:
  -h, --help            Show help message
  -o OUTPUT             Output bash script path
  --skip-lam            Skip LAM validation entirely
  --fast-mode           Use fast mode (rule guards only)
  --visualize, -v       Generate DAG visualization
  --validate-only       Only validate, don't generate script

Exit Codes

0: Success
1: Validation failed (LAM blocked)
2: RDF parsing error
3: DAG cycle detected
4: Script generation error

🧪 Testing

📘 Detailed Guide: See EXAMPLES.md for a comprehensive breakdown of all example workflows and their logic.

Example workflows are provided in tests/workflows/:

data_science_pipeline.ttl - ML workflow
document_processing_pipeline.ttl - RAG system
etl_parallel_pipeline.ttl - Data integration
llm_content_pipeline.ttl - Content generation
monitoring_pipeline.ttl - System monitoring
broken_workflow.ttl - Demonstrates LAM error detection

Run all tests:

for workflow in tests/workflows/*.ttl; do
    echo "Testing: $workflow"
    python3 symflow.py "$workflow" --validate-only
done

📄 License

MIT License - See LICENSE file for details

🙏 Acknowledgments

Phi-3 Mini by Microsoft Research
Ollama for local LLM inference
RDFLib for RDF parsing
NetworkX for graph algorithms

SymFlow: Declarative workflows meet intelligent validation 🚀

Name		Name	Last commit message	Last commit date
core		core
tests		tests
.gitignore		.gitignore
EXAMPLES.md		EXAMPLES.md
README.md		README.md
requirements.txt		requirements.txt
symflow.py		symflow.py

Latest commit

History

Folders and files

Repository files navigation

SymFlow

🎯 Core Features

1. RDF-Based Workflow Definition

2. Hybrid Language Action Model (LAM)

3. Agent Convention Support

4. DAG-Based Execution

5. Bash Script Generation

6. C3AN Platform Integration

🚀 Quick Start

Installation

Basic Usage

📖 SymFlow Pipeline Architecture

Pipeline Stages Breakdown

1. Ingestion & Adapters

2. Neuro-Symbolic Validation (LAM)

3. Graph Construction (DAG)

4. Optimization Engine

5. ACID Code Generation

C3AN Integration Module

🌐 C3AN Platform Integration

How to Use

Supported Agents

🧠 LAM (Language Action Model)

What is LAM?

How It Works

Example Output

LAM Modes

📝 RDF Workflow Syntax

Task Definition

Agent Conventions

🔍 Example Workflows

Data Science Pipeline

Document Processing Pipeline

ETL Pipeline

🛡️ Error Handling

LAM Blocking Example

🎓 Advanced Features

Custom Agents

Workflow Visualization

Validation Only

📊 Performance

🔧 Configuration

Environment Variables

Requirements

📚 Documentation

Command-Line Options

Exit Codes

🧪 Testing

📄 License

🙏 Acknowledgments

About

Resources

Stars

Watchers

Forks

Releases

Packages

Contributors

Languages