team-tasks

团队任务 — 多 Agent 流水线协调

概述

通过共享 JSON 任务文件和 AGI 调度来协调开发团队 Agent。 AGI 是指挥中心，Agent 之间不会直接交流。

两种模式：

• 模式 A（线性）： 固定流水线顺序 code → test → docs → monitor
• 模式 B（DAG）： 任务声明依赖关系，依赖满足时并行调度

任务管理器 CLI

所有命令的格式为：python3 <skill-dir>/scripts/task_manager.py <command> [args]

其中 <skill-dir> 是包含此 SKILL.md 的目录。

快速参考

命令	模式	用法	说明
init	两者	init <project> -g "goal" [-m linear|dag]	创建项目
add	DAG	add <project> <task-id> -a <agent> -d <deps>	添加带依赖的任务
status	两者	status <project> [--json]	显示进度
assign	两者	assign <project> <task> "desc"	设置任务说明
update	两者	update <project> <task> <status>	更改状态
next	线性	next <project> [--json]	获取下一阶段
ready	DAG	ready <project> [--json]	获取所有可调度任务
graph	DAG	graph <project>	显示依赖树
log	两者	log <project> <task> "msg"	添加日志条目
result	两者	result <project> <task> "output"	保存输出
reset	两者	reset <project> [task] [--all]	重置为待处理
list	两者	list	列出所有项目

状态值

• pending — 等待调度
• in-progress — Agent 正在处理
• done — 阶段已完成
• failed — 阶段失败（流水线受阻）
• skipped — 已主动跳过

流水线工作流（模式 A：线性）

步骤 1：初始化项目

python3 scripts/task_manager.py init my-project \
  -g "Build a REST API with tests and docs" \
  -p "code-agent,test-agent,docs-agent,monitor-bot"

默认流水线顺序：code-agent → test-agent → docs-agent → monitor-bot

步骤 2：为所有阶段分配任务

python3 scripts/task_manager.py assign my-project code-agent "Implement REST API with Flask: GET/POST/DELETE /items"
python3 scripts/task_manager.py assign my-project test-agent "Write pytest tests for all endpoints, target 90%+ coverage"
python3 scripts/task_manager.py assign my-project docs-agent "Write README.md with API docs, setup guide, examples"
python3 scripts/task_manager.py assign my-project monitor-bot "Verify code quality, check for security issues, validate deployment readiness"

步骤 3：依次调度 Agent

对于每个阶段，AGI 按照以下循环执行：

