---
id: daily-edict-multi-agent-orchestration
name: "edict-multi-agent-orchestration"
url: https://skills.yangsir.net/skill/daily-edict-multi-agent-orchestration
author: aradotso
domain: ai-agent-orchestration-collaboration
tags: ["multi-agent", "ai-orchestration", "agent-framework", "workflow-automation", "decision-making"]
install_count: 1300
rating: 4.30 (59 reviews)
github: https://github.com/aradotso/trending-skills
---

# edict-multi-agent-orchestration

> 实现唐代三省六部制治理模型的多代理架构，12 个专业代理形成制衡管道：太子（分诊）→中书省（起草）→门下省（审议）→尚书省（执行），实现决策的分工和监督

**Stats**: 1,300 installs · 4.3/5 (59 reviews)

## Before / After 对比

### 使用效果对比

**Before**:

手动完成实现唐代三省六部制治理模型的多相关任务，需要反复操作和确认，整个过程大约需要99小时，容易出错且效率低下

**After**:

使用该 Skill 自动化处理，智能分析和执行，10小时内完成全部工作，准确率高且流程标准化

| Metric | Before | After | Change |
|---|---|---|---|
| 完成速度 | 99小时 | 10小时 | -90% |

## Readme

# edict-multi-agent-orchestration

# Edict (三省六部) Multi-Agent Orchestration

Skill by [ara.so](https://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](https://openclaw.ai), 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](https://openclaw.ai) 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](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 Installs270Repository[aradotso/trending-skills](https://github.com/aradotso/trending-skills)GitHub Stars6First Seen5 days agoSecurity Audits[Gen Agent Trust HubFail](/aradotso/trending-skills/edict-multi-agent-orchestration/security/agent-trust-hub)[SocketWarn](/aradotso/trending-skills/edict-multi-agent-orchestration/security/socket)[SnykWarn](/aradotso/trending-skills/edict-multi-agent-orchestration/security/snyk)Installed onkimi-cli269gemini-cli269amp269cline269github-copilot269codex269

---
*Source: https://skills.yangsir.net/skill/daily-edict-multi-agent-orchestration*
*Markdown mirror: https://skills.yangsir.net/api/skill/daily-edict-multi-agent-orchestration/markdown*