multi-agent-task-orchestrator – skill Agent & Tự động hóa cho agent AI

Multi-Agent Task Orchestrator

Overview

A coordination pattern for coordinating multiple AI agents through a single orchestrator. Instead of letting agents work independently (and conflict), one orchestrator decomposes tasks, routes them to specialists, prevents duplicate work, and verifies results before marking anything done. No workload/measurement evidence is bundled for production performance claims.

When to Use This Skill

• Use when you have 3+ specialized agents that need to coordinate on complex tasks
• Use when agents are doing duplicate or conflicting work
• Use when you need audit trails showing who did what and when
• Use when agent output quality is inconsistent and needs verification gates

How It Works

Step 1: Define the Orchestrator Identity

The orchestrator must know what it IS and what it IS NOT. This prevents it from doing work instead of delegating:

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

Role boundaries can clarify scope; no quantified drift-reduction result is established here.

Step 2: Build a Task Registry

Before assigning work, check if anyone is already doing this task:

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

Step 3: Route Tasks to Specialists

Use keyword scoring to match tasks to the best 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"

Step 4: Enforce Quality Gates

Agent output is a CLAIM. Test output is EVIDENCE.

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.

Step 5: Monitor authorized work

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

The code is illustrative: tasks table creation and atomic transaction/unique claim/file-ownership enforcement are not implemented here. String similarity is only a duplicate candidate signal; review actual scope/dependencies before skipping work. Keyword scores do not establish expertise. Recurring heartbeat setup needs an explicit user request and supported scheduler; do not create background work from this guide. Delegate only within user/host permissions and use proportionate verification (read-only findings need not modify files).

Examples

Example 1: Delegating a Code Task

[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

Example 2: Handling a Duplicate

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

Best Practices

• Always define NOT-blocks for every agent (what they must refuse to do)
• Use SQLite for the task registry (lightweight, no server needed)
• Set similarity threshold at 55% for anti-duplication (lower = too many false positives)
• Require evidence-based quality gates (not just agent claims)
• Log every delegation with: task ID, agent, scope, deadline, verification command

Common Pitfalls

• Problem: Orchestrator starts doing work instead of delegating

Solution: Add explicit NOT-blocks and role boundaries

• Problem: Two agents modify the same file simultaneously

Solution: Task registry with file-level locking and queue system

• Problem: Agent claims "done" without actual changes

Solution: Quality gate checks git diff before accepting completion

• Problem: Tasks pile up without progress

Solution: 30-minute heartbeat catches stale assignments and reassigns

Related Skills

• @code-review - For reviewing code changes after delegation
• @test-driven-development - For ensuring quality in agent output
• @project-management - For tracking multi-agent project progress

Limitations

• Use this skill only when the task clearly matches the scope described above.
• Do not treat the output as a substitute for environment-specific validation, testing, or expert review.
• Stop and ask for clarification if required inputs, permissions, safety boundaries, or success criteria are missing.