Python

主题

任务分解策略详解 ​

概述 ​

任务分解是将复杂任务拆分为可管理、可执行的子任务的过程。本章详细介绍各种任务分解策略的原理、实现和最佳实践。

分解策略分类 ​

策略特点适用场景
层次分解树状结构，逐层细化任务结构清晰
功能分解按功能模块划分功能独立性强
数据分解按数据划分数据并行处理
流水线分解按处理阶段划分顺序处理任务

1. 层次分解 ​

1.1 基础层次分解 ​

python
from typing import Dict, List, Any, Optional
from dataclasses import dataclass, field
import uuid

@dataclass
class TaskNode:
    """任务节点"""
    id: str = field(default_factory=lambda: str(uuid.uuid4()))
    name: str = ""
    description: str = ""
    level: int = 0                    # 层级
    parent_id: Optional[str] = None
    children_ids: List[str] = field(default_factory=list)
    status: str = "pending"           # pending/running/completed/failed
    result: Any = None
    assigned_to: Optional[str] = None

class HierarchicalTaskDecomposer:
    """层次任务分解器"""
    
    def __init__(self, llm=None):
        self.llm = llm
        self.tasks: Dict[str, TaskNode] = {}
    
    def decompose(self, task_description: str, max_depth: int = 3, 
                  max_children: int = 5) -> TaskNode:
        """层次分解任务"""
        # 创建根任务
        root = TaskNode(
            name="根任务",
            description=task_description,
            level=0
        )
        self.tasks[root.id] = root
        
        # 递归分解
        self._decompose_recursive(root, max_depth, max_children)
        
        return root
    
    def _decompose_recursive(self, parent: TaskNode, max_depth: int, 
                            max_children: int):
        """递归分解"""
        if parent.level >= max_depth:
            return
        
        # 生成子任务
        subtasks = self._generate_subtasks(parent, max_children)
        
        for subtask_desc in subtasks:
            child = TaskNode(
                name=subtask_desc["name"],
                description=subtask_desc["description"],
                level=parent.level + 1,
                parent_id=parent.id
            )
            
            parent.children_ids.append(child.id)
            self.tasks[child.id] = child
            
            # 递归分解子任务
            self._decompose_recursive(child, max_depth, max_children)
    
    def _generate_subtasks(self, parent: TaskNode, max_children: int) -> List[Dict]:
        """生成子任务"""
        # 简单的子任务生成逻辑
        # 实际应用中应使用LLM
        
        if parent.level == 0:
            return [
                {"name": "需求分析", "description": "分析任务需求"},
                {"name": "方案设计", "description": "设计实现方案"},
                {"name": "开发实现", "description": "进行开发实现"},
                {"name": "测试验证", "description": "测试验证结果"},
                {"name": "部署上线", "description": "部署到生产环境"}
            ]
        elif parent.level == 1:
            return [
                {"name": f"{parent.name}-子任务1", "description": f"{parent.description}的子任务1"},
                {"name": f"{parent.name}-子任务2", "description": f"{parent.description}的子任务2"}
            ]
        
        return []
    
    def get_task_tree(self, root_id: str) -> Dict:
        """获取任务树"""
        root = self.tasks[root_id]
        
        def build_tree(node: TaskNode) -> Dict:
            tree = {
                "id": node.id,
                "name": node.name,
                "status": node.status,
                "children": []
            }
            
            for child_id in node.children_ids:
                child = self.tasks[child_id]
                tree["children"].append(build_tree(child))
            
            return tree
        
        return build_tree(root)
    
    def get_leaf_tasks(self, root_id: str) -> List[TaskNode]:
        """获取所有叶子任务"""
        root = self.tasks[root_id]
        leaves = []
        
        def collect_leaves(node: TaskNode):
            if not node.children_ids:
                leaves.append(node)
            else:
                for child_id in node.children_ids:
                    collect_leaves(self.tasks[child_id])
        
        collect_leaves(root)
        return leaves

# 使用示例
decomposer = HierarchicalTaskDecomposer()

