Goal-Seeking Agent Pattern Skill

目标导向Agent模式技能

1. What Are Goal-Seeking Agents?

1. 什么是目标导向Agent？

Goal-seeking agents are autonomous AI agents that execute multi-phase objectives by:

Understanding High-Level Goals: Accept natural language objectives without explicit step-by-step instructions
Planning Execution: Break goals into phases with dependencies and success criteria
Autonomous Execution: Make decisions and adapt behavior based on intermediate results
Self-Assessment: Evaluate progress against success criteria and adjust approach
Resilient Operation: Handle failures gracefully and explore alternative solutions

目标导向Agent是自主AI Agent，通过以下方式执行多阶段目标：

理解高层目标：接受自然语言目标，无需明确的分步指令
规划执行流程：将目标分解为带有依赖关系和成功标准的阶段
自主执行：根据中间结果做出决策并调整行为
自我评估：对照成功标准评估进度并调整方法
弹性运行：优雅处理故障并探索替代解决方案

Core Characteristics

核心特性

Autonomy: Agents decide HOW to achieve goals, not just follow prescriptive steps

Adaptability: Adjust strategy based on runtime conditions and intermediate results

Goal-Oriented: Focus on outcomes (what to achieve) rather than procedures (how to achieve)

Multi-Phase: Complex objectives decomposed into manageable phases with dependencies

Self-Monitoring: Track progress, detect failures, and course-correct autonomously

自主性：Agent决定如何实现目标，而非仅仅遵循预设步骤

适应性：根据运行时条件和中间结果调整策略

目标导向：聚焦于结果（要实现什么）而非流程（如何实现）

多阶段：复杂目标被分解为具有依赖关系的可管理阶段

自我监控：跟踪进度、检测故障并自主调整方向

Distinction from Traditional Agents

与传统Agent的区别

Traditional Agent	Goal-Seeking Agent
Follows fixed workflow	Adapts workflow to context
Prescriptive steps	Outcome-oriented objectives
Human intervention on failure	Autonomous recovery attempts
Single-phase execution	Multi-phase with dependencies
Rigid decision tree	Dynamic strategy adjustment

传统Agent	目标导向Agent
遵循固定工作流	根据上下文调整工作流
预设步骤	面向结果的目标
故障时需要人工干预	自主尝试恢复
单阶段执行	多阶段且带有依赖关系
刚性决策树	动态调整策略

When Goal-Seeking Makes Sense

何时适合使用目标导向模式

Goal-seeking agents excel when:

Problem space is large: Many possible paths to success
Context varies: Runtime conditions affect optimal approach
Failures are expected: Need autonomous recovery without human intervention
Objectives are clear: Success criteria well-defined but path is flexible
Multi-step complexity: Requires coordination across phases with dependencies

目标导向Agent在以下场景中表现出色：

问题空间大：存在多种可能的成功路径
上下文多变：运行时条件影响最优方法
故障可预期：无需人工干预即可自主恢复
目标明确：成功标准清晰但实现路径灵活
多步骤复杂度：需要跨阶段协调且阶段间存在依赖关系

When to Avoid Goal-Seeking

何时避免使用目标导向模式

Use traditional agents or scripts when:

Single deterministic path: Only one way to achieve goal
Latency-critical: Need fastest possible execution (no decision overhead)
Safety-critical: Human verification required at each step
Simple workflow: Complexity of goal-seeking exceeds benefit
Audit requirements: Need deterministic, reproducible execution

当以下情况时，使用传统Agent或脚本更合适：

单一确定性路径：实现目标只有一种方式
延迟敏感：需要最快的执行速度（无决策开销）
安全关键：每个步骤都需要人工验证
工作流简单：目标导向的复杂度超过收益
审计要求：需要确定性、可复现的执行流程

2. When to Use This Pattern

2. 何时使用该模式

Problem Indicators

问题指标

Use goal-seeking agents when you observe these patterns:

当观察到以下模式时，使用目标导向Agent：

Pattern 1: Workflow Variability

模式1：工作流可变性

Indicators:

Same objective requires different approaches based on context
Manual decisions needed at multiple points
"It depends" answers when mapping workflow

Example: Release workflow that varies by:

Environment (staging vs production)
Change type (hotfix vs feature)
Current system state (healthy vs degraded)

Solution: Goal-seeking agent evaluates context and adapts workflow

指标：

相同目标需要根据上下文采用不同方法
多个决策点需要人工判断
映射工作流时频繁出现“视情况而定”的答案

示例：发布工作流因以下因素而异：

环境（ staging 与 production）
变更类型（热修复与功能更新）
当前系统状态（健康与降级）

解决方案：目标导向Agent评估上下文并调整工作流

Pattern 2: Multi-Phase Complexity

模式2：多阶段复杂度

Indicators:

Objective requires 3-5+ distinct phases
Phases have dependencies (output of phase N feeds phase N+1)
Parallel execution opportunities exist
Success criteria differ per phase

Example: Data pipeline with phases:

Data collection (multiple sources, parallel)
Transformation (depends on collection results)
Validation (depends on transformation output)
Publishing (conditional on validation pass)

Solution: Goal-seeking agent orchestrates phases, handles dependencies

指标：

目标需要3-5个以上不同阶段
阶段之间存在依赖关系（第N阶段的输出作为第N+1阶段的输入）
存在并行执行的机会
各阶段的成功标准不同

示例：数据管道包含以下阶段：

数据收集（多源、并行）
数据转换（依赖收集结果）
数据验证（依赖转换输出）
数据发布（取决于验证是否通过）

解决方案：目标导向Agent编排各阶段，处理依赖关系

Pattern 3: Autonomous Recovery Needed

模式3：需要自主恢复

Indicators:

Failures are expected and recoverable
Multiple retry/fallback strategies exist
Human intervention is expensive or slow
Can verify success programmatically

Example: CI diagnostic workflow:

Test failures (retry with different approach)
Environment issues (reconfigure and retry)
Dependency conflicts (resolve and rerun)

Solution: Goal-seeking agent tries strategies until success or escalation

指标：

故障可预期且可恢复
存在多种重试/回退策略
人工干预成本高或速度慢
可通过编程方式验证成功

示例：CI诊断工作流：

测试失败（采用不同方法重试）
环境问题（重新配置并重试）
依赖冲突（解决后重新运行）

解决方案：目标导向Agent尝试多种策略直至成功或升级至人工处理

Pattern 4: Adaptive Decision Making

模式4：自适应决策

Indicators:

Need to evaluate trade-offs at runtime
Multiple valid solutions with different characteristics
Optimization objectives (speed vs quality vs cost)
Context-dependent best practices

Example: Fix agent pattern matching:

QUICK mode for obvious issues
DIAGNOSTIC mode for unclear problems
COMPREHENSIVE mode for complex solutions

Solution: Goal-seeking agent selects strategy based on problem analysis

指标：

需要在运行时评估权衡
存在多个具有不同特性的有效解决方案
优化目标（速度 vs 质量 vs 成本）
最佳实践取决于上下文

示例：修复Agent模式匹配：

针对明显问题的QUICK模式
针对不明确问题的DIAGNOSTIC模式
针对复杂解决方案的COMPREHENSIVE模式

解决方案：目标导向Agent根据问题分析选择策略

Pattern 5: Domain Expertise Required

模式5：需要领域专业知识

Indicators:

Requires specialized knowledge to execute
Multiple domain-specific tools/approaches
Best practices vary by domain
Coordination of specialized sub-agents

Example: AKS SRE automation:

Azure-specific operations (ARM, CLI)
Kubernetes expertise (kubectl, YAML)
Networking knowledge (CNI, ingress)
Security practices (RBAC, Key Vault)

Solution: Goal-seeking agent with domain expertise coordinates specialized actions

指标：

执行需要专业知识
存在多种领域特定工具/方法
最佳实践因领域而异
需要协调专业子Agent

示例：AKS SRE自动化：

Azure特定操作（ARM、CLI）
Kubernetes专业知识（kubectl、YAML）
网络知识（CNI、ingress）
安全实践（RBAC、Key Vault）

解决方案：具有领域专业知识的目标导向Agent协调专业操作

Decision Framework

决策框架

Use this 5-question framework to evaluate goal-seeking applicability:

使用以下5个问题的框架评估是否适用目标导向模式：

Question 1: Is the objective well-defined but path flexible?

问题1：目标是否明确但实现路径灵活？

YES if:

Clear success criteria exist
Multiple valid approaches
Runtime context affects optimal path

NO if:

Only one correct approach
Path is deterministic
Success criteria ambiguous

Example YES: "Ensure AKS cluster is production-ready" (many paths, clear criteria) Example NO: "Run specific kubectl command" (one path, prescriptive)

是，如果：

存在清晰的成功标准
存在多种有效方法
运行时上下文影响最优路径

否，如果：

只有一种正确方法
路径是确定性的
成功标准模糊

示例（是）：“确保AKS集群已准备好投入生产”（多种路径，标准清晰） 示例（否）：“运行特定的kubectl命令”（单一路径，预设性强）

Question 2: Are there multiple phases with dependencies?

问题2：是否存在多个带有依赖关系的阶段？

YES if:

Objective naturally decomposes into 3-5+ phases
Phase outputs feed subsequent phases
Some phases can execute in parallel
Failures in one phase affect downstream phases

NO if:

Single-phase execution sufficient
No inter-phase dependencies
Purely sequential with no branching

Example YES: Data pipeline (collect → transform → validate → publish) Example NO: Format code with ruff (single atomic operation)

是，如果：

目标自然分解为3-5个以上阶段
阶段输出为后续阶段提供输入
某些阶段可并行执行
某一阶段的故障会影响下游阶段

否，如果：

单阶段执行即可满足需求
阶段间无依赖关系
纯顺序执行且无分支

示例（是）：数据管道（收集→转换→验证→发布） 示例（否）：使用ruff格式化代码（单一原子操作）

Question 3: Is autonomous recovery valuable?

问题3：自主恢复是否有价值？

YES if:

Failures are common and expected
Multiple recovery strategies exist
Human intervention is expensive/slow
Can verify success automatically

NO if:

Failures are rare edge cases
Manual investigation always required
Safety-critical (human verification needed)
Cannot verify success programmatically

Example YES: CI diagnostic workflow (try multiple fix strategies) Example NO: Deploy to production (human approval required)

是，如果：

故障常见且可预期
存在多种恢复策略
人工干预成本高/速度慢
可自动验证成功

否，如果：

故障是罕见的边缘情况
始终需要人工调查
安全关键（需要人工验证）
无法通过编程方式验证成功

示例（是）：CI诊断工作流（尝试多种修复策略） 示例（否）：部署到生产环境（需要人工批准）

Question 4: Does context significantly affect approach?

问题4：上下文是否显著影响方法？

YES if:

Environment differences change strategy
Current system state affects decisions
Trade-offs vary by situation (speed vs quality vs cost)
Domain-specific best practices apply

NO if:

Same approach works for all contexts
No environmental dependencies
No trade-off decisions needed

Example YES: Fix agent (quick vs diagnostic vs comprehensive based on issue) Example NO: Generate UUID (context-independent)

是，如果：

环境差异会改变策略
当前系统状态影响决策
权衡因情况而异（速度 vs 质量 vs 成本）
适用领域特定最佳实践

否，如果：

相同方法适用于所有上下文
无环境依赖
无需权衡决策

示例（是）：修复Agent（根据问题选择快速/诊断/全面模式） 示例（否）：生成UUID（与上下文无关）

Question 5: Is the complexity justified?

问题5：复杂度是否合理？

YES if:

Problem is repeated frequently (2+ times/week)
Manual execution takes 30+ minutes
High value from automation
Maintenance cost is acceptable

