AI Agent Engineer

AI Agent Engineer

项目：可靠任务队列

10 min read

项目：可靠任务队列

你将证明什么

这是核心基础综合项目。你不调用模型，也不依赖网络，而是先证明 Agent 工程更基础的能力：把不可信 JSON 变成经过校验的对象，使计算与 I/O 分离，给失败稳定分类，通过 CLI 暴露行为，并用测试验证边界。重试、日志和异步并发留作明确的进阶扩展，不把一个小项目伪装成生产任务执行器。

建议先创建自己的空目录实现，再对照本笔记相邻的 examples/。直接阅读成品并运行不等于完成项目。

项目目标

把一组 JSON 任务读入 Python，拒绝非法记录，统计状态并生成 JSON 报告。整个程序必须在本地离线运行，不读取环境密钥，不写入输入文件，且对同一输入产生确定性结果。

配套文件：

examples/
├── tasks.json
├── task_queue.py
└── test_task_queue.py

配套代码采用单文件，是为了让初学者看清边界；它不是所有生产项目都应复制的目录模板。需要发布 CLI 或模块增多时，再按 项目结构、CLI 与可复现运行 拆包。

输入与输出契约

每个任务必须包含：

  • id：非空字符串，且在文件内唯一；
  • title：非空字符串；
  • status：只能是 pending、running、done 或 failed。

输出不是原数据的复制，而是一个可供下一步消费的报告：

{
  "total": 3,
  "by_status": {
    "done": 1,
    "pending": 2
  },
  "unfinished_ids": ["task-002", "task-003"]
}

输入文件最大 1,000,000 bytes。未知字段、缺失字段、重复 ID、非法状态、非 UTF-8、非法 JSON、根节点非数组和大小超限都应报错。三个字符串字段均拒绝空值和首尾空白，不做静默 strip。不要“尽量猜”；在工程边界上，显式失败比静默修正更容易追踪。

by_status 的键按状态字符串排序；unfinished_ids 保留合法输入记录的原始顺序。若指定 --output，输出路径不得与输入文件指向同一文件，避免覆盖输入。

失败契约

  • 输入和读取失败统一转换为 TaskValidationError，保留原异常因果链；
  • CLI 对预期输入错误写 stderr 并返回退出码 2；
  • 不把 traceback 当作普通用户提示，但测试或受控日志仍可查看根因；
  • 输出路径冲突或写入失败属于执行失败，写 stderr 并返回退出码 1，不应伪装成输入数据错误。

从零实现步骤

  1. 先写 Task 数据类，明确三个字段及其类型；记住类型提示不会自动校验 JSON。
  2. 写 parse_task(value, index)，只负责把一个不可信对象变成 Task 或抛出带数组位置的 TaskValidationError。
  3. 写 load_tasks(path)：先限制字节大小，再读取 UTF-8、解析 JSON、确认根节点是数组、检查 ID 唯一。
  4. 写 summarize(tasks)：只根据已校验对象产生统计，不读写文件。
  5. 写 write_report(report, path)，明确何时发生文件副作用。
  6. 写命令行入口，把输入、输出路径作为参数，而不是硬编码本机绝对路径；预期输入错误返回稳定退出码。
  7. 先写最小成功测试，再逐项加入失败矩阵；每补一个行为都让测试先失败一次。

这种拆分让“解析”“业务计算”“I/O”分别测试。未来把任务来源换成 API、数据库或模型工具时，核心统计函数仍可复用。

运行方式

从 vault 根目录运行：

python -B '.\Knowledge\AI Agent Engineer\docs\Python基础\Agent工程路线\04-项目与自测\examples\task_queue.py' `
  '.\Knowledge\AI Agent Engineer\docs\Python基础\Agent工程路线\04-项目与自测\examples\tasks.json'

不指定 --output 时只向 stdout 写 JSON，不创建文件。要测试文件输出：

python -B '.\Knowledge\AI Agent Engineer\docs\Python基础\Agent工程路线\04-项目与自测\examples\task_queue.py' `
  '.\Knowledge\AI Agent Engineer\docs\Python基础\Agent工程路线\04-项目与自测\examples\tasks.json' `
  --output "$env:TEMP\task-report.json"

运行测试：

python -B -m unittest discover `
  -s '.\Knowledge\AI Agent Engineer\docs\Python基础\Agent工程路线\04-项目与自测\examples' `
  -p 'test_*.py' `
  -v

-B 禁止生成 .pyc；测试使用临时目录，不会在 vault 中留下报告。

验收矩阵

维度最少验证
成功空数组、典型任务、中文、确定性状态排序
字段缺字段、未知字段、空白字符串、非法状态
文件不存在、非法 UTF-8、非法 JSON、根类型错误、大小超限
集合重复 ID，未完成 ID 的顺序契约
输出UTF-8 JSON 可再次解析；stdout 与文件模式一致
CLI成功为 0，预期输入错误为 2，输出冲突/写失败为 1；失败时不输出成功 JSON

可选进阶扩展

完成必做验收后再选择一项，不要破坏核心确定性：

  1. 为“模拟瞬时依赖”增加有限重试，注入 sleep，测试不真实等待；
  2. 把诊断日志写 stderr，加入不含任务正文的 operation ID，保证 stdout 仍是纯 JSON；
  3. 对多个互不依赖的只读任务模拟有限异步并发，验证并发上限、取消和部分失败策略。

必做练习

  1. 删除一条任务的 title，确认程序拒绝并指出数组位置。
  2. 制造重复 id，先写失败测试，再修复实现。
  3. 新增可选 owner 字段：先明确 null、缺失和空字符串分别表示什么，再改数据类、校验和测试。
  4. 新增 cancelled 状态，同时修改校验、统计和测试；不要只改其中一处。
  5. 增加 --fail-on-unfinished：存在未完成任务时返回非零退出码，但仍输出合法报告；写清与输入错误的区别。
  6. 解释：如果程序成为 Agent 工具，哪些错误可让模型修正参数，哪些需要重试，哪些必须中止并交给人工？

常见错误与排查

  • 直接对原字典做业务计算：未校验值会把错误拖到更远的位置。先转换成受约束对象。
  • except Exception: pass：会让坏数据像成功一样继续流动。只捕获能处理的异常。
  • 函数同时读文件、联网、计算、写文件：测试困难且副作用不清。把 I/O 放在边界。
  • 测试只看“没有报错”：还要断言输出值和错误类型。
  • 把路径写死为本机目录：改用参数与 pathlib.Path。
  • 只测成功样例：输入边界的价值主要体现在失败行为；按验收矩阵逐项补齐。
  • 把本项目误当任务执行器：它只校验和汇总任务，不调度、不并发执行，也没有持久队列语义。

自测与掌握检查

  • 能解释数据类、字典和 JSON 对象的区别。
  • 能说出 parse_task 为什么比在主函数里连续取字段更易测试。
  • 能预测空数组、根节点为对象、重复 ID 和非法状态的结果。
  • 能证明重复运行同一输入会产生相同报告。
  • 能指出程序的唯一持久写入发生在哪里。
  • 能区分 stdout、stderr 和退出码各自传递的信息。
  • 能说明当前示例没有覆盖哪些生产问题：原子写、并发写、真实队列持久化、鉴权和恢复。
  • 能在不看成品的情况下重写一个等价实现，并让全部测试通过。

完成后进入 全库自测与掌握检查，再回到 Python基础目录。

参考资料

获取日期：2026-07-14。

Backlinks