📌 前置知识：已完成第一课至第九课
🎯 本课目标：用依赖图组织多个原子动作，实现有顺序、可并行的复杂任务编排
💡 核心概念：依赖图（DAG）/ 依赖解析 / 拓扑排序 / 验证执行

前言

前九课，我们一步步搭出了一个能力越来越强的 Agent：

agent = Agent(model="qwen2.5:7b")

# 第八课：生成计划
plan = agent.create_plan("研究并写一篇 AI 技术博客")
# → {"steps": ["研究主题", "确定大纲", "写初稿", "审校"]}

# 第九课：把步骤变成原子动作
for step in plan["steps"]:
    atomic = agent.create_atomic_action(step)
    # "写初稿" → {"action": "generate_text", "inputs": {"topic": "...", "length": "..."}}

看起来完美。但有一个问题——

"审校"必须在"写初稿"之后，"写初稿"必须在"确定大纲"之后。但"研究主题"和"确定大纲"真的不能同时做吗？

{
    "steps": [
        "研究 AI Agent 最新进展",   ← 这两步其实可以并行
        "确定文章大纲",             ← 和上面这步
        "撰写初稿",                 ← 必须等上面两步都完成
        "审校修改"                  ← 必须等初稿完成
    ]
}

第八课的执行是简单遍历列表——从头到尾，一个接一个。没有处理步骤之间的依赖关系，更别提并行了。

这就像一个项目经理收到 10 个任务，不管谁先谁后、谁依赖谁，就让团队排成一列一个一个做。效率极低。

本课要解决这个问题。引入依赖图（DAG），让 Agent 不仅知道"要做什么"，还能推理出"谁先做、谁后做、哪些可以同时做"。

一、为什么需要依赖图？

1.1 顺序执行的局限

第八课的执行方式：

for step in plan["steps"]:
    execute(step)

这种方式假设所有步骤都必须按顺序执行。但现实中的任务往往不是这样的：

任务：搭建一个网站

  "注册域名"          ← 可以马上做
  "购买服务器"         ← 可以马上做，和注册域名并行
  "设计页面"          ← 可以马上做，和上面两个也并行
  "开发后端 API"      ← 需要等"购买服务器"
  "前端联调"          ← 需要等"开发后端 API" + "设计页面"
  "部署上线"          ← 需要等"前端联调"

如果按顺序做，6 个任务串行。但如果用依赖图，前 3 个可以同时开工，总时间缩短一半以上。

1.2 什么是 AoT（Atom of Thought）？

AoT（思维原子）就是把规划的结果表示为一张依赖图。

每个节点是一个原子动作（第九课），节点之间有明确的依赖关系：

{
    "nodes": [
        {"id": "1", "action": "research_ai_progress", "depends_on": []},
        {"id": "2", "action": "create_outline", "depends_on": []},
        {"id": "3", "action": "write_draft", "depends_on": ["1", "2"]},
        {"id": "4", "action": "review_edit", "depends_on": ["3"]}
    ]
}

注意看：

  • 节点 1 和节点 2 的 depends_on 是空的 → 可以并行执行
  • 节点 3 依赖 1 和 2 → 必须等 1 和 2 都完成
  • 节点 4 依赖 3 → 必须等 3 完成

这就是一张有向无环图（DAG）。"无环"意味着没有循环依赖——如果 A 等 B，B 等 A，那就死锁了。

1.3 第九课 vs 第十课

第九课：步骤列表 → 原子动作列表
  "写初稿" → {"action": "generate_text", "inputs": {...}}
  每个原子动作是独立的，不知道自己和其他动作的关系

第十课：目标 → 依赖图
  "写博客" → {
    nodes: [
      {action: "research", depends_on: []},
      {action: "outline", depends_on: []},
      {action: "draft", depends_on: ["research", "outline"]},
      {action: "review", depends_on: ["draft"]}
    ]
  }
  每个原子动作知道自己的依赖，整体形成一个执行图
