任务分解与依赖图

更新 2026-10-06

正在读取学习记录…

任务分解是把大目标拆成有明确输入和产物的小任务。依赖图用连线说明：某个任务必须等另一项产物到位，才能开始。节点表示任务，箭头表示“后者需要前者的输出”。

分解要落到具体工作：读取订单、找到退款规则、判断资格、整理说明。每一步都有输入、输出和完成条件，交出的东西能被下一步使用。一个子任务只叫“思考得更深入”，安排起来和喊“加油”差不多。

退款说明为什么不能随便并行

客户想知道一笔120元、尚未发货的订单能否退款。读取订单和读取规则可以同时进行：一项得到付款与发货状态，另一项得到允许退款的条件。

资格判断需要两份资料，不能在订单还没查完时先宣布“可以退款”。生成草稿又需要资格判断的结论。由此得到三批任务：先读两份资料，再判断资格，最后生成草稿。

这种依赖结构常叫DAG，中文是有向无环图。“有向”表示箭头有方向：订单指向资格判断，意思是判断要使用订单资料。“无环”表示沿箭头走下去，走不回原来的任务。若订单等待草稿，而草稿又要等订单资料，任务就绕成了环。双方可以一起开会，却没有谁能先拿出结果。

程序会计算批次，再按批次顺序执行。批次内用循环读固定数据；同批可并行表示依赖允许并行，运行代码没有启用线程。

保存完整程序并运行

在自己的学习目录打开PowerShell，创建一个新的子目录：

mkdir l6-decomposition
cd l6-decomposition

把下面完整代码保存为 decomposition_demo.py。固定数据和运行入口都已包含，文件名保留 .py 后缀。

"""从明确依赖生成批次，并传递真实子任务产物。"""
DEPENDENCIES = {
    "order": set(),
    "policy": set(),
    "eligibility": {"order", "policy"},
    "draft": {"eligibility"},
}

def batches(dependencies):
    names = set(dependencies)
    if any(not deps <= names for deps in dependencies.values()):
        raise ValueError("存在未定义的依赖")
    remaining = {name: set(deps) for name, deps in dependencies.items()}
    done, result = set(), []
    while remaining:
        # 只有前面批次已产出的资料，才能作为当前批次的输入。
        ready = sorted(name for name, deps in remaining.items() if deps <= done)
        if not ready:
            raise ValueError("循环依赖，任务无法启动")
        result.append(ready)
        done.update(ready)
        for name in ready:
            del remaining[name]
    return result

def perform(name, inputs):
    if name == "order":
        return {"paid": 120, "shipped": False}
    if name == "policy":
        return {"refund_before_shipping": True}
    if name == "eligibility":
        allowed = inputs["policy"]["refund_before_shipping"] and not inputs["order"]["shipped"]
        return {"allowed": allowed, "amount": inputs["order"]["paid"] if allowed else 0}
    if name == "draft":
        value = inputs["eligibility"]
        return f"可申请退款{value['amount']}元" if value["allowed"] else "请人工核查"
    raise ValueError("没有对应执行器")

def execute(dependencies):
    results, trace = {}, []
    for group in batches(dependencies):
        produced = {}
        for name in group:
            inputs = {dep: results[dep] for dep in dependencies[name]}
            produced[name] = perform(name, inputs)
            trace.append(name)
        # 同批任务完成后统一发布，避免偷偷读取同批未完成的输出。
        results.update(produced)
    return results, trace

def main():
    for index, group in enumerate(batches(DEPENDENCIES), 1):
        print(f"批次{index}：" + "、".join(group))
    results, _ = execute(DEPENDENCIES)
    print("eligibility产物：" + str(results["eligibility"]))
    print("最终草稿：" + results["draft"])
    broken = {"order": {"draft"}, "draft": {"order"}}
    try:
        batches(broken)
    except ValueError as error:
        print("错误计划：" + str(error))

if __name__ == "__main__":
    main()

保存后执行 dir，确认能看到 decomposition_demo.py，再运行：

ls
python -X utf8 decomposition_demo.py

程序输出：

批次1：order、policy
批次2：eligibility
批次3：draft
eligibility产物：{'allowed': True, 'amount': 120}
最终草稿：可申请退款120元
错误计划：循环依赖，任务无法启动

输出中的三批任务如何形成

batches先检查每个依赖名称是否存在。然后从未安排的任务中找出依赖全部属于done的节点。第一轮done为空，order和policy都没有依赖，因此同时进入批次1。

这两项安排完成后，done包含order与policy，eligibility符合条件，进入批次2。draft等待eligibility，进入批次3。每批用sorted排列，保证同一份计划每次得到稳定顺序。

沿着程序里的任务名，可以看到每份资料怎样传到下一步：

批次与任务	开始前需要什么	交给下一步的结果
1：order，读取订单	固定订单数据	paid为120，表示已付120元；shipped为False，表示尚未发货
1：policy，读取规则	固定退款规则	refund_before_shipping为True，表示发货前允许退款
2：eligibility，判断资格	上面的订单与规则两份结果	allowed为True，表示符合条件；amount为120，表示候选退款金额
3：draft，生成草稿	资格判断的结果	“可申请退款120元”

perform 返回这些实际数据，execute 把结果交给后续任务。在 perform 的订单返回值中，把 "paid": 120 改为 "paid": 80，保存并运行 python -X utf8 decomposition_demo.py。资格判断的amount会变成80，“最终草稿”那一行会变成“可申请退款80元”，三批任务的顺序保持相同。然后把80恢复为120。这样能同时观察数据变化和依赖顺序。

