任务分解与依赖建模

多 Agent 协作的第一步是将一个复杂的大任务拆解为多个可独立执行的子任务，并明确子任务之间的数据与控制依赖关系。只有通过严谨的任务分解和依赖建模，才能构建出无环、可拓扑排序的执行计划，为后续的 Agent 分配与调度奠定基础。

大任务拆分方法

常用的拆分策略有三种，每种方法适用于不同的业务场景和系统架构约束。

任务依赖类型

子任务之间常见的依赖关系分为三类，正确识别依赖类型是构建依赖图的前提。

• 数据依赖（Data Dependency）： 一个任务的输出是另一个任务的输入。最常见的依赖类型，必须严格按顺序执行。
• 控制依赖（Control Dependency）： 一个任务的完成状态（成功/失败）决定另一个任务是否执行。用于分支、条件触发和异常处理场景。
• 资源依赖（Resource Dependency）： 多个任务共享同一独占资源（GPU、API 配额、文件锁）。虽无数据传递，但需串行化访问。

任务依赖图构建

构建任务依赖图（Task Dependency Graph, TDG）一般采用以下四步流程：

1. 列出所有子任务： 基于拆分方法生成任务清单，每个任务有唯一标识。
2. 标注输入/输出： 明确每个任务需要哪些数据、产生哪些数据。
3. 建立连接边： 若任务 A 的输出是任务 B 的输入，则添加有向边 A→B。
4. 检查循环依赖： 使用 DFS 或拓扑排序检测是否存在环，若存在则必须重新拆分或合并任务。

DAG 表示与拓扑排序

任务依赖图最终被建模为有向无环图（DAG）。在 DAG 中：

• 节点 表示子任务，包含执行参数和资源需求。
• 有向边 表示依赖方向，从前置任务指向后置任务。
• 无环 确保存在至少一种线性执行顺序。

拓扑排序 是将 DAG 中的节点排列成线性序列，使得对于每一条有向边 u→v，u 在序列中都出现在 v 之前。常用的算法是 Kahn 算法（基于入度队列）。

代码实现：DAG 构建与拓扑排序

下面使用 Python 实现一个通用的任务 DAG 类，并基于 Kahn 算法计算拓扑排序。该类可用于多 Agent 编排引擎的前置依赖分析环节。

from collections import defaultdict, deque

class TaskDAG:
    """任务依赖有向无环图"""

    def __init__(self):
        self.graph = defaultdict(list)
        self.indegree = defaultdict(int)
        self.tasks = set()

    def add_task(self, task_id: str):
        """注册任务节点"""
        self.tasks.add(task_id)
        if task_id not in self.indegree:
            self.indegree[task_id] = 0

    def add_dependency(self, from_task: str, to_task: str):
        """
        添加依赖边 from_task → to_task
        表示 to_task 依赖 from_task
        """
        self.graph[from_task].append(to_task)
        self.indegree[to_task] += 1
        self.tasks.add(from_task)
        self.tasks.add(to_task)

    def topological_sort(self) -> list:
        """
        Kahn 算法实现拓扑排序
        返回任务的线性执行顺序
        """
        queue = deque([t for t in self.tasks if self.indegree[t] == 0])
        result = []
        while queue:
            task = queue.popleft()
            result.append(task)
            for neighbor in self.graph[task]:
                self.indegree[neighbor] -= 1
                if self.indegree[neighbor] == 0:
                    queue.append(neighbor)
        if len(result) != len(self.tasks):
            raise ValueError("图中存在环，无法进行拓扑排序")
        return result

# 示例：构建文档处理任务的依赖图
dag = TaskDAG()
dag.add_task("文档加载")
dag.add_task("文本清洗")
dag.add_task("实体识别")
dag.add_task("关系抽取")
dag.add_task("知识存储")
dag.add_task("报表生成")

dag.add_dependency("文档加载", "文本清洗")
dag.add_dependency("文本清洗", "实体识别")
dag.add_dependency("文本清洗", "关系抽取")
dag.add_dependency("实体识别", "知识存储")
dag.add_dependency("关系抽取", "知识存储")
dag.add_dependency("知识存储", "报表生成")

order = dag.topological_sort()
print("执行顺序:", " → ".join(order))

执行顺序: 文档加载 → 文本清洗 → 实体识别 → 关系抽取 → 知识存储 → 报表生成

配置化依赖描述

在实际的多 Agent 系统中，任务依赖通常以 JSON 或 YAML 格式描述，便于编排引擎动态解析。以下是一段符合 DAG 模型的 JSON 配置示例：

{
  "tasks": [
    {"id": "data_fetch",    "agent": "crawler",   "deps": []},
    {"id": "text_clean",    "agent": "cleaner",   "deps": ["data_fetch"]},
    {"id": "ner",           "agent": "extractor", "deps": ["text_clean"]},
    {"id": "relation_ext",  "agent": "extractor", "deps": ["text_clean"]},
    {"id": "knowledge_store","agent": "storage",   "deps": ["ner", "relation_ext"]},
    {"id": "report_gen",    "agent": "reporter",  "deps": ["knowledge_store"]}
  ]
}

编排引擎读取该配置后，自动构建 DAG 并执行拓扑排序，将无依赖的任务优先分配给空闲 Agent，实现高效的并行调度。

实践要点与常见问题

• 粒度控制： 子任务粒度过细会导致 DAG 节点膨胀，增加调度开销；粒度过粗则丧失并行潜力。建议每个子任务执行耗时在 100ms～10s 之间。
• 环检测前置： 在任务提交阶段就进行环检测（如使用 DFS 回溯标记），避免运行时出现死锁。
• 动态依赖： 某些任务在执行过程中才能确定后续依赖（如条件分支），需要编排引擎支持运行时图修改。
• 监控与追溯： 为每个节点记录执行状态、开始时间、结束时间，便于生成 DAG 执行轨迹图，用于性能分析和故障定位。