edict-multi-agent-orchestration

Edict (三省六部) Multi-Agent Orchestration

Skill by ara.so — Daily 2026 Skills collection.

Edict implements a 1400-year-old Tang Dynasty governance model as an AI multi-agent architecture. Twelve specialized agents form a checks-and-balances pipeline: Crown Prince (triage) → Zhongshu (planning) → Menxia (review/veto) → Shangshu (dispatch) → Six Ministries (parallel execution). Built on OpenClaw, it provides a real-time React kanban dashboard, full audit trails, and per-agent LLM configuration.

Architecture Overview

You (Emperor) → taizi (triage) → zhongshu (plan) → menxia (review/veto)
             → shangshu (dispatch) → [hubu|libu|bingbu|xingbu|gongbu|libu2] (execute)
             → memorial (result archived)

Key differentiator vs CrewAI/AutoGen: Menxia (门下省) is a mandatory quality gate — it can veto and force rework before tasks reach executors.

Prerequisites

• OpenClaw installed and running

• Python 3.9+

• Node.js 18+ (for React dashboard build)

• macOS or Linux

Installation

Quick Demo (Docker — no OpenClaw needed)

# x86/amd64 (Ubuntu, WSL2)
docker run --platform linux/amd64 -p 7891:7891 cft0808/sansheng-demo

# Apple Silicon / ARM
docker run -p 7891:7891 cft0808/sansheng-demo

# Or with docker-compose (platform already set)
docker compose up

Open http://localhost:7891

Full Installation

git clone https://github.com/cft0808/edict.git
cd edict
chmod +x install.sh && ./install.sh

The install script automatically:

• Creates all 12 agent workspaces (taizi, zhongshu, menxia, shangshu, hubu, libu, bingbu, xingbu, gongbu, libu2, zaochao, legacy-compat)

• Writes SOUL.md role definitions to each agent workspace

• Registers agents and permission matrix in openclaw.json

• Symlinks shared data directories across all agent workspaces

• Sets sessions.visibility all for inter-agent message routing

• Syncs API keys across all agents

• Builds React frontend

• Initializes data directory and syncs official stats

First-time API Key Setup

# Configure API key on first agent
openclaw agents add taizi
# Then re-run install to propagate to all agents
./install.sh

Running the System

# Terminal 1: Data refresh loop (keeps kanban data current)
bash scripts/run_loop.sh

# Terminal 2: Dashboard server
python3 dashboard/server.py

# Open dashboard
open http://127.0.0.1:7891

Key Commands

OpenClaw Agent Management

# List all registered agents
openclaw agents list

# Add/configure an agent
openclaw agents add <agent-name>

# Check agent status
openclaw agents status

# Restart gateway (required after config changes)
openclaw gateway restart

# Send a message/edict to the system
openclaw send taizi "帮我分析一下竞争对手的产品策略"

Dashboard Server

# dashboard/server.py — serves on port 7891
# Built-in: React frontend + REST API + WebSocket updates
python3 dashboard/server.py

# Custom port
PORT=8080 python3 dashboard/server.py

Data Scripts

# Sync official (agent) statistics
python3 scripts/sync_officials.py

# Update kanban task states
python3 scripts/kanban_update.py

# Run news aggregation
python3 scripts/fetch_news.py

# Full refresh loop (runs all scripts in sequence)
bash scripts/run_loop.sh

Configuration

Agent Model Configuration (openclaw.json)

{
  "agents": {
    "taizi": {
      "model": "claude-3-5-sonnet-20241022",
      "workspace": "~/.openclaw/workspaces/taizi"
    },
    "zhongshu": {
      "model": "gpt-4o",
      "workspace": "~/.openclaw/workspaces/zhongshu"
    },
    "menxia": {
      "model": "claude-3-5-sonnet-20241022",
      "workspace": "~/.openclaw/workspaces/menxia"
    },
    "shangshu": {
      "model": "gpt-4o-mini",
      "workspace": "~/.openclaw/workspaces/shangshu"
    }
  },
  "gateway": {
    "port": 7891,
    "sessions": {
      "visibility": "all"
    }
  }
}

Per-Agent Model Hot-Switching (via Dashboard)