第九课（原子动作） 第十课（AoT）
数据结构 动作列表（一维） 依赖图（二维）
步骤间关系 无（独立动作） 有（depends_on）
执行顺序 顺序遍历 拓扑排序 + 并行
能并行吗 不能 能（无依赖的节点并行）
能验证结构吗 能（schema） 能（schema + 依赖一致性）

第十课是对第九课的升级——同样的原子动作，但多了"谁先谁后"的关系信息。

二、核心概念

2.1 依赖图（DAG）

DAG = Directed Acyclic Graph = 有向无环图。

  • 有向：依赖关系有方向（A 依赖 B，不是 B 依赖 A）
  • 无环：没有循环依赖（A→B→C→A 这种不允许）
  • 图：节点 + 边，节点是动作，边是依赖关系
  节点1（研究）    节点2（大纲）
       ↘              ↙
        节点3（写初稿）
            ↓
        节点4（审校）

这张图清晰地表达了：研究和大纲可以并行做，写初稿必须等它们完成，审校最后做。

2.2 拓扑排序

拓扑排序 = 把有依赖关系的节点排成一个合法的执行顺序。

算法很简单：

  1. 找出所有没有依赖的节点（入度为 0）
  2. 执行它们
  3. 把它们从图中移除
  4. 重复，直到所有节点都执行完
第 1 轮：节点1、节点2（无依赖）→ 执行 → 移除
第 2 轮：节点3（依赖已满足）→ 执行 → 移除
第 3 轮：节点4（依赖已满足）→ 执行 → 移除
完成！

这就是本课 execute_graph() 的核心逻辑。

2.3 验证执行

在执行之前，必须验证图的结构完整性：

检查项 说明 不通过的后果
节点格式 每个节点必须有 id、action、depends_on 无法执行
ID 唯一 不能有两个节点用同一个 id 执行混乱
依赖有效 depends_on 里的 id 必须存在 死等
无环 不能有 A→B→A 这种循环 死锁

核心洞察：AoT 不是更聪明的思考，是更好的结构。依赖图让执行可预测、可验证、可调试。

三、代码实现

3.1 AoT 图生成器：agent/planner.py

在第九课的 planner.py 中新增 create_aot_graph() 函数：

def create_aot_graph(llm, goal: str) -> dict | None:
    """
    生成 AoT 执行图。

    用于：第十课

    Args:
        llm: 语言模型实例
        goal: 要实现的目标

    Returns:
        带 nodes 的 AoT 图，失败返回 None
    """
    from shared.utils import extract_json_from_text

    prompt = f"""为以下目标创建一个执行图。只返回有效的 JSON。

规则：
1. 只返回有效的 JSON
2. 不要任何解释，不要 Markdown
3. 直接以 {{ 开头，以 }} 结尾
4. 每个节点必须有 id、action、depends_on
5. depends_on 列表中的 id 必须是其他节点的 id
6. 不存在循环依赖
7. 没有依赖的节点可以并行执行

JSON 格式：
{{"nodes": [{{"id": "1", "action": "具体动作描述", "depends_on": []}}, {{"id": "2", "action": "具体动作描述", "depends_on": ["1"]}}]}}

目标：{goal}

请返回 JSON："""

    for attempt in range(3):
        response = llm.generate(prompt, temperature=0.0)
        graph = extract_json_from_text(response)

        if graph and "nodes" in graph and isinstance(graph["nodes"], list):
            # 验证节点结构
            node_ids = set()
            for node in graph["nodes"]:
                if "id" not in node or "action" not in node or "depends_on" not in node:
                    break
                node_ids.add(node["id"])
            else:
                # 检查依赖引用是否有效
                for node in graph["nodes"]:
                    for dep in node.get("depends_on", []):
                        if dep not in node_ids:
                            break
                    else:
                        continue
                    break
                else:
                    return graph

    return None

逐段拆解：

① Prompt 设计

核心变化在 JSON 格式——从第九课的 {"action": "...", "inputs": {...}} 变成了 {"nodes": [{"id": "1", "action": "...", "depends_on": []}]}。

多了两个关键字段：

  • id：节点标识符，用于建立依赖关系
  • depends_on：前置依赖列表，决定了执行顺序

加了一条新规则：“没有依赖的节点可以并行执行”。引导 LLM 思考哪些步骤可以同时做。

