Orchestration Planner by daffy0208/ai-dev-standards
npx skills add https://github.com/daffy0208/ai-dev-standards --skill 'Orchestration Planner'利用能力图谱和 Codex 驱动的目标分解来规划多步骤工作流
接收高层次目标,并利用能力图谱将其分解为可执行的工作流。使用 Codex 来理解目标语义、查找匹配的能力、验证前置条件,并生成包含备选方案和评分的分层任务网络(HTN)计划。
inputs:
goal: string # 用户目标(例如,"implement RAG")
project_state: object # 当前项目状态(文件、依赖项、环境变量)
capability_graph: string # capability-graph.json 的路径
preferences: object # 用户偏好(cost_weight, risk_tolerance 等)
context: array # 最近使用的能力(用于冷却)
广告位招租
在这里展示您的产品或服务
触达数万 AI 开发者,精准高效
# 使用 Codex 理解目标并提取所需效果
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
# 查询能力图谱以寻找匹配的能力
python3 <<EOF
import json
# 加载目标分析
with open('/tmp/goal-analysis.json') as f:
goal_analysis = json.load(f)
# 加载能力图谱
with open('META/capability-graph.json') as f:
graph_data = json.load(f)
graph = graph_data['graph']
# 根据效果查找能力
candidates = []
for effect in goal_analysis['required_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:
candidates.append({
'capability': capability,
'effect': effect,
'required': True,
'node': node
})
# 查找可选能力
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
})
# 写入候选列表
with open('/tmp/candidates.json', 'w') as f:
json.dump(candidates, f, indent=2)
EOF
# 针对每个候选能力,检查前置条件是否满足
python3 <<EOF
import json
import os
import subprocess
# 加载候选列表
with open('/tmp/candidates.json') as f:
candidates = json.load(f)
# 加载项目状态
with open('/tmp/project-state.json') as f:
project_state = json.load(f)
def evaluate_precondition(check, project_state):
"""根据项目状态评估前置条件检查"""
# 处理 file_exists('path')
if check.startswith('file_exists('):
path = check[12:-2] # 从函数调用中提取路径
return os.path.exists(path)
# 处理 not file_exists('path')
if check.startswith('not file_exists('):
path = check[16:-2]
return not os.path.exists(path)
# 处理 has_dependency('package')
if check.startswith('has_dependency('):
package = check[15:-2]
return package in project_state.get('dependencies', {})
# 处理 env_var_set('VAR')
if check.startswith('env_var_set('):
var = check[12:-2]
return var in project_state.get('env_vars', {})
# 处理 OR 条件
if ' or ' in check:
parts = check.split(' or ')
return any(evaluate_precondition(p.strip(), project_state) for p in parts)
# 处理 AND 条件
if ' and ' in check:
parts = check.split(' and ')
return all(evaluate_precondition(p.strip(), project_state) for p in parts)
# 未知检查类型
return False
# 验证每个候选能力
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
# 写入验证后的候选列表
with open('/tmp/candidates-validated.json', 'w') as f:
json.dump(candidates, f, indent=2)
EOF
# 使用 Codex 构建分层任务网络
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
# 使用效用函数为计划评分
python3 <<EOF
import json
with open('/tmp/htn-plan.json') as f:
plan = json.load(f)
# 评分权重(可由用户偏好配置)
COST_WEIGHT = 0.3
LATENCY_WEIGHT = 0.2
RISK_WEIGHT = 0.3
DIVERSITY_WEIGHT = 0.2
# 将定性值映射为分数
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}
# 计算分数
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)
# 多样性奖励(使用不同的领域/种类)
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)
# 总效用
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
}
# 写入评分后的计划
with open('/tmp/plan-scored.json', 'w') as f:
json.dump(plan, f, indent=2)
print(f"Plan utility score: {utility:.2f}")
EOF
# 创建可解释的决策日志
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) {
// 来自清单的基础分数
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
// 多样性奖励:偏好来自代表性不足领域的能力
const domainCount = context.usedDomains[capability.domains[0]] || 0
const diversityBonus = 1.0 / (1 + domainCount)
score += diversityBonus * context.diversityWeight
// 冷却惩罚:降低最近使用过的能力的分数
const lastUsed = context.recentlyUsed[capability.name]
if (lastUsed) {
const stepsSince = context.currentStep - lastUsed
if (stepsSince < 3) {
score *= 0.7 // 30% 惩罚
}
}
// 新颖性奖励:偏好此计划中尚未使用的能力
if (!context.usedCapabilities.has(capability.name)) {
score *= 1.2 // 20% 奖励
}
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
}
}
查询能力图谱以查找匹配的能力和关系。
使用清单元数据(前置条件、效果、成本、延迟、风险)进行规划。
将集成到 scripts/brain/plan 命令中以进行交互式规划。
每周安装次数
0
代码仓库
GitHub 星标数
18
首次出现
1970年1月1日
安全审计
Plan multi-step workflows using capability graph and Codex-powered goal decomposition
Takes high-level goals and decomposes them into executable workflows using the capability graph. Uses Codex to understand goal semantics, find matching capabilities, validate preconditions, and generate Hierarchical Task Network (HTN) plans with alternatives and scoring.
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)
# 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
# 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
# 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
# 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
# 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
# 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
}
}
Queries the capability graph to find matching capabilities and relationships.
Uses manifest metadata (preconditions, effects, cost, latency, risk) for planning.
Will integrate into scripts/brain/plan command for interactive planning.
Weekly Installs
0
Repository
GitHub Stars
18
First Seen
Jan 1, 1970
Security Audits
AI Elements:基于shadcn/ui的AI原生应用组件库,快速构建对话界面
60,400 周安装