《AI Agent 开发入门》11.1 任务分解：把大问题拆成子任务 DAG

检索 → 抽取A → 抽取B → 综合

# minagent/plan.py（任务类型）
from __future__ import annotations

from dataclasses import dataclass, field


@dataclass
class Task:
    id: str
    description: str
    depends_on: list[str] = field(default_factory=list)
    tool_hint: str | None = None


Plan = list[Task]

# minagent/plan.py（提示词常量）
把目标拆成子任务，只输出 JSON 数组。每项字段：
  id          短英文标识，全局唯一
  description 一句话说清这一步做什么
  depends_on  依赖的任务 id 列表，无依赖写 []
  tool_hint   建议使用的工具名（可选）
后置任务的 depends_on 必须列出它需要的前置任务 id。

# minagent/plan.py（解析）
import json


def parse_plan(text: str) -> Plan:
    raw = json.loads(text)
    return [
        Task(
            id=t["id"],
            description=t["description"],
            depends_on=list(t.get("depends_on", [])),
            tool_hint=t.get("tool_hint"),
        )
        for t in raw
    ]

# minagent/plan.py（校验）
def validate_plan(plan: Plan) -> list[str]:
    errors: list[str] = []
    ids = [t.id for t in plan]
    id_set = set(ids)
    dup = sorted({i for i in ids if ids.count(i) > 1})
    if dup:
        errors.append(f"重复的任务 id: {dup}")
    for t in plan:
        for d in t.depends_on:
            if d not in id_set:
                errors.append(f"任务 {t.id} 依赖不存在的 {d}")
            elif d == t.id:
                errors.append(f"任务 {t.id} 依赖自己")
    indeg = {t.id: 0 for t in plan}
    adj: dict[str, list[str]] = {t.id: [] for t in plan}
    for t in plan:
        for d in t.depends_on:
            if d in id_set and d != t.id:
                adj[d].append(t.id)
                indeg[t.id] += 1
    queue = [i for i, n in indeg.items() if n == 0]
    seen = 0
    while queue:
        cur = queue.pop()
        seen += 1
        for nxt in adj[cur]:
            indeg[nxt] -= 1
            if indeg[nxt] == 0:
                queue.append(nxt)
    if seen != len(plan):
        errors.append("计划存在环（无法拓扑排序）")
    return errors

# demo111.py
from minagent.fake import FakeModel
from minagent.plan import TASK_SCHEMA_PROMPT, parse_plan, topo_order, validate_plan
from minagent.types import Message

PLAN_JSON = """[
  {"id": "retrieve",   "description": "检索与主题相关的文档", "depends_on": [], "tool_hint": "search"},
  {"id": "extract_a",  "description": "从检索结果抽取方法 A 的要点", "depends_on": ["retrieve"]},
  {"id": "extract_b",  "description": "从检索结果抽取方法 B 的要点", "depends_on": ["retrieve"]},
  {"id": "synthesize", "description": "综合 A、B 要点写成对比结论", "depends_on": ["extract_a", "extract_b"]}
]"""

model = FakeModel([PLAN_JSON])
resp = model.complete([Message(role="user", content=TASK_SCHEMA_PROMPT + "\n目标：对比方法 A 与方法 B")], [])
plan = parse_plan(resp.message.content)

print("任务 DAG：")
for t in plan:
    deps = ", ".join(t.depends_on) or "(无)"
    hint = f"  [{t.tool_hint}]" if t.tool_hint else ""
    print(f"  {t.id:<10} <- {deps:<22} {t.description}{hint}")
print("校验:", validate_plan(plan) or "通过")
print("拓扑序:", " -> ".join(topo_order(plan)))

任务 DAG：
  retrieve   <- (无)                    检索与主题相关的文档  [search]
  extract_a  <- retrieve               从检索结果抽取方法 A 的要点
  extract_b  <- retrieve               从检索结果抽取方法 B 的要点
  synthesize <- extract_a, extract_b   综合 A、B 要点写成对比结论
校验: 通过
拓扑序: retrieve -> extract_a -> extract_b -> synthesize

BAD_JSON = """[
  {"id": "a", "description": "A", "depends_on": ["c"]},
  {"id": "b", "description": "B", "depends_on": ["a", "x"]},
  {"id": "c", "description": "C", "depends_on": ["b"]}
]"""
bad = parse_plan(BAD_JSON)
print("坏计划校验:", validate_plan(bad))

坏计划校验: ['任务 b 依赖不存在的 x', '计划存在环（无法拓扑排序）']