goal-seeking-agent-pattern

目标导向 Agent 模式技能

1. 什么是目标导向 Agent？

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

1. 理解高层目标：接受自然语言目标，无需明确的逐步指令
2. 规划执行过程：将目标拆分为包含依赖关系和成功标准的阶段
3. 自主执行：根据中间结果做出决策并调整行为
4. 自我评估：根据成功标准评估进度并调整方法
5. 弹性运行：妥善处理失败并探索替代方案

核心特征

自主性：Agent 决定如何实现目标，而不只是遵循规定的步骤

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

目标导向：关注结果（要实现什么），而非流程（如何实现）

多阶段：将复杂目标拆分为包含依赖关系的可管理阶段

自我监控：跟踪进度、检测失败并自主纠正方向

与传统 Agent 的区别

传统 Agent	目标导向 Agent
遵循固定工作流	根据上下文调整工作流
规定好的步骤	以结果为导向的目标
失败时需要人工干预	自主尝试恢复
单阶段执行	包含依赖关系的多阶段执行
僵化的决策树	动态调整策略

适合使用目标导向模式的情况

在以下情况下，目标导向 Agent 尤为出色：

• 问题空间庞大：有许多可能的成功路径
• 上下文多变：运行时条件会影响最优方法
• 可能发生失败：需要在没有人工干预的情况下自主恢复
• 目标明确：成功标准定义清晰，但路径灵活
• 多步骤复杂：需要协调包含依赖关系的多个阶段

不适合使用目标导向模式的情况

在以下情况下，应使用传统 Agent 或脚本：

• 只有一条确定性路径：实现目标只有一种方法
• 延迟敏感：需要尽可能快速地执行（不应有决策开销）
• 安全关键：每一步都需要人工验证
• 工作流简单：目标导向模式带来的复杂性超过其收益
• 审计要求：需要确定且可复现的执行过程

2. 何时使用此模式

问题表现

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

模式 1：工作流多变

表现：

• 同一目标需要根据上下文采取不同方法
• 多个环节都需要人工决策
• 梳理工作流时，答案总是“视情况而定”

示例：发布工作流会根据以下因素变化：

• 环境（预发布环境与生产环境）
• 变更类型（热修复与功能）
• 当前系统状态（健康与降级）

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

模式 2：多阶段复杂性

表现：

• 目标需要 3–5 个或更多不同阶段
• 阶段之间存在依赖关系（第 N 阶段的输出会传递给第 N+1 阶段）
• 存在并行执行的机会
• 每个阶段的成功标准各不相同

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

1. 数据收集（多个来源，并行执行）
2. 转换（依赖收集结果）
3. 验证（依赖转换输出）
4. 发布（验证通过后才执行）

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

模式 3：需要自主恢复

指标：

• 预计会发生故障，且可以恢复
• 存在多种重试／回退策略
• 人工干预成本高或耗时长
• 可以通过程序验证是否成功

示例：CI 诊断工作流：

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

解决方案：目标导向的 Agent 尝试各种策略，直到成功或升级处理

模式 4：自适应决策

指标：

• 需要在运行时评估权衡
• 存在多种有效解决方案，且各有特点
• 有优化目标（速度与质量与成本）
• 最佳实践取决于上下文

示例：修复 Agent 的模式匹配：

• 明显问题使用 QUICK 模式
• 问题不明确时使用 DIAGNOSTIC 模式
• 复杂解决方案使用 COMPREHENSIVE 模式

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

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

指标：

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

示例：AKS SRE 自动化：

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

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

决策框架

使用这个 5 个问题的框架评估目标导向方法是否适用：

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

如果满足以下条件，则回答 YES：

• 有明确的成功标准
• 存在多种有效方法
• 运行时上下文会影响最佳路径

如果满足以下条件，则回答 NO：

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

YES 示例：“确保 AKS 集群已达到生产就绪状态”（路径多样，标准明确） NO 示例：“运行指定的 kubectl 命令”（只有一条路径，且规定明确）

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

