Edict (三省六部) 多智能体编排 (Multi-Agent Orchestration)

技能来自 ara.so — Daily 2026 Skills 合集

Edict 将 1400 年前的唐代治理模式实现为 AI 多智能体架构。十二个专业智能体形成一个制衡流水线：太子（分流）→ 中书省（规划）→ 门下省（审查/否决）→ 尚书省（分派）→ 六部（并行执行）。基于 OpenClaw 构建，提供实时 React 看板、完整审计追踪和每个智能体的 LLM 配置。

架构概览 (Architecture Overview)

您（皇帝）→ 太子 (taizi)（分流）→ 中书省 (zhongshu)（规划）→ 门下省 (menxia)（审查/否决）
             → 尚书省 (shangshu)（分派）→ [户部|礼部|兵部|刑部|工部|吏部] (执行)
             → 奏折 (memorial)（结果归档）

与 CrewAI/AutoGen 的关键区别 (Key differentiator vs CrewAI/AutoGen): 门下省 (menxia) 是一个强制性的质量关卡——它可以在任务到达执行者之前否决并要求返工。

前提条件 (Prerequisites)

• OpenClaw 已安装并运行
• Python 3.9+
• Node.js 18+（用于 React 仪表板构建）(for React dashboard build)
• macOS 或 Linux

安装 (Installation)