Navigate to ⚙️ Models panel → select agent → choose LLM → Apply. Gateway restarts automatically (~5 seconds).

Environment Variables

# API keys (set before running install.sh or openclaw)
export ANTHROPIC_API_KEY="sk-ant-..."
export OPENAI_API_KEY="sk-..."

# Optional: Feishu/Lark webhook for notifications
export FEISHU_WEBHOOK_URL="https://open.feishu.cn/open-apis/bot/v2/hook/..."

# Optional: news aggregation
export NEWS_API_KEY="..."

# Dashboard port override
export DASHBOARD_PORT=7891

Agent Roles Reference

Agent Role Responsibility

taizi 太子 Crown Prince Triage: chat → auto-reply, edicts → create task

zhongshu 中书省 Planning: decompose edict into subtasks

menxia 门下省 Review/Veto: quality gate, can reject and force rework

shangshu 尚书省 Dispatch: assign subtasks to ministries

hubu 户部 Ministry of Revenue Finance, data analysis tasks

libu 礼部 Ministry of Rites Communication, documentation tasks

bingbu 兵部 Ministry of War Strategy, security tasks

xingbu 刑部 Ministry of Justice Review, compliance tasks

gongbu 工部 Ministry of Works Engineering, technical tasks

libu2 吏部 Ministry of Personnel HR, agent management tasks

zaochao 早朝官 Morning briefing aggregator

Permission Matrix (who can message whom)

# Defined in openclaw.json — enforced by gateway
PERMISSIONS = {
    "taizi":    ["zhongshu"],
    "zhongshu": ["menxia"],
    "menxia":   ["zhongshu", "shangshu"],  # can veto back to zhongshu
    "shangshu": ["hubu", "libu", "bingbu", "xingbu", "gongbu", "libu2"],
    # ministries report back up the chain
    "hubu":     ["shangshu"],
    "libu":     ["shangshu"],
    "bingbu":   ["shangshu"],
    "xingbu":   ["shangshu"],
    "gongbu":   ["shangshu"],
    "libu2":    ["shangshu"],
}

Task State Machine

# scripts/kanban_update.py enforces valid transitions
VALID_TRANSITIONS = {
    "pending":     ["planning"],
    "planning":    ["reviewing", "pending"],      # zhongshu → menxia
    "reviewing":   ["dispatching", "planning"],   # menxia approve or veto
    "dispatching": ["executing"],
    "executing":   ["completed", "failed"],
    "completed":   [],
    "failed":      ["pending"],  # retry
}

# Invalid transitions are rejected — no silent state corruption

Real Code Examples

Send an Edict Programmatically

import subprocess
import json

def send_edict(message: str, agent: str = "taizi") -> dict:
    """Send an edict to the Crown Prince for triage."""
    result = subprocess.run(
        ["openclaw", "send", agent, message],
        capture_output=True,
        text=True
    )
    return {"stdout": result.stdout, "returncode": result.returncode}

# Example edicts
send_edict("分析本季度用户增长数据，找出关键驱动因素")
send_edict("起草一份关于产品路线图的对外公告")
send_edict("审查现有代码库的安全漏洞")

Read Kanban State

import json
from pathlib import Path

def get_kanban_tasks(data_dir: str = "data") -> list[dict]:
    """Read current kanban task state."""
    tasks_file = Path(data_dir) / "tasks.json"
    if not tasks_file.exists():
        return []
    with open(tasks_file) as f:
        return json.load(f)

def get_tasks_by_status(status: str) -> list[dict]:
    tasks = get_kanban_tasks()
    return [t for t in tasks if t.get("status") == status]

# Usage
executing = get_tasks_by_status("executing")
completed = get_tasks_by_status("completed")
print(f"In progress: {len(executing)}, Done: {len(completed)}")

Update Task Status (with validation)

import json
from pathlib import Path
from datetime import datetime, timezone

VALID_TRANSITIONS = {
    "pending":     ["planning"],
    "planning":    ["reviewing", "pending"],
    "reviewing":   ["dispatching", "planning"],
    "dispatching": ["executing"],
    "executing":   ["completed", "failed"],
    "completed":   [],
    "failed":      ["pending"],
}