1. Check next stage:   task_manager.py next <project> --json
2. Mark in-progress:   task_manager.py update <project> <agent> in-progress
3. Dispatch agent:     sessions_send(sessionKey="agent:<agent>:telegram:group:<id>", message=<task>)
4. Wait for reply      (sessions_send returns the agent's response)
5. Save result:        task_manager.py result <project> <agent> "<summary>"
6. Mark done:          task_manager.py update <project> <agent> done
7. Repeat from 1       (currentStage auto-advances)

步骤 4：处理失败

如果 Agent 失败：

python3 scripts/task_manager.py update my-project code-agent failed
python3 scripts/task_manager.py log my-project code-agent "Failed: syntax error in main.py"

重试方法：

python3 scripts/task_manager.py reset my-project code-agent
python3 scripts/task_manager.py update my-project code-agent in-progress
# Re-dispatch...

步骤 5：随时查看进度

python3 scripts/task_manager.py status my-project

输出：

📋 Project: my-project
🎯 Goal: Build a REST API with tests and docs
📊 Status: active
▶️  Current: test-agent

  ✅ code-agent: done
     Task: Implement REST API with Flask
     Output: Created /home/ubuntu/projects/my-project/app.py
  🔄 test-agent: in-progress
     Task: Write pytest tests for all endpoints
  ⬜ docs-agent: pending
  ⬜ monitor-bot: pending

  Progress: [██░░] 2/4

Agent 调度详情

会话密钥（开发团队）

Agent	会话密钥
code-agent	agent:code-agent:telegram:group:-5189558203
test-agent	agent:test-agent:telegram:group:-5218382533
docs-agent	agent:docs-agent:telegram:group:-5253138320
monitor-bot	agent:monitor-bot:telegram:group:-5193935559

调度模板

向 Agent 调度任务时，请包含：

1. 项目背景 — 项目内容
2. 具体任务 — 该 Agent 应执行的任务
3. 工作目录 — 文件的创建或查找位置
4. 上一阶段的输出 — 如有需要（例如，test-agent 需要了解 code-agent 构建了什么）

调度消息示例：

Project: my-project
Goal: Build a REST API with tests and docs
Your task: Write pytest tests for all endpoints in /home/ubuntu/projects/my-project/app.py
Target: 90%+ coverage, test GET/POST/DELETE /items
Working directory: /home/ubuntu/projects/my-project/
Previous stage (code-agent) output: Created app.py with Flask REST API, 3 endpoints

修复消息投递上下文

⚠️ 如果 Agent 的会话最初是通过 sessions_send 创建的，其 deliveryContext 为 webchat，而非 telegram。Agent 的回复不会出现在 Telegram 群组中。

解决方法：通过 sessions_send 获取 Agent 的回复后，使用 message 工具将关键结果转发到群组：

message(action="send", channel="telegram", target="-5189558203", message="✅ code-agent 完成: Created app.py")

模式 B：DAG 工作流（并行依赖）

步骤 1：初始化 DAG 项目

python3 scripts/task_manager.py init my-project -m dag -g "Build REST API with parallel workstreams"

步骤 2：添加带依赖的任务

TM="python3 scripts/task_manager.py"
# Root tasks (no deps — can run in parallel)
$TM add my-project design     -a docs-agent  --desc "Write API spec"
$TM add my-project scaffold   -a code-agent  --desc "Create project skeleton"

# Tasks with dependencies (blocked until deps are done)
$TM add my-project implement  -a code-agent  -d "design,scaffold" --desc "Implement API"
$TM add my-project write-tests -a test-agent -d "design"          --desc "Write test cases from spec"

# Fan-in: depends on multiple tasks
$TM add my-project run-tests  -a test-agent  -d "implement,write-tests" --desc "Run all tests"
$TM add my-project write-docs -a docs-agent  -d "implement"             --desc "Write final docs"

# Final gate
$TM add my-project review     -a monitor-bot -d "run-tests,write-docs"  --desc "Final review"

步骤 3：查看 DAG 图

$TM graph my-project

├─ ⬜ design [docs-agent]
│  ├─ ⬜ implement [code-agent]
│  │  ├─ ⬜ run-tests [test-agent]
│  │  │  └─ ⬜ review [monitor-bot]
│  │  └─ ⬜ write-docs [docs-agent]
│  └─ ⬜ write-tests [test-agent]
└─ ⬜ scaffold [code-agent]
   └─ ⬜ implement (↑ see above)

步骤 4：调度就绪任务

$TM ready my-project    # Shows all tasks whose deps are met

对于每个就绪任务，AGI 按照以下循环执行：

1. Get ready tasks:     task_manager.py ready <project> --json
2. For each ready task (can dispatch in parallel):
   a. Mark in-progress: task_manager.py update <project> <task> in-progress
   b. Dispatch agent:   sessions_send(sessionKey=..., message=<task + dep outputs>)
3. When agent replies:
   a. Save result:      task_manager.py result <project> <task> "<summary>"
   b. Mark done:        task_manager.py update <project> <task> done
   c. Check newly unblocked tasks (printed automatically)
4. Repeat until all done

DAG 的主要功能

• 并行分发：ready 会返回所有依赖项均已满足的任务 — 同时分发这些任务
• 转发依赖项输出：ready --json 会包含 depOutputs — 供 Agent 传递的上一阶段结果
• 自动解除阻塞通知：任务完成时，会显示新解除阻塞的任务
• 循环检测：add 会拒绝会造成循环依赖的任务
• 部分失败：如果某个任务失败，不相关的分支会继续运行；只有下游任务会被阻塞
• 图形可视化：graph 会显示带状态图标和去重标记的树形视图

自定义工作流

线性（模式 A）

# Code + test only
python3 scripts/task_manager.py init quick-fix -g "Hotfix" -p "code-agent,test-agent"

# Docs first, then code
python3 scripts/task_manager.py init spec-driven -g "Spec-driven dev" -p "docs-agent,code-agent,test-agent"

DAG（模式 B）

# Diamond pattern: 2 parallel branches merge for review
$TM init diamond -m dag -g "Parallel dev"
$TM add diamond code -a code-agent --desc "Write code"
$TM add diamond test -a test-agent --desc "Write tests"
$TM add diamond integrate -a code-agent -d "code,test" --desc "Integration"
$TM add diamond review -a monitor-bot -d "integrate" --desc "Final review"

如何选择模式

	模式 A（线性）	模式 B（DAG）
适用场景	顺序任务、简单流程	并行工作流、复杂依赖
分发方式	一次一个，自动推进	同时分发多个，由依赖关系驱动
设置方式	init -p agents（一条命令）	init -m dag + 每个任务执行一次 add
最适合	Bug 修复、简单功能	大型功能、规范驱动的开发

数据位置

任务文件：/home/ubuntu/clawd/data/team-tasks/<project>.json

⚠️ 常见陷阱

模式 A：阶段 ID 是 Agent 名称，不是数字

在线性模式中，阶段 ID 是 Agent 名称（例如 code-agent），而不是 1、2、3 这样的数字索引。

# ❌ WRONG — will error "stage '1' not found"
python3 scripts/task_manager.py assign my-project 1 "Build API"
python3 scripts/task_manager.py update my-project 1 done

# ✅ CORRECT — use agent name as stage ID
python3 scripts/task_manager.py assign my-project code-agent "Build API"
python3 scripts/task_manager.py update my-project code-agent done
python3 scripts/task_manager.py result my-project code-agent "Created main.py"

这适用于所有引用阶段的命令：assign、update、result、log、reset。

工作流顺序由 init 时的 -p 定义（例如 -p "code-agent,test-agent,docs-agent"），而 next 会按顺序自动推进 — 但始终要使用 Agent 名称来引用阶段。

提示

• 每个任务一个项目 — 保持范围聚焦；创建多个项目以并行处理工作
• 使用有意义的项目标识 — rest-api-v2、bug-fix-auth、refactor-db（不要用 project1）
• 保存结果 — 执行 update done 前始终先执行 result；这是 Agent 间的上下文
• 多记录日志 — log 的成本很低；有助于调试失败的工作流
• 重置并重试 — 使用 reset --all 进行干净的重新运行；使用 reset <stage> 针对特定阶段重试
• DAG 扇出 — 一个根任务可以解除多个并行任务的阻塞
• DAG 扇入 — 一个任务可以依赖多个前置任务（所有前置任务都必须完成）