如何用Python实现未知依赖关系的任务分配与工作流自动化？

一、核心问题拆解 #

你的场景需要解决两个关键问题：

1. 任务分配：将m个任务（m>n）分配给n个员工，保证负载均衡
2. 工作流编排：识别未知的任务依赖关系，生成合法的任务执行顺序

二、分步解决方案及Python实现 #

1. 任务分配策略 #

因为员工数少于任务数，优先采用负载均衡分配逻辑：按任务数量均分，剩余任务依次分配给前几个员工，确保每个员工的任务量尽可能接近。

Python实现示例

def assign_tasks(tasks, employees):
    # tasks：任务列表，每个元素含task_id、description字段
    # employees：员工ID列表，如["e1", "e2", "e3"]
    task_count = len(tasks)
    emp_count = len(employees)
    
    base_tasks = task_count // emp_count
    remainder = task_count % emp_count
    
    assignment = {emp: [] for emp in employees}
    task_idx = 0
    
    # 先给前remainder个员工多分配1个任务
    for emp in employees[:remainder]:
        assignment[emp] = tasks[task_idx:task_idx + base_tasks + 1]
        task_idx += base_tasks + 1
    
    # 给剩余员工分配基础任务量
    for emp in employees[remainder:]:
        assignment[emp] = tasks[task_idx:task_idx + base_tasks]
        task_idx += base_tasks
    
    # 输出格式化分配结果
    print("任务分配结果：")
    print("| Task ID | Task Description | Employee |")
    print("|---------|------------------|----------|")
    for emp, emp_tasks in assignment.items():
        for task in emp_tasks:
            print(f"| {task['task_id']} | {task['description']} | {emp} |")
    return assignment

# 测试数据
tasks = [
    {"task_id": 1, "description": "draw a apple"},
    {"task_id": 2, "description": "draw a banana"},
    {"task_id": 3, "description": "draw a orange"},
    {"task_id": 4, "description": "edit fruit images"},
    {"task_id": 5, "description": "publish fruit artwork"}
]
employees = ["e1", "e2", "e3"]

assign_tasks(tasks, employees)

2. 未知依赖识别与工作流编排 #

依赖未知时，先通过任务描述语义分析提取关联逻辑，再用有向无环图(DAG) 建模依赖关系，最后通过拓扑排序生成合法执行顺序。

步骤说明：

1. 语义分析：用NLP工具识别任务描述中的动作关联（如"绘制"任务是"编辑"任务的前置条件）
2. DAG建模：节点为任务ID，有向边代表依赖关系（如Task2 → Task4表示Task2完成后才能执行Task4）
3. 拓扑排序：基于Kahn算法（入度表+队列）生成无环执行顺序，避免循环依赖

Python实现示例

from collections import deque

def extract_dependencies(tasks):
    # 初始化入度表和邻接表
    in_degree = {task["task_id"]: 0 for task in tasks}
    adj = {task["task_id"]: [] for task in tasks}
    
    # 示例逻辑：基于任务描述关键词识别依赖
    draw_tasks = [t["task_id"] for t in tasks if "draw" in t["description"]]
    edit_tasks = [t["task_id"] for t in tasks if "edit" in t["description"]]
    publish_tasks = [t["task_id"] for t in tasks if "publish" in t["description"]]
    
    # 模拟示例中的Task2 -> Task1、Task3
    for task_id in draw_tasks:
        if task_id != 2:
            adj[2].append(task_id)
            in_degree[task_id] += 1
    
    # 编辑任务依赖所有绘制任务
    for edit_id in edit_tasks:
        for draw_id in draw_tasks:
            adj[draw_id].append(edit_id)
            in_degree[edit_id] += 1
    
    # 发布任务依赖编辑任务
    for publish_id in publish_tasks:
        for edit_id in edit_tasks:
            adj[edit_id].append(publish_id)
            in_degree[publish_id] += 1
    
    return adj, in_degree

def topological_sort(adj, in_degree):
    queue = deque()
    # 入度为0的任务优先入队
    for task_id, degree in in_degree.items():
        if degree == 0:
            queue.append(task_id)
    
    execution_order = []
    while queue:
        current_task = queue.popleft()
        execution_order.append(current_task)
        
        # 更新邻接任务的入度
        for neighbor in adj[current_task]:
            in_degree[neighbor] -= 1
            if in_degree[neighbor] == 0:
                queue.append(neighbor)
    
    # 检查循环依赖
    if len(execution_order) != len(in_degree):
        raise ValueError("任务间存在循环依赖，无法生成合法执行顺序")
    
    return execution_order

# 测试依赖提取与拓扑排序
adj, in_degree = extract_dependencies(tasks)
execution_order = topological_sort(adj, in_degree)
print("\n全局任务执行顺序：", execution_order)

3. 全流程自动化整合 #

将任务分配与工作流编排结合，输出每个员工的专属任务执行顺序：

def full_workflow_automation(tasks, employees):
    # 1. 任务分配
    assignment = assign_tasks(tasks, employees)
    # 2. 提取依赖关系
    adj, in_degree = extract_dependencies(tasks)
    # 3. 生成全局执行顺序
    global_order = topological_sort(adj, in_degree)
    # 4. 生成每个员工的任务执行顺序
    emp_task_order = {}
    for emp in employees:
        emp_task_ids = [t["task_id"] for t in assignment[emp]]
        emp_task_order[emp] = [tid for tid in global_order if tid in emp_task_ids]
    
    print("\n各员工任务执行顺序：")
    for emp, order in emp_task_order.items():
        print(f"{emp}: {order}")
    return assignment, emp_task_order

# 执行全流程
full_workflow_automation(tasks, employees)

三、关键算法说明 #

• 负载均衡分配：通过均分+余数分配保证员工任务量均衡，适合无优先级的任务分配场景
• 有向无环图(DAG)：用于建模任务依赖关系，从根源避免循环依赖问题
• 拓扑排序（Kahn算法）：基于入度表和队列实现，高效生成符合依赖规则的任务执行顺序

内容的提问来源于stack exchange，提问作者Josh-money