如果满足以下条件，则回答 YES：

• 目标自然拆分为 3-5 个或更多阶段
• 各阶段的输出会供后续阶段使用
• 部分阶段可以并行执行
• 某个阶段的故障会影响下游阶段

如果满足以下条件，则回答 NO：

• 单阶段执行即可满足需求
• 阶段之间没有依赖关系
• 纯顺序执行，没有分支

YES 示例：数据流水线（收集 → 转换 → 验证 → 发布） NO 示例：使用 ruff 格式化代码（单个原子操作）

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

如果满足以下条件，则回答 YES：

• 故障常见且可预期
• 存在多种恢复策略
• 人工干预成本高或耗时长
• 可以自动验证是否成功

如果满足以下条件，则回答 NO：

• 故障属于罕见的边缘情况
• 始终需要人工调查
• 安全至关重要（需要人工验证）
• 无法通过程序验证是否成功

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

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

如果满足以下条件，则回答 YES：

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

如果满足以下条件，则回答 NO：

• 所有上下文都适用相同的方法
• 不受环境依赖影响
• 无需权衡决策

YES 示例：修复 Agent（根据问题使用快速、诊断或全面模式） NO 示例：生成 UUID（与上下文无关）

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

是，如果：

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

否，如果：

• 问题是一次性的或很少出现
• 手动执行很快（少于 5 分钟）
• 简单脚本就足够
• 维护成本超过收益

是的示例：诊断 CI 故障（频繁、耗时、价值高） 否的示例：一次性数据迁移（很少发生，脚本足以应对）

决策矩阵

| 5 项全是 | 使用目标导向 Agent | | 4 项是，1 项否 | 可能适合使用目标导向 Agent | | 3 项是，2 项否 | 考虑使用更简单的 Agent 或混合方案 | | 2 项是，3 项否 | 传统 Agent 可能更合适 | | 0-1 项是 | 脚本或简单自动化 |

3. 架构模式

组件架构

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

# 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

执行流程

┌─────────────────────────────────────────────────────────────┐
│ 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]              │
└─────────────────────────────────────────────────────────────┘

阶段依赖管理

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

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

Phase A → Phase B → Phase C

并行执行：阶段可以并发运行

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

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

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

状态管理

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

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

错误处理

三种错误恢复策略：

退避重试：采用相同方法，并按指数间隔延迟

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)

替代策略：采用不同方法实现相同目标

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

优雅降级：接受部分成功

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

4. 与 goal_agent_generator 集成

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

核心 API

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"
)

CLI 集成

# 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

PromptAnalyzer 详情

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

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"}

领域分类：

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

复杂度判定：

• simple：单阶段，少于 50 个词，基本操作
• moderate：2-4 个阶段，50-150 个词，需要一定协调
• complex：5 个以上阶段，超过 150 个词，需要复杂编排

ObjectivePlanner 详情

生成多阶段执行计划：

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}")

按领域划分的阶段模板：

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

SkillSynthesizer 详情

将能力映射到技能：

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}")

能力映射：

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

AgentAssembler 详情

将组件组合成可执行包：

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

自动模式配置：

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

GoalAgentPackager 详情

打包部署所需的组件：

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

5. Amplihack 最新示例

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

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

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

目标驱动的解决方案：