② 双层验证

# 第一层：节点结构验证
if "id" not in node or "action" not in node or "depends_on" not in node:
    break

# 第二层：依赖引用验证
for dep in node.get("depends_on", []):
    if dep not in node_ids:
        break

第一层检查每个节点是否完整（id、action、depends_on 三个字段）。
第二层检查 depends_on 里引用的节点 ID 是否真实存在。

两层都通过，才返回图。验证越来越严格——从第九课的"有 action 就行"，到第十课的"结构完整 + 引用有效"。

③ 还是老三样

  • extract_json_from_text() —— 老朋友
  • 重试 3 次 —— 老规矩
  • temperature=0.0 —— 图结构需要确定性

3.2 图执行器：execute_graph()

这是本课新增的核心函数：

def execute_graph(graph: dict, action_fn) -> list:
    """
    按依赖关系执行图。

    Args:
        graph: AoT 图
        action_fn: 执行单个动作的函数

    Returns:
        执行结果列表
    """
    nodes = graph["nodes"]
    node_map = {n["id"]: n for n in nodes}
    completed = set()
    results = []

    while len(completed) < len(nodes):
        # 找出所有依赖已满足的节点
        ready = [
            n for n in nodes
            if n["id"] not in completed
            and all(d in completed for d in n.get("depends_on", []))
        ]

        if not ready:
            # 没有可执行的节点，说明有循环依赖
            remaining = [n["id"] for n in nodes if n["id"] not in completed]
            print(f"  ⚠️ 检测到循环依赖，未完成节点：{remaining}")
            break

        for node in ready:
            result = action_fn(node["action"])
            results.append({
                "id": node["id"],
                "action": node["action"],
                "result": result,
                "status": "completed"
            })
            completed.add(node["id"])

    return results

这段代码就是拓扑排序的实现：

while 循环：
  1. 找出所有"依赖已满足"的节点 → ready
  2. 如果没有 ready 节点 → 循环依赖，终止
  3. 执行 ready 节点
  4. 标记为完成
  5. 重复，直到所有节点都完成

关键点：每一轮可以执行多个节点（ready 列表可能有多个）。 这就是并行执行的潜力——虽然本课用循环模拟，但逻辑上这些节点是"同时就绪"的。后续可以用多线程/异步真正并行。

3.3 Agent 中的 AoT 方法

在 agent/agent.py 中新增：

def create_aot_plan(self, goal: str) -> dict | None:
    """
    生成 AoT 执行图（第十课核心方法）。

    Args:
        goal: 要实现的目标

    Returns:
        带原子节点和依赖的 AoT 图
    """
    from agent.planner import create_aot_graph

    graph = create_aot_graph(self.llm, goal)

    if graph:
        total = len(graph["nodes"])
        parallel = sum(1 for n in graph["nodes"] if not n.get("depends_on"))
        print(f"🧠 AoT 图生成完成：{total} 个节点，{parallel} 个可并行")

    return graph

def execute_aot_plan(self, graph: dict) -> list:
    """
    尊重依赖关系执行 AoT 图（第十课核心方法）。

    Args:
        graph: AoT 图

    Returns:
        执行结果列表
    """
    def execute_action(action: str):
        print(f"  ✅ 执行：{action}")
        return f"Executed: {action}"

    return execute_graph(graph, execute_action)

注意设计细节：

① 委托给 planner.py

和第八课、第九课一样，planner.py 负责生成逻辑，Agent 类只做调度。职责分离贯穿始终。

② 生成时打印摘要

print(f"🧠 AoT 图生成完成：{total} 个节点，{parallel} 个可并行")

告诉你图有多大、有多少节点可以并行——一眼看出复杂度和并行潜力。

③ execute_graph() 是独立函数

图执行逻辑不在 Agent 类里，而是一个独立的 execute_graph() 函数。这是因为图的执行和 Agent 类本身无关——它接收一个图和一个执行函数，做拓扑排序。你可以把这个函数拿去用在任何地方。

四、运行示例

4.1 基础场景：生成并执行 AoT 图

