複数エージェントをどう束ねるか: 並列化する仕事と直列に残す仕事の線引き

ファン構造で考える

並列と直列の混在は、扇（ファン）のかたちで設計すると見通しがよくなります。入口で1本、そこから複数に広げ(ファンアウト)、最後にまた1本へ束ねる(ファンイン)。

            ┌─ Agent A: UIコンポーネント ─┐
入口(計画) ─┼─ Agent B: APIルート       ─┼─ ゲート集約 → デプロイ
            └─ Agent C: スキーマ定義     ─┘
                 (ファンアウト=並列)        (ファンイン=直列)

入口の「計画」は1つのエージェントが担い、何を作るかと分担を決めます。ここを並列にしてはいけません。全体の整合は1か所で取る必要があるからです。中間の実装は、上の3軸で衝突しないものだけを並列に広げます。そして出口の検証とデプロイは、必ず1本に束ね直します。

この「出口を直列にする」のが肝心です。並列で作ったものを、まとめて検証する場所を1つ用意する。私は記事の自動生成でも、生成は分散させつつ、品質ゲートとデプロイは必ず最後に1か所へ集約しています。複数の検証を各所に散らすより、1か所に集約する方を推奨します。

ゲートを集約する実装の骨格

ファンインの部分は、各エージェントの成果がそろってから検証を走らせる形にします。擬似コードで骨格を示します。

import asyncio
 
async def run_agent(name, task):
    # 各エージェントを並列に起動する
    result = await dispatch_agent(name, task)
    return name, result
 
async def orchestrate(plan):
    # ファンアウト: 衝突しない作業だけを並列に投げる
    parallel = [
        run_agent("ui", plan["ui"]),
        run_agent("api", plan["api"]),
        run_agent("schema", plan["schema"]),
    ]
    results = dict(await asyncio.gather(*parallel))
 
    # ファンイン: 全部そろってから検証を1か所で行う
    if not all_outputs_present(results):
        raise RuntimeError("一部のエージェント出力が欠落。デプロイ中止")
 
    gate_ok = run_quality_gates(results)   # ここは直列で1回だけ
    if not gate_ok:
        raise RuntimeError("品質ゲート不合格。差し戻し")
 
    return deploy(results)

asyncio.gather で並列起動しつつ、all_outputs_present で全部そろったことを確認してからゲートに進む。この「そろうまで待つ」一手が、並列実行で生成物が虫食いになる事故を防ぎます。ゲートを各エージェント内に分散させたくなりますが、それをやると合否の判断基準が少しずつズレていくエラーの温床になるので、私は1か所に集約する方を選んでいます。

どこを直列に残したか、という実例

私の自動運用では、4つのサイトを順に処理しています。サイトごとの記事生成は互いに独立なので並列にできますが、実際には完全並列にはしていません。理由は共有リソースです。生成の途中で同じ作業ディスクや同じ外部レート上限を使うため、全部を同時に走らせると、ディスク逼迫やレート制限で共倒れになる時間帯が出てきます。これがまさに「たまにしか再現しない壊れ方」で、原因の切り分けに何度も時間を取られました。

そこで、サイト内では分担を並列に、サイト間はゆるやかにずらして直列寄りにしています。並列度を上げる方向にばかり目が行きがちですが、「あえて直列に残す」判断こそが、長く安定して回る運用の背骨になります。Antigravity 2.0 がエージェントのオーケストレーション基盤になったことで、こうした線引きを設計として明示的に書けるようになりました。

次に設計を始めるなら、まず手元のワークフローを1枚の扇の図にしてみてください。どこがファンアウトでき、どこをファンインで束ねるべきかが見えた瞬間に、並列化の議論はぐっと地に足がついたものになります。お読みいただきありがとうございました。