# 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.
"""

# 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

主要特点：

• 自主运行：无需逐步指示即可检查多个系统
• 自适应：根据发现结果调整调查深度
• 多阶段：并行执行安全、网络和监控检查，按顺序生成报告
• 领域专家：内置 Azure 和 Kubernetes 知识
• 自我评估：验证每项检查并汇总结果

实现：

# 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

# Domain-specific customization:
plan.phases[0].required_capabilities = [
    "rbac-audit", "network-policy-check", "key-vault-integration"
]

经验总结：

• 对于复杂基础设施，领域专业知识至关重要
• 并行执行可显著缩短总耗时
• 可执行的建议能提升 Agent 的价值
• 全面的知识库（问答格式）有助于自主决策

示例 2：CI 诊断工作流

问题：CI 失败需要手动诊断并反复尝试修复

目标驱动的解决方案：

# 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)

主要特点：

• 迭代执行：重复修复流程，直到成功
• 自主恢复：尝试多种修复策略
• 状态管理：跟踪已尝试的修复，避免重复失败的操作
• 模式匹配：识别常见的 CI 失败类型
• 升级处理：达到最大迭代次数后向用户报告

实现：

# 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")

经验总结：

• 限制迭代次数可避免无限循环
• 模式匹配（测试、lint、类型）有助于针对性修复
• 智能等待（指数退避）可减少等待时间
• 绝不自动合并：始终需要人工批准

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

问题：提交前钩子失败，但错误信息不明确

目标驱动的解决方案：

# 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

主要特点：

• 专注推送前检查：在推送到 CI 前修复问题
• 工具版本管理：确保本地版本与提交前配置一致
• 钩子专属修复：根据钩子类型采用定制方法
• 快速迭代：无需等待 CI，即时获得反馈

实现：

# 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_FIXES[hook]()
    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!")

经验总结：

• 提交前修复比 CI 迭代更快
• 工具版本不匹配是常见原因
• 80% 的情况可通过自动修复解决
• 对剩余 20% 的情况进行升级处理，并提供清晰的诊断信息

示例 4：Fix-Agent 模式匹配

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

目标驱动的解决方案：

# 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)
"""

# 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

主要特点：

• 感知上下文：根据问题分析选择策略
• 多模式：针对不同复杂度提供三种修复模式
• 模式识别：从过往修复中学习
• 自适应：初始模式失败时升级处理复杂度

实现：

# 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

经验总结：

• 模式匹配有助于使用基于模板的修复（覆盖 80% 的情况）
• 模式选择可减少过度设计，采用恰到好处的方法
• 对于不明确的问题，诊断模式至关重要（根因分析）
• 使用数据有助于确定模板优先级

6. 设计检查清单

设计目标驱动型 Agent 时，请使用此检查清单：

目标定义

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

阶段设计

• 已拆分为 3-5 个阶段（不过于细化，也不过于粗略）
• 阶段依赖关系明确
• 已确定并行执行的机会
• 每个阶段都有明确的成功指标
• 已估算各阶段所需时长

技能映射

• 已确定各阶段所需的能力
• 已将技能映射到现有 Agent 或工具
• 已指定委派对象
• 不存在能力缺口

错误处理

• 已定义重试策略（最大尝试次数、退避策略）
• 已确定替代策略
• 升级标准明确（何时寻求帮助）
• 已考虑优雅降级方案（备用方法）

状态管理

• 跟踪各阶段的状态
• 存储阶段结果以供后续使用
• 保留失败历史
• 适当地共享上下文

测试

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

文档

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

理念合规性

• 极致简洁（不引入不必要的复杂性）
• 每个阶段各司其职
• 避免过度设计（解决方案规模适当）
• 可重新生成（规范清晰）

7. Agent SDK 集成（未来）

集成 Agent SDK Skill 后，目标驱动型 Agent 可以利用：

增强自主性

# 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)

工具发现

# 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"}
)

记忆管理

# 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)

Agent 委派

# 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")
])

可观测性

# 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)

集成示例

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

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

8. 权衡分析

目标驱动型 Agent 与传统 Agent 的对比

维度	目标驱动型 Agent	传统 Agent
灵活性	高——可根据上下文调整	低——工作流固定
开发时间	中等——定义目标和阶段	低——编写步骤脚本
执行时间	较长——存在决策开销	较短——直接执行
维护成本	较低——可自适应	较高——需要手动更新
可调试性	较难——行为动态变化	较容易——流程可预测
可复用性	高——同一 Agent，适用于不同上下文	低——依赖特定上下文
故障处理	自主恢复	需要人工干预
复杂性	较高——多阶段协调	较低——线性执行

