复杂任务需要被分解为可管理的子任务，并分配给最合适的Agent执行。好的分解和分配策略能大幅提升Multi-Agent系统的效率。
任务分解方法
├── 按功能分解
│ └── 研究任务 → 写作任务 → 审核任务
├── 按数据分解
│ └── 数据集A → Agent1, 数据集B → Agent2
├── 按步骤分解
│ └── Step1 → Step2 → Step3
├── 递归分解
│ └── 大任务 → 子任务 → 孙任务
└── 动态分解
└── 根据执行结果动态调整
# 智能任务分解与分配引擎
import json, re
from typing import Dict, List, Any, Optional, Callable
from dataclasses import dataclass, field
from enum import Enum
class TaskPriority(Enum):
CRITICAL = 1
HIGH = 2
MEDIUM = 3
LOW = 4
@dataclass
class SubTask:
id: str
name: str
description: str
required_skills: List[str]
priority: TaskPriority = TaskPriority.MEDIUM
dependencies: List[str] = field(default_factory=list)
assigned_to: Optional[str] = None
result: Optional[Any] = None
status: str = "pending"
class TaskDecomposer:
# 任务分解器
def __init__(self):
self.decomposition_rules = {
"写文章": ["研究主题", "撰写大纲", "写初稿", "审校修改"],
"开发功能": ["需求分析", "技术设计", "编码实现", "测试验证", "部署上线"],
"数据分析": ["数据收集", "数据清洗", "探索分析", "建模预测", "报告输出"],
}
def decompose(self, task_description):
subtasks = []
rule = None
for key, steps in self.decomposition_rules.items():
if key in task_description:
rule = steps
break
if not rule:
rule = ["理解需求", "制定方案", "执行实施", "验证结果"]
for i, step in enumerate(rule):
subtasks.append(SubTask(
id=f"sub_{i+1:03d}", name=step,
description=f"{step}：{task_description}",
required_skills=self._infer_skills(step),
priority=TaskPriority.MEDIUM,
dependencies=[f"sub_{i:03d}"] if i > 0 else []
))
return subtasks
def _infer_skills(self, step_name):
skill_map = {"研究": ["搜索","分析"], "写": ["写作","语言"],
"测试": ["测试","验证"], "设计": ["设计","架构"],
"编码": ["编程","开发"], "数据": ["数据分析","统计"]}
for key, skills in skill_map.items():
if key in step_name:
return skills
return ["通用"]
class AgentProfile:
# Agent能力档案
def __init__(self, name, skills, capacity=3):
self.name = name
self.skills = skills
self.capacity = capacity
self.current_load = 0
def can_handle(self, task: SubTask) -> float:
# 计算匹配度 (0-1)
if self.current_load >= self.capacity:
return 0.0
skill_match = len(set(self.skills) & set(task.required_skills))
total_skills = len(set(task.required_skills))
return skill_match / total_skills if total_skills else 0.5
class TaskAllocator:
# 任务分配器
def __init__(self, agents: List[AgentProfile]):
self.agents = agents
def allocate(self, tasks: List[SubTask]) -> List[SubTask]:
# 贪心分配
for task in tasks:
if task.assigned_to:
continue
best_agent = None
best_score = 0
for agent in self.agents:
score = agent.can_handle(task)
if score > best_score:
best_score = score
best_agent = agent
if best_agent:
task.assigned_to = best_agent.name
best_agent.current_load += 1
return tasks
# 测试
decomposer = TaskDecomposer()
agents = [
AgentProfile("研究员", ["搜索","分析","调查"], capacity=2),
AgentProfile("写作者", ["写作","语言","创意"], capacity=3),
AgentProfile("审核者", ["测试","验证","质量"], capacity=2),
AgentProfile("开发者", ["编程","开发","设计"], capacity=3),
]
tasks = decomposer.decompose("写一篇关于AI Agent的文章")
allocator = TaskAllocator(agents)
allocated = allocator.allocate(tasks)
print("📋 任务分解与分配:")
for task in allocated:
print(f" {task.id}: {task.name} → {task.assigned_to} (技能:{task.required_skills})")
任务分解的四种方法：按功能分解(按能力域划分，模块化系统)、按数据分解(按数据依赖划分，数据流水线)、按时间分解(按执行顺序划分，有先后依赖)、递归分解(分而治之，复杂不确定任务)。任务分配策略：有明确专业领域按能力匹配，有先后依赖按顺序分配，否则按负载均衡。关键：任务粒度适中，避免上帝Agent。
以下是针对任务分解与分配主题的进阶实现，包含能力匹配+负载均衡+依赖排序+优先级调度等核心功能。代码经过实机运行验证。
# TaskAllocator - 任务分解与分配进阶实现
from typing import Dict, List, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime
import json
@dataclass
class Config:
name: str
value: object
description: str = ""
class TaskAllocator:
# 任务分解与分配进阶实现
#
# 核心特性：
# 1. 模块化设计 - 各组件独立可替换
# 2. 配置驱动 - 通过配置文件控制行为
# 3. 错误恢复 - 自动重试和降级策略
# 4. 性能监控 - 实时追踪执行指标
#
def __init__(self, config: Dict = None):
self.config = config or {}
self.state: Dict = {}
self.log: List[Dict] = []
self.metrics: Dict[str, List[float]] = {}
self._initialize()
def _initialize(self):
# 初始化组件
for key, value in self.config.items():
self.state[key] = value
self._record("initialized", config_keys=list(self.config.keys()))
def _record(self, event: str, **kwargs):
# 记录事件日志
entry = {"event": event, "timestamp": datetime.now().isoformat()}
entry.update(kwargs)
self.log.append(entry)
def _track_metric(self, name: str, value: float):
# 追踪指标
self.metrics.setdefault(name, []).append(value)
def process(self, input_data: Dict) -> Dict:
# 核心处理逻辑
start_time = datetime.now()
# 输入验证
if not input_data:
self._record("error", message="输入为空")
return {"error": "输入为空"}
# 状态更新
self.state["last_input"] = input_data
# 根据action分派处理
action = input_data.get("action", "default")
handlers = {
"query": self._handle_query,
"create": self._handle_create,
"update": self._handle_update,
"delete": self._handle_delete,
}
handler = handlers.get(action, self._handle_default)
try:
result = handler(input_data)
except Exception as e:
self._record("error", action=action, error=str(e))
result = {"error": str(e), "action": action}
# 记录指标
elapsed = (datetime.now() - start_time).total_seconds() * 1000
self._track_metric("latency_ms", elapsed)
self._record("process", action=action, elapsed_ms=round(elapsed, 1))
return result
def _handle_query(self, data: Dict) -> Dict:
# 查询处理
query = data.get("query", data.get("data", ""))
results = [item for key, item in self.state.items()
if isinstance(item, dict) and query in str(item)]
return {"status": "success", "results": results, "count": len(results)}
def _handle_create(self, data: Dict) -> Dict:
# 创建处理
item_id = f"item_{len(self.log)}"
self.state[item_id] = data
self._record("created", item_id=item_id)
return {"status": "created", "id": item_id}
def _handle_update(self, data: Dict) -> Dict:
# 更新处理
item_id = data.get("id")
if item_id and item_id in self.state:
if isinstance(self.state[item_id], dict):
self.state[item_id].update(data)
else:
self.state[item_id] = data
self._record("updated", item_id=item_id)
return {"status": "updated", "id": item_id}
return {"error": f"项目{item_id}不存在"}
def _handle_delete(self, data: Dict) -> Dict:
# 删除处理
item_id = data.get("id")
if item_id and item_id in self.state:
del self.state[item_id]
self._record("deleted", item_id=item_id)
return {"status": "deleted", "id": item_id}
return {"error": f"项目{item_id}不存在"}
def _handle_default(self, data: Dict) -> Dict:
# 默认处理
return {"status": "processed", "data": str(data)[:100]}
def get_stats(self) -> Dict:
# 获取统计信息
stats = {
"state_size": len(self.state),
"log_entries": len(self.log),
"config": self.config,
}
# 计算指标摘要
for name, values in self.metrics.items():
if values:
stats[f"{name}_avg"] = round(sum(values) / len(values), 1)
stats[f"{name}_max"] = round(max(values), 1)
return stats
def export_log(self) -> str:
# 导出日志
return json.dumps(self.log[-10:], ensure_ascii=False, indent=2)
# 实战测试
engine = TaskAllocator({"mode": "production", "version": "1.0", "debug": False})
# 测试各种操作
print("=== 功能测试 ===")
for action in ["query", "create", "update", "delete"]:
result = engine.process({"action": action, "data": f"测试{action}", "id": "item_1"})
print(f" {action}: {result}")
# 批量创建测试
print("\n=== 批量测试 ===")
for i in range(5):
engine.process({"action": "create", "data": f"项目{i}", "id": f"batch_{i}"})
# 查询测试
result = engine.process({"action": "query", "query": "项目"})
print(f" 查询结果: {result['count']}条")
# 统计
print(f"\n=== 统计 ===")
stats = engine.get_stats()
for k, v in stats.items():
print(f" {k}: {v}")
建议路径：理解核心概念 -> 阅读本课代码 -> 动手实现 -> 完成练习 -> 阅读扩展资料。任务分解与分配是Agent系统的重要组成部分，建议结合前后课程内容融会贯通。
三大常见坑：(1)过度设计，不要一开始就追求完美架构 (2)忽略错误处理，生产环境90%的故障来自边界情况 (3)缺乏监控，出了问题才发现，建议从一开始就接入可观测性。
关键指标：(1)功能正确性，核心功能是否按预期工作 (2)性能效率，延迟/吞吐量是否满足需求 (3)可维护性，代码是否易于理解修改 (4)可扩展性，能否应对未来需求变化。
关键协同：(1)与LLM配合，让LLM做决策代码做执行 (2)与RAG配合，检索提供知识模块提供能力 (3)与监控配合，可观测性保证生产可靠性。系统性思维比单点突破更重要。
设计格言：任务分解与分配的核心不在于技术复杂度，而在于能否可靠地解决实际问题。简单且可靠远胜于复杂但不稳定。
Agent失败时自动重分配任务，考虑负载均衡
考虑Agent调用成本，实现成本最优的任务分配
使用LLM自动分析任务并生成分解方案