from agent.agent import Agent

agent = Agent(model="qwen2.5:7b")

# 生成 AoT 图
graph = agent.create_aot_plan("研究并写一篇关于 AI Agent 的技术博客")

if graph:
    print("\n📋 执行图结构：\n")
    for node in graph["nodes"]:
        deps = node.get("depends_on", [])
        dep_str = f"（依赖：{', '.join(deps)}）" if deps else "（无依赖，可并行）"
        print(f"  [{node['id']}] {node['action']} {dep_str}")

    # 执行
    print("\n🚀 开始执行\n")
    results = agent.execute_aot_plan(graph)

    print(f"\n📊 执行完成，共 {len(results)} 个节点")

预期输出（类似）：

🧠 AoT 图生成完成：4 个节点，2 个可并行

📋 执行图结构：

  [1] 研究 AI Agent 最新技术和应用案例 （无依赖，可并行）
  [2] 确定文章结构和读者目标 （无依赖，可并行）
  [3] 撰写技术博客初稿 （依赖：1, 2）
  [4] 审校并优化文章质量 （依赖：3）

🚀 开始执行

  ✅ 执行：研究 AI Agent 最新技术和应用案例
  ✅ 执行：确定文章结构和读者目标
  ✅ 执行：撰写技术博客初稿
  ✅ 执行：审校并优化文章质量

📊 执行完成，共 4 个节点

注意看：节点 1 和节点 2 同时就绪，被连续执行（本课是顺序模拟，逻辑上可以并行）。节点 3 等它们完成后再执行。

4.2 复杂场景：多依赖关系

graph = agent.create_aot_plan("搭建一个电商网站")

if graph:
    print("\n📋 依赖关系图：\n")
    for node in graph["nodes"]:
        deps = node.get("depends_on", [])
        dep_str = f" → 等待 [{', '.join(deps)}]" if deps else " → 可立即执行"
        print(f"  [{node['id']}] {node['action']}{dep_str}")

可能的输出：

📋 依赖关系图：

  [1] 调研主流电商技术方案 → 可立即执行
  [2] 注册域名 → 可立即执行
  [3] 购买服务器 → 可立即执行
  [4] 设计数据库结构 → 等待 [1]
  [5] 开发后端 API → 等待 [3, 4]
  [6] 设计页面 UI → 等待 [1]
  [7] 开发前端页面 → 等待 [5, 6]
  [8] 部署上线 → 等待 [2, 7]

8 个节点，依赖关系清晰。节点 1、2、3 可以同时开工，节点 7 必须等 5 和 6 都完成，最后的节点 8 要等域名注册和前端开发都搞定。

这就是 AoT 的价值——一眼看出任务的并行潜力和关键路径。

4.3 验证：循环依赖检测

# 手动构造一个有循环依赖的图（演示用）
bad_graph = {
    "nodes": [
        {"id": "1", "action": "步骤A", "depends_on": ["2"]},
        {"id": "2", "action": "步骤B", "depends_on": ["1"]}
    ]
}

print("执行有循环依赖的图：")
results = agent.execute_aot_plan(bad_graph)
# → ⚠️ 检测到循环依赖，未完成节点：['1', '2']

execute_graph() 的安全机制——当没有可执行的节点时，检测到死锁并终止，不会无限等待。

4.4 交互模式

cd lesson10
python complete_example.py

会进入交互模式，你可以输入任意目标，观察 Agent 如何生成依赖图并按序执行。

五、与前九课的关系

5.1 第九课 → 第十课：从"独立动作"到"关系动作"

第九课：
  步骤1 → 原子动作1（独立）
  步骤2 → 原子动作2（独立）
  步骤3 → 原子动作3（独立）
  三个动作之间没有关系

第十课：
  节点1（研究）  ← 无依赖
  节点2（大纲）  ← 无依赖
  节点3（初稿）  ← 依赖 1、2
  节点4（审校）  ← 依赖 3
  四个动作之间有明确的先后关系
第九课（原子动作） 第十课（AoT）
输入 单个步骤字符串 整体目标
输出 单个动作 JSON 完整依赖图
动作间关系 无 有（depends_on）
执行方式 逐个处理 拓扑排序
能并行吗 ❌ ✅
结构验证 action + inputs action + id + depends_on + 引用有效