如何选择

适合选择目标驱动型 Agent 的情况：

• 问题空间广阔，存在多种有效方法
• 不同执行过程的上下文差异显著
• 自主恢复能带来价值
• 在不同上下文中复用很重要
• 值得投入开发时间

适合选择传统 Agent 的情况：

• 存在单一路径且具有确定性
• 性能至关重要（需要低延迟）
• 简洁性至上
• 仅执行一次或很少执行
• 调试和可审计性至关重要

成本效益分析

目标驱动型 Agent 的成本：

• 开发时间更长（定义目标、阶段和能力）
• 执行时间更长（存在决策开销）
• 测试更复杂（行为动态变化）
• 调试更困难（路径不确定）

目标驱动型 Agent 的优势：

• 自主运行（减少人工干预）
• 适应上下文（可在不同条件下工作）
• 可在不同问题间复用（同一 Agent，不同目标）
• 可自行恢复（从容处理故障）

盈亏平衡点：符合以下条件时，采用目标驱动型方案才合理：

• 每周重复执行 2 次或以上，或
• 手动执行耗时 30 分钟或以上，或
• 需要难以记录的专业知识，或
• 自主恢复能带来很高价值

9. 何时升级处理

出现以下情况时，目标驱动型 Agent 应升级至人工处理：

达到硬性限制

超过最大迭代次数：

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

超过超时时间：

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

安全边界

破坏性操作：

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

生产环境变更：

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

不确定性检测

置信度较低：

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

策略冲突：

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
    )

意外情况

无法识别的错误：

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

环境不匹配：

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

升级处理最佳实践

提供上下文：

• 尝试了什么
• 失败了什么以及原因
• 考虑过哪些替代方案
• 当前系统状态

建议操作：

• 推荐后续步骤
• 提供诊断命令
• 提供手动干预点
• 必要时建议回滚

支持恢复：

• 保存执行状态
• 记录失败情况
• 提供恢复执行的能力
• 提供手动覆盖选项

升级处理示例：

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. 示例工作流

完整示例：构建一个用于数据管道自动化的目标驱动型 Agent

第 1 步：定义目标

# 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

## Constraints

- 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)

第 2 步：使用 PromptAnalyzer 进行分析

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"
# ]

第 3 步：使用 ObjectivePlanner 生成计划

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

第 4 步：合成技能

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)

第 5 步：组装 Agent

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"

第 6 步：打包 Agent

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

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

# 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:

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"]
)

第 8 步：监控执行

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)

第 9 步：检查结果

# Pipeline Execution Report

## Summary

- **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)

### 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)

### Phase 3: Quality Validation

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

### Phase 4: Data Publishing

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

## Issues Encountered

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

## Recommendations

1. Investigate schema mismatches in API data
2. Add validation for API data format
3. Consider increasing timeout for API calls

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

如果管道失败，Agent 会进行调整：

# 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")

# 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")

11. 相关模式

目标驱动型 Agent 与其他模式相关并可集成：

辩论模式（多 Agent 决策）

组合使用的时机：

• 目标驱动型 Agent 面临存在权衡的复杂决策
• 存在多种有效方案
• 需要汇集不同视角达成共识

示例：

# 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

N 版本模式（冗余实现）

组合使用的时机：

• 目标驱动型 Agent 正在执行关键阶段
• 错误代价很高
• 可以进行多个独立实现

示例：

# 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)

级联模式（回退策略）

组合使用的时机：

• 目标驱动型 Agent 有首选方案，但需要备用方案
• 存在质量与性能之间的权衡
• 希望实现平稳降级

示例：

# 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)

调查工作流（知识发现）

组合使用的时机：

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

示例：

# 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)

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

组合使用的时机：

• 目标驱动型 Agent 会生成或修改代码
• 明确的规范可防止偏离
• 文档是唯一事实来源