快速演示（Docker — 无需 OpenClaw）(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

# 或使用 docker-compose（平台已设置）(Or with docker-compose - platform already set)
docker compose up

打开 http://localhost:7891

完整安装 (Full Installation)

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

安装脚本自动执行以下操作：

• 创建所有 12 个智能体工作区（taizi, zhongshu, menxia, shangshu, hubu, libu, bingbu, xingbu, gongbu, libu2, zaochao, legacy-compat）
• 将 SOUL.md 角色定义写入每个智能体工作区
• 在 openclaw.json 中注册智能体和权限矩阵
• 在所有智能体工作区之间建立共享数据目录的符号链接
• 设置 sessions.visibility all 以实现智能体间消息路由
• 在所有智能体间同步 API 密钥
• 构建 React 前端
• 初始化数据目录并同步官方统计数据

首次 API 密钥设置 (First-time API Key Setup)

# 在第一个智能体上配置 API 密钥 (Configure API key on first agent)
openclaw agents add taizi
# 然后重新运行安装以传播到所有智能体 (Then re-run install to propagate to all agents)
./install.sh

运行系统 (Running the System)

# 终端 1：数据刷新循环（保持看板数据最新）(Terminal 1: Data refresh loop - keeps kanban data current)
bash scripts/run_loop.sh

# 终端 2：仪表板服务器 (Terminal 2: Dashboard server)
python3 dashboard/server.py

# 打开仪表板 (Open dashboard)
open http://127.0.0.1:7891

关键命令 (Key Commands)

OpenClaw 智能体管理 (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 — 在端口 7891 提供服务 (serves on port 7891)
# 内置：React 前端 + REST API + WebSocket 更新
# (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)

智能体模型配置 (openclaw.json) (Agent Model Configuration)

{
  "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)

导航到 ⚙️ Models 面板 → 选择智能体 → 选择 LLM → 应用 (Apply)。网关会自动重启（约 5 秒）。

环境变量 (Environment Variables)

# API 密钥（在运行 install.sh 或 openclaw 前设置）(API keys - set before running install.sh or openclaw)
export ANTHROPIC_API_KEY="sk-ant-..."
export OPENAI_API_KEY="sk-..."

# 可选：飞书/Lark webhook 用于通知 (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)

权限矩阵（谁可以给谁发消息）(Permission Matrix – who can message whom)

# 在 openclaw.json 中定义 — 由网关强制执行 (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("执行中")
completed = get_tasks_by_status("已完成")
print(f"进行中: {len(executing)}, 已完成: {len(completed)}")

更新任务状态（带验证）(Update Task Status – with validation)

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

VALID_TRANSITIONS = {
    "待处理":     ["规划中"],
    "规划中":    ["审查中", "待处理"],
    "审查中":   ["分派中", "规划中"],
    "分派中": ["执行中"],
    "执行中":   ["已完成", "失败"],
    "已完成":   [],
    "失败":      ["待处理"],
}

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_id} 未找到 (Task not found)")

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

    if new_status not in allowed:
        raise ValueError(
            f"无效转换: {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

仪表板 REST API 客户端 (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)   = 心跳在 2 分钟内 (heartbeat within 2 min)
    🟡 陈旧 (stale)    = 心跳在 2-10 分钟前 (heartbeat 2-10 min ago)
    🔴 离线 (offline)  = 心跳超过 10 分钟前或缺失 (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}")

自定义 SOUL.md（智能体个性）(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. 返回结构化结果：{ "status": "...", "output": "...", "artifacts": [] } (Return structured results)
3. 立即标记障碍 — 不要静默失败 (Flag blockers immediately — do not silently fail)
4. 在开始前估算复杂度：小/中/大/特大 (S/M/L/XL) (Estimate complexity: S/M/L/XL before starting)

## 输出格式 (Output Format)
始终以有效的 JSON 响应。包含一个不超过 50 个字符的 `summary` 字段用于看板显示。
(Always respond with valid JSON. Include a `summary` field ≤ 50 chars for kanban display.)

仪表板面板 (Dashboard Panels)

常见模式 (Common Patterns)

模式 1：并行六部执行 (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)

# 所有部门并行执行；尚书省等待所有 4 个完成，然后汇总
# (All execute in parallel; shangshu waits for all 4, then aggregates)

模式 2：门下省否决循环 (Pattern 2: Menxia Veto Loop)

# 如果门下省拒绝中书省的规划 (If menxia rejects zhongshu's plan):
# menxia → zhongshu: "子任务拆解不完整，缺少风险评估维度，请补充"
# zhongshu 修订并重新提交给 menxia (zhongshu revises and resubmits to menxia)
# 循环持续直到门下省批准 (Loop continues until menxia approves)
# 最大迭代次数可在 openclaw.json 中配置: "max_review_cycles": 3

模式 3：新闻聚合 + 推送 (Pattern 3: News Aggregation + Push)

# scripts/fetch_news.py → data/news.json → 仪表板 #news 面板 (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 在 Docker 中 (in Docker)

# 在 x86/amd64 上强制指定平台 (Force platform on x86/amd64)
docker run --platform linux/amd64 -p 7891:7891 cft0808/sansheng-demo

智能体未收到消息 (Agents not receiving messages)

# 确保会话可见性设置为 "all" (Ensure sessions visibility is set to "all")
openclaw config set sessions.visibility all
openclaw gateway restart
# 或重新运行 install.sh — 它会自动设置此项 (Or re-run install.sh — it sets this automatically)
./install.sh

API 密钥未传播到所有智能体 (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)

# 确保 run_loop.sh 正在运行 (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 前端未构建 (React frontend not built)

# 需要 Node.js 18+ (Requires Node.js 18+)
cd dashboard/frontend
npm install && npm run build
# 然后 server.py 将提供构建的资源 (server.py will then serve the built assets)

无效状态转换错误 (Invalid state transition error)

# kanban_update.py 强制执行状态机 (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)

# 编辑 openclaw.json 中的模型部分后 (After editing openclaw.json models section)
openclaw gateway restart
# 等待约 5 秒让智能体重连 (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           # 仅使用标准库的 HTTP + WebSocket 服务器（端口 7891）(stdlib-only HTTP + WebSocket server - port 7891)
│   ├── dashboard.html      # 备用单文件仪表板 (Fallback single-file dashboard)
│   └── frontend/           # React 18 源码（构建到 server.py 资源）(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