グラフエンジニアリング：1 つのプロンプトから 1,000 個の AI エージェントを並列実行する方法

1何かを試す → 結果を確認する → 調整する → もう一度試す

1「このコードベースを読んで、変更ログを書いて」
2「価格ページを取得して、競合他社の機能を要約して」

1from dataclasses import dataclass
2
3@dataclass
4class TaskNode:
5    id: str
6    prompt: str
7    depends_on: list[str]  # IDs of nodes this one actually needs
8
9def has_real_edge(node_a: TaskNode, node_b: TaskNode) -> bool:
10    """
11    コア グラフ エンジニアリング テスト：
12    node_b のプロンプトは実際に node_a の出力を必要としますか？
13    """
14    return node_a.id in node_b.depends_on
15
16# 例：ほとんどの「チェーン」は 2〜3 の実際の依存関係グループに集約される
17nodes = [
18    TaskNode("audit_routes", "List all API route files", []),
19    TaskNode("check_auth", "Check auth middleware coverage", []),
20    TaskNode("fetch_weather", "Get today's weather", []),
21    TaskNode("summarize", "Summarize route + auth findings",
22              depends_on=["audit_routes", "check_auth"]),
23]
24
25# audit_routes、check_auth、fetch_weather の間にはエッジがない
26# これらは並列実行される。実際のエッジを持つのは "summarize" だけです -- 待機が発生します。

1このコードベース内のすべてのルート ファイルを監査するワークフローを作成します。
2
3各ルート ファイルについて、独立してチェックします：
4- 認証ミドルウェアが存在する
5- すべてのパラメータに入力検証がある
6- レート制限が設定されている
7- エラーハンドリングがスタック トレースを漏洩しない
8
9これらのチェックをすべてのルート ファイルにわたって並列実行します —
10互いに依存していません。
11
12すべてのファイルのチェックが完了したら、重要度（critical、warning、info）ごとにグループ化された
131 つの統合レポートを生成します。
14
15統合ステップは、すべてのチェックが完了するのを待つ必要があります。
16その前のすべてのステップは待つべきではありません。

1import asyncio
2from anthropic import Anthropic
3
4client = Anthropic()
5
6async def audit_route_file(filepath: str) -> dict:
7    """1 つのノード。他のすべてのルート ファイルから独立して実行されます。"""
8    response = await client.messages.create(
9        model="claude-sonnet-5",
10        max_tokens=1000,
11        messages=[{
12            "role": "user",
13            "content": f"""Audit this route file for:
14            - auth middleware, input validation,
15              rate limiting, error handling
16
17            File: {filepath}
18
19            Return JSON: {{"file": "", "issues": [], "severity": ""}}"""
20        }]
21    )
22    return {"file": filepath, "result": response.content[0].text}
23
24async def consolidate(results: list[dict]) -> str:
25    """唯一の実際のエッジ -- すべての監査ノードが完了するのを待ちます。"""
26    response = await client.messages.create(
27        model="claude-opus-4-8",
28        max_tokens=2000,
29        messages=[{
30            "role": "user",
31            "content": f"""Consolidate these {len(results)} route audits
32            into one report grouped by severity:
33
34            {results}"""
35        }]
36    )
37    return response.content[0].text
38
39async def run_graph(route_files: list[str]):
40    # ファンアウト -- すべての独立したノードが並行して実行される
41    audit_tasks = [audit_route_file(f) for f in route_files]
42    results = await asyncio.gather(*audit_tasks)
43
44    # ファンイン -- 実際の依存関係を持つ唯一のノード
45    report = await consolidate(results)
46    return report
47
48# 40 個のルート ファイル、1 つのプロンプト、1 回の並列パス
49results = asyncio.run(run_graph([
50    f"routes/{f}.py" for f in ["auth", "users", "billing", "orders"]
51    # ...36 more

1async def layered_consolidate(results: list[dict], batch_size: int = 30):
2    """階層的なファンイン -- 大規模な生の出力を合成しないでください。"""
3    batches = [results[i:i+batch_size]
4               for i in range(0, len(results), batch_size)]
5
6    batch_summaries = await asyncio.gather(*[
7        summarize_batch(batch) for batch in batches
8    ])
9
10    # 最終的な統合は、1,000 の生の結果ではなく、要約に対して機能します
11    return await consolidate(batch_summaries)

1async def safe_consolidate(results: list[dict], expected_count: int):
2    if len(results) < expected_count:
3        missing = expected_count - len(results)
4        print(f"WARNING: {missing} 個のノードがサイレントに失敗しました。"
5              f"レポートは不完全になります。")
6    return await consolidate(results)

1                    オーケストレーター
2                         |
3        +--------+-------+-------+--------+
4        v        v       v       v        v
5     ノード 1  ノード 2 ノード 3  ...   ノード N
6   (並列、相互にエッジなし)
7        |        |       |               |
8        +--------+-------+-------+-------+
9                         v
10                 バッチ要約  <- 階層的ファンイン
11                 (30個ずつのグループ)
12                         v
13                 最終レポート   <- 唯一の真のエッジ

1async def orchestrate(task: str, resources: list[str]):
2    """
3    オーケストレーター ノード -- 分解は行うが、実行はしない。
4    """
5    plan = await client.messages.create(
6        model="claude-opus-4-8",
7        max_tokens=2000,
8        messages=[{
9            "role": "user",
10            "content": f"""Task: {task}
11            Resources available: {resources}
12
13            Decompose into a graph:
14            - List each independent node (no shared edges)
15            - List any real dependencies between nodes
16            - Group nodes into fan-in batches if count > 50
17
18            Return JSON with: nodes, edges, batch_groups"""
19        }]
20    )
21
22    graph = parse_plan(plan.content[0].text)
23
24    # 独立したノードを並列実行
25    node_results = await asyncio.gather(*[
26        execute_node(n) for n in graph["nodes"] if not n["depends_on"]
27    ])
28
29    # 次に、実際のエッジのみを尊重して依存ノードを実行
30    final = await execute_dependent_chain(graph["edges"], node_results)
31
32    return final