NO if:

One-off or rare problem
Quick manual execution (< 5 minutes)
Simple script suffices
Maintenance cost exceeds benefit

Example YES: CI failure diagnosis (frequent, time-consuming, high value) Example NO: One-time data migration (rare, script sufficient)

是，如果：

问题重复频繁（每周2次以上）
人工执行需要30分钟以上
自动化带来高价值
维护成本可接受

否，如果：

一次性或罕见问题
人工执行快速（<5分钟）
简单脚本即可满足需求
维护成本超过收益

示例（是）：CI故障诊断（频繁、耗时、高价值） 示例（否）：一次性数据迁移（罕见，脚本即可满足）

Decision Matrix

决策矩阵

3. Architecture Pattern

3. 架构模式

Component Architecture

组件架构

Goal-seeking agents have four core components:

python

undefined

目标导向Agent包含四个核心组件：

python

undefined

Component 1: Goal Definition

class GoalDefinition: """Structured representation of objective""" raw_prompt: str # Natural language goal goal: str # Extracted primary objective domain: str # Problem domain (security, data, automation, etc.) constraints: list[str] # Technical/operational constraints success_criteria: list[str] # How to verify success complexity: str # simple, moderate, complex context: dict # Additional metadata

Component 2: Execution Plan

class ExecutionPlan: """Multi-phase plan with dependencies""" goal_id: uuid.UUID phases: list[PlanPhase] total_estimated_duration: str required_skills: list[str] parallel_opportunities: list[list[str]] # Phases that can run parallel risk_factors: list[str]

Component 3: Plan Phase

class PlanPhase: """Individual phase in execution plan""" name: str description: str required_capabilities: list[str] estimated_duration: str dependencies: list[str] # Names of prerequisite phases parallel_safe: bool # Can execute in parallel success_indicators: list[str] # How to verify phase completion

Component 4: Skill Definition

class SkillDefinition: """Capability needed for execution""" name: str description: str capabilities: list[str] implementation_type: str # "native" or "delegated" delegation_target: str # Agent to delegate to

undefined

class SkillDefinition: """Capability needed for execution""" name: str description: str capabilities: list[str] implementation_type: str # "native" or "delegated" delegation_target: str # Agent to delegate to

undefined

Execution Flow

执行流程

┌─────────────────────────────────────────────────────────────┐
│ 1. GOAL ANALYSIS                                            │
│                                                             │
│  Input: Natural language objective                         │
│  Process: Extract goal, domain, constraints, criteria      │
│  Output: GoalDefinition                                    │
│                                                             │
│  [PromptAnalyzer.analyze_text(prompt)]                    │
└─────────────────────────────────────────────────────────────┘
                            ↓
┌─────────────────────────────────────────────────────────────┐
│ 2. PLANNING                                                 │
│                                                             │
│  Input: GoalDefinition                                     │
│  Process: Decompose into phases, identify dependencies     │
│  Output: ExecutionPlan                                     │
│                                                             │
│  [ObjectivePlanner.generate_plan(goal_definition)]        │
└─────────────────────────────────────────────────────────────┘
                            ↓
┌─────────────────────────────────────────────────────────────┐
│ 3. SKILL SYNTHESIS                                          │
│                                                             │
│  Input: ExecutionPlan                                      │
│  Process: Map capabilities to skills, identify agents      │
│  Output: list[SkillDefinition]                            │
│                                                             │
│  [SkillSynthesizer.synthesize(execution_plan)]            │
└─────────────────────────────────────────────────────────────┘
                            ↓
┌─────────────────────────────────────────────────────────────┐
│ 4. AGENT ASSEMBLY                                           │
│                                                             │
│  Input: GoalDefinition, ExecutionPlan, Skills              │
│  Process: Combine into executable bundle                   │
│  Output: GoalAgentBundle                                   │
│                                                             │
│  [AgentAssembler.assemble(goal, plan, skills)]            │
└─────────────────────────────────────────────────────────────┘
                            ↓
┌─────────────────────────────────────────────────────────────┐
│ 5. EXECUTION (Auto-Mode)                                    │
│                                                             │
│  Input: GoalAgentBundle                                    │
│  Process: Execute phases, monitor progress, adapt          │
│  Output: Success or escalation                             │
│                                                             │
│  [Auto-mode with initial_prompt from bundle]              │
└─────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────┐
│ 1. 目标分析                                                │
│                                                             │
│  输入：自然语言目标                                        │
│  处理：提取目标、领域、约束条件、成功标准                  │
│  输出：GoalDefinition                                       │
│                                                             │
│  [PromptAnalyzer.analyze_text(prompt)]                    │
└─────────────────────────────────────────────────────────────┘
                            ↓
┌─────────────────────────────────────────────────────────────┐
│ 2. 规划                                                     │
│                                                             │
│  输入：GoalDefinition                                      │
│  处理：分解为阶段，识别依赖关系                            │
│  输出：ExecutionPlan                                        │
│                                                             │
│  [ObjectivePlanner.generate_plan(goal_definition)]        │
└─────────────────────────────────────────────────────────────┘
                            ↓
┌─────────────────────────────────────────────────────────────┐
│ 3. 技能合成                                                │
│                                                             │
│  输入：ExecutionPlan                                       │
│  处理：将能力映射到技能，识别Agent                         │
│  输出：list[SkillDefinition]                               │
│                                                             │
│  [SkillSynthesizer.synthesize(execution_plan)]            │
└─────────────────────────────────────────────────────────────┘
                            ↓
┌─────────────────────────────────────────────────────────────┐
│ 4. Agent组装                                               │
│                                                             │
│  输入：GoalDefinition、ExecutionPlan、Skills               │
│  处理：组合为可执行包                                      │
│  输出：GoalAgentBundle                                      │
│                                                             │
│  [AgentAssembler.assemble(goal, plan, skills)]            │
└─────────────────────────────────────────────────────────────┘
                            ↓
┌─────────────────────────────────────────────────────────────┐
│ 5. 执行（自动模式）                                        │
│                                                             │
│  输入：GoalAgentBundle                                     │
│  处理：执行阶段、监控进度、自适应调整                      │
│  输出：成功或升级至人工处理                                │
│                                                             │
│  [Auto-mode with initial_prompt from bundle]              │
└─────────────────────────────────────────────────────────────┘

Phase Dependency Management

阶段依赖管理

Phases can have three relationship types:

Sequential Dependency: Phase B depends on Phase A completion

Phase A → Phase B → Phase C

Parallel Execution: Phases can run concurrently

Phase A ──┬→ Phase B ──┐
          └→ Phase C ──┴→ Phase D

Conditional Branching: Phase selection based on results

Phase A → [Decision] → Phase B (success path)
                    └→ Phase C (recovery path)

阶段之间可以有三种关系类型：

顺序依赖：阶段B依赖于阶段A完成

阶段A → 阶段B → 阶段C

并行执行：阶段可以同时运行

阶段A ──┬→ 阶段B ──┐
          └→ 阶段C ──┴→ 阶段D

条件分支：根据结果选择阶段

阶段A → [决策] → 阶段B（成功路径）
                    └→ 阶段C（恢复路径）

State Management

状态管理

Goal-seeking agents maintain state across phases:

python

class AgentState:
    """Runtime state for goal-seeking agent"""
    current_phase: str
    completed_phases: list[str]
    phase_results: dict[str, Any]  # Output from each phase
    failures: list[FailureRecord]  # Track what didn't work
    retry_count: int
    total_duration: timedelta
    context: dict                  # Shared context across phases

目标导向Agent在各阶段之间维护状态：

python

class AgentState:
    """Runtime state for goal-seeking agent"""
    current_phase: str
    completed_phases: list[str]
    phase_results: dict[str, Any]  # Output from each phase
    failures: list[FailureRecord]  # Track what didn't work
    retry_count: int
    total_duration: timedelta
    context: dict                  # Shared context across phases

Error Handling

错误处理

Three error recovery strategies:

Retry with Backoff: Same approach, exponential delay

python

for attempt in range(MAX_RETRIES):
    try:
        result = execute_phase(phase)
        break
    except RetryableError as e:
        wait_time = INITIAL_DELAY * (2 ** attempt)
        sleep(wait_time)

Alternative Strategy: Different approach to same goal

python

for strategy in STRATEGIES:
    try:
        result = execute_phase(phase, strategy)
        break
    except StrategyFailedError:
        continue  # Try next strategy
else:
    escalate_to_human("All strategies exhausted")

Graceful Degradation: Accept partial success

python

try:
    result = execute_phase_optimal(phase)
except OptimalFailedError:
    result = execute_phase_fallback(phase)  # Lower quality but works

三种错误恢复策略：

带退避的重试：相同方法，指数延迟

python

for attempt in range(MAX_RETRIES):
    try:
        result = execute_phase(phase)
        break
    except RetryableError as e:
        wait_time = INITIAL_DELAY * (2 ** attempt)
        sleep(wait_time)

替代策略：实现同一目标的不同方法

python

for strategy in STRATEGIES:
    try:
        result = execute_phase(phase, strategy)
        break
    except StrategyFailedError:
        continue  # Try next strategy
else:
    escalate_to_human("All strategies exhausted")

优雅降级：接受部分成功

python

try:
    result = execute_phase_optimal(phase)
except OptimalFailedError:
    result = execute_phase_fallback(phase)  # Lower quality but works

4. Integration with goal_agent_generator

4. 与goal_agent_generator集成

The

goal_agent_generator

module provides the implementation for goal-seeking agents. Here's how to integrate:

goal_agent_generator

模块提供目标导向Agent的实现。以下是集成方式：

Core API

核心API

python

from amplihack.goal_agent_generator import (
    PromptAnalyzer,
    ObjectivePlanner,
    SkillSynthesizer,
    AgentAssembler,
    GoalAgentPackager,
)

python

from amplihack.goal_agent_generator import (
    PromptAnalyzer,
    ObjectivePlanner,
    SkillSynthesizer,
    AgentAssembler,
    GoalAgentPackager,
)

Step 1: Analyze natural language goal

analyzer = PromptAnalyzer() goal_definition = analyzer.analyze_text(""" Automate AKS cluster production readiness verification. Check security, networking, monitoring, and compliance. Generate report with actionable recommendations. """)

Step 2: Generate execution plan

planner = ObjectivePlanner() execution_plan = planner.generate_plan(goal_definition)

Step 3: Synthesize required skills

synthesizer = SkillSynthesizer() skills = synthesizer.synthesize(execution_plan)

Step 4: Assemble complete agent

assembler = AgentAssembler() agent_bundle = assembler.assemble( goal_definition=goal_definition, execution_plan=execution_plan, skills=skills, bundle_name="aks-readiness-checker" )

Step 5: Package for deployment

packager = GoalAgentPackager() packager.package( bundle=agent_bundle, output_dir=".claude/agents/goal-driven/aks-readiness-checker" )

undefined

packager = GoalAgentPackager() packager.package( bundle=agent_bundle, output_dir=".claude/agents/goal-driven/aks-readiness-checker" )

undefined

CLI Integration

CLI集成

bash

undefined

bash

undefined

Generate agent from prompt file

amplihack goal-agent-generator create
--prompt ./prompts/aks-readiness.md
--output .claude/agents/goal-driven/aks-readiness-checker

Generate agent from inline prompt

amplihack goal-agent-generator create
--inline "Automate CI failure diagnosis and fix iteration"
--output .claude/agents/goal-driven/ci-fixer

List generated agents

amplihack goal-agent-generator list

Test agent execution

amplihack goal-agent-generator test
--agent-path .claude/agents/goal-driven/ci-fixer
--dry-run

undefined

