Contents
1. 基本アーキテクチャ
オーケストレーターはタスク全体の流れを司り、サブエージェントは独立したコンテキストで個別処理を行います。この分離により コンテキスト汚染 を防ぎつつ、必要なだけエージェントを増減できる柔軟性が得られます。
1‑1. オーケストレーターとサブエージェントの役割
オーケストレーターは次の三つの主要責務を持ちます。
| 責務 | 内容 |
|---|---|
| タスク分配 | 受信したリクエストを適切なサブエージェントへ振り分ける |
| 結果統合 | 各サブエージェントの出力を集約し、最終レスポンスを生成する |
| ライフサイクル管理 | サブエージェントの起動・停止・再起動を API で制御する |
※オーケストレーターが提供する API は公式リポジトリの
README.mdに記載されています（2024 年 12 月時点）。
サブエージェントは 与えられた情報だけ を参照し、独自のコンテキストウィンドウで処理を完結させます。
1‑2. コンテキスト分離がもたらす利点
| 利点 | 説明 |
|---|---|
| 汚染防止 | 他タスクの履歴が混在しないため、プロンプトの一貫性が保たれる |
| スケーラビリティ | 必要に応じてエージェント数を増減でき、リソース最適化が可能 |
| 障害切り分け | 1 エージェントが失敗しても他は継続できるため、システム全体の可用性が向上 |
2. config.toml の設定方法
正しい構成ファイルはサブエージェントの安定稼働に不可欠です。本節では必須項目と実務で推奨される設定例を示します。
2‑1. 必須パラメータ一覧
以下の項目は すべて記述 が必要です。型やフォーマットは公式スキーマに合わせます（codex-agent-schema.json を参照）。
| パラメータ | 型 | 説明 |
|---|---|---|
agent.id |
string | オーケストレーター側の一意識別子 |
orchestrator.endpoint |
url | REST エンドポイント（例: https://api.codex.openai.com/v1/orchestrator） |
security.token |
string (JWT) | 認証トークン。環境変数から注入することが推奨されます |
subagents.max_concurrency |
integer | 同時に起動できるサブエージェント上限（デフォルト 10） |
logging.level |
enum(debug,info,warn,error) |
ログ出力レベル。運用開始直後は debug が便利です |
2‑2. 推奨設定例（PR レビュー自動化向け）
以下は プルリクエストレビュー を自動化するシナリオで実際に使用できるサンプルです。コメント行で各項目の意味を補足しています。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 |
# ------------------------------------------------- # 基本情報 # ------------------------------------------------- agent.id = "codex-orchestrator-prreview" orchestrator.endpoint = "https://api.codex.openai.com/v1/orchestrator" security.token = "${CODEX_ORCHESTRATOR_TOKEN}" # 環境変数から取得し、平文保存は避ける # ------------------------------------------------- # サブエージェント設定 # ------------------------------------------------- [subagents] max_concurrency = 5 # 同時に最大 5 件の PR を処理 default_timeout_seconds = 180 # 各タスクのタイムアウト上限（3 分） # ------------------------------------------------- # ロギング & モニタリング # ------------------------------------------------- [logging] level = "debug" output = "stdout" # 本番環境では CloudWatch / Stackdriver 推奨 format = "json" # ------------------------------------------------- # 拡張ポイント（公式拡張 API の例） # ------------------------------------------------- [extensions] worktree.enabled = true # Worktree 並列実行を有効化 cloud_tasks.queue_name = "codex-prreview-queue" |
設定項目の詳細は OpenAI Codex 公式ドキュメント（
https://platform.openai.com/docs/codex/agents）をご確認ください。
3. Agents SDK を使ったサブエージェント操作
SDK は Python と TypeScript の両方で提供され、config.toml の内容を自動的に読み込んでクライアントインスタンスを生成できます。本節では 共通ハンドリング を意識したコード例を示します。
3‑1. SDK のインストールと認証
Python
|
1 2 |
pip install openai-codex-agents-sdk |
|
1 2 3 4 5 6 7 8 9 10 |
import os from codex_agents import OrchestratorClient # 環境変数からトークン取得（ベストプラクティス） token = os.getenv("CODEX_ORCHESTRATOR_TOKEN") client = OrchestratorClient( endpoint="https://api.codex.openai.com/v1/orchestrator", auth_token=token, ) |
TypeScript
|
1 2 |
npm install @openai/codex-agents-sdk |
|
1 2 3 4 5 6 7 8 |
import { OrchestratorClient } from '@openai/codex-agents-sdk'; const token = process.env.CODEX_ORCHESTRATOR_TOKEN!; const client = new OrchestratorClient({ endpoint: 'https://api.codex.openai.com/v1/orchestrator', authToken: token, }); |
3‑2. サブエージェントの登録・起動（共通ハンドリング）
以下は エラーハンドリングと再試行ロジック を関数化した例です。重複コードを排除し、可読性を高めています。
Python
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 |
import tomllib import time from typing import Any def load_config(path: str = "config.toml") -> dict[str, Any]: with open(path, "rb") as f: return tomllib.load(f) def retry(operation, retries: int = 3, delay: float = 2.0): for attempt in range(1, retries + 1): try: return operation() except Exception as e: if attempt == retries: raise time.sleep(delay) cfg = load_config() # 登録と起動をまとめて実行 def launch_subagent(name: str): client.register_subagent( id=name, max_concurrency=cfg["subagents"]["max_concurrency"], timeout_seconds=cfg["subagents"]["default_timeout_seconds"], ) retry(lambda: client.start_subagent(name)) launch_subagent("pr-reviewer") |
TypeScript
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 |
import * as fs from 'fs'; import * as toml from '@iarna/toml'; interface Config { subagents: { max_concurrency: number; default_timeout_seconds: number }; } const cfg = toml.parse(fs.readFileSync('config.toml', 'utf-8')) as Config; async function retry<T>(fn: () => Promise<T>, retries = 3, delay = 2000): Promise<T> { for (let i = 0; i < retries; i++) { try { return await fn(); } catch (e) { if (i === retries - 1) throw e; await new Promise(r => setTimeout(r, delay)); } } // TypeScript のコンパイラが到達不能コードとみなすためのダミー throw new Error('unreachable'); } async function launchSubagent(name: string) { await client.registerSubagent({ id: name, maxConcurrency: cfg.subagents.max_concurrency, timeoutSeconds: cfg.subagents.default_timeout_seconds, }); await retry(() => client.startSubagent(name)); } launchSubagent('pr-reviewer').catch(console.error); |
3‑3. ライフサイクル管理 API の利用例
公式 SDK が提供するメソッドは以下のようにまとめられます。エンドポイント名は 公式リファレンス に準拠しています。
| 操作 | SDK メソッド | 主な用途 |
|---|---|---|
| 状態取得 | client.get_subagent_status(id) |
実行中・完了・エラーを確認 |
| 再起動 | client.restart_subagent(id) |
障害復旧や設定変更後のリロード |
| 完全停止 | client.delete_subagent(id) |
不要になったサブエージェントの削除 |
|
1 2 3 4 5 6 |
status = client.get_subagent_status("pr-reviewer") print(f"PR reviewer の現在状態: {status.state}") if status.state == "error": client.restart_subagent("pr-reviewer") |
4. マルチタスク実装パターン：Worktree と Cloud Tasks
ここでは Git Worktree を使った並列レビューと、Google Cloud Tasks（または同等のキューサービス）へのジョブオフロード例を示します。
4‑1. Worktree を用いた並列レビュー
Worktree によりリポジトリの複数ブランチを同時にチェックアウトでき、サブエージェントごとに独立した作業領域が確保されます。以下はロック取得からクリーンアップまでをシェルスクリプトで表現した例です。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 |
#!/usr/bin/env bash # ファイルベースの排他ロックで同時実行数を制御 LOCK_DIR="/tmp/worktree.lock" if mkdir "${LOCK_DIR}" 2>/dev/null; then # ロック取得成功 → 作業開始 PR_NUM=$1 WORKTREE_PATH="/tmp/wt/pr-${PR_NUM}" git worktree add "${WORKTREE_PATH}" "refs/pull/${PR_NUM}/head" # Codex エージェントにレビューを依頼（仮想コマンド） codex review --path "${WORKTREE_PATH}" # 後処理 git worktree remove "${WORKTREE_PATH}" rmdir "${LOCK_DIR}" else echo "ロック取得失敗: 他のエージェントが作業中です。5 秒後に再試行してください。" >&2 exit 1 fi |
Worktree の詳細は OpenAI Codex 公式ガイド（
/guides/worktree.md）をご参照ください。
4‑2. Cloud Tasks へのジョブ投入と結果取得
長時間実行が予想されるタスクはキューにオフロードし、サブエージェントは「投げ込み」だけを担当します。以下は Python の例です（google-cloud-tasks ライブラリ使用）。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 |
import json from datetime import datetime, timedelta from google.cloud import tasks_v2 client = tasks_v2.CloudTasksClient() queue_path = client.queue_path('my-project', 'us-central1', 'codex-prreview-queue') def enqueue_review(pr_number: int): payload = {"pr": pr_number} task = { "app_engine_http_request": { # App Engine 環境を想定 "http_method": tasks_v2.HttpMethod.POST, "relative_uri": "/tasks/review", "body": json.dumps(payload).encode(), }, "schedule_time": (datetime.utcnow() + timedelta(seconds=5)).isoformat("T") + "Z", } client.create_task(request={"parent": queue_path, "task": task}) |
タスク実行側（例：Cloud Run）では結果を ステータス API で取得し、オーケストレーターが最終的に集約します。
5. 実務シナリオ：PR レビューと Issue トリアジングの同時自動化
この章では、実際の開発フローに組み込める 「PR が作成されたらコードレビューと Issue の分類を同時に走らせる」 パイプラインを設計します。
5‑1. フロー全体図と主要ステップ
| ステップ | 内容 |
|---|---|
| 1. Webhook 捕捉 | GitHub → Orchestrator の /webhook/pr エンドポイントで PR 情報を取得 |
| 2. サブエージェント A（レビュー） | pr-reviewer が Worktree を作成し、コード解析・コメント生成 |
| 3. サブエージェント B（トリアジング） | issue-triager が新規 Issue の取得とラベル付与・担当者割り当てを実施 |
| 4. 結果統合 | Orchestrator が両方の出力をマージし、GitHub API でコメント・ラベルを投稿 |
この構成により タスクごとのコンテキストは完全に分離 され、相互干渉が起きません。
5‑2. コンテキスト分離のベストプラクティス
| 項目 | 推奨設定 / 手順 |
|---|---|
| 環境変数の分離 | PR_REVIEWER_TOKEN と ISSUE_TRIAGER_TOKEN を別々に管理し、サブエージェント起動時に注入 |
| プロンプトスコープ限定 | 各エージェントは「対象 PR のみ」や「新規 Issue（バグ／機能要望）だけ」と明示的に指示 |
| タイムアウト設定 | レビュー → 180 秒、トリアジング → 120 秒で個別に調整 |
| ロック機構の併用 | Worktree 用はファイルベースロック、Issue キュー用は Cloud Tasks の lease 機能を活用 |
5‑3. デバッグ・モニタリング手法
- 統一ログ形式：
logging.format = "json"に設定し、agent_idとtask_typeを必ず付与。Logstash / Fluentd で集約すると検索が容易です。 - ステータス取得エンドポイント
bash
GET https://api.codex.openai.com/v1/orchestrator/status?task=pr-reviewer
→running,completed,errorが返ります。 - エラーハンドリング例（Python）
python
status = client.get_subagent_status("pr-reviewer")
if status.state == "error":
logger.error(f"PRレビュー失敗: {status.detail}")
client.restart_subagent("pr-reviewer") # 再起動で自動復旧を試みる
- 可視化：Grafana ダッシュボードに
codex_agent_requests_totalとcodex_agent_error_secondsを登録し、閾値超過時は PagerDuty に通知。
おわりに
本稿で紹介した オーケストレーター + サブエージェント の構造と、公式 SDK・設定ファイルのベストプラクティスを組み合わせることで、以下が実現できます。
- コンテキスト汚染のない安全なマルチタスク処理
- 必要に応じたエージェント数の柔軟なスケーリング
- 障害時の自動復旧と可視化による運用効率向上
まずは config.toml を作成し、SDK でサブエージェントを起動してみましょう。公式リポジトリ（https://github.com/openai/codex-agents）にサンプルコードとテストスイートが用意されていますので、ローカル環境で動作確認した後に本番環境へ展開してください。
参考リンク
- OpenAI Codex 公式ドキュメント:https://platform.openai.com/docs/codex/agents
- SDK リポジトリ（Python / TypeScript）:https://github.com/openai/codex-agents-sdk