# 分解任务
root = decomposer.decompose("开发一个电商网站", max_depth=3)

# 获取任务树
tree = decomposer.get_task_tree(root.id)
print("任务树:")
import json
print(json.dumps(tree, indent=2, ensure_ascii=False))

# 获取叶子任务（可执行任务）
leaves = decomposer.get_leaf_tasks(root.id)
print(f"\n可执行任务数量: {len(leaves)}")
for leaf in leaves:
    print(f"  - {leaf.name}: {leaf.description}")

1.2 LLM驱动的层次分解 ​

python
from typing import Dict, List, Any
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.pydantic_v1 import BaseModel, Field

class SubTask(BaseModel):
    """子任务"""
    name: str = Field(description="任务名称")
    description: str = Field(description="任务描述")
    complexity: str = Field(description="复杂度: low/medium/high")

class DecompositionResult(BaseModel):
    """分解结果"""
    original_task: str = Field(description="原始任务")
    subtasks: List[SubTask] = Field(description="子任务列表")
    decomposition_strategy: str = Field(description="分解策略")

class LLMHierarchicalDecomposer:
    """LLM驱动的层次分解器"""
    
    def __init__(self, llm=None):
        self.llm = llm or ChatOpenAI(model="gpt-4")
        self.structured_llm = self.llm.with_structured_output(DecompositionResult)
    
    def decompose(self, task: str, context: str = "", 
                  num_subtasks: int = 5) -> DecompositionResult:
        """使用LLM分解任务"""
        prompt = ChatPromptTemplate.from_messages([
            ("system", """你是一个任务分解专家。请将复杂任务分解为可执行的子任务。

分解原则:
1. 每个子任务应该是独立可执行的
2. 子任务之间应该有清晰的依赖关系
3. 子任务的粒度应该适中
4. 考虑任务的复杂度和优先级

上下文信息: {context}"""),
            ("human", "请将以下任务分解为 {num_subtasks} 个子任务:\n\n{task}")
        ])
        
        chain = prompt | self.structured_llm
        result = chain.invoke({
            "task": task,
            "context": context,
            "num_subtasks": num_subtasks
        })
        
        return result
    
    def decompose_recursive(self, task: str, max_depth: int = 2) -> Dict:
        """递归分解"""
        result = self.decompose(task)
        
        tree = {
            "task": task,
            "subtasks": []
        }
        
        for subtask in result.subtasks:
            subtask_node = {
                "name": subtask.name,
                "description": subtask.description,
                "complexity": subtask.complexity,
                "children": []
            }
            
            # 如果复杂度高且未达到最大深度，继续分解
            if subtask.complexity == "high" and max_depth > 0:
                child_result = self.decompose_recursive(
                    subtask.description, 
                    max_depth - 1
                )
                subtask_node["children"] = child_result.get("subtasks", [])
            
            tree["subtasks"].append(subtask_node)
        
        return tree

# 使用示例
decomposer = LLMHierarchicalDecomposer()

# 分解任务
result = decomposer.decompose(
    "开发一个智能客服系统",
    context="使用Python和LangChain，支持多轮对话",
    num_subtasks=5
)

print(f"原始任务: {result.original_task}")
print(f"分解策略: {result.decomposition_strategy}")
print("\n子任务:")
for i, subtask in enumerate(result.subtasks, 1):
    print(f"  {i}. {subtask.name} (复杂度: {subtask.complexity})")
    print(f"     {subtask.description}")

2. 功能分解 ​

2.1 基于功能模块的分解 ​

python
from typing import Dict, List, Any, Optional
from dataclasses import dataclass, field
from enum import Enum

class FunctionType(str, Enum):
    """功能类型"""
    INPUT = "input"           # 输入处理
    PROCESS = "process"       # 核心处理
    OUTPUT = "output"         # 输出处理
    STORAGE = "storage"       # 存储
    COMMUNICATION = "communication"  # 通信

