edict-multi-agent-orchestration

name	edict-multi-agent-orchestration
description	Install and use the Edict (三省六部) multi-agent orchestration system with 12 specialized AI agents, real-time kanban dashboard, and audit trails
triggers	["set up edict multi agent system","configure sansheng liubu agents","install edict orchestration","use openclaw multi agent dashboard","set up AI agent pipeline with review gates","configure edict kanban board","how to use edict agent workflow","deploy three provinces six ministries AI system"]

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": 
     
     
       
    
  

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:  → . "
            
        )

    task[] = new_status
    task[] = datetime.now(timezone.utc).isoformat()
    task.setdefault(, []).append({
        : current,
        : new_status,
        : task[]
    })

    tasks_file.write_text(json.dumps(tasks, ensure_ascii=, indent=))
     

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": ,
    : 
})

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