Turn multi-step AI tasks into structured, observable, retryable workflows — then let SoloFlow learn repeated patterns and turn them into reusable skills.
AI agents fail in predictable ways:
| Problem | SoloFlow Solution |
|---|---|
| No structure — multi-step chains run as ad-hoc scripts | DAG + FSM engine — dependency-ordered, parallel, state-machine validated |
| No retry/timeout discipline | Per-step retry with exponential backoff, per-step timeouts |
| Amnesiac agents — every invocation starts from zero | Three-tier memory — Working (LRU), Episodic (SQLite FTS5), Semantic (templates) |
| Repeated patterns stay manual | Skill Evolution — observe → detect → package → score → install |
SoloFlow is honest about its scope. Everything below exists in this repo and is tested:
- ✅ DAG engine — Kahn's algorithm layering, DFS cycle detection, ready-step computation
- ✅ FSM state machine — validated transitions for workflow and step states
- ✅ Async scheduler — parallel step execution, exponential-backoff retries, per-step timeouts, pluggable step executor
- ✅ SQLite persistence — workflows, steps, edges, layers, episodic memory, templates
- ✅ Three-tier memory — WorkingMemory (LRU), EpisodicMemory (FTS5 full-text), SemanticMemory (workflow → template extraction)
- ✅ Discipline routing — TaskClassifier (keyword-scored, EN+中文) + DisciplineRouter (one executor per discipline, fallback); the Scheduler dispatches each step to its discipline's executor when no direct executor is set
- ✅ Skill evolution — PatternDetector (fingerprinting), SkillPackager (SKILL.md + plugin.py generation), QualityScorer (4-dimension A-F grading)
- ✅ Hermes plugin —
plugins/soloflow.pywith/soloflow *commands and a memory-provider integration (hermes-plugin/__init__.py, 8 tool schemas) - ✅ Human approval —
hermes-plugin/human/approval.py
Out of scope for now (not in this repo): standalone MCP server, trace/span observability, Ebbinghaus forgetting curve, governance/RBAC.
git clone https://github.com/SonicBotMan/SoloFlow.git
cd SoloFlow
python examples/01_basic_workflow.py # zero dependencies, runs as-isimport 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())Prefer manual control? start_workflow → loop over get_ready_steps → advance_step — see examples/01_basic_workflow.py.
The scheduler decides when to run steps; you decide how:
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"])- Retries:
max_retriesper step (default 2), exponential backoff - Timeouts:
timeout_secondsper step (default 300) - Persistence: every result/error lands in SQLite with full step history
SoloFlow includes a Hermes plugin that watches your workflows and automatically generates reusable skills.
bash install.shOr manually:
cp plugins/soloflow.py ~/.hermes/plugins/
cp -r skills/meta/soloflow ~/.hermes/skills/meta/
cp -r evolution ~/.hermes/plugins/
hermes skills reloadtool_call events → WorkflowBuilder (aggregate) → PatternDetector (fingerprint)
↓
Pattern (2+ occurrences)
↓
SkillPackager → SKILL.md + plugin.py
↓
QualityScorer → grade (A-F)
- WorkflowBuilder accumulates consecutive
tool_callevents 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
- DAG engine integration:
WorkflowService.set_on_complete(callback)feeds completed workflows straight into pattern detection
| 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 |
Tell Hermes naturally — no commands needed:
- "Save this as a skill"
- "Remember how to do this"
- "I always do this manually..."
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)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
pip install -r requirements-dev.txt # pytest + pytest-asyncio
python -m pytest tests/ -v96 tests, all passing.
See CONTRIBUTING.md for guidelines.
MIT License - see LICENSE
- Inspired by LangGraph, AutoGen, and the Agent Harness Engineering research
- Built with ❤️ for the AI Agent community