@dataclass
class FunctionModule:
    """功能模块"""
    id: str = field(default_factory=lambda: str(uuid.uuid4()))
    name: str = ""
    function_type: FunctionType = FunctionType.PROCESS
    inputs: List[str] = field(default_factory=list)      # 输入接口
    outputs: List[str] = field(default_factory=list)     # 输出接口
    dependencies: List[str] = field(default_factory=list)  # 依赖的模块ID
    implementation: str = ""  # 实现描述

class FunctionalDecomposer:
    """功能分解器"""
    
    def __init__(self):
        self.modules: Dict[str, FunctionModule] = {}
    
    def decompose(self, system_description: str) -> Dict[str, FunctionModule]:
        """功能分解"""
        # 根据系统描述分解功能模块
        modules = self._identify_modules(system_description)
        
        # 建立模块关系
        self._establish_relationships(modules)
        
        return modules
    
    def _identify_modules(self, description: str) -> Dict[str, FunctionModule]:
        """识别功能模块"""
        # 简单的模块识别逻辑
        # 实际应用中应使用LLM
        
        modules = {}
        
        # 输入模块
        input_module = FunctionModule(
            name="用户输入处理",
            function_type=FunctionType.INPUT,
            outputs=["parsed_input"],
            implementation="解析用户输入"
        )
        modules[input_module.id] = input_module
        
        # 核心处理模块
        process_module = FunctionModule(
            name="核心处理",
            function_type=FunctionType.PROCESS,
            inputs=["parsed_input"],
            outputs=["processing_result"],
            dependencies=[input_module.id],
            implementation="执行核心业务逻辑"
        )
        modules[process_module.id] = process_module
        
        # 输出模块
        output_module = FunctionModule(
            name="结果输出",
            function_type=FunctionType.OUTPUT,
            inputs=["processing_result"],
            dependencies=[process_module.id],
            implementation="格式化并输出结果"
        )
        modules[output_module.id] = output_module
        
        return modules
    
    def _establish_relationships(self, modules: Dict[str, FunctionModule]):
        """建立模块关系"""
        # 根据输入输出建立依赖关系
        pass
    
    def get_dependency_graph(self) -> Dict[str, List[str]]:
        """获取依赖图"""
        graph = {}
        for module_id, module in self.modules.items():
            graph[module_id] = module.dependencies
        return graph

# 使用示例
decomposer = FunctionalDecomposer()
modules = decomposer.decompose("智能客服系统")

print("功能模块:")
for module_id, module in modules.items():
    print(f"  {module.name} ({module.function_type.value})")
    print(f"    输入: {module.inputs}")
    print(f"    输出: {module.outputs}")

3. 数据分解 ​

3.1 数据并行分解 ​

python
from typing import Dict, List, Any, Callable
from dataclasses import dataclass, field
import uuid

@dataclass
class DataChunk:
    """数据块"""
    id: str = field(default_factory=lambda: str(uuid.uuid4()))
    data: Any = None
    index: int = 0
    total_chunks: int = 1
    metadata: Dict = field(default_factory=dict)

class DataDecomposer:
    """数据分解器"""
    
    def __init__(self, chunk_size: int = 100):
        self.chunk_size = chunk_size
    
    def decompose_list(self, data: List[Any]) -> List[DataChunk]:
        """分解列表数据"""
        chunks = []
        total_chunks = (len(data) + self.chunk_size - 1) // self.chunk_size
        
        for i in range(0, len(data), self.chunk_size):
            chunk = DataChunk(
                data=data[i:i+self.chunk_size],
                index=len(chunks),
                total_chunks=total_chunks
            )
            chunks.append(chunk)
        
        return chunks
    
    def decompose_dict(self, data: Dict[str, Any], 
                       keys_per_chunk: int = 10) -> List[DataChunk]:
        """分解字典数据"""
        chunks = []
        keys = list(data.keys())
        total_chunks = (len(keys) + keys_per_chunk - 1) // keys_per_chunk
        
        for i in range(0, len(keys), keys_per_chunk):
            chunk_keys = keys[i:i+keys_per_chunk]
            chunk_data = {k: data[k] for k in chunk_keys}
            
            chunk = DataChunk(
                data=chunk_data,
                index=len(chunks),
                total_chunks=total_chunks
            )
            chunks.append(chunk)
        
        return chunks
    
    def decompose_by_field(self, data: List[Dict], field: str) -> Dict[str, List[DataChunk]]:
        """按字段值分解"""
        # 按字段分组
        groups = {}
        for item in data:
            key = item.get(field, "unknown")
            if key not in groups:
                groups[key] = []
            groups[key].append(item)
        
        # 对每个分组进行分解
        result = {}
        for key, items in groups.items():
            result[key] = self.decompose_list(items)
        
        return result