amplihack goal-agent-generator test
--agent-path .claude/agents/goal-driven/ci-fixer
--dry-run

undefined

PromptAnalyzer Details

PromptAnalyzer详情

Extracts structured information from natural language:

python

from amplihack.goal_agent_generator import PromptAnalyzer
from pathlib import Path

analyzer = PromptAnalyzer()

从自然语言中提取结构化信息：

python

from amplihack.goal_agent_generator import PromptAnalyzer
from pathlib import Path

analyzer = PromptAnalyzer()

From file

goal_def = analyzer.analyze(Path("./prompts/my-goal.md"))

From text

goal_def = analyzer.analyze_text("Deploy and monitor microservices to AKS")

GoalDefinition contains:

print(goal_def.goal) # "Deploy and monitor microservices to AKS" print(goal_def.domain) # "deployment" print(goal_def.constraints) # ["Zero downtime", "Rollback capability"] print(goal_def.success_criteria) # ["All pods running", "Metrics visible"] print(goal_def.complexity) # "moderate" print(goal_def.context) # {"priority": "high", "scale": "medium"}


Domain classification:

- `data-processing`: Data transformation, analysis, ETL
- `security-analysis`: Vulnerability scanning, audits
- `automation`: Workflow automation, scheduling
- `testing`: Test generation, validation
- `deployment`: Release, publishing, distribution
- `monitoring`: Observability, alerting
- `integration`: API connections, webhooks
- `reporting`: Dashboards, metrics, summaries

Complexity determination:

- `simple`: Single-phase, < 50 words, basic operations
- `moderate`: 2-4 phases, 50-150 words, some coordination
- `complex`: 5+ phases, > 150 words, sophisticated orchestration

print(goal_def.goal) # "Deploy and monitor microservices to AKS" print(goal_def.domain) # "deployment" print(goal_def.constraints) # ["Zero downtime", "Rollback capability"] print(goal_def.success_criteria) # ["All pods running", "Metrics visible"] print(goal_def.complexity) # "moderate" print(goal_def.context) # {"priority": "high", "scale": "medium"}


领域分类：

- `data-processing`：数据转换、分析、ETL
- `security-analysis`：漏洞扫描、审计
- `automation`：工作流自动化、调度
- `testing`：测试生成、验证
- `deployment`：发布、部署、分发
- `monitoring`：可观测性、告警
- `integration`：API连接、Webhook
- `reporting`：仪表板、指标、摘要

复杂度判定：

- `simple`：单阶段、<50词、基础操作
- `moderate`：2-4阶段、50-150词、需要一定协调
- `complex`：5+阶段、>150词、复杂编排

ObjectivePlanner Details

ObjectivePlanner详情

Generates multi-phase execution plans:

python

from amplihack.goal_agent_generator import ObjectivePlanner

planner = ObjectivePlanner()
plan = planner.generate_plan(goal_definition)

生成多阶段执行计划：

python

from amplihack.goal_agent_generator import ObjectivePlanner

planner = ObjectivePlanner()
plan = planner.generate_plan(goal_definition)

ExecutionPlan contains:

for i, phase in enumerate(plan.phases, 1): print(f"Phase {i}: {phase.name}") print(f" Description: {phase.description}") print(f" Duration: {phase.estimated_duration}") print(f" Capabilities: {', '.join(phase.required_capabilities)}") print(f" Dependencies: {', '.join(phase.dependencies)}") print(f" Parallel Safe: {phase.parallel_safe}") print(f" Success Indicators: {phase.success_indicators}")

print(f"\nTotal Duration: {plan.total_estimated_duration}") print(f"Required Skills: {', '.join(plan.required_skills)}") print(f"Parallel Opportunities: {plan.parallel_opportunities}") print(f"Risk Factors: {plan.risk_factors}")


Phase templates by domain:

- **data-processing**: Collection → Transformation → Analysis → Reporting
- **security-analysis**: Reconnaissance → Vulnerability Detection → Risk Assessment → Reporting
- **automation**: Setup → Workflow Design → Execution → Validation
- **testing**: Test Planning → Implementation → Execution → Results Analysis
- **deployment**: Pre-deployment → Deployment → Verification → Post-deployment
- **monitoring**: Setup Monitors → Data Collection → Analysis → Alerting

for i, phase in enumerate(plan.phases, 1): print(f"Phase {i}: {phase.name}") print(f" Description: {phase.description}") print(f" Duration: {phase.estimated_duration}") print(f" Capabilities: {', '.join(phase.required_capabilities)}") print(f" Dependencies: {', '.join(phase.dependencies)}") print(f" Parallel Safe: {phase.parallel_safe}") print(f" Success Indicators: {phase.success_indicators}")

print(f"\nTotal Duration: {plan.total_estimated_duration}") print(f"Required Skills: {', '.join(plan.required_skills)}") print(f"Parallel Opportunities: {plan.parallel_opportunities}") print(f"Risk Factors: {plan.risk_factors}")


按领域划分的阶段模板：

- **data-processing**：收集→转换→分析→报告
- **security-analysis**：侦察→漏洞检测→风险评估→报告
- **automation**：设置→工作流设计→执行→验证
- **testing**：测试规划→实现→执行→结果分析
- **deployment**：部署前→部署→验证→部署后
- **monitoring**：监控设置→数据收集→分析→告警

SkillSynthesizer Details

SkillSynthesizer详情

Maps capabilities to skills:

python

from amplihack.goal_agent_generator import SkillSynthesizer

synthesizer = SkillSynthesizer()
skills = synthesizer.synthesize(execution_plan)

将能力映射到技能：

python

from amplihack.goal_agent_generator import SkillSynthesizer

synthesizer = SkillSynthesizer()
skills = synthesizer.synthesize(execution_plan)

list[SkillDefinition]

for skill in skills: print(f"Skill: {skill.name}") print(f" Description: {skill.description}") print(f" Capabilities: {', '.join(skill.capabilities)}") print(f" Type: {skill.implementation_type}") if skill.implementation_type == "delegated": print(f" Delegates to: {skill.delegation_target}")


Capability mapping:

- `data-*` → `data-processor` skill
- `security-*`, `vulnerability-*` → `security-analyzer` skill
- `test-*` → `tester` skill
- `deploy-*` → `deployer` skill
- `monitor-*`, `alert-*` → `monitor` skill
- `report-*`, `document-*` → `documenter` skill

for skill in skills: print(f"Skill: {skill.name}") print(f" Description: {skill.description}") print(f" Capabilities: {', '.join(skill.capabilities)}") print(f" Type: {skill.implementation_type}") if skill.implementation_type == "delegated": print(f" Delegates to: {skill.delegation_target}")


能力映射：

- `data-*` → `data-processor`技能
- `security-*`, `vulnerability-*` → `security-analyzer`技能
- `test-*` → `tester`技能
- `deploy-*` → `deployer`技能
- `monitor-*`, `alert-*` → `monitor`技能
- `report-*`, `document-*` → `documenter`技能

AgentAssembler Details

AgentAssembler详情

Combines components into executable bundle:

python

from amplihack.goal_agent_generator import AgentAssembler

assembler = AgentAssembler()
bundle = assembler.assemble(
    goal_definition=goal_definition,
    execution_plan=execution_plan,
    skills=skills,
    bundle_name="custom-agent"  # Optional, auto-generated if omitted
)

将组件组合为可执行包：

python

from amplihack.goal_agent_generator import AgentAssembler

assembler = AgentAssembler()
bundle = assembler.assemble(
    goal_definition=goal_definition,
    execution_plan=execution_plan,
    skills=skills,
    bundle_name="custom-agent"  # Optional, auto-generated if omitted
)

GoalAgentBundle contains:

print(bundle.id) # UUID print(bundle.name) # "custom-agent" or auto-generated print(bundle.version) # "1.0.0" print(bundle.status) # "ready" print(bundle.auto_mode_config) # Configuration for auto-mode execution print(bundle.metadata) # Domain, complexity, skills, etc.

Auto-mode configuration

config = bundle.auto_mode_config print(config["max_turns"]) # Based on complexity print(config["initial_prompt"]) # Generated execution prompt print(config["success_criteria"]) # From goal definition print(config["constraints"]) # From goal definition


Auto-mode configuration:

- `max_turns`: 5 (simple), 10 (moderate), 15 (complex), +20% per extra phase
- `initial_prompt`: Full markdown prompt with goal, plan, success criteria
- `working_dir`: Current directory
- `sdk`: "claude" (default)
- `ui_mode`: False (headless by default)

config = bundle.auto_mode_config print(config["max_turns"]) # Based on complexity print(config["initial_prompt"]) # Generated execution prompt print(config["success_criteria"]) # From goal definition print(config["constraints"]) # From goal definition


自动模式配置：

- `max_turns`：5（简单）、10（中等）、15（复杂），每增加一个阶段额外增加20%
- `initial_prompt`：包含目标、计划、成功标准的完整Markdown提示
- `working_dir`：当前目录
- `sdk`："claude"（默认）
- `ui_mode`：False（默认无头模式）

GoalAgentPackager Details

GoalAgentPackager详情

Packages bundle for deployment:

python

from amplihack.goal_agent_generator import GoalAgentPackager
from pathlib import Path

packager = GoalAgentPackager()
packager.package(
    bundle=agent_bundle,
    output_dir=Path(".claude/agents/goal-driven/my-agent")
)

将包打包用于部署：

python

from amplihack.goal_agent_generator import GoalAgentPackager
from pathlib import Path

packager = GoalAgentPackager()
packager.package(
    bundle=agent_bundle,
    output_dir=Path(".claude/agents/goal-driven/my-agent")
)

Creates:

.claude/agents/goal-driven/my-agent/

├── agent.md # Agent definition

├── prompt.md # Initial prompt

├── metadata.json # Bundle metadata

├── plan.yaml # Execution plan

└── skills.yaml # Required skills

undefined

undefined

5. Recent Amplihack Examples

5. 近期Amplihack示例

Real goal-seeking agents from the amplihack project:

来自amplihack项目的真实目标导向Agent：

