Orchestration Planner

オーケストレーションプランナー

能力グラフと Codex による目標分解を使用して、複数ステップのワークフローを計画します

目的

能力グラフを使用して、高レベルの目標を実行可能なワークフローに分解します。Codex を使用して目標の意味を理解し、一致する能力を見つけ、前提条件を検証し、代替案とスコアリングを含む階層型タスクネットワーク（HTN）プランを生成します。

使用するタイミング

• ユーザーの目標を実行可能なプランに変換する場合（例：「RAG システムを構築する」）
• 複雑なタスクに最適な能力の順序を見つける場合
• プロジェクトの状態が能力をサポートしているか検証する場合
• 主な経路が妨げられたときに代替案を生成する場合
• 特定の能力が推奨される理由を説明する場合

主な機能

• 目標の分解：Codex を使用して、目標を必要な効果に分解します
• 能力のマッチング：望ましい効果を生み出す能力を見つけます
• 前提条件の検証：現在のプロジェクトの状態が要件を満たしているか確認します
• HTN 計画：サブタスクを含む階層型タスクネットワークを構築します
• スコアリングと順位付け：コスト、レイテンシー、リスク、多様性に基づいて経路を評価します
• 代替案の生成：主な経路が失敗したときに、代替オプションを提示します
• 意思決定ログ：能力選択の根拠を記録します

入力

inputs:
  goal: string # User goal (e.g., "implement RAG")
  project_state: object # Current project state (files, dependencies, env vars)
  capability_graph: string # Path to capability-graph.json
  preferences: object # User preferences (cost_weight, risk_tolerance, etc.)
  context: array # Recently used capabilities (for cooldown)

プロセス

ステップ 1：Codex による目標の分析

# Use Codex to understand goal and extract required effects
codex exec "
Analyze this goal and determine what effects are needed:

GOAL: ${USER_GOAL}

Examples of effects:
- creates_vector_index
- adds_auth_middleware
- configures_database
- implements_api_endpoint
- adds_tests

Task: Extract the effects needed to achieve this goal.

Output JSON:
{
  \"goal\": \"original goal\",
  \"required_effects\": [\"effect1\", \"effect2\"],
  \"optional_effects\": [\"effect3\"],
  \"domains\": [\"rag\", \"api\"],
  \"reasoning\": \"explanation\"
}
" > /tmp/goal-analysis.json

ステップ 2：候補となる能力の検索

# Query capability graph for matching capabilities
python3 <<EOF
import json

# Load goal analysis
with open('/tmp/goal-analysis.json') as f:
    goal_analysis = json.load(f)

# Load capability graph
with open('META/capability-graph.json') as f:
    graph_data = json.load(f)
    graph = graph_data['graph']

# Find capabilities by effect
candidates = []
for effect in goal_analysis['required_effects']:
    if effect in graph['effects']:
        for capability in graph['effects'][effect]:
            # Get full capability data
            node = next((n for n in graph['nodes'] if n['id'] == capability), None)
            if node:
                candidates.append({
                    'capability': capability,
                    'effect': effect,
                    'required': True,
                    'node': node
                })

# Find optional capabilities
for effect in goal_analysis.get('optional_effects', []):
    if effect in graph['effects']:
        for capability in graph['effects'][effect]:
            node = next((n for n in graph['nodes'] if n['id'] == capability), None)
            if node and capability not in [c['capability'] for c in candidates]:
                candidates.append({
                    'capability': capability,
                    'effect': effect,
                    'required': False,
                    'node': node
                })

# Write candidates
with open('/tmp/candidates.json', 'w') as f:
    json.dump(candidates, f, indent=2)
EOF

ステップ 3：前提条件の検証

# For each candidate, check if preconditions are satisfied
python3 <<EOF
import json
import os
import subprocess

# Load candidates
with open('/tmp/candidates.json') as f:
    candidates = json.load(f)

# Load project state
with open('/tmp/project-state.json') as f:
    project_state = json.load(f)