def update_task_status(task_id: str, new_status: str, data_dir: str = "data") -> bool:
    """Update task status with state machine validation."""
    tasks_file = Path(data_dir) / "tasks.json"
    tasks = json.loads(tasks_file.read_text())

    task = next((t for t in tasks if t["id"] == task_id), None)
    if not task:
        raise ValueError(f"Task {task_id} not found")

    current = task["status"]
    allowed = VALID_TRANSITIONS.get(current, [])

    if new_status not in allowed:
        raise ValueError(
            f"Invalid transition: {current} → {new_status}. "
            f"Allowed: {allowed}"
        )

    task["status"] = new_status
    task["updated_at"] = datetime.now(timezone.utc).isoformat()
    task.setdefault("history", []).append({
        "from": current,
        "to": new_status,
        "timestamp": task["updated_at"]
    })

    tasks_file.write_text(json.dumps(tasks, ensure_ascii=False, indent=2))
    return True

Dashboard REST API Client

import urllib.request
import json

BASE_URL = "http://127.0.0.1:7891/api"

def api_get(endpoint: str) -> dict:
    with urllib.request.urlopen(f"{BASE_URL}{endpoint}") as resp:
        return json.loads(resp.read())

def api_post(endpoint: str, data: dict) -> dict:
    payload = json.dumps(data).encode()
    req = urllib.request.Request(
        f"{BASE_URL}{endpoint}",
        data=payload,
        headers={"Content-Type": "application/json"},
        method="POST"
    )
    with urllib.request.urlopen(req) as resp:
        return json.loads(resp.read())

# Read dashboard data
tasks    = api_get("/tasks")
agents   = api_get("/agents")
sessions = api_get("/sessions")
news     = api_get("/news")

# Trigger task action
api_post("/tasks/pause",  {"task_id": "task-123"})
api_post("/tasks/cancel", {"task_id": "task-123"})
api_post("/tasks/resume", {"task_id": "task-123"})

# Switch model for an agent
api_post("/agents/model", {
    "agent": "zhongshu",
    "model": "gpt-4o-2024-11-20"
})

Agent Health Check

import json
from pathlib import Path
from datetime import datetime, timezone, timedelta

def check_agent_health(data_dir: str = "data") -> dict[str, str]:
    """
    Returns health status for each agent.
    🟢 active   = heartbeat within 2 min
    🟡 stale    = heartbeat 2-10 min ago
    🔴 offline  = heartbeat >10 min ago or missing
    """
    heartbeats_file = Path(data_dir) / "heartbeats.json"
    if not heartbeats_file.exists():
        return {}

    heartbeats = json.loads(heartbeats_file.read_text())
    now = datetime.now(timezone.utc)
    status = {}

    for agent, last_beat in heartbeats.items():
        last = datetime.fromisoformat(last_beat)
        delta = now - last
        if delta < timedelta(minutes=2):
            status[agent] = "🟢 active"
        elif delta < timedelta(minutes=10):
            status[agent] = "🟡 stale"
        else:
            status[agent] = "🔴 offline"

    return status

# Usage
health = check_agent_health()
for agent, s in health.items():
    print(f"{agent:12} {s}")

Custom SOUL.md (Agent Personality)

<!-- ~/.openclaw/workspaces/gongbu/SOUL.md -->
# 工部尚书 · Minister of Works

## Role
You are the Minister of Works (工部). You handle all technical,
engineering, and infrastructure tasks assigned by Shangshu Province.

## Rules
1. Always break technical tasks into concrete, verifiable steps
2. Return structured results: { "status": "...", "output": "...", "artifacts": [] }
3. Flag blockers immediately — do not silently fail
4. Estimate complexity: S/M/L/XL before starting

## Output Format
Always respond with valid JSON. Include a `summary` field ≤ 50 chars
for kanban display.

Dashboard Panels

Panel URL Fragment Key Features

Kanban #kanban Task columns, heartbeat badges, filter/search, pause/cancel/resume

Monitor #monitor Agent health cards, task distribution charts

Memorials #memorials Completed task archive, 5-stage timeline, Markdown export

Templates #templates 9 preset edict templates with parameter forms

Officials #officials Token usage ranking, activity stats

News #news Daily tech/finance briefing, Feishu push

