SoloFlow

Pure Python, zero dependencies

```

Quick Start

```bash git clone https://github.com/SonicBotMan/SoloFlow.git cd SoloFlow

Usage with Hermes Agent

```yaml

Finish with token usage

collector.finish_span( step.span_id, status=SpanStatus.SUCCESS, token_usage=TokenUsage(prompt_tokens=100, completion_tokens=200), )

config.yaml

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

---

The Brain Behind AI Workflow Orchestration

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

---

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():
    # Initialize
    store = SQLiteStore(Path("soloflow.db"))
    store.initialize()
    ws = WorkflowService(store)
    ws.set_scheduler(Scheduler(store, ws))

    # Create workflow
    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"),
            ("search_a", "outline"), ("search_b", "outline"),
            ("outline", "write"), ("write", "review"),
        ],
    )

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

asyncio.run(main())

---

Run specific module

python -m pytest tests/mcp/ -v python -m pytest tests/trace/ -v python -m pytest tests/memory/ -v python -m pytest tests/routing/ -v python -m pytest tests/evolution/ -v

Workflow Visualization (NEW)

Generate Mermaid diagrams from workflows:

```python from hermes_plugin.visualization import WorkflowVisualizer

visualizer = WorkflowVisualizer()