def evaluate_precondition(check, project_state):
    """Evaluate a precondition check against project state"""
    # Handle file_exists('path')
    if check.startswith('file_exists('):
        path = check[12:-2]  # Extract path from function call
        return os.path.exists(path)

    # Handle not file_exists('path')
    if check.startswith('not file_exists('):
        path = check[16:-2]
        return not os.path.exists(path)

    # Handle has_dependency('package')
    if check.startswith('has_dependency('):
        package = check[15:-2]
        return package in project_state.get('dependencies', {})

    # Handle env_var_set('VAR')
    if check.startswith('env_var_set('):
        var = check[12:-2]
        return var in project_state.get('env_vars', {})

    # Handle OR conditions
    if ' or ' in check:
        parts = check.split(' or ')
        return any(evaluate_precondition(p.strip(), project_state) for p in parts)

    # Handle AND conditions
    if ' and ' in check:
        parts = check.split(' and ')
        return all(evaluate_precondition(p.strip(), project_state) for p in parts)

    # Unknown check type
    return False

# Validate each candidate
for candidate in candidates:
    node = candidate['node']
    preconditions = node.get('preconditions', [])

    satisfied = []
    unsatisfied = []

    for precond in preconditions:
        check = precond['check']
        required = precond.get('required', True)
        result = evaluate_precondition(check, project_state)

        if result:
            satisfied.append(precond)
        else:
            unsatisfied.append(precond)
            if required:
                candidate['blocked'] = True
                candidate['missing_precondition'] = precond

    candidate['satisfied_preconditions'] = satisfied
    candidate['unsatisfied_preconditions'] = unsatisfied

# Write validated candidates
with open('/tmp/candidates-validated.json', 'w') as f:
    json.dump(candidates, f, indent=2)
EOF

ステップ 4：Codex による HTN プランの構築

# Use Codex to build hierarchical task network
CANDIDATES=$(cat /tmp/candidates-validated.json)
GRAPH=$(cat META/capability-graph.json)

codex exec "
Build a hierarchical task network (HTN) plan to achieve this goal.

GOAL:
$(cat /tmp/goal-analysis.json)

AVAILABLE CAPABILITIES:
$CANDIDATES

CAPABILITY GRAPH:
$GRAPH

Task: Create an HTN plan with ordered steps, alternatives, and dependencies.

HTN Structure:
{
  \"goal\": \"original goal\",
  \"plan\": [
    {
      \"step\": 1,
      \"capability\": \"capability-name\",
      \"effect\": \"what this achieves\",
      \"required\": true,
      \"blocked\": false,
      \"alternatives\": [\"alt-capability-1\", \"alt-capability-2\"],
      \"dependencies\": [\"step-0\"],
      \"reasoning\": \"why this capability\"
    }
  ],
  \"total_cost\": \"medium\",
  \"total_latency\": \"slow\",
  \"max_risk\": \"low\",
  \"parallel_steps\": [[1, 2], [3, 4]]
}

Instructions:
1. Order capabilities by dependencies (requires/enables)
2. For each capability, list alternatives with similar effects
3. Mark parallel-executable steps
4. Validate each step's preconditions
5. Calculate aggregate cost/latency/risk
6. Explain reasoning for each choice

Output ONLY valid JSON.
" > /tmp/htn-plan.json

ステップ 5：プランのスコアリングと順位付け

# Score plan using utility function
python3 <<EOF
import json

with open('/tmp/htn-plan.json') as f:
    plan = json.load(f)

# Scoring weights (configurable by user preferences)
COST_WEIGHT = 0.3
LATENCY_WEIGHT = 0.2
RISK_WEIGHT = 0.3
DIVERSITY_WEIGHT = 0.2

# Map qualitative values to scores
cost_scores = {'free': 1.0, 'low': 0.8, 'medium': 0.5, 'high': 0.2}
latency_scores = {'instant': 1.0, 'fast': 0.7, 'slow': 0.3}
risk_scores = {'safe': 1.0, 'low': 0.8, 'medium': 0.5, 'high': 0.2, 'critical': 0.0}

# Calculate scores
cost_score = cost_scores.get(plan['total_cost'], 0.5)
latency_score = latency_scores.get(plan['total_latency'], 0.5)
risk_score = risk_scores.get(plan['max_risk'], 0.5)

# Diversity bonus (using different domains/kinds)
capabilities = [step['capability'] for step in plan['plan']]
unique_domains = len(set([cap.split('-')[0] for cap in capabilities]))
diversity_score = min(unique_domains / 5.0, 1.0)

# Total utility
utility = (
    cost_score * COST_WEIGHT +
    latency_score * LATENCY_WEIGHT +
    risk_score * RISK_WEIGHT +
    diversity_score * DIVERSITY_WEIGHT
)

plan['scores'] = {
    'utility': utility,
    'cost_score': cost_score,
    'latency_score': latency_score,
    'risk_score': risk_score,
    'diversity_score': diversity_score
}

