27.3 自主规划算法
约 2524 字大约 8 分钟
自主规划算法概述
自主规划算法是Agentic AI系统的核心能力之一，它允许AI系统根据目标自主分解任务、制定执行计划、监控执行进度，并根据实际情况调整计划。
规划算法的基本概念
1. 什么是自主规划
自主规划是指AI系统在没有人类干预的情况下，根据给定的目标，自动生成和执行任务计划的过程。
自主规划的特点:
- 目标驱动: 以目标为导向进行规划
- 任务分解: 将复杂目标分解为可执行的子任务
- 动态调整: 根据执行情况动态调整计划
- 资源管理: 合理分配和利用资源
2. 规划算法分类
任务分解算法
1. 层次任务分解
示例：层次任务分解
用户请求： "实现一个层次任务分解算法"
Claude Code 生成的代码：
````python
from typing import Dict, List, Any, Optional
from dataclasses import dataclass, field
from enum import Enum
import logging
logger = logging.getLogger(__name__)
class TaskStatus(Enum):
"""任务状态"""
PENDING = "pending"
IN_PROGRESS = "in_progress"
COMPLETED = "completed"
FAILED = "failed"
CANCELLED = "cancelled"
class TaskPriority(Enum):
"""任务优先级"""
CRITICAL = 1
HIGH = 2
MEDIUM = 3
LOW = 4
@dataclass
class Task:
"""任务"""
id: str
name: str
description: str
status: TaskStatus = TaskStatus.PENDING
priority: TaskPriority = TaskPriority.MEDIUM
dependencies: List[str] = field(default_factory=list)
subtasks: List['Task'] = field(default_factory=list)
estimated_duration: Optional[float] = None
actual_duration: Optional[float] = None
result: Optional[Any] = None
error: Optional[str] = None
metadata: Dict[str, Any] = field(default_factory=dict)
class HierarchicalTaskPlanner:
"""层次任务规划器"""
def __init__(self):
self.tasks: Dict[str, Task] = {}
self.task_graph: Dict[str, List[str]] = {}
def add_task(self, task: Task):
"""添加任务"""
self.tasks[task.id] = task
self.task_graph[task.id] = task.dependencies
logger.info(f"Task added: {task.id}")
def decompose_task(self, task_id: str, max_depth: int = 3) -> List[Task]:
"""分解任务"""
task = self.tasks.get(task_id)
if not task:
raise ValueError(f"Task not found: {task_id}")
if max_depth <= 0:
return [task]
# 生成子任务
subtasks = self._generate_subtasks(task)
# 递归分解子任务
all_tasks = [task]
for subtask in subtasks:
self.add_task(subtask)
task.subtasks.append(subtask)
all_tasks.extend(self.decompose_task(subtask.id, max_depth - 1))
# 更新任务图
self.task_graph[task.id] = [subtask.id for subtask in subtasks]
return all_tasks
def _generate_subtasks(self, task: Task) -> List[Task]:
"""生成子任务"""
subtasks = []
# 根据任务类型生成子任务
task_type = task.metadata.get('type', 'default')
if task_type == 'code_generation':
subtasks = self._generate_code_generation_subtasks(task)
elif task_type == 'code_review':
subtasks = self._generate_code_review_subtasks(task)
elif task_type == 'testing':
subtasks = self._generate_testing_subtasks(task)
elif task_type == 'deployment':
subtasks = self._generate_deployment_subtasks(task)
else:
subtasks = self._generate_default_subtasks(task)
return subtasks
def _generate_code_generation_subtasks(self, task: Task) -> List[Task]:
"""生成代码生成子任务"""
subtasks = [
Task(
id=f"{task.id}_analyze_requirements",
name="Analyze Requirements",
description="Analyze the requirements for code generation",
priority=TaskPriority.HIGH,
metadata={'type': 'analysis'}
),
Task(
id=f"{task.id}_design_architecture",
name="Design Architecture",
description="Design the system architecture",
priority=TaskPriority.HIGH,
dependencies=[f"{task.id}_analyze_requirements"],
metadata={'type': 'design'}
),
Task(
id=f"{task.id}_generate_code",
name="Generate Code",
description="Generate the code",
priority=TaskPriority.HIGH,
dependencies=[f"{task.id}_design_architecture"],
metadata={'type': 'code_generation'}
),
Task(
id=f"{task.id}_review_code",
name="Review Code",
description="Review the generated code",
priority=TaskPriority.MEDIUM,
dependencies=[f"{task.id}_generate_code"],
metadata={'type': 'code_review'}
)
]
return subtasks
def _generate_code_review_subtasks(self, task: Task) -> List[Task]:
"""生成代码审查子任务"""
subtasks = [
Task(
id=f"{task.id}_static_analysis",
name="Static Analysis",
description="Perform static code analysis",
priority=TaskPriority.HIGH,
metadata={'type': 'analysis'}
),
Task(
id=f"{task.id}_security_check",
name="Security Check",
description="Check for security vulnerabilities",
priority=TaskPriority.HIGH,
dependencies=[f"{task.id}_static_analysis"],
metadata={'type': 'security'}
),
Task(
id=f"{task.id}_performance_check",
name="Performance Check",
description="Check for performance issues",
priority=TaskPriority.MEDIUM,
dependencies=[f"{task.id}_static_analysis"],
metadata={'type': 'performance'}
),
Task(
id=f"{task.id}_generate_report",
name="Generate Report",
description="Generate review report",
priority=TaskPriority.MEDIUM,
dependencies=[
f"{task.id}_security_check",
f"{task.id}_performance_check"
],
metadata={'type': 'reporting'}
)
]
return subtasks
def _generate_testing_subtasks(self, task: Task) -> List[Task]:
"""生成测试子任务"""
subtasks = [
Task(
id=f"{task.id}_unit_tests",
name="Unit Tests",
description="Write and run unit tests",
priority=TaskPriority.HIGH,
metadata={'type': 'testing'}
),
Task(
id=f"{task.id}_integration_tests",
name="Integration Tests",
description="Write and run integration tests",
priority=TaskPriority.HIGH,
dependencies=[f"{task.id}_unit_tests"],
metadata={'type': 'testing'}
),
Task(
id=f"{task.id}_e2e_tests",name="E2E Tests", description="Write and run end-to-end tests", priority=TaskPriority.MEDIUM, dependencies=[f"{task.id}_integration_tests"], metadata={'type': 'testing'} ) ]
return subtasks
def _generate_deployment_subtasks(self, task: Task) -> List[Task]: """生成部署子任务""" subtasks = [ Task( id=f"{task.id}_build", name="Build", description="Build the application", priority=TaskPriority.HIGH, metadata={'type': 'build'} ), Task( id=f"{task.id}_test", name="Test", description="Run tests", priority=TaskPriority.HIGH, dependencies=[f"{task.id}_build"], metadata={'type': 'testing'} ), Task( id=f"{task.id}_deploy", name="Deploy", description="Deploy to production", priority=TaskPriority.HIGH, dependencies=[f"{task.id}_test"], metadata={'type': 'deployment'} ), Task( id=f"{task.id}_verify", name="Verify", description="Verify deployment", priority=TaskPriority.HIGH, dependencies=[f"{task.id}_deploy"], metadata={'type': 'verification'} ) ]
return subtasks
def _generate_default_subtasks(self, task: Task) -> List[Task]: """生成默认子任务""" subtasks = [ Task( id=f"{task.id}_prepare", name="Prepare", description="Prepare for task execution", priority=TaskPriority.HIGH, metadata={'type': 'preparation'} ), Task( id=f"{task.id}_execute", name="Execute", description="Execute the task", priority=TaskPriority.HIGH, dependencies=[f"{task.id}_prepare"], metadata={'type': 'execution'} ), Task( id=f"{task.id}_finalize", name="Finalize", description="Finalize the task", priority=TaskPriority.MEDIUM, dependencies=[f"{task.id}_execute"], metadata={'type': 'finalization'} ) ]
return subtasks
def get_execution_plan(self, task_id: str) -> List[Task]: """获取执行计划""" plan = [] visited = set()
def dfs(current_task_id): if current_task_id in visited: return visited.add(current_task_id)
先执行依赖任务
for dep_id in self.task_graph.get(current_task_id, []): dfs(dep_id)
添加当前任务
task = self.tasks.get(current_task_id) if task: plan.append(task)
dfs(task_id)
return plan
def visualize_plan(self, task_id: str) -> str: """可视化执行计划""" plan = self.get_execution_plan(task_id)
visualization = "Execution Plan:\n" for i, task in enumerate(plan, 1): status_icon = { TaskStatus.PENDING: "○", TaskStatus.IN_PROGRESS: "◐", TaskStatus.COMPLETED: "●", TaskStatus.FAILED: "✗", TaskStatus.CANCELLED: "⊘" }.get(task.status, "?")
visualization += f"{i}. {status_icon} {task.name}\n"
if task.subtasks: for j, subtask in enumerate(task.subtasks, 1): subtask_icon = { TaskStatus.PENDING: "○", TaskStatus.IN_PROGRESS: "◐", TaskStatus.COMPLETED: "●", TaskStatus.FAILED: "✗", TaskStatus.CANCELLED: "⊘" }.get(subtask.status, "?")
visualization += f" {j}. {subtask_icon} {subtask.name}\n"
return visualization
使用示例
planner = HierarchicalTaskPlanner()
添加主任务
main_task = Task( id="main", name="Develop User Authentication System", description="Develop a complete user authentication system", priority=TaskPriority.HIGH, metadata={'type': 'code_generation'} )
planner.add_task(main_task)
分解任务
planner.decompose_task("main", max_depth=2)
获取执行计划
plan = planner.get_execution_plan("main")
print("Execution Plan:") for task in plan: print(f" - {task.name}")
可视化计划
print("\n" + planner.visualize_plan("main"))
2. 动态任务调整#
示例：动态任务调整
用户请求： "实现动态任务调整算法" Claude Code 生成的代码：
`python
from typing import Dict, List, Any, Optional
from datetime import datetime, timedelta
import logging
logger = logging.getLogger(__name__)
class DynamicTaskPlanner(HierarchicalTaskPlanner):
"""动态任务规划器"""
def __init__(self):
super().__init__()
self.execution_history: Dict[str, List[Dict[str, Any]]] = {}
self.performance_metrics: Dict[str, Dict[str, float]] = {}
def monitor_execution(self, task_id: str, status: TaskStatus, result: Optional[Any] = None, error: Optional[str] = None):
"""监控任务执行"""
task = self.tasks.get(task_id)
if not task:
return
# 更新任务状态
task.status = status
task.result = result
task.error = error
# 记录执行历史
if task_id not in self.execution_history:
self.execution_history[task_id] = []
self.execution_history[task_id].append({
'timestamp': datetime.utcnow(),
'status': status,
'result': result,
'error': error
})
logger.info(f"Task {task_id} status updated to {status}")
def analyze_performance(self, task_id: str) -> Dict[str, float]:
"""分析任务性能"""
if task_id not in self.execution_history:
return {}
history = self.execution_history[task_id]
# 计算性能指标
metrics = {
'total_executions': len(history),
'success_rate': 0.0,
'avg_duration': 0.0,
'failure_rate': 0.0
}
successful = [h for h in history if h['status'] == TaskStatus.COMPLETED]
failed = [h for h in history if h['status'] == TaskStatus.FAILED]
if len(history) > 0:
metrics['success_rate'] = len(successful) / len(history)
metrics['failure_rate'] = len(failed) / len(history)
self.performance_metrics[task_id] = metrics
return metrics
def adjust_plan(self, task_id: str) -> List[Task]:
"""调整执行计划"""
task = self.tasks.get(task_id)
if not task:
return []
# 分析性能
metrics = self.analyze_performance(task_id)
# 根据性能调整计划
if metrics.get('failure_rate', 0) > 0.5:
# 失败率高，添加重试任务
return self._add_retry_tasks(task)
elif metrics.get('avg_duration', 0) > 3600:
# 执行时间长，优化任务
return self._optimize_tasks(task)
else:
# 正常情况，返回原计划
return self.get_execution_plan(task_id)
def _add_retry_tasks(self, task: Task) -> List[Task]:
"""添加重试任务"""
retry_task = Task(
id=f"{task.id}_retry",
name=f"Retry {task.name}",
description=f"Retry failed task: {task.description}",
priority=TaskPriority.HIGH,
dependencies=[task.id],
metadata={'type': 'retry', 'original_task_id': task.id}
)
self.add_task(retry_task)
# 更新任务图
self.task_graph[task.id].append(retry_task.id)
return self.get_execution_plan(task.id)
def _optimize_tasks(self, task: Task) -> List[Task]:
"""优化任务"""
# 分解长时间运行的任务
if task.subtasks:
# 已经有子任务，优化子任务
for subtask in task.subtasks:
if subtask.estimated_duration and subtask.estimated_duration > 1800:
self.decompose_task(subtask.id, max_depth=1)
else:
# 分解当前任务
self.decompose_task(task.id, max_depth=2)
return self.get_execution_plan(task.id)
def predict_completion_time(self, task_id: str) -> Optional[float]:
"""预测完成时间"""
plan = self.get_execution_plan(task_id)
total_time = 0.0
for task in plan:
if task.estimated_duration:
total_time += task.estimated_duration
elif task.id in self.performance_metrics:
# 使用历史平均时间
avg_duration = self.performance_metrics[task.id].get('avg_duration', 0)
total_time += avg_duration
return total_time if total_time > 0 else None
# 使用示例
planner = DynamicTaskPlanner()
# 添加任务
task = Task(
id="task1",
name="Process Large Dataset",
description="Process a large dataset",
priority=TaskPriority.HIGH,
estimated_duration=7200,
metadata={'type': 'data_processing'}
)
planner.add_task(task)
# 分解任务
planner.decompose_task("task1", max_depth=2)
# 模拟执行
planner.monitor_execution("task1", TaskStatus.IN_PROGRESS)
# 模拟失败
planner.monitor_execution("task1", TaskStatus.FAILED, error="Timeout")
# 分析性能
metrics = planner.analyze_performance("task1")
print(f"Performance metrics: {metrics}")
# 调整计划
adjusted_plan = planner.adjust_plan("task1")
print("\nAdjusted plan:")
for task in adjusted_plan:
print(f" - {task.name}")
```> **动态任务调整特点**:
> - 实时监控任务执行
> - 分析任务性能
> - 根据性能调整计划
> - 预测完成时间规划算法优化
1. 并行任务调度
示例：并行任务调度
用户请求： "实现并行任务调度算法" Claude Code 生成的代码：
`python
from typing import Dict, List, Set, Optional
from collections import defaultdict, deque
import logging
logger = logging.getLogger(__name__)
class ParallelTaskScheduler:
"""并行任务调度器"""
def __init__(self, max_workers: int = 4):
self.max_workers = max_workers
self.tasks: Dict[str, Task] = {}
self.dependencies: Dict[str, Set[str]] = defaultdict(set)
self.dependents: Dict[str, Set[str]] = defaultdict(set)
self.ready_tasks: Set[str] = set()
self.running_tasks: Set[str] = set()
self.completed_tasks: Set[str] = set()
def add_task(self, task: Task):
"""添加任务"""
self.tasks[task.id] = task
# 添加依赖关系
for dep_id in task.dependencies:
self.dependencies[task.id].add(dep_id)
self.dependents[dep_id].add(task.id)
# 如果没有依赖，标记为就绪
if not task.dependencies:
self.ready_tasks.add(task.id)
logger.info(f"Task added: {task.id}")
def get_ready_tasks(self) -> List[Task]:
"""获取就绪任务"""
available = self.ready_tasks - self.running_tasks
return [self.tasks[task_id] for task_id in available]
def start_task(self, task_id: str):
"""开始任务"""
if task_id in self.ready_tasks and task_id not in self.running_tasks:
self.running_tasks.add(task_id)
self.ready_tasks.remove(task_id)
logger.info(f"Task started: {task_id}")
def complete_task(self, task_id: str, result: Optional[Any] = None, error: Optional[str] = None):
"""完成任务"""
if task_id not in self.running_tasks:
logger.warning(f"Task not running: {task_id}")
return
# 更新任务状态
task = self.tasks[task_id]
task.status = TaskStatus.COMPLETED if not error else TaskStatus.FAILED
task.result = result
task.error = error
# 更新集合
self.running_tasks.remove(task_id)
self.completed_tasks.add(task_id)
# 检查依赖此任务的任务
for dependent_id in self.dependents[task_id]:
self.dependencies[dependent_id].remove(task_id)
# 如果所有依赖都完成，标记为就绪
if not self.dependencies[dependent_id]:
self.ready_tasks.add(dependent_id)
logger.info(f"Task completed: {task_id}")
def get_schedule(self) -> List[List[str]]:
"""获取调度计划"""
schedule = []
remaining = set(self.tasks.keys())
while remaining:
# 获取当前可以执行的任务
current_batch = []
for task_id in remaining:
task = self.tasks[task_id]
if all(dep in self.completed_tasks for dep in task.dependencies):
current_batch.append(task_id)
# 限制并发数
current_batch = current_batch[:self.max_workers]
if not current_batch:
logger.warning("Circular dependency detected")
break
schedule.append(current_batch)
# 更新剩余任务
remaining -= set(current_batch)
self.completed_tasks.update(current_batch)
return schedule
def visualize_schedule(self) -> str:
"""可视化调度"""
schedule = self.get_schedule()
visualization = "Task Schedule:\n"
for i, batch in enumerate(schedule, 1):
visualization += f"Batch {i} (parallel):\n"
for task_id in batch:
task = self.tasks[task_id]
visualization += f" - {task.name}\n"
return visualization
# 使用示例
scheduler = ParallelTaskScheduler(max_workers=3)
# 添加任务
tasks = [
Task(id="A", name="Task A", description="Task A", priority=TaskPriority.HIGH),
Task(id="B", name="Task B", description="Task B", priority=TaskPriority.HIGH, dependencies=["A"]),
Task(id="C", name="Task C", description="Task C", priority=TaskPriority.HIGH, dependencies=["A"]),
Task(id="D", name="Task D", description="Task D", priority=TaskPriority.HIGH, dependencies=["B", "C"]),
Task(id="E", name="Task E", description="Task E", priority=TaskPriority.HIGH, dependencies=["B"]),
Task(id="F", name="Task F", description="Task F", priority=TaskPriority.HIGH, dependencies=["C"]),
Task(id="G", name="Task G", description="Task G", priority=TaskPriority.HIGH, dependencies=["D", "E", "F"])
]
for task in tasks:
scheduler.add_task(task)
# 获取调度计划
schedule = scheduler.get_schedule()
print("Schedule:")
for i, batch in enumerate(schedule, 1):
print(f"Batch {i}: {[scheduler.tasks[task_id].name for task_id in batch]}")
# 可视化调度
print("\n" + scheduler.visualize_schedule())
```> **并行任务调度特点**:
> - 识别可并行执行的任务
> - 限制并发任务数量
> - 处理任务依赖关系
> - 生成调度计划
```
## 总结
自主规划算法包括：
1. **规划算法的基本概念**: 什么是自主规划、规划算法分类
2. **任务分解算法**: 层次任务分解、动态任务调整
3. **规划算法优化**: 并行任务调度
通过自主规划算法，Claude Code可以自主分解任务、制定执行计划，并根据实际情况动态调整。
在下一节中，我们将探讨记忆系统设计。
```
```