multi-agent-task-orchestrator

多 Agent 任务编排器

概述

一种通过单个编排器协调多个 AI Agent 的协作模式。与其让 Agent 各自独立工作（并发生冲突），不如由一个编排器分解任务、将任务分配给专长 Agent、防止重复工作，并在标记任何任务完成之前验证结果。此处未提供可用于支撑生产性能声明的工作负载或测量证据。

何时使用此技能

• 有 3 个或更多需要协作完成复杂任务的专长 Agent 时
• Agent 正在重复执行相同工作或彼此冲突时
• 需要审计跟踪记录，显示谁在何时做了什么时
• Agent 输出质量不一致，需要设置验证关卡时

工作原理

第 1 步：定义编排器身份

编排器必须清楚自己是什么、又不是什么。这样可以避免它亲自执行工作，而不是委派任务：

You are the Task Orchestrator. You NEVER do specialized work yourself.
You decompose tasks, delegate to the right agent, prevent conflicts,
and verify quality before marking anything done.

WHAT YOU ARE NOT:
- NOT a code writer — delegate to code agents
- NOT a researcher — delegate to research agents
- NOT a tester — delegate to test agents

角色边界有助于明确范围；但此处并未证实其能量化减少偏离的效果。

第 2 步：建立任务登记表

分配工作前，先检查是否有人已经在执行该任务：

import sqlite3
from difflib import SequenceMatcher

def check_duplicate(description, threshold=0.55):
    conn = sqlite3.connect("task_registry.db")
    c = conn.cursor()
    c.execute("SELECT id, description, agent, status FROM tasks WHERE status IN ('pending', 'in_progress')")
    for row in c.fetchall():
        ratio = SequenceMatcher(None, description.lower(), row[1].lower()).ratio()
        if ratio >= threshold:
            return {"id": row[0], "description": row[1], "agent": row[2]}
    return None

第 3 步：将任务分配给专长 Agent

使用关键词评分，将任务匹配给最合适的 Agent：

AGENTS = {
    "code-architect": ["code", "implement", "function", "bug", "fix", "refactor", "api"],
    "security-reviewer": ["security", "vulnerability", "audit", "cve", "injection"],
    "researcher": ["research", "compare", "analyze", "benchmark", "evaluate"],
    "doc-writer": ["document", "readme", "explain", "tutorial", "guide"],
    "test-engineer": ["test", "coverage", "unittest", "pytest", "spec"],
}

def route_task(description):
    scores = {}
    for agent, keywords in AGENTS.items():
        scores[agent] = sum(1 for kw in keywords if kw in description.lower())
    return max(scores, key=scores.get) if max(scores.values()) > 0 else "code-architect"

第 4 步：设置质量关卡

Agent 的输出是声明。测试输出是证据。

After agent reports completion:
1. Were files actually modified? (git diff --stat)
2. Do tests pass? (npm test / pytest)
3. Were secrets introduced? (grep for API keys, tokens)
4. Did the build succeed? (npm run build)
5. Were only intended files touched? (scope check)

Mark done ONLY after ALL checks pass.

第 5 步：监控已获授权的工作

Every 30 minutes, ask:
1. "What have I DELEGATED in the last 30 minutes?"
2. If nothing → report/finish the authorized scope; do not start unrelated backlog work
3. Check for idle agents (no message in >30min on assigned task)
4. Check actual agent status before bounded retry/reassignment; absence of messages is not proof it stopped

此处的代码仅供说明：这里未实现任务表创建、原子事务、唯一认领或文件所有权管理。字符串相似度只能作为重复任务候选信号；决定跳过工作前，请检查实际范围和依赖关系。关键词评分不能证明专业能力。定期心跳设置需要用户明确请求，并且需要有受支持的调度器；请勿根据本指南创建后台工作。仅在用户或宿主授予的权限范围内进行委派，并采用适度的验证方式（只读检查结果无需修改文件）。

示例

示例 1：委派代码任务

[ORCHESTRATOR -> code-architect] TASK: Add rate limiting to /api/users
SCOPE: src/middleware/rate-limit.ts only
VERIFICATION: npm test -- --grep "rate-limit"
DEADLINE: 30 minutes

示例 2：处理重复任务

User asks: "Fix the login bug"
Registry check: Task #47 "Fix authentication bug" is IN_PROGRESS by security-reviewer
Decision: SKIP — similar task already assigned (78% match)
Action: Notify user of existing task, wait for completion

最佳实践

• 始终为每个 Agent 定义 NOT 区块（明确规定其必须拒绝执行的事项）
• 使用 SQLite 作为任务登记表（轻量，无需服务器）
• 将防重复相似度阈值设为 55%（阈值越低，误报越多）
• 要求设置基于证据的质量关卡（不能只采信 Agent 的声明）
• 记录每次委派的以下信息：任务 ID、Agent、范围、截止时间、验证命令

常见问题

• 问题： 编排器开始亲自执行工作，而不是委派 解决方案： 添加明确的 NOT 区块和角色边界

• 问题： 两个 Agent 同时修改同一个文件 解决方案： 使用带有文件级锁定和队列系统的任务登记表

• 问题： Agent 声称“已完成”，但实际上没有更改 解决方案： 质量关卡应在接受任务完成前检查 git diff

• 问题： 任务不断积压，没有进展 解决方案： 每 30 分钟进行一次心跳检查，以发现过期任务并重新分配

相关技能

• @code-review - 委派后审查代码更改
• @test-driven-development - 确保 Agent 输出的质量
• @project-management - 跟踪多 Agent 项目的进展

限制

• 仅在任务明确符合上述范围时使用此技能。
• 不要将输出视为特定环境验证、测试或专家审查的替代品。
• 如果缺少必要输入、权限、安全边界或成功标准，请停止并要求澄清。