# Write scored plan
with open('/tmp/plan-scored.json', 'w') as f:
    json.dump(plan, f, indent=2)

print(f"Plan utility score: {utility:.2f}")
EOF

ステップ 6：意思決定ログの生成

# Create explainable decision log
cat > /tmp/decision-log.json <<EOF
{
  "goal": "$(jq -r .goal /tmp/goal-analysis.json)",
  "timestamp": "$(date -u +%Y-%m-%dT%H:%M:%SZ)",
  "selected_plan": $(cat /tmp/plan-scored.json),
  "alternatives_considered": [
    {
      "capability": "alternative-1",
      "reason_rejected": "Higher cost",
      "score": 0.65
    }
  ],
  "precondition_checks": $(cat /tmp/candidates-validated.json),
  "reasoning": "Selected capabilities optimizing for low risk and moderate cost"
}
EOF

スコアリング関数

function scoreCapability(capability, context) {
  // Base scores from manifest
  const costScores = { free: 1.0, low: 0.8, medium: 0.5, high: 0.2 }
  const latencyScores = { instant: 1.0, fast: 0.7, slow: 0.3 }
  const riskScores = { safe: 1.0, low: 0.8, medium: 0.5, high: 0.2, critical: 0.0 }

  let score = 0
  score += costScores[capability.cost] * context.costWeight
  score += latencyScores[capability.latency] * context.latencyWeight
  score += riskScores[capability.risk_level] * context.riskWeight

  // Diversity bonus: prefer capabilities from underrepresented domains
  const domainCount = context.usedDomains[capability.domains[0]] || 0
  const diversityBonus = 1.0 / (1 + domainCount)
  score += diversityBonus * context.diversityWeight

  // Cooldown penalty: reduce score for recently used capabilities
  const lastUsed = context.recentlyUsed[capability.name]
  if (lastUsed) {
    const stepsSince = context.currentStep - lastUsed
    if (stepsSince < 3) {
      score *= 0.7 // 30% penalty
    }
  }

  // Novelty bonus: prefer capabilities not yet used in this plan
  if (!context.usedCapabilities.has(capability.name)) {
    score *= 1.2 // 20% bonus
  }

  return score
}

出力例

{
  "goal": "Build RAG system for documentation search",
  "plan": [
    {
      "step": 1,
      "capability": "openai-integration",
      "effect": "enables_embedding_generation",
      "required": true,
      "blocked": false,
      "alternatives": ["anthropic-integration", "local-embeddings"],
      "dependencies": [],
      "reasoning": "Provides embeddings API for vector generation",
      "preconditions_met": true
    },
    {
      "step": 2,
      "capability": "pinecone-mcp",
      "effect": "creates_vector_index",
      "required": true,
      "blocked": false,
      "alternatives": ["weaviate-mcp", "qdrant-mcp"],
      "dependencies": [1],
      "reasoning": "Managed vector database with low latency",
      "preconditions_met": true
    },
    {
      "step": 3,
      "capability": "rag-implementer",
      "effect": "configures_retrieval_pipeline",
      "required": true,
      "blocked": false,
      "alternatives": [],
      "dependencies": [1, 2],
      "reasoning": "Orchestrates embedding + retrieval workflow",
      "preconditions_met": true
    }
  ],
  "total_cost": "medium",
  "total_latency": "slow",
  "max_risk": "low",
  "parallel_steps": [[1, 2]],
  "scores": {
    "utility": 0.78,
    "cost_score": 0.65,
    "latency_score": 0.7,
    "risk_score": 0.8,
    "diversity_score": 0.8
  }
}

統合

capability-graph-builder との連携

能力グラフを照会して、一致する能力とその関係を見つけます。

manifest-generator との連携

計画には、マニフェストのメタデータ（前提条件、効果、コスト、レイテンシー、リスク）を使用します。

Repository Brain との連携

対話型の計画作成用に、scripts/brain/plan コマンドへ統合する予定です。

成功指標

• ✅ テスト目標の 90％以上をプランで正しく分解できる
• ✅ 前提条件の検証で、互換性のない能力を検出できる
• ✅ 代替案の生成により、各ステップで 2 つ以上のオプションを提示できる
• ✅ 意思決定ログが説明可能で、追跡可能である
• ✅ スコアリング関数が妥当な順位を生成する

関連スキル

• capability-graph-builder：照会可能なグラフを提供します
• manifest-generator：能力のメタデータを提供します
• system-diagnostician：推奨のためにプランナーを使用します