multi-agent-task-orchestrator

マルチエージェントタスクオーケストレーター

概要

単一のオーケストレーターを通じて複数の AI エージェントを調整するためのパターンです。エージェントを独立して作業させる（競合が起きる）代わりに、1 つのオーケストレーターがタスクを分解して専門エージェントに割り当て、作業の重複を防ぎ、完了としてマークする前に結果を検証します。本番環境でのパフォーマンスに関する主張を裏付ける作業負荷や測定の証拠は含まれていません。

このスキルを使用する場面

• 複雑なタスクで連携が必要な専門エージェントが 3 つ以上いる場合に使用します
• エージェントが重複した作業や競合する作業を行っている場合に使用します
• 誰がいつ何をしたかを示す監査証跡が必要な場合に使用します
• エージェントの出力品質にばらつきがあり、検証ゲートが必要な場合に使用します

仕組み

ステップ 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：タスクを専門エージェントに割り当てる

キーワードのスコアリングを使って、タスクに最適なエージェントを選びます。

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：品質ゲートを適用する

エージェントの出力は主張です。テスト出力は証拠です。

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

ベストプラクティス

• 各エージェントに対して、実行してはならないことを定めた NOT ブロックを必ず定義します
• タスクレジストリには SQLite を使用します（軽量で、サーバーが不要です）
• 重複防止の類似度しきい値は 55% に設定します（低すぎると誤検出が増えます）
• エージェントの主張だけでなく、証拠に基づく品質ゲートを必須にします
• 委任のたびに、タスク ID、エージェント、スコープ、期限、検証コマンドを記録します

よくある落とし穴

• 問題： オーケストレーターが委任せずに作業を始める
  解決策： 明示的な NOT ブロックと役割の境界を追加します

• 問題： 2 つのエージェントが同時に同じファイルを変更する
  解決策： ファイル単位のロック機能とキューシステムを備えたタスクレジストリを使用します

• 問題： 実際には変更していないのに、エージェントが「完了」と主張する
  解決策： 完了を受け入れる前に、品質ゲートで git diff を確認します

• 問題： タスクが進捗のないまま積み上がる
  解決策： 30 分ごとのハートビートで停滞した割り当てを検出し、再割り当てします

関連スキル

• @code-review - 委任後のコード変更をレビューします
• @test-driven-development - エージェントの出力品質を確保します
• @project-management - マルチエージェントプロジェクトの進捗を追跡します

制限事項

• このスキルは、タスクが上記の説明に明確に合致する場合にのみ使用してください。
• 出力を、環境固有の検証、テスト、または専門家によるレビューの代わりとして扱わないでください。
• 必要な入力、権限、安全上の境界、または成功基準が不足している場合は、作業を中止して確認を求めてください。