SoloFlow

KO | EN

gitlite — search

#python #javascript #cpp #r #php #typescript #data-visualization #data #golang #csharp #kotlin #go

★ 67 Open GitHub ↗

Complete ETCLOVG framework for AI Agent workflows - DAG+FSM orchestration, Ebbinghaus memory, discipline routing, skill evolution, trace system, governance. 80+ tests, zero deps, 7/7 layers.

// repository documentation

Was this content helpful?

★ 0 (0 ratings)

# SoloFlow ⚡

### The Brain Behind AI Workflow Orchestration

**Turn chaotic multi-step AI tasks into structured, observable, retryable workflows — with cognitive memory, discipline-aware routing, and automatic skill evolution.**

[![MIT License](https://img.shields.io/badge/license-MIT-blue.svg)](https://raw.githubusercontent.com/SonicBotMan/SoloFlow/main/LICENSE)
[![Tests](https://img.shields.io/badge/tests-68%20passing-brightgreen.svg)](https://raw.githubusercontent.com/SonicBotMan/SoloFlow/main/tests)
[![Python](https://img.shields.io/badge/python-3.10+-blue.svg)](https://www.python.org/)
[![Dependencies](https://img.shields.io/badge/dependencies-0-brightgreen.svg)](https://raw.githubusercontent.com/SonicBotMan/SoloFlow/main/requirements.txt)

---

## Why SoloFlow?

AI Agents fail in predictable ways:

| Problem | SoloFlow Solution |
|---------|-------------------|
| **No Observability** — 8-step chain fails at step 5, no trace, no resume | **Trace System** — nested spans, token tracking, JSON export |
| **Amnesiac Agents** — every invocation starts from zero | **Ebbinghaus Memory** — three-tier memory with forgetting curve |
| **One-Size-Fits-All** — simple tasks waste deep reasoning | **Discipline Routing** — auto-classify to quick/deep/visual/ultrabrain |
| **No Learning** — repeated patterns stay manual | **Skill Evolution** — observe → detect → package → install |

---

## Four Pillars

### 1. DAG + FSM Hybrid Architecture

```
expressiveness(DAG) + rigor(FSM) = reliability
```

- Kahn algorithm for topological sorting
- Parallel execution where possible, sequential where required
- Automatic retry with exponential backoff

### 2. Cognitive Memory System

```
R(t) = base × e^(-t / stability)
```

- **Working Memory** — LRU cache for current context
- **Episodic Memory** — SQLite + FTS5 for event history
- **Semantic Memory** — pattern extraction and template storage
- **Ebbinghaus Forgetting Curve** — automatic memory consolidation

### 3. Discipline-Aware Routing

```
quick (~2s) → deep (~30s) → visual (~30s) → ultrabrain (~120s)
```

- Auto-classify tasks by complexity
- Route to appropriate agent discipline
- Fallback to default when uncertain

### 4. Skill Auto-Evolution

```
observe → fingerprint → detect → package → install
```

- **Passive observation** via `hermes.on("tool_call")` event hooks
- **Multi-step workflow aggregation** — consecutive tool calls grouped automatically
- **Rich step descriptions** — extracts key args into human-readable steps
- **4-dimension quality scoring** — reliability, efficiency, maturity, reusability
- **Auto-generate** SKILL.md + plugin.py and install to `~/.hermes/skills/`

---

## Quick Start

```bash
git clone https://github.com/SonicBotMan/SoloFlow.git
cd SoloFlow
# Pure Python, zero dependencies
```

### Create and Execute a Workflow

```python
import asyncio
from pathlib import Path
from hermes_plugin.store.sqlite_store import SQLiteStore
from hermes_plugin.services.workflow_service import WorkflowService
from hermes_plugin.services.scheduler import Scheduler

async def main():
    store = SQLiteStore(Path("soloflow.db"))
    store.initialize()
    ws = WorkflowService(store)
    ws.set_scheduler(Scheduler(store, ws))

# Create a DAG workflow with parallel branches
    wf = await ws.create_workflow(
        name="research-report",
        description="行业调研报告",
        steps=[
            {"id": "topic",    "name": "选题",   "discipline": "deep",  "prompt": "确定研究方向"},
            {"id": "search_a", "name": "学术搜索", "discipline": "quick", "prompt": "搜索学术资料"},
            {"id": "search_b", "name": "行业搜索", "discipline": "quick", "prompt": "搜索行业报告"},
            {"id": "outline",  "name": "大纲",   "discipline": "deep",  "prompt": "整理大纲"},
            {"id": "write",    "name": "撰写",   "discipline": "deep",  "prompt": "写正文"},
            {"id": "review",   "name": "审校",   "discipline": "quick", "prompt": "审校发布"},
        ],
        edges=[
            ("topic", "search_a"), ("topic", "search_b"),     # parallel branches
            ("search_a", "outline"), ("search_b", "outline"), # merge
            ("outline", "write"), ("write", "review"),
        ],
    )

await ws.start_workflow(wf["id"])
    status = await ws.get_status(wf["id"])
    print(f"State: {status['state']}, Progress: {status['progress']}")

asyncio.run(main())
```

---

## SoloFlow Plugin — Automatic Skill Detection

SoloFlow includes a Hermes plugin that watches your workflows and automatically generates reusable skills.

### Install

```bash
bash install.sh
```

Or manually:

```bash
cp plugins/soloflow.py ~/.hermes/plugins/
cp -r skills/meta/soloflow ~/.hermes/skills/meta/
cp -r evolution ~/.hermes/plugins/
hermes skills reload
```

### How It Works

```
tool_call events → WorkflowBuilder (aggregate) → PatternDetector (fingerprint)
                                                       ↓
                                              Pattern (2+ occurrences)
                                                       ↓
                                              SkillPackager → SKILL.md + plugin.py
                                                       ↓
                                              QualityScorer → grade (A-F)
```

- **WorkflowBuilder** accumulates consecutive `tool_call` events into multi-step workflows (auto-flushes after 60s idle)
- **PatternDetector** fingerprints workflow structure (step names + edges + tools) and groups identical executions
- **SkillPackager** generates Hermes-native SKILL.md and plugin.py with rich step descriptions
- **QualityScorer** rates skills on 4 dimensions: reliability, efficiency, maturity, reusability

### Commands

| Command | Description |
|---------|-------------|
| `/soloflow begin [name]` | Mark workflow start |
| `/soloflow end [name]` | Mark workflow end, record pattern |
| `/soloflow propose` | Analyze session, propose top skill |
| `/soloflow generate [name]` | Generate and install a skill |
| `/soloflow list` | List detected patterns |
| `/soloflow skills` | List generated skills |
| `/soloflow status` | Show tracking status |
| `/soloflow queue` | Show pending proposals |
| `/soloflow clear` | Clear session log |

### Natural Language Triggers

Tell Hermes naturally — no commands needed:

- *"Save this as a skill"*
- *"Remember how to do this"*
- *"Turn this workflow into a reusable skill"*
- *"I always do this manually..."*
- *"Let's automate this"*

### DAG Engine Integration

When a workflow completes through the DAG engine, SoloFlow automatically feeds the execution data to PatternDetector:

```python
from hermes_plugin.services.workflow_service import WorkflowService

ws = WorkflowService(store)
ws.set_on_complete(lambda wf_id, success, duration, wf_def: ...)
# Completed workflows are automatically recorded for pattern detection
```

---

## MCP Tools

5 MCP tools for integration with AI agents:

| Tool | Description |
|------|-------------|
| `soloflow_create` | Create a new workflow with steps and DAG edges |
| `soloflow_run` | Execute a workflow with DAG parallelism |
| `soloflow_status` | Get workflow status and progress |
| `soloflow_list` | List workflows with optional state filter |
| `soloflow_cancel` | Cancel a running workflow |

```yaml
# config.yaml
tools:
  mcp:
    servers:
      soloflow:
        command: python
        args: ["-m", "mcp.server"]
```

---

## Trace System

Track every workflow execution with nested spans:

```python
from trace.collector import TraceCollector
from trace.exporter import TraceExporter
from trace.span import SpanStatus, TokenUsage

collector = TraceCollector(db_path=Path("traces.db"))
exporter = TraceExporter(collector)

span = collector.start_span(operation="workflow", node_name="research")
step = collector.start_span(
    operation="step", node_name="search",
    parent_id=span.span_id, trace_id=span.trace_id,
)
collector.finish_span(
    step.span_id,
    status=SpanStatus.SUCCESS,
    token_usage=TokenUsage(prompt_tokens=100, completion_tokens=200),
)
print(exporter.format_trace_tree(span.trace_id))
```

---

## Ebbinghaus Memory

Memory system with automatic consolidation:

```python
from memory.forgetting.consolidation import MemoryConsolidator

consolidator = MemoryConsolidator(db_path=Path("memory.db"))

await consolidator.add_memory(
    key="user_preference",
    content={"theme": "dark"},
    tier="episodic",
    stability=1.0,
)

entry = await consolidator.get_memory("user_preference")
stats = await consolidator.consolidate_all()
```

---

## Human-in-the-Loop

Approval system for sensitive workflow steps:

```python
from hermes_plugin.human import HumanApprovalManager

manager = HumanApprovalManager()
request = manager.create_request(
    workflow_id="wf_123",
    step_id="review",
    prompt="Please review and approve",
)
result = await manager.wait_for_approval(request.request_id)
```

---

## Governance

Role-based permissions, audit logging, and policy enforcement:

```python
from hermes_plugin.governance import GovernanceManager, Permission

governance = GovernanceManager()
governance.grant_permission("user_1", Permission.EXECUTE)
has_perm = governance.check_permission("user_1", Permission.EXECUTE)
governance.log_audit(
    action=AuditAction.WORKFLOW_STARTED,
    workflow_id="wf_123",
    user_id="user_1",
)
```

---

## Architecture

```
SoloFlow/
├── hermes-plugin/          # Core engine
│   ├── core/               # DAG + FSM
│   ├── services/           # WorkflowService + Scheduler
│   ├── memory/             # Three-tier memory
│   ├── store/              # SQLite persistence
│   ├── checkpoint/         # LangGraph: resumable execution
│   ├── dispatch/           # DeerFlow: sub-agent dispatch
│   ├── roles/              # CrewAI: permission boundaries
│   ├── output/             # PydanticAI: typed contracts
│   ├── boundary/           # Mastra: workflow vs agent control
│   ├── handoff/            # OpenAI Agents SDK: control transfer
│   ├── session/            # Google ADK: session + context budget
│   ├── hooks/              # Claude Agent SDK: lifecycle hooks
│   ├── pipeline/           # Haystack: component orchestration
│   ├── context/            # Microsoft: pluggable context providers
│   ├── governance/         # Permissions + audit
│   ├── human/              # Human approval
│   └── visualization/      # Mermaid diagrams
├── plugins/                # Hermes plugins
│   └── soloflow.py         # Skill detection plugin
├── skills/                 # Hermes skills
│   └── meta/soloflow/      # AI behavior guidance
├── evolution/              # Skill auto-evolution
│   ├── pattern_detector.py # Fingerprint + detect
│   ├── skill_packager.py   # Generate SKILL.md + plugin.py
│   └── quality_scorer.py   # 4-dimension scoring
├── mcp/                    # MCP Tool Layer
├── trace/                  # Observability
├── memory/forgetting/      # Ebbinghaus forgetting curve
├── routing/                # Discipline-aware routing
├── install.sh              # One-command installer
└── tests/                  # Test suite (68 tests)
```

---

## ETCLOVG Coverage

| Layer | Component | Status |
|-------|-----------|--------|
| **T** | MCP Tool Layer | ✅ 5 tools |
| **C** | Ebbinghaus Memory + Context Providers | ✅ Forgetting curve + pluggable context |
| **L** | DAG + FSM Engine + Pipeline | ✅ Core + Haystack-style components |
| **O** | Trace System + Hooks | ✅ Nested spans + lifecycle hooks |
| **V** | Quality Scorer + Output Validation | ✅ 4-dimension scoring + typed contracts |
| **E** | Execution + Dispatch + Handoff | ✅ Sub-agent dispatch + control transfer |
| **G** | Governance + Roles + Session + Boundary | ✅ Role permissions + session mgmt |

**Coverage: 7/7 layers (100%)**

---

## Testing

```bash
# Run all tests
python3.11 -m pytest tests/ -v

# Run specific module
python3.11 -m pytest tests/evolution/ -v
python3.11 -m pytest tests/hermes-plugin/ -v
python3.11 -m pytest tests/mcp/ -v
```

**68 tests, all passing.**

---

## Contributing

See [CONTRIBUTING.md](./CONTRIBUTING.md) for guidelines.

---

## License

MIT License - see [LICENSE](./LICENSE)

---

## Acknowledgments

- Inspired by LangGraph, AutoGen, and the Agent Harness Engineering research
- Built with ❤️ for the AI Agent community

SoloFlow

Explore Similar Repositories

Was this content helpful?

Recent Feedback

Download README