class ParallelProcessor:
    """并行处理器"""
    
    def __init__(self, num_workers: int = 4):
        self.num_workers = num_workers
    
    def process_chunks(self, chunks: List[DataChunk], 
                       processor: Callable) -> List[Any]:
        """并行处理数据块"""
        results = []
        
        # 简单的顺序处理（实际应使用多线程/多进程）
        for chunk in chunks:
            result = processor(chunk.data)
            results.append(result)
        
        return results
    
    def merge_results(self, results: List[Any]) -> Any:
        """合并结果"""
        # 简单的合并逻辑
        if isinstance(results[0], list):
            merged = []
            for result in results:
                merged.extend(result)
            return merged
        elif isinstance(results[0], dict):
            merged = {}
            for result in results:
                merged.update(result)
            return merged
        
        return results

# 使用示例
decomposer = DataDecomposer(chunk_size=50)

# 分解列表数据
data = list(range(500))
chunks = decomposer.decompose_list(data)

print(f"数据分解: {len(data)} 条数据 -> {len(chunks)} 个块")

# 并行处理
processor = ParallelProcessor(num_workers=4)

def process_chunk(chunk_data):
    """处理数据块"""
    return [x * 2 for x in chunk_data]

results = processor.process_chunks(chunks, process_chunk)
merged = processor.merge_results(results)

print(f"处理结果: {len(merged)} 条数据")

4. 流水线分解 ​

4.1 阶段流水线 ​

python
from typing import Dict, List, Any, Callable, Optional
from dataclasses import dataclass, field
from abc import ABC, abstractmethod
import uuid

@dataclass
class PipelineStage:
    """流水线阶段"""
    id: str = field(default_factory=lambda: str(uuid.uuid4()))
    name: str = ""
    description: str = ""
    processor: Optional[Callable] = None
    input_type: str = "Any"
    output_type: str = "Any"

class Pipeline:
    """流水线"""
    
    def __init__(self, name: str):
        self.name = name
        self.stages: List[PipelineStage] = []
        self.results: Dict[str, Any] = {}
    
    def add_stage(self, stage: PipelineStage):
        """添加阶段"""
        self.stages.append(stage)
        print(f"[{self.name}] 添加阶段: {stage.name}")
    
    def execute(self, input_data: Any) -> Any:
        """执行流水线"""
        print(f"\n{'='*50}")
        print(f"执行流水线: {self.name}")
        print(f"阶段数: {len(self.stages)}")
        print(f"{'='*50}\n")
        
        current_data = input_data
        
        for i, stage in enumerate(self.stages):
            print(f"阶段 {i+1}: {stage.name}")
            
            if stage.processor:
                current_data = stage.processor(current_data)
                self.results[stage.id] = current_data
                print(f"  完成，输出类型: {type(current_data).__name__}")
            else:
                print(f"  跳过（无处理器）")
        
        print(f"\n{'='*50}")
        print(f"流水线完成")
        print(f"{'='*50}\n")
        
        return current_data

class PipelineBuilder:
    """流水线构建器"""
    
    def __init__(self, name: str):
        self.pipeline = Pipeline(name)
    
    def add_stage(self, name: str, processor: Callable, 
                  description: str = "") -> 'PipelineBuilder':
        """添加阶段"""
        stage = PipelineStage(
            name=name,
            description=description,
            processor=processor
        )
        self.pipeline.add_stage(stage)
        return self
    
    def build(self) -> Pipeline:
        """构建流水线"""
        return self.pipeline

