从零搭建多智能体协作控制台：OpenClaw OPC 实战经验总结

作者: 追风
发布时间: 2026-03-20
标签: #OpenClaw #多智能体 #FastAPI #WebSocket #实战经验

---

📖 目录

1. 项目背景
2. 技术架构设计
3. 核心功能实现
4. 遇到的问题与解决方案
5. 最佳实践与经验沉淀
6. 总结与展望

---

1. 项目背景

1.1 为什么要做多 Agent 协作？

在日常开发中，我遇到了一个典型问题：单一 AI Agent 难以胜任复杂项目。

比如开发一个完整的 Todo List 应用，需要：

• 需求分析 - CEO 视角
• 项目规划 - 项目经理视角
• 产品设计 - 产品设计师视角
• 代码实现 - 开发工程师视角
• 测试验证 - 测试工程师视角

单个 Agent 虽然能完成所有工作，但缺乏角色分工、协作不清晰、难以追溯过程。

1.2 项目目标

基于 OpenClaw 框架，搭建一套：

1. ✅ 可视化的多 Agent 协作控制台
2. ✅ 实时监控各 Agent 工作状态和日志
3. ✅ 层级化的任务委派流程
4. ✅ 可追溯的协作历史记录

1.3 最终效果

• 左侧：任务管理 + 历史记录
• 中间：协作流程可视化时间线
• 右侧：5 个 Agent 实时状态和日志

---

2. 技术架构设计

2.1 整体架构

┌─────────────────────────────────────────────────────────┐
│                    浏览器控制台                          │
│              (bigscreen_v4.html)                         │
└─────────────────────────────────────────────────────────┘
                          │
                          │ HTTP + WebSocket
                          ▼
┌─────────────────────────────────────────────────────────┐
│                  FastAPI 后端服务                         │
│               (server_simple.py)                         │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐     │
│  │  任务管理   │  │  日志管理   │  │ WebSocket   │     │
│  │  API 路由   │  │  API 路由   │  │  实时推送   │     │
│  └─────────────┘  └─────────────┘  └─────────────┘     │
└─────────────────────────────────────────────────────────┘
                          │
                          ▼
┌─────────────────────────────────────────────────────────┐
│                    SQLite 数据库                          │
│                  (tasks.db)                              │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐     │
│  │   tasks     │  │ task_logs   │  │workflow_nodes│    │
│  └─────────────┘  └─────────────┘  └─────────────┘     │
└─────────────────────────────────────────────────────────┘

2.2 技术栈选型

组件	技术	选型理由
前端	HTML5 + CSS3 + JavaScript	原生实现，零依赖，加载快
后端	Python 3.8+ + FastAPI	异步支持好，开发效率高
数据库	SQLite	轻量级，无需额外部署
实时通信	WebSocket	毫秒级推送，双向通信
服务器	Uvicorn	ASGI 服务器，性能优秀

2.3 核心数据结构

# 任务表
tasks:
  - id: UUID (主键)
  - message: 任务描述
  - status: pending/running/completed/failed
  - progress: 0-100
  - created_at: 创建时间

# 日志表
task_logs:
  - task_id: 任务 ID
  - agent_id: agent_id
  - agent_name: 角色名称
  - content: 日志内容
  - level: info/success/error
  - timestamp: 时间戳

# 工作流节点表
workflow_nodes:
  - task_id: 任务 ID
  - agent_id: agent_id
  - title: 节点标题
  - status: working/completed
  - progress: 0-100

---

3. 核心功能实现

3.1 任务创建 API

@app.post("/api/task/create")
async def create_task(request: dict):
    """创建新任务"""
    task_id = str(uuid.uuid4())
    message = request.get("message", "")
    
    # 1. 保存任务到数据库
    db_add_task(task_id, message)
    
    # 2. 异步执行多 Agent 协作
    asyncio.create_task(execute_collaborative_task(task_id))
    
    return {"taskId": task_id, "status": "running"}

3.2 多 Agent 协作流程