示例：

# 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)

提交前 / CI 诊断（质量门禁）

组合使用的时机：

• 目标驱动型 Agent 会修改代码
• 需要确保提交或推送前的质量
• 需要自动验证和修复

示例：

# 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)

12. 质量标准

目标驱动型 Agent 必须满足以下质量标准：

正确性

成功标准验证：

• Agent 在完成前验证所有成功标准
• 验证各中间阶段的结果
• 不存在静默失败（记录并处理所有错误）

测试覆盖：

• 测试正常路径（满足所有成功标准）
• 测试失败场景（阶段失败、重试）
• 识别并测试边界情况
• 验证与真实系统的集成

弹性

错误处理：

• 使用指数退避的重试逻辑
• 针对常见故障提供替代策略
• 无法采用最优路径时平稳降级
• 明确升级处理标准

状态管理：

• 在阶段边界之间持久化状态
• 失败后具备恢复执行的能力
• 执行具有幂等性（可安全重新运行）
• 中止时执行清理

性能

效率：

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

延迟：

• 决策开销符合使用场景的要求
• 不进行阻塞等待（尽可能采用异步方式）
• 报告进度（避免出现黑箱时段）

可观测性

日志记录：

• 记录阶段转换
• 记录决策及其理由
• 记录错误及其上下文
• 记录结果及其指标

指标：

• 跟踪各阶段耗时
• 跟踪成功率和失败率
• 监控资源使用情况
• 报告质量指标

追踪：

• 可追踪执行流程
• 保持各阶段之间的关联
• 提供充足的调试信息

易用性

文档：

• 清晰说明目标
• 记录成功标准
• 提供使用示例
• 完整的集成指南

用户体验：

• 清晰报告进度
• 提供可执行的错误消息
• 输出易于阅读
• 易于调用和监控

遵循理念

极致简洁：

• 没有不必要的阶段或复杂性
• 采用可行的最简单方案
• 不进行过早优化

单一职责：

• 每个阶段都有一项明确的任务
• 职责没有重叠
• 阶段边界清晰

模块化：

• 技能可在各 Agent 之间复用
• 阶段相互独立
• 接口（输入/输出）清晰

可再生成：

• 可以根据规范重新构建
• 没有硬编码的魔法值
• 配置外置

13. 入门

快速入门：构建你的第一个目标驱动型 Agent

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

pip install amplihack

步骤 2：编写目标提示词

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

步骤 3：生成 Agent

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

步骤 4：检查生成的计划

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

步骤 5：执行 Agent

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

常见用例

用例 1：工作流自动化

# 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

用例 2：数据管道

# 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

用例 3：诊断工作流

# 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

学习资源

文档：

• 查看 ~/.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 中的集成指南

练习：

1. 从简单开始：构建单阶段 Agent（例如，文件格式化工具）
2. 增加复杂度：构建多阶段 Agent（例如，测试生成器 + 运行器）
3. 增加自主性：构建具备错误恢复能力的 Agent（例如，CI 修复器）
4. 构建生产级 Agent：构建完整的目标驱动型 Agent（例如，部署管道）

获取帮助：

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

后续步骤

构建第一个目标驱动型 Agent 后：

1. 全面测试：覆盖成功、失败和边缘情况
2. 在生产环境中监控：跟踪指标、日志和故障
3. 持续迭代：根据实际使用情况进行改进
4. 记录经验：将心得更新到 DISCOVERIES.md
5. 分享模式：将成功方案添加到 PATTERNS.md

成功指标：

• Agent 在 80% 以上的情况下能够自主完成目标
• 故障升级时附带清晰的上下文
• 执行时间在可接受范围内
• 用户信任 Agent 能自主运行

---

请记住：目标驱动型 Agent 应当极致简洁，专注于明确的目标，并能根据上下文灵活调整。从简单方案开始，只在确有必要时增加复杂性，并始终依据成功标准进行验证。