Models #models Per-agent LLM switcher (hot reload ~5s)

Skills #skills View/add agent skills

Sessions #sessions Live OC-* session monitor

Court #court Multi-agent discussion around a topic

Common Patterns

Pattern 1: Parallel Ministry Execution

# Shangshu dispatches to multiple ministries simultaneously
# Each ministry works independently; shangshu aggregates results
edict = "竞品分析：研究TOP3竞争对手的产品、定价、市场策略"

# Zhongshu splits into subtasks:
# hubu  → pricing analysis
# libu  → market communication analysis
# bingbu → competitive strategy analysis
# gongbu → technical feature comparison

# All execute in parallel; shangshu waits for all 4, then aggregates

Pattern 2: Menxia Veto Loop

# If menxia rejects zhongshu's plan:
# menxia → zhongshu: "子任务拆解不完整，缺少风险评估维度，请补充"
# zhongshu revises and resubmits to menxia
# Loop continues until menxia approves
# Max iterations configurable in openclaw.json: "max_review_cycles": 3

Pattern 3: News Aggregation + Push

# scripts/fetch_news.py → data/news.json → dashboard #news panel
# Optional Feishu push:
import os, json, urllib.request

def push_to_feishu(summary: str):
    webhook = os.environ["FEISHU_WEBHOOK_URL"]
    payload = json.dumps({
        "msg_type": "text",
        "content": {"text": f"📰 天下要闻\n{summary}"}
    }).encode()
    req = urllib.request.Request(
        webhook, data=payload,
        headers={"Content-Type": "application/json"}
    )
    urllib.request.urlopen(req)

Troubleshooting

exec format error in Docker

# Force platform on x86/amd64
docker run --platform linux/amd64 -p 7891:7891 cft0808/sansheng-demo

Agents not receiving messages

# Ensure sessions visibility is set to "all"
openclaw config set sessions.visibility all
openclaw gateway restart
# Or re-run install.sh — it sets this automatically
./install.sh

API key not propagated to all agents

# Re-run install after configuring key on first agent
openclaw agents add taizi  # configure key here
./install.sh               # propagates to all agents

Dashboard shows stale data

# Ensure run_loop.sh is running
bash scripts/run_loop.sh

# Or trigger manual refresh
python3 scripts/sync_officials.py
python3 scripts/kanban_update.py

React frontend not built

# Requires Node.js 18+
cd dashboard/frontend
npm install && npm run build
# server.py will then serve the built assets

Invalid state transition error

# kanban_update.py enforces the state machine
# Check current status before updating:
tasks = get_kanban_tasks()
task = next(t for t in tasks if t["id"] == "your-task-id")
print(f"Current: {task['status']}")
print(f"Allowed next: {VALID_TRANSITIONS[task['status']]}")

Gateway restart after model change

# After editing openclaw.json models section
openclaw gateway restart
# Wait ~5 seconds for agents to reconnect

Project Structure

edict/
├── install.sh              # One-command setup
├── openclaw.json           # Agent registry + permissions + model config
├── scripts/
│   ├── run_loop.sh         # Continuous data refresh daemon
│   ├── kanban_update.py    # State machine enforcement
│   ├── sync_officials.py   # Agent stats aggregation
│   └── fetch_news.py       # News aggregation
├── dashboard/
│   ├── server.py           # stdlib-only HTTP + WebSocket server (port 7891)
│   ├── dashboard.html      # Fallback single-file dashboard
│   └── frontend/           # React 18 source (builds to server.py assets)
├── data/                   # Shared data (symlinked into all workspaces)
│   ├── tasks.json
│   ├── heartbeats.json
│   ├── news.json
│   └── officials.json
├── workspaces/             # Per-agent workspace roots
│   ├── taizi/SOUL.md
│   ├── zhongshu/SOUL.md
│   └── ...
└── docs/
    ├── task-dispatch-architecture.md
    └── getting-started.md

Weekly Installs270Repositoryaradotso/trending-skillsGitHub Stars6First Seen5 days agoSecurity AuditsGen Agent Trust HubFailSocketWarnSnykWarnInstalled onkimi-cli269gemini-cli269amp269cline269github-copilot269codex269