5.2 AoT 不是凭空冒出来的

回顾前九课，AoT 的每个概念都已经有铺垫：

第三课：结构化输出（JSON 解析 + 验证）    ← AoT 的图结构就是 JSON
第五课：工具调用（action + 参数）           ← AoT 的节点就是 action
第八课：规划（目标 → 步骤列表）             ← AoT 把步骤列表升级为图
第九课：原子动作（步骤 → 带参数的动作）     ← AoT 的节点就是原子动作
第十课：AoT（动作 + 依赖关系）             ← 把上面所有能力串起来

核心洞察：AoT 是前面所有课程的自然组合。它不是什么高级算法，就是"原子动作 + 依赖关系"——第九课的产物加上了第八课的规划能力，再加上一个图结构。

六、关键洞察

6.1 AoT 是结构，不是魔法

很多人把 Agent 的"推理能力"想得很神秘。但看我们的实现：

create_aot_graph() = LLM + Prompt + extract_json_from_text()
execute_graph()    = while 循环 + 集合操作

没有推理引擎，没有搜索算法，没有神经网络。 就是一个让 LLM 输出 JSON 图结构，然后用最基础的集合操作做拓扑排序。

Agent 系统不是"思维"，是"结构"。好的结构让执行可预测、可验证、可调试。

6.2 依赖关系是效率的关键

同样的 8 个任务：

顺序执行：task1 → task2 → ... → task8 = 8 个时间单位

依赖图执行：
  第 1 轮：task1 + task2 + task3（并行）= 1 个时间单位
  第 2 轮：task4 + task6（并行）         = 1 个时间单位
  第 3 轮：task5                        = 1 个时间单位
  第 4 轮：task7                        = 1 个时间单位
  第 5 轮：task8                        = 1 个时间单位
  总计 = 5 个时间单位（节省 37.5%）

依赖关系不仅是"顺序保证"，更是并行优化的前提。

6.3 验证在执行之前（再次强调）

第九课的核心原则在第十课被加强——执行前验证图结构：

验证项：
  ✅ 每个节点有 id、action、depends_on
  ✅ 所有 id 唯一
  ✅ depends_on 引用的节点存在
  ✅ 无循环依赖（通过 execute_graph 的终止条件间接保证）

任何一条不满足，图就不执行。在错误发生之前拦截，而不是在执行过程中崩掉。

6.4 构建块可以组合出复杂系统

单个原子动作很简单。但加上依赖关系之后，它们能组合出任意复杂的工作流：

简单的：A → B → C          （线性）
中等的：A + B → C → D      （并行 + 串行）
复杂的：A → B → C          （多分支）
          ↘ D ↗
          E → F

复杂度不来自单个动作的复杂度，而来自动作之间的组合方式。 这就是"简单组件 + 明确规则 = 复杂系统"的工程哲学。

七、常见问题

Q：循环依赖怎么检测？

A：本课的 execute_graph() 通过终止条件间接检测——当没有可执行节点但还有未完成节点时，说明存在循环依赖。更严格的实现可以用 DFS 染色法在建图时就检测环。本课先保持简单，后续可以加强。

Q：depends_on 里的 ID 不存在怎么办？

A：create_aot_graph() 的第二层验证已经处理了这个问题——检查每个依赖引用的 ID 是否在 node_ids 集合中。如果不存在，验证失败，重试或返回 None。

Q：执行顺序看起来不对怎么办？

A：检查两件事。第一，LLM 生成的 depends_on 关系是否正确（它可能错误地认为 A 依赖 B）。第二，execute_graph() 的拓扑排序是否正确。可以通过打印 ready 列表来调试——每轮打印哪些节点就绪了。

Q：可以真正并行执行吗？

A：本课的 execute_graph() 用 for node in ready 循环模拟并行，实际是顺序执行。要真正并行，可以用 Python 的 concurrent.futures.ThreadPoolExecutor 或 asyncio，把 for 换成线程池提交即可。逻辑不变，只是执行方式变了。