上方订单与规则独立读取，再判断资格并生成草稿；下方新增草稿到订单的依赖，形成订单、资格、草稿互相等待的环 — 上方是正常的三批执行；下方对应后面给order增加draft依赖的改参，形成三节点环。main里的broken则是另一个更小的两节点环，两者都无法启动。

分解到什么粒度才有价值

order产出两项业务事实；policy产出条件；eligibility产出判断和候选金额；draft产出用户可读文字。这四个产物之间有清楚的数据关系，所以能测试某个上游字段改变后，下游是否改变。

粒度过粗时，一个执行者同时查十种资料，难以定位遗漏；粒度过细时，每个字段都派一个Agent，通信与汇总可能比查询本身更贵。可以先按独立数据源、独立权限或可单独核对的判断拆开，再观察调用次数与等待时间。

依赖图还要区分“读取数据”和“修改外部系统”。生成退款候选不等于付款动作。增加付款节点时，它应依赖资格、审批和幂等标识等具体产物；不能只因为草稿任务完成就执行。

动态计划也需要校验。模型可能生成不存在的节点名、重复编号、错误依赖或环。把模型建议的计划转换为程序认识的任务契约后，再安排执行。图能排出来，也不代表任务权限成立，两项应分别检查。

完成一个节点与完成整件事

execute按每批统一发布输出，后续只能读已经发布的结果。这避免同批任务偷读对方尚未完成的资料。真正并行执行时，仍应等待这一批需要的任务结束，或按各个依赖就绪程度继续推进。

如果订单读取失败，资格任务应保持未就绪或进入受控失败处理。把失败包装为空订单再继续计算，会把“没有查到资料”变成一个看似正常的业务结论。任务状态和业务产物应分别存储。

给依赖加一条边，再运行

在DEPENDENCIES中把policy的set()改为{"order"}，运行python -X utf8 decomposition_demo.py。批次从三批变成四批：order、policy、eligibility、draft。这条边表示规则读取也必须等订单结果；即使两项原本可以独立读取，新增依赖也会延长流程。

把order的set()改为{"draft"}再运行，程序会报“循环依赖，任务无法启动”，因为order等待draft，而draft通过eligibility又等待order。删除该边后恢复正常；增加线程不会解开逻辑上的环。

完成后将order和policy两项都恢复为set()，保存并重跑，确认重新得到三批任务。否则前一项新增依赖仍会留下四批结果，与原始输出和配套检查不同。

修改规则条件时，直接观察eligibility的allowed和amount，再看草稿。先查上游产物，再看下游结论，通常比盯着最终文字猜哪位Agent出错更有效。

参考阅读：Python graphlib中的TopologicalSorter与CycleError，比较就绪节点和已完成节点；LangGraph Graph API阅读节点、边与状态更新的关系。

运行配套检查

把下面测试完整保存为 test_decomposition_demo.py，与主程序放在同一目录。6项检查覆盖批次、真实数据传递、循环、缺失节点、空图和发货状态变化。

查看测试代码：test_decomposition_demo.py

"""验证拓扑安排与数据依赖，而不是角色数量。"""
import sys
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent))
import decomposition_demo as demo

class DecompositionTests(unittest.TestCase):
    def test_independent_reads_share_batch(self):
        self.assertEqual(demo.batches(demo.DEPENDENCIES),
                         [["order", "policy"], ["eligibility"], ["draft"]])

    def test_results_feed_later_task(self):
        results, trace = demo.execute(demo.DEPENDENCIES)
        self.assertEqual(results["eligibility"], {"allowed": True, "amount": 120})
        self.assertEqual(results["draft"], "可申请退款120元")
        self.assertLess(trace.index("eligibility"), trace.index("draft"))

    def test_cycle_rejected(self):
        with self.assertRaisesRegex(ValueError, "循环"):
            demo.batches({"A": {"B"}, "B": {"A"}})

    def test_missing_dependency_rejected(self):
        with self.assertRaisesRegex(ValueError, "未定义"):
            demo.batches({"A": {"missing"}})

    def test_empty_graph(self):
        self.assertEqual(demo.execute({}), ({}, []))

    def test_rule_changes_downstream_result(self):
        result = demo.perform("eligibility", {
            "policy": {"refund_before_shipping": True},
            "order": {"paid": 120, "shipped": True},
        })
        self.assertEqual(demo.perform("draft", {"eligibility": result}), "请人工核查")

if __name__ == "__main__":
    unittest.main()

python -X utf8 -m unittest -v test_decomposition_demo.py

测试名称后应显示 ok，末尾显示 Ran 6 tests 和 OK。若找不到模块，先用 dir 确认两个文件保存在同一目录。

课后练习展开习题收起习题

正在读取本地记录…

不调用模型，不执行代码。简答、代码和场景题仅保存答案并展示自查标准，不生成评分、徽章或认证。输入内容不会自动保存，请点击提交；不要填写个人信息或密钥。

练习 1 / 4 · v1.1.0

选择题 · 查订单A和查政策B互不依赖，计算C需要两份结果。首批可执行哪些任务

查订单A和查政策B互不依赖，计算C需要两份结果。首批可执行哪些任务？

对应知识点：L6-decomposition

选择一个答案A. 只有CB. A、B、C一起C. A和B

练习 2 / 4 · v1.1.0

简答题 · 解释机制

依赖图中的箭头A→C表示什么？

对应知识点：L6-decomposition

你的回答

练习 3 / 4 · v1.1.0

代码题 · 核对给定数据

已完成集合是{'A'}，C依赖{'A','B'}。写出判断C是否可运行的表达式。

对应知识点：L6-decomposition

你的代码或伪代码

练习 4 / 4 · v1.1.0

场景题 · 分析具体变化

A等待B、B又等待A时，程序应怎样解释没有可执行任务？

对应知识点：L6-decomposition

你的回答