《AI Agent 开发入门》11.1 任务分解:把大问题拆成子任务 DAG
把 5.2 的线性计划升级为带依赖的 DAG:定义 Task 与 Plan,给出让模型输出任务图的 JSON 协议,实现 validate_plan 检测环与悬空依赖,并用假模型脚本化「检索→并行抽取→综合」的 4 任务 DAG,真跑打印 DAG 与拓扑序,最后给出线性计划与 DAG 计划的取舍表。
category
把 5.2 的线性计划升级为带依赖的 DAG:定义 Task 与 Plan,给出让模型输出任务图的 JSON 协议,实现 validate_plan 检测环与悬空依赖,并用假模型脚本化「检索→并行抽取→综合」的 4 任务 DAG,真跑打印 DAG 与拓扑序,最后给出线性计划与 DAG 计划的取舍表。
前两节讲了 Claude Agent SDK 怎么用,这一节算总账。把它的每项能力逐条映射回卷一 minagent 的自研实现,看清它内置省下的四块重活(subagent 调度、hook、权限、session 持久化),也看清它锁定的边界:依赖 claude CLI、绑定 Anthropic 模型、可观测与扩展受限、假模型测试被掐断。文末给对照表与选型清单。
上一节把权限一笔带过,本节拆开讲透。Claude Agent SDK 用四道闸门控制工具:tools 决定有哪些工具、allowed_tools / disallowed_tools 做静态白黑名单、permission_mode 定档位、can_use_tool 回调做动态人工审批。全部字段与签名现场核对 0.2.165,并给出「只读研究助手 + 写操作二次确认」的完整骨架。
Claude Agent SDK 不是模型 SDK,而是把 Claude Code 的 Agent 运行时(subagent / hooks / 权限 / session)打包成 Python 接口。本节现场核对 0.2.165 的 API,讲清 AgentDefinition、HookMatcher 与 session,并搭出多 subagent 加 hook 的研究任务骨架。
Agent 会执行模型写出来的代码、SQL 与命令,必须隔离。本节讲三层隔离:进程内不可信、子进程加资源限制、容器加只读挂载与无网络。实现 sandboxed_python 真跑正常代码、死循环超时、写文件三例,如实说明 subprocess 不是安全边界,再用 Docker 只读无网络容器真跑对照。
工具设计有五条可复用模式:读写分离(只读随便调、写入要审批加幂等键)、幂等、可观测、单一粒度、描述即提示词。本节先用假模型做「描述含糊 vs 描述具体」的选中率对照(实测 67% 对 100%),再用 pydantic 约束对照万能 sql 与专用工具的危险调用拦截率(实测 0/6 对 5/6)。
本节把 MCP Server 当成工具源:写一个同步外壳的 MCPToolSource,连上本地 stdio Server、list_tools,把每个远程工具适配成 minagent 的 Tool 注册进 ToolRegistry,并讲清命名冲突、错误映射与连接生命周期;最后用假模型驱动 Agent 真跑一次 MCP 调用(Python 3.14.6 + mcp 2.3.0 实测)。
要求 Agent 输出时对每个论断标注来源 [doc_id#chunk_id],并给出引用校验函数:先查引用是否指向真实存在的块(存在性),再查被引块是否真的支持该论断(用论断关键词是否出现在被引块做简易支持度)。用假模型脚本化一个带引用、夹一个伪造引用的回答,真跑校验并打印有效/无效引用,讲清「可核对性」为什么是 Agent 可信度的关键。
把 10.1 切好的块做成两个工具:search(query) 返回命中块(doc_id、chunk_id、片段),read(doc_id, chunk_id) 取回完整块。两者注册进第 4 章的 ToolRegistry,再用假模型脚本化一条多跳检索轨迹并真跑。本节聚焦「检索作为工具」,检索算法复用第 9 章的关键词 + 哈希向量混合思路。
检索增强的第一步是把一堆文件变成可检索的块。本节先剥掉 frontmatter,再用带重叠的 chunk(text, size, overlap) 固定长度切块,并拿本仓库 content/posts/python/ 下 6 篇真实文章实测:文件数、总块数、平均/最长/最短块长全部给出。最后对比按字符切与按语义切的取舍,代码在 Python 3.14.6 跑通。
同一任务写两版:卷一契约的 minagent 版与 PydanticAI 版,都真跑并断言。实测本次任务脚本两版行数基本持平(minagent 34 逻辑行 vs PydanticAI 36 逻辑行),真正差别在引擎本体的 122 行由谁维护。给出代码量、断言入口、失败信号、扩展成本的对照表,并说明框架赢在「你不再拥有的代码」而非「你少写的代码」。
本节把 Agent 的 output_type 从 str 换成 Pydantic 模型,让回答自带类型与校验:citations 至少一条、confidence 落在 0~1。再用 FunctionModel 注入一次非法输出(citations 为空),真跑出框架的校验重试——模型被调用三次、第二次的 retry-prompt 里带着 pydantic 的报错。最后给出重试上限与失败行为。
本节用 PydanticAI 的 Agent 重写卷二单跳问答路径:用 deps_type 声明依赖、用 RunContext 把假检索器注入工具、用 FunctionModel 与 TestModel 本地确定性跑通「检索→带引用回答」。实测 Agent 的输出参数是 output_type 而非旧版 result_type,并给出框架与自研引擎的让渡清单。
SSE 断线后客户端带 Last-Event-ID 重连,服务端从该 id 之后续推,呼应 3.2 的事件 id。讲清为什么必须服务端保留事件缓冲、缓冲多久、缓冲丢了怎么降级为全量重取;用 httpx 真跑「消费 5 条断开 → 重连补齐 11 条」并断言无重复无缺失。
把 9.1 的事件流变成可见界面:实现纯函数归约器 reduce(state, event) 与渲染器 render(state),token 追加气泡、tool 开折叠块、final 收尾;在 Node 里真跑一串事件、逐步打印状态快照与最终渲染。再讲增量与全量重渲染的取舍,最后给一段接 DOM 的最小代码。
把 3.2 的 SSE 事件抽成一份语言无关的 schema:一个 type 判别字段加各自字段。给出 TypeScript 判别联合类型与 Python pydantic 判别联合两份定义,用同一份 JSON 样本在两侧真跑解析;再对比手写双份、JSON Schema 生成、共享契约文件三种「共享类型」做法的代价。
本节把 9.2 的混合检索封装成 recall(query) 工具注册进 ToolRegistry,让 Agent 在循环里自己决定何时回忆。用假模型脚本化「用户提问 → 调用 recall → 拿到记忆 → 给出回答」并真跑打印轨迹,最后对比记忆即工具与自动注入在自主性和确定性上的取舍。
本节在 9.1 的向量检索之上再叠两路信号:用 BM25 做关键词召回,与向量相似度一起用 RRF 倒数排名融合,最后按标题命中做简单重排。我们会真跑对比纯关键词、纯向量、混合三路在同一批记忆上的排名差异,给出重排前后的实测表格,并说明生产系统为何最终用交叉编码器替代这条简单规则。
本节把 minagent 的记忆从关键词匹配推进到向量检索:定义 Embedder 协议,用纯标准库加 numpy 实现确定性的哈希嵌入 HashingEmbedder,再写一个余弦相似度的极简 VectorStore,索引 8 条记忆并实测 top-3 召回,最后说清哈希嵌入无语义的局限。
本节把框架选型从拍脑袋变成可计算:给出可观测、持久化、生态、锁定度四维权重表(含每项理由)与 1–5 分评分标准,配一个可运行的最小打分脚本,输入六个框架的分数、输出加权总分与排序,并做权重重排的敏感度分析,真跑贴输出,最后给出按场景调权的决策清单。