Q：AoT 和第五课的工具调用有什么区别？

A：第五课是 Agent 运行时动态选择调用什么工具，AoT 是规划阶段预先确定所有动作和依赖。第五课是"边想边做"，AoT 是"先规划好整个流程再做"。两者可以结合——AoT 规划出整体结构，执行时每个节点内部用第五课的工具调用能力。

八、十课演进线

把前十课放在一起看，一个完整的 Agent 系统的构建路径：

第一课：能对话了（LLM + Ollama）
第二课：有角色了 + 能多轮对话（System Prompt + History）
第三课：能输出 JSON 了（结构化输出 + 验证 + 重试）
第四课：能做选择了（意图理解 → 动作路由）
第五课：能调用工具了（工具选择 + 参数提取 + 安全执行）
第六课：能循环了（Agent Loop + 状态追踪 + 终止条件）
第七课：能记住了（跨对话记忆存储 + 检索 + 显式管理）
第八课：能规划了（计划生成 → 验证 → 逐步执行）
第九课：能拆解了（模糊步骤 → 原子动作 → Schema 验证）
第十课：能编排了（依赖图 + 拓扑排序 + 并行潜力）

从"裸模型"到"有角色、有工具、有循环、有记忆、有规划、有编排"的智能体。

十个课，没有一行代码是"魔法"。 全部都是：LLM + 结构化输出 + 验证 + 明确的执行逻辑。

九、系列总结

到这一课，我们的 Agent 已经具备了所有核心能力：

能力 实现课程 核心机制
对话 第一、二课 LLM + System Prompt + 对话历史
结构化输出 第三课 JSON + extract_json + 重试
决策 第四课 意图分类 + 动作路由
工具调用 第五课 工具选择 + 参数提取 + 安全执行
循环执行 第六课 Agent Loop + 状态追踪 + 终止条件
记忆 第七课 存储 + 检索 + 显式管理
规划 第八课 目标 → 步骤列表 → 验证 → 执行
原子动作 第九课 模糊步骤 → 带参数的动作 → Schema 验证
任务编排 第十课 依赖图 + 拓扑排序 + 并行执行

你可能注意到了：每个能力的底层都是同一套模式——让 LLM 输出结构化数据，然后你的代码来验证和处理。

这不是巧合。Agent 系统的核心不是"让 AI 更聪明"，而是把 AI 的能力嵌入到可靠的工程结构中。

LLM 负责理解和生成，你的代码负责验证和执行。

这就是"手搓 AI Agent"的全貌——没有黑盒，没有魔法，每一行代码你都理解。

完整代码获取

本课涉及的完整代码包括：

  • agent/planner.py——规划器模块（新增 create_aot_graph() 函数）
  • agent/agent.py——Agent 类（新增 create_aot_plan() 和 execute_aot_plan()）
  • complete_example.py——演示模式（4 个场景）+ 交互模式

标签

#Python #AI Agent #LLM #AoT #依赖图 #DAG #Ollama #Qwen #大模型 #手搓Agent

本文为《手搓 AI Agent 从 0 到 1》系列教程第 10 课

# 人工智能 # python # 自然语言处理
Logo
https://edu.csdn.net/learn/39067/627173?utm_source=2019755004

汇聚全球AI编程工具，助力开发者即刻编程。

更多推荐

初识智能体

在过去两年中，几乎所有人都体验过大语言模型（如 ChatGPT、Claude、DeepSeek）的强大能力：写周报、润色邮件、解答编程疑难。被动响应：你问一句，它答一句。如果你的 Prompt 没有写清楚约束条件，它无法主动追问或修正目标。知行脱节：大模型能用流畅的文字告诉你“如何部署一个高可用 K8s 集群”，但它无法替你登录服务器敲下命令、查看报错日志并自动回滚。信息孤岛：大模型的知识止步于预

avatar AI编程社区
cover

Windows 下部署 New API 中转站全过程实录（含 6 个真实踩坑）

avatar AI编程社区
cover

Agent 敢开写权限吗？Python 文件写入沙箱守卫实战

avatar AI编程社区