# 使用示例
def extract_data(data):
    """提取数据"""
    print("    提取数据...")
    return {"extracted": data, "count": len(data)}

def transform_data(data):
    """转换数据"""
    print("    转换数据...")
    data["transformed"] = True
    return data

def validate_data(data):
    """验证数据"""
    print("    验证数据...")
    data["valid"] = True
    return data

def load_data(data):
    """加载数据"""
    print("    加载数据...")
    data["loaded"] = True
    return data

# 构建ETL流水线
pipeline = (PipelineBuilder("数据处理流水线")
    .add_stage("提取", extract_data, "从数据源提取数据")
    .add_stage("转换", transform_data, "转换数据格式")
    .add_stage("验证", validate_data, "验证数据质量")
    .add_stage("加载", load_data, "加载到目标系统")
    .build())

# 执行流水线
result = pipeline.execute([1, 2, 3, 4, 5])

print(f"\n最终结果: {result}")

5. 智能分解 ​

5.1 自适应分解 ​

python
from typing import Dict, List, Any, Optional
from dataclasses import dataclass, field
from enum import Enum

class DecompositionStrategy(str, Enum):
    """分解策略"""
    HIERARCHICAL = "hierarchical"
    FUNCTIONAL = "functional"
    DATA = "data"
    PIPELINE = "pipeline"
    HYBRID = "hybrid"

class AdaptiveDecomposer:
    """自适应分解器"""
    
    def __init__(self, llm=None):
        self.llm = llm
        self.strategy_selectors = {
            "开发": DecompositionStrategy.FUNCTIONAL,
            "分析": DecompositionStrategy.DATA,
            "处理": DecompositionStrategy.PIPELINE,
            "设计": DecompositionStrategy.HIERARCHICAL
        }
    
    def analyze_task(self, task: str) -> Dict[str, Any]:
        """分析任务特征"""
        features = {
            "keywords": self._extract_keywords(task),
            "complexity": self._estimate_complexity(task),
            "parallelizability": self._estimate_parallelizability(task),
            "data_intensive": self._is_data_intensive(task)
        }
        
        return features
    
    def select_strategy(self, task: str) -> DecompositionStrategy:
        """选择分解策略"""
        features = self.analyze_task(task)
        
        # 根据特征选择策略
        if features["data_intensive"]:
            return DecompositionStrategy.DATA
        elif features["parallelizability"] > 0.7:
            return DecompositionStrategy.PIPELINE
        elif features["complexity"] > 0.8:
            return DecompositionStrategy.HIERARCHICAL
        else:
            return DecompositionStrategy.FUNCTIONAL
    
    def decompose(self, task: str) -> Dict[str, Any]:
        """自适应分解"""
        strategy = self.select_strategy(task)
        
        print(f"任务: {task}")
        print(f"选择策略: {strategy.value}")
        
        # 根据策略选择分解器
        if strategy == DecompositionStrategy.HIERARCHICAL:
            decomposer = HierarchicalTaskDecomposer()
            result = decomposer.decompose(task)
            return {"strategy": strategy, "result": decomposer.get_task_tree(result.id)}
        
        elif strategy == DecompositionStrategy.FUNCTIONAL:
            decomposer = FunctionalDecomposer()
            result = decomposer.decompose(task)
            return {"strategy": strategy, "modules": {k: v.name for k, v in result.items()}}
        
        elif strategy == DecompositionStrategy.DATA:
            decomposer = DataDecomposer()
            # 假设任务数据是列表
            result = decomposer.decompose_list(list(range(100)))
            return {"strategy": strategy, "chunks": len(result)}
        
        elif strategy == DecompositionStrategy.PIPELINE:
            # 创建简单流水线
            pipeline = PipelineBuilder(task).build()
            return {"strategy": strategy, "stages": len(pipeline.stages)}
        
        return {"strategy": strategy, "result": None}
    
    def _extract_keywords(self, task: str) -> List[str]:
        """提取关键词"""
        # 简单实现
        return task.split()[:5]
    
    def _estimate_complexity(self, task: str) -> float:
        """估计复杂度"""
        # 简单实现
        return min(len(task) / 100, 1.0)
    
    def _estimate_parallelizability(self, task: str) -> float:
        """估计可并行度"""
        # 简单实现
        return 0.5
    
    def _is_data_intensive(self, task: str) -> bool:
        """判断是否数据密集"""
        data_keywords = ["数据", "分析", "处理", "统计", "计算"]
        return any(keyword in task for keyword in data_keywords)