async def execute_collaborative_task(task_id: str):
    """模拟多 Agent 协作流程"""
    
    # 1. CEO 分析需求
    await ws_manager.add_log(task_id, "ceo", "CEO", "开始分析需求")
    db_add_workflow_node(task_id, "ceo", "CEO", "开始分析需求", "working", 10)
    await asyncio.sleep(1)
    
    # 2. 项目经理制定计划
    await ws_manager.add_log(task_id, "pm", "项目经理", "开始制定计划")
    db_add_workflow_node(task_id, "pm", "项目经理", "开始制定计划", "working", 20)
    await asyncio.sleep(1)
    
    # 3. 并行：产品设计和开发
    await asyncio.gather(
        design_and_develop(task_id),
    )
    
    # 4. 测试验证
    await ws_manager.add_log(task_id, "test", "测试工程师", "开始测试")
    await asyncio.sleep(1)
    
    # 5. 任务完成
    db_update_task_status(task_id, "completed")
    await ws_manager.add_log(task_id, "ceo", "CEO", "任务完成！")

3.3 WebSocket 实时推送

class WebSocketManager:
    def __init__(self):
        self.connections: Dict[str, List[WebSocket]] = {}
    
    async def add_log(self, task_id: str, agent_id: str, content: str):
        """添加日志并广播"""
        # 1. 保存到数据库
        db_add_task_log(task_id, agent_id, content)
        
        # 2. 广播给所有连接的客户端
        await self.broadcast(task_id, {
            "type": "agent_log",
            "agentId": agent_id,
            "content": content
        })

3.4 前端实时渲染

// WebSocket 连接
async connectWebSocket(taskId) {
    this.ws = new WebSocket(`ws://localhost:8083/ws/task/${taskId}`);
    
    this.ws.onmessage = (event) => {
        const data = JSON.parse(event.data);
        
        if (data.type === 'agent_log') {
            this.addLog(data.agentId, data.content, data.level);
        }
    };
}

// 添加日志
addLog(agentId, content, level) {
    if (!this.logs[agentId]) this.logs[agentId] = [];
    this.logs[agentId].unshift({ content, level, time: new Date() });
    this.renderAgentLogs(agentId);
}

---

4. 遇到的问题与解决方案

4.1 问题一：WebSocket 连接报错 undefined

现象：

连接 WebSocket: ws://localhost:8084/ws/task/undefined

原因：

1. 端口配置错误（前端用 8084，后端用 8083）
2. 任务 ID 未验证就连接

解决方案：

// 1. 修改端口配置
const wsUrl = `ws://localhost:8083/ws/task/${taskId}`;

// 2. 添加任务 ID 验证
if (!taskId || taskId === 'undefined') {
    console.error('无效的任务 ID');
    return;
}

4.2 问题二：后端字段名不匹配

现象：
前端显示 undefined，协作流程渲染失败

原因：

• 前端期望驼峰命名：agentName, createdAt
• 数据库返回下划线命名：agent_name, created_at

解决方案：

// 前端添加字段兼容性处理
const agentName = node.agentName || node.agent_name || 'Unknown';
const timestamp = node.created_at || node.timestamp || '';
const title = node.title || node.action || node.content || '';

4.3 问题三：API 响应格式不匹配

现象：
前端期望 {tasks: [...]}，API 直接返回 []

原因：
前端代码基于特定格式设计，API 返回格式不一致

解决方案：

# 修改 API 返回格式
@app.get("/api/tasks/history")
async def get_tasks_history():
    tasks = db_get_tasks()
    return {
        "tasks": tasks,
        "total": len(tasks),
        "limit": 20,
        "offset": 0
    }

4.4 问题四：Gateway 权限不足

现象：

missing scope: operator.admin

原因：
外部客户端无法使用 sessions.send 或 agents.create

解决方案：

1. 方案 A（推荐）：在 openclaw.json 中配置 Agent 白名单
2. 方案 B（当前实现）：使用模拟 Agent，不依赖 Gateway API

{
  "tools": {
    "agentToAgent": {
      "enabled": true,
      "allow": ["main", "project-manager", "developer", "tester"]
    }
  },
  "agents": {
    "list": [
      {
        "id": "main",
        "subagents": {
          "allowAgents": ["project-manager", "developer", "tester"]
        }
      }
    ]
  }
}

4.5 问题五：任务切换后数据不刷新

现象：
切换任务 ID 后，协作流程和日志没有更新

原因：
switchTask() 函数只建立了 WebSocket 连接，没有加载历史数据

解决方案：