Example 1: AKS SRE Automation (Issue #1293)

示例1：AKS SRE自动化（Issue #1293）

Problem: Manual AKS cluster operations are time-consuming and error-prone

Goal-Seeking Solution:

python

undefined

问题：手动AKS集群操作耗时且容易出错

目标导向解决方案：

python

undefined

Goal: Automate AKS production readiness verification

goal = """ Verify AKS cluster production readiness:

Security: RBAC, network policies, Key Vault integration
Networking: Ingress, DNS, load balancers
Monitoring: Container Insights, alerts, dashboards
Compliance: Azure Policy, resource quotas Generate actionable report with recommendations. """

goal = """ Verify AKS cluster production readiness:

Security: RBAC, network policies, Key Vault integration
Networking: Ingress, DNS, load balancers
Monitoring: Container Insights, alerts, dashboards
Compliance: Azure Policy, resource quotas Generate actionable report with recommendations. """

Agent decomposes into phases:

1. Security Audit (parallel): RBAC check, network policies, Key Vault

2. Networking Validation (parallel): Ingress test, DNS resolution, LB health

3. Monitoring Verification (parallel): Metrics, logs, alerts configured

4. Compliance Check (depends on 1-3): Azure Policy, quotas, best practices

5. Report Generation (depends on 4): Markdown report with findings

Agent adapts based on findings:

- If security issues found: Suggest fixes, offer to apply

- If monitoring missing: Generate alert templates

- If compliance violations: List remediation steps


**Key Characteristics**:

- **Autonomous**: Checks multiple systems without step-by-step instructions
- **Adaptive**: Investigation depth varies by findings
- **Multi-Phase**: Parallel security/networking/monitoring, sequential reporting
- **Domain Expert**: Azure + Kubernetes knowledge embedded
- **Self-Assessing**: Validates each check, aggregates results

**Implementation**:

```python


**关键特性**：

- **自主性**：无需分步指令即可检查多个系统
- **适应性**：调查深度根据发现的问题而变化
- **多阶段**：安全/网络/监控并行执行，报告顺序执行
- **领域专家**：嵌入Azure + Kubernetes知识
- **自我评估**：验证每个检查，汇总结果

**实现**：

```python

Located in: .claude/agents/amplihack/specialized/azure-kubernetes-expert.md

Uses knowledge base: .claude/data/azure_aks_expert/

Integrates with goal_agent_generator:

from amplihack.goal_agent_generator import ( PromptAnalyzer, ObjectivePlanner, AgentAssembler )

analyzer = PromptAnalyzer() goal_def = analyzer.analyze_text(goal)

planner = ObjectivePlanner() plan = planner.generate_plan(goal_def) # Generates 5-phase plan

from amplihack.goal_agent_generator import ( PromptAnalyzer, ObjectivePlanner, AgentAssembler )

analyzer = PromptAnalyzer() goal_def = analyzer.analyze_text(goal)

planner = ObjectivePlanner() plan = planner.generate_plan(goal_def) # Generates 5-phase plan

Domain-specific customization:

plan.phases[0].required_capabilities = [ "rbac-audit", "network-policy-check", "key-vault-integration" ]


**Lessons Learned**:

- Domain expertise critical for complex infrastructure
- Parallel execution significantly reduces total time
- Actionable recommendations increase agent value
- Comprehensive knowledge base (Q&A format) enables autonomous decisions

plan.phases[0].required_capabilities = [ "rbac-audit", "network-policy-check", "key-vault-integration" ]


**经验教训**：

- 领域专业知识对于复杂基础设施至关重要
- 并行执行显著减少总时间
- 可操作的建议提升Agent价值
- 全面的知识库（问答格式）支持自主决策

Example 2: CI Diagnostic Workflow

示例2：CI诊断工作流

Problem: CI failures require manual diagnosis and fix iteration

Goal-Seeking Solution:

python

undefined

问题：CI故障需要手动诊断和迭代修复

目标导向解决方案：

python

undefined

Goal: Diagnose CI failure and iterate fixes until success

goal = """ CI pipeline failing after push. Diagnose failures, apply fixes, push updates, monitor CI. Iterate until all checks pass. Stop at mergeable state without auto-merging. """

Agent decomposes into phases:

1. CI Status Monitoring: Check current CI state

2. Failure Diagnosis: Analyze logs, compare environments

3. Fix Application: Apply fixes based on failure patterns

4. Push and Wait: Commit fixes, push, wait for CI re-run

5. Success Verification: Confirm all checks pass

Iterative loop:

Phases 2-4 repeat until success or max iterations (5)


**Key Characteristics**:

- **Iterative**: Repeats fix cycle until success
- **Autonomous Recovery**: Tries multiple fix strategies
- **State Management**: Tracks attempted fixes, avoids repeating failures
- **Pattern Matching**: Recognizes common CI failure types
- **Escalation**: Reports to user after max iterations

**Implementation**:

```python


**关键特性**：

- **迭代性**：重复修复周期直至成功
- **自主恢复**：尝试多种修复策略
- **状态管理**：跟踪已尝试的修复，避免重复失败
- **模式匹配**：识别常见CI故障类型
- **升级处理**：达到最大迭代次数后向用户报告

**实现**：

```python

Located in: .claude/agents/amplihack/specialized/ci-diagnostic-workflow.md

Fix iteration loop:

MAX_ITERATIONS = 5 iteration = 0

while iteration < MAX_ITERATIONS: status = check_ci_status()

if status["conclusion"] == "success":
    break

# Diagnose failures
failures = analyze_ci_logs(status)

# Apply pattern-matched fixes
for failure in failures:
    if "test" in failure["type"]:
        fix_test_failure(failure)
    elif "lint" in failure["type"]:
        fix_lint_failure(failure)
    elif "type" in failure["type"]:
        fix_type_failure(failure)

# Commit and push
git_commit_and_push(f"fix: CI iteration {iteration + 1}")

# Wait for CI re-run
wait_for_ci_completion()

iteration += 1

if iteration >= MAX_ITERATIONS: escalate_to_user("CI still failing after 5 iterations")


**Lessons Learned**:

- Iteration limits prevent infinite loops
- Pattern matching (test/lint/type) enables targeted fixes
- Smart waiting (exponential backoff) reduces wait time
- Never auto-merge: human approval always required

MAX_ITERATIONS = 5 iteration = 0

while iteration < MAX_ITERATIONS: status = check_ci_status()

if status["conclusion"] == "success":
    break

# Diagnose failures
failures = analyze_ci_logs(status)

# Apply pattern-matched fixes
for failure in failures:
    if "test" in failure["type"]:
        fix_test_failure(failure)
    elif "lint" in failure["type"]:
        fix_lint_failure(failure)
    elif "type" in failure["type"]:
        fix_type_failure(failure)

# Commit and push
git_commit_and_push(f"fix: CI iteration {iteration + 1}")

# Wait for CI re-run
wait_for_ci_completion()

iteration += 1

if iteration >= MAX_ITERATIONS: escalate_to_user("CI still failing after 5 iterations")


**经验教训**：

- 迭代限制防止无限循环
- 模式匹配（测试/ lint /类型）实现针对性修复
- 智能等待（指数退避）减少等待时间
- 绝不自动合并：始终需要人工批准

Example 3: Pre-Commit Diagnostic Workflow

示例3：预提交诊断工作流

Problem: Pre-commit hooks fail with unclear errors

Goal-Seeking Solution:

python

undefined

问题：预提交钩子失败且错误不明确

目标导向解决方案：

python

undefined

Goal: Fix pre-commit hook failures before commit

goal = """ Pre-commit hooks failing. Diagnose issues (formatting, linting, type checking). Apply fixes locally, re-run hooks. Ensure all hooks pass before allowing commit. """

Agent decomposes into phases:

1. Hook Failure Analysis: Identify which hooks failed

2. Environment Check: Compare local vs pre-commit versions

3. Targeted Fixes: Apply fixes per hook type

4. Hook Re-run: Validate fixes, iterate if needed

5. Commit Readiness: Confirm all hooks pass


**Key Characteristics**:

- **Pre-Push Focus**: Fixes issues before pushing to CI
- **Tool Version Management**: Ensures local matches pre-commit config
- **Hook-Specific Fixes**: Tailored approach per hook type
- **Fast Iteration**: No wait for CI, immediate feedback

**Implementation**:

```python


**关键特性**：

- **预推送聚焦**：在推送到CI之前修复问题
- **工具版本管理**：确保本地版本与预提交配置匹配
- **钩子特定修复**：针对每种钩子类型采用定制方法
- **快速迭代**：无需等待CI，即时反馈

**实现**：

```python

Located in: .claude/agents/amplihack/specialized/pre-commit-diagnostic.md

Hook failure patterns:

HOOK_FIXES = { "ruff": lambda: subprocess.run(["ruff", "check", "--fix", "."]), "black": lambda: subprocess.run(["black", "."]), "mypy": lambda: add_type_ignores(), "trailing-whitespace": lambda: subprocess.run(["pre-commit", "run", "trailing-whitespace", "--all-files"]), }

Execution:

failed_hooks = detect_failed_hooks()

for hook in failed_hooks: if hook in HOOK_FIXES: HOOK_FIXEShook else: generic_fix(hook)

failed_hooks = detect_failed_hooks()

for hook in failed_hooks: if hook in HOOK_FIXES: HOOK_FIXEShook else: generic_fix(hook)

Re-run to verify

rerun_result = subprocess.run(["pre-commit", "run", "--all-files"]) if rerun_result.returncode == 0: print("All hooks passing, ready to commit!")


**Lessons Learned**:

- Pre-commit fixes are faster than CI iteration
- Tool version mismatches are common culprit
- Automated fixes for 80% of cases
- Remaining 20% escalate with clear diagnostics

rerun_result = subprocess.run(["pre-commit", "run", "--all-files"]) if rerun_result.returncode == 0: print("All hooks passing, ready to commit!")


**经验教训**：

- 预提交修复比CI迭代更快
- 工具版本不匹配是常见问题
- 自动修复80%的案例
- 剩余20%的问题需提供明确诊断后升级至人工处理

Example 4: Fix-Agent Pattern Matching

示例4：修复Agent模式匹配

Problem: Different issues require different fix approaches

Goal-Seeking Solution:

python

undefined

问题：不同问题需要不同的修复方法

目标导向解决方案：

python

undefined

Goal: Select optimal fix strategy based on problem context

goal = """ Analyze issue and select fix mode:

QUICK: Obvious fixes (< 5 min)
DIAGNOSTIC: Unclear root cause (investigation)
COMPREHENSIVE: Complex issues (full workflow) """

goal = """ Analyze issue and select fix mode:

QUICK: Obvious fixes (< 5 min)
DIAGNOSTIC: Unclear root cause (investigation)
COMPREHENSIVE: Complex issues (full workflow) """

Agent decomposes into phases:

1. Issue Analysis: Classify problem type and complexity

2. Mode Selection: Choose QUICK/DIAGNOSTIC/COMPREHENSIVE

3. Fix Execution: Apply mode-appropriate strategy

4. Validation: Verify fix resolves issue


**Key Characteristics**:

- **Context-Aware**: Selects strategy based on problem analysis
- **Multi-Mode**: Three fix modes for different complexity levels
- **Pattern Recognition**: Learns from past fixes
- **Adaptive**: Escalates complexity if initial mode fails

**Implementation**:

```python


**关键特性**：

- **上下文感知**：根据问题分析选择策略
- **多模式**：针对不同复杂度级别的三种修复模式
- **模式识别**：从过去的修复中学习
- **适应性**：如果初始模式失败，升级复杂度

**实现**：

```python

Located in: .claude/agents/amplihack/specialized/fix-agent.md

Mode selection logic:

def select_fix_mode(issue: Issue) -> FixMode: if issue.is_obvious() and issue.scope == "single-file": return FixMode.QUICK elif issue.root_cause_unclear(): return FixMode.DIAGNOSTIC elif issue.is_complex() or issue.requires_architecture_change(): return FixMode.COMPREHENSIVE else: return FixMode.DIAGNOSTIC # Default to investigation

Pattern frequency (from real usage):

FIX_PATTERNS = { "import": 0.15, # Import errors (15%) "config": 0.12, # Configuration issues (12%) "test": 0.18, # Test failures (18%) "ci": 0.20, # CI/CD problems (20%) "quality": 0.25, # Code quality (linting, types) (25%) "logic": 0.10, # Logic errors (10%) }

Template-based fixes for common patterns:

if issue.pattern == "import": apply_template("import-fix-template", issue) elif issue.pattern == "config": apply_template("config-fix-template", issue)

... etc


**Lessons Learned**:

- Pattern matching enables template-based fixes (80% coverage)
- Mode selection reduces over-engineering (right-sized approach)
- Diagnostic mode critical for unclear issues (root cause analysis)
- Usage data informs template priorities


**经验教训**：

- 模式匹配实现基于模板的修复（覆盖80%）
- 模式选择减少过度设计（合适的解决方案）
- 诊断模式对于不明确问题至关重要（根本原因分析）
- 使用数据指导模板优先级

6. Design Checklist

6. 设计检查表

Use this checklist when designing goal-seeking agents:

设计目标导向Agent时使用此检查表：

Goal Definition

目标定义

Objective is clear and well-defined
Success criteria are measurable and verifiable
Constraints are explicit (time, resources, safety)
Domain is identified (impacts phase templates)
Complexity is estimated (simple/moderate/complex)

目标明确且定义清晰
成功标准可衡量且可验证
约束条件明确（时间、资源、安全）
已识别领域（影响阶段模板）
已估算复杂度（简单/中等/复杂）

Phase Design

阶段设计

Decomposed into 3-5 phases (not too granular, not too coarse)
Phase dependencies are explicit
Parallel execution opportunities identified
Each phase has clear success indicators
Phase durations are estimated

分解为3-5个阶段（不过于精细，也不过于粗糙）
阶段依赖关系明确
已识别并行执行机会
每个阶段都有明确的成功指标
已估算阶段持续时间

Skill Mapping

技能映射

Required capabilities identified per phase
Skills mapped to existing agents or tools
Delegation targets specified
No missing capabilities

已识别每个阶段所需的能力
技能已映射到现有Agent或工具
已指定委托目标
无缺失能力

Error Handling

错误处理

Retry strategies defined (max attempts, backoff)
Alternative strategies identified
Escalation criteria clear (when to ask for help)
Graceful degradation options (fallback approaches)

已定义重试策略（最大尝试次数、退避）
已识别替代策略
升级标准清晰（何时寻求帮助）
已提供优雅降级选项（回退方法）

State Management

状态管理

State tracked across phases
Phase results stored for downstream use
Failure history maintained
Context shared appropriately

跨阶段跟踪状态
阶段结果存储以供下游使用
维护故障历史
适当共享上下文

Testing

测试

Success scenarios tested
Failure recovery tested
Edge cases identified
Performance validated (duration, resource usage)

已测试成功场景
已测试故障恢复
已识别边缘情况
已验证性能（持续时间、资源使用）

Documentation

文档

Goal clearly documented
Phase descriptions complete
Usage examples provided
Integration points specified

目标已清晰记录
阶段描述完整
提供使用示例
指定集成点

Philosophy Compliance

理念合规性

Ruthless simplicity (no unnecessary complexity)
Single responsibility per phase
No over-engineering (right-sized solution)
Regeneratable (clear specifications)

极致简洁（无不必要的复杂度）
每个阶段单一职责
无过度设计（合适的解决方案）
可重新生成（清晰的规范）

7. Agent SDK Integration (Future)

7. Agent SDK集成（未来）

When the Agent SDK Skill is integrated, goal-seeking agents can leverage:

集成Agent SDK技能后，目标导向Agent可利用：

Enhanced Autonomy

增强自主性

python

undefined

python

undefined

Agent SDK provides enhanced context management

from claude_agent_sdk import AgentContext, Tool

class GoalSeekingAgent: def init(self, context: AgentContext): self.context = context self.state = {}

async def execute_phase(self, phase: PlanPhase):
    # SDK provides tools, memory, delegation
    tools = self.context.get_tools(phase.required_capabilities)
    memory = self.context.get_memory()

    # Execute with SDK support
    result = await phase.execute(tools, memory)

    # Store in context for downstream phases
    self.context.store_result(phase.name, result)

undefined

from claude_agent_sdk import AgentContext, Tool

class GoalSeekingAgent: def init(self, context: AgentContext): self.context = context self.state = {}

async def execute_phase(self, phase: PlanPhase):
    # SDK provides tools, memory, delegation
    tools = self.context.get_tools(phase.required_capabilities)
    memory = self.context.get_memory()

    # Execute with SDK support
    result = await phase.execute(tools, memory)

    # Store in context for downstream phases
    self.context.store_result(phase.name, result)

undefined

Tool Discovery

工具发现

python

undefined

python

undefined

SDK enables dynamic tool discovery

available_tools = context.discover_tools(capability="data-processing")

Select optimal tool for task

tool = context.select_tool( capability="data-transformation", criteria={"performance": "high", "accuracy": "required"} )

undefined

tool = context.select_tool( capability="data-transformation", criteria={"performance": "high", "accuracy": "required"} )

undefined

Memory Management

内存管理

python

undefined

python

undefined

SDK provides persistent memory across sessions

context.memory.store("deployment-history", deployment_record) previous = context.memory.retrieve("deployment-history")

Enables learning from past executions

if previous and previous.failed: # Avoid previous failure strategy strategy = select_alternative_strategy(previous.failure_reason)

undefined

if previous and previous.failed: # Avoid previous failure strategy strategy = select_alternative_strategy(previous.failure_reason)

undefined

Agent Delegation

Agent委托

python

undefined

python

undefined

SDK simplifies agent-to-agent delegation

result = await context.delegate( agent="security-analyzer", task="audit-rbac-policies", input={"cluster": cluster_name} )

Parallel delegation

results = await context.delegate_parallel([ ("security-analyzer", "audit-rbac-policies"), ("network-analyzer", "validate-ingress"), ("monitoring-validator", "check-metrics") ])

undefined

results = await context.delegate_parallel([ ("security-analyzer", "audit-rbac-policies"), ("network-analyzer", "validate-ingress"), ("monitoring-validator", "check-metrics") ])

undefined

Observability

可观测性

python

undefined

python

undefined

SDK provides built-in tracing and metrics

with context.trace("data-transformation"): result = transform_data(input_data)

context.metrics.record("transformation-duration", duration) context.metrics.record("transformation-accuracy", accuracy)

undefined

with context.trace("data-transformation"): result = transform_data(input_data)

context.metrics.record("transformation-duration", duration) context.metrics.record("transformation-accuracy", accuracy)

undefined

Integration Example

集成示例

python

from claude_agent_sdk import AgentContext, create_agent
from amplihack.goal_agent_generator import GoalAgentBundle

python

from claude_agent_sdk import AgentContext, create_agent
from amplihack.goal_agent_generator import GoalAgentBundle

Create SDK-enabled goal-seeking agent

def create_goal_agent(bundle: GoalAgentBundle) -> Agent: context = AgentContext( name=bundle.name, version=bundle.version, capabilities=bundle.metadata["required_capabilities"] )

# Register phases as agent tasks
for phase in bundle.execution_plan.phases:
    context.register_task(
        name=phase.name,
        capabilities=phase.required_capabilities,
        executor=create_phase_executor(phase)
    )

# Create agent with SDK
agent = create_agent(context)

# Execute goal
return agent

def create_goal_agent(bundle: GoalAgentBundle) -> Agent: context = AgentContext( name=bundle.name, version=bundle.version, capabilities=bundle.metadata["required_capabilities"] )

# Register phases as agent tasks
for phase in bundle.execution_plan.phases:
    context.register_task(
        name=phase.name,
        capabilities=phase.required_capabilities,
        executor=create_phase_executor(phase)
    )

# Create agent with SDK
agent = create_agent(context)

# Execute goal
return agent

Usage:

agent = create_goal_agent(agent_bundle) result = await agent.execute(bundle.auto_mode_config["initial_prompt"])

undefined

agent = create_goal_agent(agent_bundle) result = await agent.execute(bundle.auto_mode_config["initial_prompt"])

undefined

8. Trade-Off Analysis

8. 权衡分析

Goal-Seeking vs Traditional Agents

目标导向Agent vs 传统Agent

Dimension	Goal-Seeking Agent	Traditional Agent
Flexibility	High - adapts to context	Low - fixed workflow
Development Time	Moderate - define goals & phases	Low - script steps
Execution Time	Higher - decision overhead	Lower - direct execution
Maintenance	Lower - self-adapting	Higher - manual updates
Debuggability	Harder - dynamic behavior	Easier - predictable flow
Reusability	High - same agent, different contexts	Low - context-specific
Failure Handling	Autonomous recovery	Manual intervention
Complexity	Higher - multi-phase coordination	Lower - linear execution

维度	目标导向Agent	传统Agent
灵活性	高 - 适应上下文	低 - 固定工作流
开发时间	中等 - 定义目标和阶段	低 - 编写脚本步骤
执行时间	较长 - 决策开销	较短 - 直接执行
维护	较低 - 自适应	较高 - 手动更新
可调试性	较难 - 动态行为	较易 - 可预测流程
可重用性	高 - 同一Agent，不同上下文	低 - 特定上下文
故障处理	自主恢复	人工干预
复杂度	较高 - 多阶段协调	较低 - 线性执行

When to Choose Each

何时选择每种Agent

Choose Goal-Seeking when:

Problem space is large with many valid approaches
Context varies significantly across executions
Autonomous recovery is valuable
Reusability across contexts is important
Development time investment is justified

Choose Traditional when:

Single deterministic path exists
Performance is critical (low latency required)
Simplicity is paramount
One-off or rare execution
Debugging and auditability are critical

选择目标导向Agent当：

问题空间大，存在多种有效方法
上下文在不同执行中差异显著
自主恢复具有价值
跨上下文的可重用性很重要
开发时间投入合理

选择传统Agent当：

存在单一确定性路径
性能至关重要（需要低延迟）
简单性是首要考虑因素
一次性或罕见执行
调试和可审计性至关重要

Cost-Benefit Analysis

成本效益分析

Goal-Seeking Costs:

Higher development time (define goals, phases, capabilities)
Increased execution time (decision overhead)
More complex testing (dynamic behavior)
Harder debugging (non-deterministic paths)

Goal-Seeking Benefits:

Autonomous operation (less human intervention)
Adaptive to context (works in varied conditions)
Reusable across problems (same agent, different goals)
Self-recovering (handles failures gracefully)

Break-Even Point: Goal-seeking justified when problem is:

Repeated 2+ times per week, OR
Takes 30+ minutes manual execution, OR
Requires expert knowledge hard to document, OR
High value from autonomous recovery

目标导向Agent成本：

更高的开发时间（定义目标、阶段、能力）
更长的执行时间（决策开销）
更复杂的测试（动态行为）
更难调试（非确定性路径）

目标导向Agent收益：

自主运行（减少人工干预）
适应上下文（在各种条件下工作）
跨问题可重用（同一Agent，不同目标）
自我恢复（优雅处理故障）

收支平衡点：当问题满足以下条件时，目标导向Agent是合理的：

每周重复2次以上，或
人工执行需要30分钟以上，或
需要难以记录的专业知识，或
自主恢复带来高价值

9. When to Escalate

9. 何时升级至人工处理

Goal-seeking agents should escalate to humans when:

当出现以下情况时，目标导向Agent应升级至人工处理：

Hard Limits Reached

达到硬限制

Max Iterations Exceeded:

python

if iteration_count >= MAX_ITERATIONS:
    escalate(
        reason="Reached maximum iterations without success",
        context={
            "iterations": iteration_count,
            "attempted_strategies": attempted_strategies,
            "last_error": last_error
        }
    )

Timeout Exceeded:

python

if elapsed_time > MAX_DURATION:
    escalate(
        reason="Execution time exceeded limit",
        context={
            "elapsed": elapsed_time,
            "max_allowed": MAX_DURATION,
            "completed_phases": completed_phases
        }
    )

超过最大迭代次数：

python

if iteration_count >= MAX_ITERATIONS:
    escalate(
        reason="Reached maximum iterations without success",
        context={
            "iterations": iteration_count,
            "attempted_strategies": attempted_strategies,
            "last_error": last_error
        }
    )

超过超时时间：

python

if elapsed_time > MAX_DURATION:
    escalate(
        reason="Execution time exceeded limit",
        context={
            "elapsed": elapsed_time,
            "max_allowed": MAX_DURATION,
            "completed_phases": completed_phases
        }
    )

Safety Boundaries

安全边界

Destructive Operations:

python

if operation.is_destructive() and not operation.has_approval():
    escalate(
        reason="Destructive operation requires human approval",
        operation=operation.description,
        impact=operation.estimate_impact()
    )

Production Changes:

python

if target_environment == "production":
    escalate(
        reason="Production deployments require human verification",
        changes=proposed_changes,
        rollback_plan=rollback_strategy
    )

破坏性操作：

python

if operation.is_destructive() and not operation.has_approval():
    escalate(
        reason="Destructive operation requires human approval",
        operation=operation.description,
        impact=operation.estimate_impact()
    )

生产环境变更：

python

if target_environment == "production":
    escalate(
        reason="Production deployments require human verification",
        changes=proposed_changes,
        rollback_plan=rollback_strategy
    )

Uncertainty Detection

不确定性检测

Low Confidence:

python

if decision_confidence < CONFIDENCE_THRESHOLD:
    escalate(
        reason="Confidence below threshold for autonomous decision",
        decision=decision_description,
        confidence=decision_confidence,
        alternatives=alternative_options
    )

Conflicting Strategies:

python

if len(viable_strategies) > 1 and not clear_winner:
    escalate(
        reason="Multiple viable strategies, need human judgment",
        strategies=viable_strategies,
        trade_offs=strategy_trade_offs
    )

低置信度：

python

if decision_confidence < CONFIDENCE_THRESHOLD:
    escalate(
        reason="Confidence below threshold for autonomous decision",
        decision=decision_description,
        confidence=decision_confidence,
        alternatives=alternative_options
    )

策略冲突：

python

if len(viable_strategies) > 1 and not clear_winner:
    escalate(
        reason="Multiple viable strategies, need human judgment",
        strategies=viable_strategies,
        trade_offs=strategy_trade_offs
    )

Unexpected Conditions

意外情况

Unrecognized Errors:

python

if error_type not in KNOWN_ERROR_PATTERNS:
    escalate(
        reason="Encountered unknown error pattern",
        error=error_details,
        context=execution_context,
        recommendation="Manual investigation required"
    )

Environment Mismatch:

python

if detected_environment != expected_environment:
    escalate(
        reason="Environment mismatch detected",
        expected=expected_environment,
        detected=detected_environment,
        risk="Potential for incorrect behavior"
    )

未识别的错误：

python

if error_type not in KNOWN_ERROR_PATTERNS:
    escalate(
        reason="Encountered unknown error pattern",
        error=error_details,
        context=execution_context,
        recommendation="Manual investigation required"
    )

环境不匹配：

python

if detected_environment != expected_environment:
    escalate(
        reason="Environment mismatch detected",
        expected=expected_environment,
        detected=detected_environment,
        risk="Potential for incorrect behavior"
    )

Escalation Best Practices

升级最佳实践

Provide Context:

What was attempted
What failed and why
What alternatives were considered
Current system state

Suggest Actions:

Recommend next steps
Provide diagnostic commands
Offer manual intervention points
Suggest rollback if needed

Enable Recovery:

Save execution state
Document failures
Provide resume capability
Offer manual override

Example Escalation:

python

escalate(
    reason="CI failure diagnosis unsuccessful after 5 iterations",
    context={
        "iterations": 5,
        "attempted_fixes": [
            "Import path corrections (iteration 1)",
            "Type annotation fixes (iteration 2)",
            "Test environment setup (iteration 3)",
            "Dependency version pins (iteration 4)",
            "Mock configuration (iteration 5)"
        ],
        "persistent_failures": [
            "test_integration.py::test_api_connection - Timeout",
            "test_models.py::test_validation - Assertion error"
        ],
        "system_state": "2 of 25 tests still failing",
        "ci_logs": "https://github.com/.../actions/runs/123456"
    },
    recommendations=[
        "Review test_api_connection timeout - may need increased timeout or mock",
        "Examine test_validation assertion - data structure may have changed",
        "Consider running tests locally with same environment as CI",
        "Check if recent changes affected integration test setup"
    ],
    next_steps={
        "manual_investigation": "Run failing tests locally with verbose output",
        "rollback_option": "git revert HEAD~5 if fixes made things worse",
        "resume_point": "Fix failures and run /amplihack:ci-diagnostic to resume"
    }
)

提供上下文：

已尝试的操作
失败的原因
已考虑的替代方案
当前系统状态

建议操作：

推荐下一步
提供诊断命令
提供人工干预点
必要时建议回滚

支持恢复：

保存执行状态
记录故障
提供恢复能力
提供手动覆盖

示例升级：

python

escalate(
    reason="CI failure diagnosis unsuccessful after 5 iterations",
    context={
        "iterations": 5,
        "attempted_fixes": [
            "Import path corrections (iteration 1)",
            "Type annotation fixes (iteration 2)",
            "Test environment setup (iteration 3)",
            "Dependency version pins (iteration 4)",
            "Mock configuration (iteration 5)"
        ],
        "persistent_failures": [
            "test_integration.py::test_api_connection - Timeout",
            "test_models.py::test_validation - Assertion error"
        ],
        "system_state": "2 of 25 tests still failing",
        "ci_logs": "https://github.com/.../actions/runs/123456"
    },
    recommendations=[
        "Review test_api_connection timeout - may need increased timeout or mock",
        "Examine test_validation assertion - data structure may have changed",
        "Consider running tests locally with same environment as CI",
        "Check if recent changes affected integration test setup"
    ],
    next_steps={
        "manual_investigation": "Run failing tests locally with verbose output",
        "rollback_option": "git revert HEAD~5 if fixes made things worse",
        "resume_point": "Fix failures and run /amplihack:ci-diagnostic to resume"
    }
)

10. Example Workflow

10. 示例工作流

Complete example: Building a goal-seeking agent for data pipeline automation

完整示例：为数据管道自动化构建目标导向Agent

Step 1: Define Goal

步骤1：定义目标

markdown

undefined

markdown

undefined

Goal: Automate Multi-Source Data Pipeline

Objective

Collect data from multiple sources (S3, database, API), transform to common schema, validate quality, publish to data warehouse.

Success Criteria

All sources successfully ingested
Data transformed to target schema
Quality checks pass (completeness, accuracy)
Data published to warehouse
Pipeline completes within 30 minutes

All sources successfully ingested
Data transformed to target schema
Quality checks pass (completeness, accuracy)
Data published to warehouse
Pipeline completes within 30 minutes

Constraints

Must handle source unavailability gracefully
No data loss (failed records logged)
Idempotent (safe to re-run)
Resource limits: 8GB RAM, 4 CPU cores

Must handle source unavailability gracefully
No data loss (failed records logged)
Idempotent (safe to re-run)
Resource limits: 8GB RAM, 4 CPU cores

Context

Daily execution (automated schedule)
Priority: High (blocking downstream analytics)
Scale: Medium (100K-1M records per source)

undefined

Daily execution (automated schedule)
Priority: High (blocking downstream analytics)
Scale: Medium (100K-1M records per source)

undefined

Step 2: Analyze with PromptAnalyzer

步骤2：使用PromptAnalyzer分析

python

from amplihack.goal_agent_generator import PromptAnalyzer

analyzer = PromptAnalyzer()
goal_definition = analyzer.analyze_text(goal_text)

python

from amplihack.goal_agent_generator import PromptAnalyzer

analyzer = PromptAnalyzer()
goal_definition = analyzer.analyze_text(goal_text)

Result:

goal_definition.goal = "Automate Multi-Source Data Pipeline"

goal_definition.domain = "data-processing"

goal_definition.complexity = "moderate"

goal_definition.constraints = [

"Must handle source unavailability gracefully",

"No data loss (failed records logged)",

"Idempotent (safe to re-run)",

"Resource limits: 8GB RAM, 4 CPU cores"

]

goal_definition.success_criteria = [

"All sources successfully ingested",

"Data transformed to target schema",

"Quality checks pass (completeness, accuracy)",

"Data published to warehouse",

"Pipeline completes within 30 minutes"

]

undefined

undefined

Step 3: Generate Plan with ObjectivePlanner

步骤3：使用ObjectivePlanner生成计划

python

from amplihack.goal_agent_generator import ObjectivePlanner

planner = ObjectivePlanner()
execution_plan = planner.generate_plan(goal_definition)

python

from amplihack.goal_agent_generator import ObjectivePlanner

planner = ObjectivePlanner()
execution_plan = planner.generate_plan(goal_definition)

Result: 4-phase plan

Phase 1: Data Collection (parallel)

- Collect from S3 (parallel-safe)

- Collect from database (parallel-safe)

- Collect from API (parallel-safe)

Duration: 15 minutes

Success: All sources attempted, failures logged

Phase 2: Data Transformation (depends on Phase 1)

- Parse raw data

- Transform to common schema

- Handle missing fields

Duration: 15 minutes

Success: All records transformed or logged as failed

Phase 3: Quality Validation (depends on Phase 2)

- Completeness check

- Accuracy validation

- Consistency verification

Duration: 5 minutes

Success: Quality thresholds met

Phase 4: Data Publishing (depends on Phase 3)

- Load to warehouse

- Update metadata

- Generate report

Duration: 10 minutes

Success: Data in warehouse, report generated

undefined

undefined

Step 4: Synthesize Skills

步骤4：合成技能

python

from amplihack.goal_agent_generator import SkillSynthesizer

synthesizer = SkillSynthesizer()
skills = synthesizer.synthesize(execution_plan)

python

from amplihack.goal_agent_generator import SkillSynthesizer

synthesizer = SkillSynthesizer()
skills = synthesizer.synthesize(execution_plan)

Result: 3 skills

Skill 1: data-collector

Capabilities: ["s3-read", "database-query", "api-fetch"]

Implementation: "native" (built-in)

Skill 2: data-transformer

Capabilities: ["parsing", "schema-mapping", "validation"]

Implementation: "native" (built-in)

Skill 3: data-publisher

Capabilities: ["warehouse-load", "metadata-update", "reporting"]

Implementation: "delegated" (delegates to warehouse tool)

undefined

undefined

Step 5: Assemble Agent

步骤5：组装Agent

python

from amplihack.goal_agent_generator import AgentAssembler

assembler = AgentAssembler()
agent_bundle = assembler.assemble(
    goal_definition=goal_definition,
    execution_plan=execution_plan,
    skills=skills,
    bundle_name="multi-source-data-pipeline"
)

python

from amplihack.goal_agent_generator import AgentAssembler

assembler = AgentAssembler()
agent_bundle = assembler.assemble(
    goal_definition=goal_definition,
    execution_plan=execution_plan,
    skills=skills,
    bundle_name="multi-source-data-pipeline"
)

Result: GoalAgentBundle

- Name: multi-source-data-pipeline

- Max turns: 12 (moderate complexity, 4 phases)

- Initial prompt: Full execution plan with phases

- Status: "ready"

undefined

undefined

Step 6: Package Agent

步骤6：打包Agent

python

from amplihack.goal_agent_generator import GoalAgentPackager
from pathlib import Path

packager = GoalAgentPackager()
packager.package(
    bundle=agent_bundle,
    output_dir=Path(".claude/agents/goal-driven/multi-source-data-pipeline")
)

python

from amplihack.goal_agent_generator import GoalAgentPackager
from pathlib import Path

packager = GoalAgentPackager()
packager.package(
    bundle=agent_bundle,
    output_dir=Path(".claude/agents/goal-driven/multi-source-data-pipeline")
)

Creates agent package:

.claude/agents/goal-driven/multi-source-data-pipeline/

├── agent.md # Agent definition

├── prompt.md # Execution prompt

├── metadata.json # Bundle metadata

├── plan.yaml # Execution plan (4 phases)

└── skills.yaml # 3 required skills

undefined

undefined

Step 7: Execute Agent (Auto-Mode)

步骤7：执行Agent（自动模式）

bash

undefined

bash

undefined

Execute via CLI

amplihack goal-agent-generator execute
--agent-path .claude/agents/goal-driven/multi-source-data-pipeline
--auto-mode
--max-turns 12

Or programmatically:


```python
from claude_code import execute_auto_mode

result = execute_auto_mode(
    initial_prompt=agent_bundle.auto_mode_config["initial_prompt"],
    max_turns=agent_bundle.auto_mode_config["max_turns"],
    working_dir=agent_bundle.auto_mode_config["working_dir"]
)


```python
from claude_code import execute_auto_mode

result = execute_auto_mode(
    initial_prompt=agent_bundle.auto_mode_config["initial_prompt"],
    max_turns=agent_bundle.auto_mode_config["max_turns"],
    working_dir=agent_bundle.auto_mode_config["working_dir"]
)

Step 8: Monitor Execution

步骤8：监控执行

Agent executes autonomously:

Phase 1: Data Collection [In Progress]
├── S3 Collection: ✓ COMPLETED (50K records, 5 minutes)
├── Database Collection: ✓ COMPLETED (75K records, 8 minutes)
└── API Collection: ✗ FAILED (timeout, retrying...)
    └── Retry 1: ✓ COMPLETED (25K records, 4 minutes)

Phase 1: ✓ COMPLETED (150K records total, 3 sources, 17 minutes)

Phase 2: Data Transformation [In Progress]
├── Parsing: ✓ COMPLETED (150K records parsed)
├── Schema Mapping: ✓ COMPLETED (148K records mapped, 2K failed)
└── Missing Fields: ✓ COMPLETED (defaults applied)

Phase 2: ✓ COMPLETED (148K records ready, 2K logged as failed, 12 minutes)

Phase 3: Quality Validation [In Progress]
├── Completeness: ✓ PASS (98.7% complete, threshold 95%)
├── Accuracy: ✓ PASS (99.2% accurate, threshold 98%)
└── Consistency: ✓ PASS (100% consistent)

Phase 3: ✓ COMPLETED (All checks passed, 4 minutes)

Phase 4: Data Publishing [In Progress]
├── Warehouse Load: ✓ COMPLETED (148K records loaded)
├── Metadata Update: ✓ COMPLETED (pipeline_run_id: 12345)
└── Report Generation: ✓ COMPLETED (report.html)

Phase 4: ✓ COMPLETED (Data published, 8 minutes)

Total Execution: ✓ SUCCESS (41 minutes, all success criteria met)

Agent自主执行：

Phase 1: Data Collection [In Progress]
├── S3 Collection: ✓ COMPLETED (50K records, 5 minutes)
├── Database Collection: ✓ COMPLETED (75K records, 8 minutes)
└── API Collection: ✗ FAILED (timeout, retrying...)
    └── Retry 1: ✓ COMPLETED (25K records, 4 minutes)

Phase 1: ✓ COMPLETED (150K records total, 3 sources, 17 minutes)

Phase 2: Data Transformation [In Progress]
├── Parsing: ✓ COMPLETED (150K records parsed)
├── Schema Mapping: ✓ COMPLETED (148K records mapped, 2K failed)
└── Missing Fields: ✓ COMPLETED (defaults applied)

Phase 2: ✓ COMPLETED (148K records ready, 2K logged as failed, 12 minutes)

Phase 3: Quality Validation [In Progress]
├── Completeness: ✓ PASS (98.7% complete, threshold 95%)
├── Accuracy: ✓ PASS (99.2% accurate, threshold 98%)
└── Consistency: ✓ PASS (100% consistent)

Phase 3: ✓ COMPLETED (All checks passed, 4 minutes)

Phase 4: Data Publishing [In Progress]
├── Warehouse Load: ✓ COMPLETED (148K records loaded)
├── Metadata Update: ✓ COMPLETED (pipeline_run_id: 12345)
└── Report Generation: ✓ COMPLETED (report.html)

Phase 4: ✓ COMPLETED (Data published, 8 minutes)

Total Execution: ✓ SUCCESS (41 minutes, all success criteria met)

Step 9: Review Results

步骤9：审查结果

markdown

undefined

markdown

undefined

Pipeline Execution Report

Summary

Status: SUCCESS
Duration: 41 minutes (estimated: 30 minutes)
Records Processed: 150K ingested, 148K published
Success Rate: 98.7%

Status: SUCCESS
Duration: 41 minutes (estimated: 30 minutes)
Records Processed: 150K ingested, 148K published
Success Rate: 98.7%

Phase Results

Phase 1: Data Collection

S3: 50K records (5 min)
Database: 75K records (8 min)
API: 25K records (4 min, 1 retry)

S3: 50K records (5 min)
Database: 75K records (8 min)
API: 25K records (4 min, 1 retry)

Phase 2: Data Transformation

Successfully transformed: 148K records
Failed transformations: 2K records (logged to failed_records.log)
Failure reasons: Schema mismatch (1.5K), Invalid data (500)

Successfully transformed: 148K records
Failed transformations: 2K records (logged to failed_records.log)
Failure reasons: Schema mismatch (1.5K), Invalid data (500)

Phase 3: Quality Validation

Completeness: 98.7% ✓
Accuracy: 99.2% ✓
Consistency: 100% ✓

Completeness: 98.7% ✓
Accuracy: 99.2% ✓
Consistency: 100% ✓

Phase 4: Data Publishing

Warehouse load: Success
Pipeline run ID: 12345
Report: report.html

Warehouse load: Success
Pipeline run ID: 12345
Report: report.html

Issues Encountered

API timeout (Phase 1): Resolved with retry
2K transformation failures: Logged for manual review

API timeout (Phase 1): Resolved with retry
2K transformation failures: Logged for manual review

Recommendations

Investigate schema mismatches in API data
Add validation for API data format
Consider increasing timeout for API calls

undefined

Investigate schema mismatches in API data
Add validation for API data format
Consider increasing timeout for API calls

undefined

Step 10: Iteration (If Needed)

步骤10：迭代（如有需要）

If pipeline fails, agent adapts:

python

undefined

如果管道失败，Agent会自适应调整：

python

undefined

Example: API source completely unavailable

if phase1_result["api"]["status"] == "unavailable": # Agent adapts: continues with partial data log_warning("API source unavailable, continuing with S3 + database") proceed_to_phase2_with_partial_data()

# Report notes partial data
add_to_report("Data incomplete: API source unavailable")

if phase1_result["api"]["status"] == "unavailable": # Agent adapts: continues with partial data log_warning("API source unavailable, continuing with S3 + database") proceed_to_phase2_with_partial_data()

# Report notes partial data
add_to_report("Data incomplete: API source unavailable")

Example: Quality validation fails

if phase3_result["completeness"] < THRESHOLD: # Agent tries recovery: fetch missing data missing_records = identify_missing_records() retry_collection_for_missing(missing_records) rerun_transformation() rerun_validation()

# If still fails after retry, escalate
if still_below_threshold:
    escalate("Quality threshold not met after retry")

undefined

if phase3_result["completeness"] < THRESHOLD: # Agent tries recovery: fetch missing data missing_records = identify_missing_records() retry_collection_for_missing(missing_records) rerun_transformation() rerun_validation()

# If still fails after retry, escalate
if still_below_threshold:
    escalate("Quality threshold not met after retry")

undefined

11. Related Patterns

11. 相关模式

Goal-seeking agents relate to and integrate with other patterns:

目标导向Agent与其他模式相关并集成：

Debate Pattern (Multi-Agent Decision Making)

辩论模式（多Agent决策）

When to Combine:

Goal-seeking agent faces complex decision with trade-offs
Multiple valid approaches exist
Need consensus from different perspectives

Example:

python

undefined

何时结合：

目标导向Agent面临复杂的权衡决策
存在多种有效方法
需要不同视角的共识

示例：

python

undefined

Goal-seeking agent reaches decision point

if len(viable_strategies) > 1: # Invoke debate pattern result = invoke_debate( question="Which data transformation approach?", perspectives=["performance", "accuracy", "simplicity"], context=current_state )

# Use debate result to select strategy
selected_strategy = result.consensus

undefined

if len(viable_strategies) > 1: # Invoke debate pattern result = invoke_debate( question="Which data transformation approach?", perspectives=["performance", "accuracy", "simplicity"], context=current_state )

# Use debate result to select strategy
selected_strategy = result.consensus

undefined

N-Version Pattern (Redundant Implementation)

N版本模式（冗余实现）

When to Combine:

Goal-seeking agent executing critical phase
Error cost is high
Multiple independent implementations possible

Example:

python

undefined

何时结合：

目标导向Agent执行关键阶段
错误成本高
可能有多个独立实现

示例：

python

undefined

Critical security validation phase

if phase.is_critical(): # Generate N versions results = generate_n_versions( phase=phase, n=3, independent=True )

# Use voting or comparison to select result
validated_result = compare_and_validate(results)

undefined

if phase.is_critical(): # Generate N versions results = generate_n_versions( phase=phase, n=3, independent=True )

# Use voting or comparison to select result
validated_result = compare_and_validate(results)

undefined

Cascade Pattern (Fallback Strategies)

级联模式（回退策略）

When to Combine:

Goal-seeking agent has preferred approach but needs fallbacks
Quality/performance trade-offs exist
Graceful degradation desired

Example:

python

undefined

何时结合：

目标导向Agent有首选方法但需要回退
存在质量/性能权衡
需要优雅降级

示例：

python

undefined

Data transformation with fallback

try: # Optimal: ML-based transformation result = ml_transform(data) except MLModelUnavailable: try: # Pragmatic: Rule-based transformation result = rule_based_transform(data) except RuleEngineError: # Minimal: Manual templates result = template_transform(data)

undefined

try: # Optimal: ML-based transformation result = ml_transform(data) except MLModelUnavailable: try: # Pragmatic: Rule-based transformation result = rule_based_transform(data) except RuleEngineError: # Minimal: Manual templates result = template_transform(data)

undefined

Investigation Workflow (Knowledge Discovery)

调查工作流（知识发现）

When to Combine:

Goal requires understanding existing system
Need to discover architecture or patterns
Knowledge excavation before execution

Example:

python

undefined

何时结合：

目标需要了解现有系统
需要发现架构或模式
执行前需要挖掘知识

示例：

python

undefined

Before automating deployment, understand current system

if goal.requires_system_knowledge(): # Run investigation workflow investigation = run_investigation_workflow( scope="deployment pipeline", depth="comprehensive" )

# Use findings to inform goal-seeking execution
adapt_plan_based_on_investigation(investigation.findings)

undefined

if goal.requires_system_knowledge(): # Run investigation workflow investigation = run_investigation_workflow( scope="deployment pipeline", depth="comprehensive" )

# Use findings to inform goal-seeking execution
adapt_plan_based_on_investigation(investigation.findings)

undefined

Document-Driven Development (Specification First)

文档驱动开发（规范优先）

When to Combine:

Goal-seeking agent generates or modifies code
Clear specifications prevent drift
Documentation is single source of truth

Example:

python

undefined

何时结合：

目标导向Agent生成或修改代码
清晰的规范防止偏差
文档是单一事实来源

示例：

python

undefined

Goal: Implement new feature

if goal.involves_code_changes(): # DDD Phase 1: Generate specifications specs = generate_specifications(goal)

# DDD Phase 2: Review and approve specs
await human_review(specs)

# Goal-seeking agent implements from specs
implementation = execute_from_specifications(specs)

undefined

if goal.involves_code_changes(): # DDD Phase 1: Generate specifications specs = generate_specifications(goal)

# DDD Phase 2: Review and approve specs
await human_review(specs)

# Goal-seeking agent implements from specs
implementation = execute_from_specifications(specs)

undefined

Pre-Commit / CI Diagnostic (Quality Gates)

预提交/CI诊断（质量门）

When to Combine:

Goal-seeking agent makes code changes
Need to ensure quality before commit/push
Automated validation and fixes

Example:

python

undefined

何时结合：

目标导向Agent进行代码变更
需要在提交/推送前确保质量
自动验证和修复

示例：

python

undefined

After goal-seeking agent generates code

if changes_made: # Run pre-commit diagnostic pre_commit_result = run_pre_commit_diagnostic()

if pre_commit_result.has_failures():
    # Agent fixes issues
    apply_pre_commit_fixes(pre_commit_result.failures)

# After push, run CI diagnostic
ci_result = run_ci_diagnostic_workflow()

if ci_result.has_failures():
    # Agent iterates fixes
    iterate_ci_fixes_until_pass(ci_result)

undefined

if changes_made: # Run pre-commit diagnostic pre_commit_result = run_pre_commit_diagnostic()

if pre_commit_result.has_failures():
    # Agent fixes issues
    apply_pre_commit_fixes(pre_commit_result.failures)

# After push, run CI diagnostic
ci_result = run_ci_diagnostic_workflow()

if ci_result.has_failures():
    # Agent iterates fixes
    iterate_ci_fixes_until_pass(ci_result)

undefined

12. Quality Standards

12. 质量标准

Goal-seeking agents must meet these quality standards:

目标导向Agent必须满足这些质量标准：

Correctness

正确性

Success Criteria Verification:

Agent verifies all success criteria before completion
Intermediate phase results validated
No silent failures (all errors logged and handled)

Testing Coverage:

Happy path tested (all success criteria met)
Failure scenarios tested (phase failures, retries)
Edge cases identified and tested
Integration with real systems validated

成功标准验证：

Agent在完成前验证所有成功标准
中间阶段结果已验证
无静默故障（所有错误已记录并处理）

测试覆盖率：

测试了快乐路径（所有成功标准均满足）
测试了故障场景（阶段故障、重试）
已识别并测试边缘情况
已验证与真实系统的集成

Resilience

弹性

Error Handling:

Retry logic with exponential backoff
Alternative strategies for common failures
Graceful degradation when optimal path unavailable
Clear escalation criteria

State Management:

State persisted across phase boundaries
Resume capability after failures
Idempotent execution (safe to re-run)
Cleanup on abort

错误处理：

带有指数退避的重试逻辑
针对常见故障的替代策略
最优路径不可用时的优雅降级
清晰的升级标准

状态管理：

跨阶段边界持久化状态
故障后恢复能力
等幂执行（安全重跑）
中止时清理

Performance

性能

Efficiency:

Phases execute in parallel when possible
No unnecessary work (skip completed phases on retry)
Resource usage within limits (memory, CPU, time)
Timeout limits enforced

Latency:

Decision overhead acceptable for use case
No blocking waits (async where possible)
Progress reported (no black box periods)

效率：

阶段尽可能并行执行
无不必要的工作（重试时跳过已完成阶段）
资源使用在限制范围内（内存、CPU、时间）
超时限制已执行

延迟：

决策开销符合用例要求
无阻塞等待（尽可能异步）
报告进度（无黑盒时段）

Observability

可观测性

Usability

可用性

Documentation:

Goal clearly stated
Success criteria documented
Usage examples provided
Integration guide complete

User Experience:

Clear progress reporting
Actionable error messages
Human-readable outputs
Easy to invoke and monitor

Philosophy Compliance

理念合规性

13. Getting Started

13. 入门指南

Quick Start: Build Your First Goal-Seeking Agent

快速开始：构建你的第一个目标导向Agent

Step 1: Install amplihack (if not already)

bash

pip install amplihack

Step 2: Write a goal prompt

bash

cat > my-goal.md << 'EOF'

步骤1：安装amplihack（如果尚未安装）

bash

pip install amplihack

步骤2：编写目标提示

bash

cat > my-goal.md << 'EOF'

Goal: Automated Security Audit

Check application for common security issues:

SQL injection vulnerabilities
XSS vulnerabilities
Insecure dependencies
Missing security headers

Generate report with severity levels and remediation steps. EOF


**Step 3**: Generate agent

```bash
amplihack goal-agent-generator create \
  --prompt my-goal.md \
  --output .claude/agents/goal-driven/security-auditor

Step 4: Review generated plan

bash

cat .claude/agents/goal-driven/security-auditor/plan.yaml

Step 5: Execute agent

bash

amplihack goal-agent-generator execute \
  --agent-path .claude/agents/goal-driven/security-auditor \
  --auto-mode

Check application for common security issues:

SQL injection vulnerabilities
XSS vulnerabilities
Insecure dependencies
Missing security headers

Generate report with severity levels and remediation steps. EOF


**步骤3**：生成Agent

```bash
amplihack goal-agent-generator create \
  --prompt my-goal.md \
  --output .claude/agents/goal-driven/security-auditor

步骤4：查看生成的计划

bash

cat .claude/agents/goal-driven/security-auditor/plan.yaml

步骤5：执行Agent

bash

amplihack goal-agent-generator execute \
  --agent-path .claude/agents/goal-driven/security-auditor \
  --auto-mode

Common Use Cases

常见用例

Use Case 1: Workflow Automation

bash

undefined

用例1：工作流自动化

bash

undefined

Create release automation agent

echo "Automate release workflow: tag, build, test, deploy to staging" |
amplihack goal-agent-generator create --inline --output .claude/agents/goal-driven/release-automator


**Use Case 2: Data Pipeline**

```bash

echo "Automate release workflow: tag, build, test, deploy to staging" |
amplihack goal-agent-generator create --inline --output .claude/agents/goal-driven/release-automator


**用例2：数据管道**

```bash

Create ETL pipeline agent

echo "Extract from sources, transform to schema, validate quality, load to warehouse" |
amplihack goal-agent-generator create --inline --output .claude/agents/goal-driven/etl-pipeline


**Use Case 3: Diagnostic Workflow**

```bash

echo "Extract from sources, transform to schema, validate quality, load to warehouse" |
amplihack goal-agent-generator create --inline --output .claude/agents/goal-driven/etl-pipeline


**用例3：诊断工作流**

```bash

Create performance diagnostic agent

echo "Diagnose application performance issues, identify bottlenecks, suggest optimizations" |
amplihack goal-agent-generator create --inline --output .claude/agents/goal-driven/perf-diagnostic

undefined

echo "Diagnose application performance issues, identify bottlenecks, suggest optimizations" |
amplihack goal-agent-generator create --inline --output .claude/agents/goal-driven/perf-diagnostic

undefined

Learning Resources

学习资源

Documentation:

Review examples in

~/.amplihack/.claude/skills/goal-seeking-agent-pattern/examples/

Read real agent implementations in

~/.amplihack/.claude/agents/amplihack/specialized/

Check integration guide in

~/.amplihack/.claude/skills/goal-seeking-agent-pattern/templates/integration_guide.md

Practice:

Start simple: Build single-phase agent (e.g., file formatter)
Add complexity: Build multi-phase agent (e.g., test generator + runner)
Add autonomy: Build agent with error recovery (e.g., CI fixer)
Build production: Build full goal-seeking agent (e.g., deployment pipeline)

Get Help:

Review decision framework (Section 2)
Check design checklist (Section 6)
Study real examples (Section 5)
Ask architect agent for guidance

文档：

查看

~/.amplihack/.claude/skills/goal-seeking-agent-pattern/examples/

中的示例

阅读

~/.amplihack/.claude/agents/amplihack/specialized/

中的真实Agent实现

查看

~/.amplihack/.claude/skills/goal-seeking-agent-pattern/templates/integration_guide.md

中的集成指南

实践：

从简单开始：构建单阶段Agent（如文件格式化工具）
增加复杂度：构建多阶段Agent（如测试生成器+运行器）
添加自主性：构建带有错误恢复的Agent（如CI修复工具）
构建生产级Agent：构建完整的目标导向Agent（如部署管道）

获取帮助：

查看决策框架（第2节）
检查设计检查表（第6节）
研究真实示例（第5节）
向架构师Agent寻求指导

Next Steps

下一步

After building your first goal-seeking agent:

Test thoroughly: Cover success, failure, and edge cases
Monitor in production: Track metrics, logs, failures
Iterate: Refine based on real usage
Document learnings: Update DISCOVERIES.md with insights
Share patterns: Add successful approaches to PATTERNS.md

Success Indicators:

Agent completes goal autonomously 80%+ of time
Failures escalate with clear context
Execution time is acceptable
Users trust agent to run autonomously

Remember: Goal-seeking agents should be ruthlessly simple, focused on clear objectives, and adaptive to context. Start simple, add complexity only when justified, and always verify against success criteria.

构建第一个目标导向Agent后：

全面测试：覆盖成功、故障和边缘情况
生产环境监控：跟踪指标、日志、故障
迭代：根据实际使用情况优化
记录经验：在DISCOVERIES.md中更新见解
分享模式：在PATTERNS.md中添加成功方法

成功指标：

Agent自主完成目标的时间占比80%+
故障升级时提供清晰上下文
执行时间可接受
用户信任Agent自主运行

请记住：目标导向Agent应极致简洁，聚焦于明确的目标，并能适应上下文。从简单开始，仅在必要时增加复杂度，并始终对照成功标准进行验证。

goal-seeking-agent-pattern

Original

Translation

Goal-Seeking Agent Pattern Skill

目标导向Agent模式技能

1. What Are Goal-Seeking Agents?

1. 什么是目标导向Agent？

Core Characteristics

核心特性

Distinction from Traditional Agents

与传统Agent的区别

When Goal-Seeking Makes Sense

何时适合使用目标导向模式

When to Avoid Goal-Seeking

何时避免使用目标导向模式

2. When to Use This Pattern

2. 何时使用该模式

Problem Indicators

问题指标

Pattern 1: Workflow Variability

模式1：工作流可变性

Pattern 2: Multi-Phase Complexity

模式2：多阶段复杂度

Pattern 3: Autonomous Recovery Needed

模式3：需要自主恢复

Pattern 4: Adaptive Decision Making

模式4：自适应决策

Pattern 5: Domain Expertise Required

模式5：需要领域专业知识

Decision Framework

决策框架

Question 1: Is the objective well-defined but path flexible?

问题1：目标是否明确但实现路径灵活？

Question 2: Are there multiple phases with dependencies?

问题2：是否存在多个带有依赖关系的阶段？

Question 3: Is autonomous recovery valuable?

问题3：自主恢复是否有价值？

Question 4: Does context significantly affect approach?

问题4：上下文是否显著影响方法？

Question 5: Is the complexity justified?

问题5：复杂度是否合理？

Decision Matrix

决策矩阵

3. Architecture Pattern

3. 架构模式

Component Architecture

组件架构

Component 1: Goal Definition

Component 1: Goal Definition

Component 2: Execution Plan

Component 2: Execution Plan

Component 3: Plan Phase

Component 3: Plan Phase

Component 4: Skill Definition

Component 4: Skill Definition

Execution Flow

执行流程

Phase Dependency Management

阶段依赖管理

State Management

状态管理

Error Handling

错误处理

4. Integration with goal_agent_generator

4. 与goal_agent_generator集成

Core API

核心API

Step 1: Analyze natural language goal

Step 1: Analyze natural language goal

Step 2: Generate execution plan

Step 2: Generate execution plan

Step 3: Synthesize required skills

Step 3: Synthesize required skills

Step 4: Assemble complete agent

Step 4: Assemble complete agent

Step 5: Package for deployment

Step 5: Package for deployment

CLI Integration

CLI集成

Generate agent from prompt file