# 使用示例
decomposer = AdaptiveDecomposer()

tasks = [
    "开发一个用户管理系统",
    "分析100万条用户行为数据",
    "处理订单从创建到完成的流程",
    "设计一个推荐系统架构"
]

for task in tasks:
    print(f"\n{'='*60}")
    result = decomposer.decompose(task)
    print(f"分解结果: {result}")

6. 分解策略对比 ​

策略优点缺点适用场景
层次分解结构清晰、易于管理可能过度分解任务结构明确
功能分解模块独立、易于复用接口设计复杂功能系统开发
数据分解支持并行、高效数据一致性难保证数据处理任务
流水线分解流程清晰、易于优化灵活性低流程化任务
自适应分解灵活、智能实现复杂复杂多变任务

最佳实践 ​

1. 分解粒度控制 ​

python
def check_granularity(task: TaskNode, min_complexity: int = 1, 
                      max_complexity: int = 10) -> bool:
    """检查分解粒度"""
    complexity = estimate_complexity(task)
    
    if complexity < min_complexity:
        print(f"警告: 任务 '{task.name}' 分解过细")
        return False
    elif complexity > max_complexity:
        print(f"警告: 任务 '{task.name}' 需要进一步分解")
        return False
    
    return True

2. 依赖管理 ​

python
class DependencyManager:
    """依赖管理器"""
    
    def __init__(self):
        self.dependencies: Dict[str, List[str]] = {}
    
    def add_dependency(self, task_id: str, depends_on: str):
        """添加依赖"""
        if task_id not in self.dependencies:
            self.dependencies[task_id] = []
        self.dependencies[task_id].append(depends_on)
    
    def get_execution_order(self) -> List[str]:
        """获取执行顺序（拓扑排序）"""
        # 实现拓扑排序
        visited = set()
        order = []
        
        def dfs(node):
            if node in visited:
                return
            visited.add(node)
            for dep in self.dependencies.get(node, []):
                dfs(dep)
            order.append(node)
        
        for node in self.dependencies:
            dfs(node)
        
        return order

3. 质量检查 ​

python
class DecompositionQualityChecker:
    """分解质量检查器"""
    
    def check(self, root: TaskNode, tasks: Dict[str, TaskNode]) -> Dict[str, Any]:
        """检查分解质量"""
        checks = {
            "completeness": self._check_completeness(root, tasks),
            "independence": self._check_independence(tasks),
            "granularity": self._check_granularity(tasks),
            "coverage": self._check_coverage(root, tasks)
        }
        
        return checks
    
    def _check_completeness(self, root: TaskNode, tasks: Dict[str, TaskNode]) -> bool:
        """检查完整性"""
        # 检查是否所有叶子任务覆盖了原始任务
        return True
    
    def _check_independence(self, tasks: Dict[str, TaskNode]) -> float:
        """检查独立性"""
        # 检查子任务之间的依赖程度
        return 0.8
    
    def _check_granularity(self, tasks: Dict[str, TaskNode]) -> float:
        """检查粒度"""
        # 检查任务粒度是否合适
        return 0.9
    
    def _check_coverage(self, root: TaskNode, tasks: Dict[str, TaskNode]) -> float:
        """检查覆盖度"""
        # 检查任务覆盖度
        return 0.95

总结 ​

分解策略核心思想实现复杂度推荐场景
层次分解树状结构逐层细化低任务结构清晰
功能分解按功能模块划分中功能系统开发
数据分解按数据划分并行处理中数据密集任务
流水线分解按处理阶段划分低流程化任务
自适应分解智能选择策略高复杂多变任务

下一步学习 ​