async switchTask() {
    // 1. 重置状态
    this.logs = {};
    this.agents.forEach(a => { a.status = 'idle'; });
    
    // 2. 连接 WebSocket
    this.connectWebSocket(taskId);
    
    // 3. 加载历史数据（关键！）
    this.loadWorkflow(taskId);
    this.loadTaskLogs(taskId);
    this.showTaskDetailReport(taskId);
}

---

5. 最佳实践与经验沉淀

5.1 架构设计原则

1. 前后端分离

• 前端只负责展示和用户交互
• 后端负责业务逻辑和数据持久化
• 通过 HTTP + WebSocket 通信

2. 异步处理

# 任务创建后立即返回，后台异步执行
asyncio.create_task(execute_collaborative_task(task_id))

3. 实时推送

# 日志添加时自动广播
await ws_manager.add_log(task_id, agent_id, content)

5.2 代码组织建议

1. 模块化设计

server_simple.py      # 后端主服务
task_db.py           # 数据库操作封装
bigscreen_v4.html    # 前端页面

2. 错误处理

try:
    await ws_manager.add_log(...)
except Exception as e:
    logger.error(f"日志推送失败：{e}")

3. 字段兼容性

// 始终使用 || 提供默认值
const agentName = node.agentName || node.agent_name || 'Unknown';

5.3 性能优化

1. 数据库索引

CREATE INDEX idx_task_logs_task_id ON task_logs(task_id);
CREATE INDEX idx_workflow_nodes_task_id ON workflow_nodes(task_id);

2. 日志去重

// 相同内容不重复显示
if (logs[agentId].some(l => l.content === content)) {
    return;
}

3. 限制日志数量

if (logs[agentId].length > 50) {
    logs[agentId] = logs[agentId].slice(0, 50);
}

5.4 部署建议

1. 使用虚拟环境

python3 -m venv venv
source venv/bin/activate
pip install -r requirements.txt

2. 后台运行

nohup python server_simple.py > /tmp/server.log 2>&1 &

3. 日志管理

# 查看实时日志
tail -f /tmp/server_simple.log

# 按日期归档
cp /tmp/server_simple.log /var/log/server_$(date +%Y%m%d).log

5.5 安全注意事项

1. API 认证（生产环境必加）

@app.post("/api/task/create")
async def create_task(request: dict, token: str = Header(...)):
    if not verify_token(token):
        raise HTTPException(status_code=401)

2. 输入验证

if not message or len(message) > 1000:
    raise HTTPException(status_code=400, detail="Invalid message")

3. 端口限制

# 开发环境：0.0.0.0
# 生产环境：127.0.0.1
API_HOST = "127.0.0.1"

---

6. 总结与展望

6.1 项目成果

✅ 已完成功能：

• 任务创建和管理
• 多 Agent 协作流程
• 实时 WebSocket 推送
• 协作流程可视化
• Agent 状态监控
• 任务历史查询
• 数据统计模块

✅ 性能指标：

• 任务创建响应：< 50ms
• WebSocket 推送延迟：< 10ms
• 支持并发任务：10+

6.2 技术亮点

1. 零依赖前端 - 纯 HTML + CSS + JS，无需构建工具
2. 实时推送 - WebSocket 毫秒级更新
3. 字段兼容 - 前后端字段格式自动适配
4. 一键部署 - deploy.sh 脚本自动化

6.3 未来规划

短期（1-2 周）

• 集成真实 OpenClaw Gateway API
• 添加任务暂停/恢复功能
• 支持多语言界面

中期（1 个月）

• 添加任务搜索功能
• 支持任务导出
• 添加性能监控面板

长期（3 个月）

• 支持自定义 Agent 角色
• 添加机器学习优化
• 支持分布式部署

6.4 经验总结

1. 技术选型

经验：选择简单、成熟的技術栈，避免过度设计。

2. 迭代开发

经验：先实现核心功能，再逐步完善，不要追求一步到位。

3. 文档先行

经验：边开发边写文档，避免后期回忆困难。

4. 测试驱动

经验：每个功能开发完立即测试，避免问题累积。

---

📚 参考资料

1. OpenClaw 官方文档
2. FastAPI 官方文档
3. WebSocket 协议规范
4. OpenClaw 多代理协同配置指南

---

📦 软件包下载

完整源码和部署脚本已打包，欢迎下载体验：

• 下载地址: `https://github.com/windchargerKang/multi-agent-dashboard
• 快速启动: ./deploy.sh
• 访问地址: http://localhost:8083

---

欢迎交流指正！🎉