SoloFlow ⚡

A DAG workflow engine for AI agents — with cognitive memory and automatic skill evolution

Why SoloFlow?

What's Inside (and what's not)

Quick Start

git clone https://github.com/SonicBotMan/SoloFlow.git
cd SoloFlow
python examples/01_basic_workflow.py # zero dependencies, runs as-is

Create and Execute a Workflow

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)
 scheduler = Scheduler(store, ws)
 ws.set_scheduler(scheduler)
 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"),
 ],
 )
 # Plug in your own step executor (LLM call, tool call, ...)
 async def my_executor(step: dict) -> str:
 return await call_your_llm(step["prompt"])
 status = await scheduler.execute_workflow(wf["id"], executor=my_executor)
 print(f"State: {status['state']}, Progress: {status['progress']['completed']}/{status['progress']['total']}")
asyncio.run(main())

The Pluggable Executor

async def my_executor(step: dict) -> str:
 """Called for every step. Raise to trigger retry/backoff; exceed
 step['timeout_seconds'] to time out. Return value is persisted."""
 return await llm.call(step["prompt"])

SoloFlow Plugin — Automatic Skill Detection

Install

bash install.sh

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)

Commands

Natural Language Triggers

Three-Tier Memory

from memory.working_memory import WorkingMemory
from memory.episodic_memory import EpisodicMemory
from memory.semantic_memory import SemanticMemory
wm = WorkingMemory(max_size=100) # LRU, instant context
wm.put("key", {"value": 42})
em = EpisodicMemory(store) # SQLite + FTS5 full-text events
await em.record(event_type="step_completed", data={"step": "search"})
hits = await em.search("search")
sm = SemanticMemory(store) # workflow → reusable template
template = await sm.extract_and_store(completed_workflow)

Architecture

SoloFlow/
├── hermes-plugin/ # Core engine (installable as Hermes memory provider)
│ ├── core/ # DAG (Kahn + cycle detection) + FSM
│ ├── services/ # WorkflowService + Scheduler (pluggable executor)
│ ├── memory/ # Three-tier memory (Working/Episodic/Semantic)
│ ├── store/ # SQLite persistence + migrations
│ ├── agent/ # MemoryProvider re-export
│ ├── human/ # Human approval manager
│ ├── models.py # Dataclasses + enums
│ └── config.py # Env-driven configuration
├── plugins/ # Hermes plugins
│ └── soloflow.py # Skill detection plugin (/soloflow commands)
├── 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 (A-F)
├── routing/ # Discipline-aware routing
│ ├── classifier.py # TaskClassifier (keyword-scored, EN+中文)
│ └── router.py # DisciplineRouter (executor per discipline)
├── examples/ # 6 runnable demos (all verified)
├── tests/ # 96 tests, zero deps beyond pytest
├── install.sh # One-command installer
└── docs/ # API.md + ARCHITECTURE.md

Testing

pip install -r requirements-dev.txt # pytest + pytest-asyncio
python -m pytest tests/ -v

Contributing

License

Acknowledgments