用 PocketFlow 构建自动化 Coding Agent：6 个工具驱动的 LLM 循环与 Patch 子流程实战

人工智能大模型AI Agent工作流自动化RAG【免费下载链接】PocketFlowPocket Flow: 100-line LLM framework. Let Agents build Agents!项目地址https://gitcode.com/gh_mirrors/poc/PocketFlow点击查看免费下载PocketFlow 生态中的 cookbook/pocketflow-coding-agent 演示了一个可直接落地的生产级 Coding Agent它基于 PocketFlow 的 Flow/Node 抽象用一个 LLM 驱动的工具循环Tool Loop自主读取骨架代码、实现函数并跑通全部测试。本文以该示例为主线结合 flow.py、nodes.py 与 PocketFlow 核心库源码讲解其工具路由、历史压缩、跨会话记忆、项目级 Skills 加载以及以Flow 即 Node实现的patch_file容错子流程读完即可复刻一个属于自己的自动化编码 Agent。一、这个 Coding Agent 是什么这是一个基于 PocketFlow 构建的自动化编程智能体给它一个让失败测试全部通过的任务它会自主完成读代码 → 写补丁 → 跑测试的闭环直到所有测试变绿。它的核心特性在 README 中定义得很清晰6 个工具list_files、grep_search、read_file、patch_file、run_command、done由 LLM 每次决策挑选一个执行跨会话记忆把过去会话的教训保存到.memory.md供后续运行复用项目级 Skills如果工作目录存在AGENTS.md自动加载其中的项目专属规则Patch 子流程patch_file以 Flow-IS-Node流程即节点的方式实现为 Read → Validate → Apply 三步子流程并用模糊匹配做错误恢复历史压缩超过上下文阈值后自动摘要旧历史防止超长对话撑爆上下文窗口一工具一节点为每个工具配备独立节点路由架构清晰、易于扩展。值得强调的是这个 Agent 的大脑是 PocketFlow 的 Flow 编排机制而不是某个大而全的 Agent 框架——所有循环、路由、重试、子流程都由 pocketflow/init.py 中约 100 行的Node/Flow抽象完成。二、快速开始5 分钟跑起来在 cookbook/pocketflow-coding-agent 目录下依次执行# 1. 安装依赖 pip install -r requirements.txt # 2. 导出 OpenAI API Key export OPENAI_API_KEYyour-api-key-here # 3. 测试 LLM 连接 python utils/call_llm.py # 4. 运行 Coding Agent python main.py其中 requirements.txt 只需要三个依赖pocketflow0.0.1、openai1.0.0、PyYAML6.0.2。启动后Agent 会通过 setup_test_project.py 自动生成一个迷你 SQL 引擎测试工程含 tokenizer、parser、executor 三部分然后自主实现所有骨架函数让测试从失败变为全部通过。main.py还支持通过命令行参数自定义任务例如python main.py --Fix the failing tests in test_executor.py only从 main.py 可以看到以--开头的参数会被整体截取为任务描述否则使用默认任务Implement the skeleton functions to make all tests pass. Run: python -m pytest test_tokenizer.py test_parser.py test_executor.py -v三、工作原理一个闭环的 LLM 工具循环README 用一张 mermaid 图清晰描述了整体架构Agent 以循环方式工作CompactHistory压缩过旧的历史记录避免上下文溢出DecideAction把任务、历史、记忆Memory、Skills 一起发给 LLM让模型选出一个工具被选中的工具节点执行动作并记录结果回到第 1 步继续循环直到 LLM 调用done或达到最大步数。3.1 路由如何实现条件转移语法这套循环完全由 PocketFlow 的条件转移语法搭建见 flow.pycompact CompactHistory() decide DecideAction(max_retries3) decide - retry compact decide - list_files ListFiles() compact decide - grep_search GrepSearch() compact decide - read_file ReadFile() compact decide - patch_file PatchFile() compact decide - run_command RunCommand() compact compact decide return Flow(startcompact)其底层机制在 pocketflow/init.pynode other等价于node.next(other)注册默认后继node - action target通过_ConditionalTransition注册带动作标签的后继。DecideAction.post的返回值list_files、patch_file、done、retry等就是动作标签Flow._orch在每轮执行后用它查找下一个节点找不到对应动作时 Flow 结束。而每个工具节点执行完后 compact保证循环回到CompactHistory从而形成决策 → 执行 → 压缩 → 再决策的闭环。四、节点级源码剖析所有节点实现都在 nodes.py 中。理解它们也就理解了用 PocketFlow 编排 Agent的全部要点。4.1 CompactHistory历史压缩节点MAX_STEPS 50 COMPACT_AFTER 30 class CompactHistory(Node): def prep(self, shared): return shared.get(history, []), shared[task] def exec(self, inputs): history, task inputs if len(history) COMPACT_AFTER: return history old history[:len(history) - COMPACT_AFTER // 2] recent history[len(history) - COMPACT_AFTER // 2:] old_text \n.join(f- {h[tool]}: {h[result]} for h in old) summary call_llm(fSummarize these past actions briefly:\n{old_text}) return [{tool: summary, args: {}, result: summary}] recent def post(self, shared, prep_res, exec_res): shared[history] exec_res逻辑非常直观当历史条目数超过COMPACT_AFTER 30时把较早的一半history[:15]交给 LLM 压缩成一条summary记录与最近的 15 条一起保留。这样无论 Agent 跑多少步喂给DecideAction的历史始终在可控长度内。这体现了 PocketFlow 的典型三步法prep从共享存储取值、exec做计算、post回写共享存储。4.2 DecideActionLLM 决策节点这是 Agent 的大脑职责是拼接完整 Prompt 并让 LLM 输出结构化的工具调用 JSONTOOL_DESC - list_files(directory.) — List all files - grep_search(pattern, path.) — Search for pattern in .py files - read_file(path, start1, endNone) — Read file with line numbers - patch_file(path, old_str, new_str) — Replace old_str with new_str - run_command(cmd) — Run shell command - done(result) — Task complete VALID_TOOLS {list_files, grep_search, read_file, patch_file, run_command}在prep中它依次加载 SkillsAGENTS.md、跨会话记忆.memory.md、历史记录并拼装成包含系统身份、项目规则、可用工具、任务、历史的 Prompt最后要求模型只输出 JSONPick ONE tool. Output ONLY json: json {tool: tool_name, args: {arg1: value1}, reason: why}exec 中解析出 JSON 后post 承担**路由判定**职责[nodes.py](https://link.gitcode.com/i/bdb28e1156ba56988e84dc1b1be32ea5#L81-L96) - 若工具为 done 或步数达到 MAX_STEPS 50写入最终结果、调用 save_memory 保存本次会话记忆返回 done 结束循环 - 若工具不在 VALID_TOOLS 中把错误信息写入历史并返回 retry让 LLM 重新决策配合 DecideAction(max_retries3)模型最多重试 3 次 - 否则把 tool_call 写入共享存储返回工具名作为动作标签触发对应工具节点。 ### 4.3 工具节点一工具一节点 所有工具节点继承自 ToolNode它统一负责从共享存储取参数、执行、回写历史 python class ToolNode(Node): def prep(self, shared): return shared[tool_call].get(args, {}), shared[workdir] def post(self, shared, prep_res, exec_res): shared.setdefault(history, []).append({ tool: shared[tool_call][tool], args: shared[tool_call].get(args, {}), result: str(exec_res), })四个文件类工具的exec实现分别为ListFilesos.walk递归列出目录过滤隐藏文件GrepSearch只搜索.py文件按文件:行号: 内容输出匹配结果无匹配返回No matchesReadFile带行号读取支持start/end参数控制区间RunCommandsubprocess.run(..., shellTrue, timeout30)执行 shell 命令同时捕获 stdout 与 stderr命令在workdir内运行——这就是 Agent 能跑pytest的原因。每个工具节点的post都会把(tool, args, result)三元组追加进history作为下一轮决策的上下文。4.4 PatchFile SubFlowFlow-IS-Node 模式的精髓patch_file是整个示例中最值得借鉴的设计它不是一个普通节点而是一个Flow 内嵌为 NodeFlow-IS-Node的子流程由三个节点串联pr, pv, pa PatchRead(), PatchValidate(), PatchApply() pr pv pa class PatchFile(Flow): def __init__(self): super().__init__(startpr) def post(self, shared, prep_res, exec_res): result shared.pop(_patch_result, ERROR: validation failed) shared.setdefault(history, []).append({ tool: patch_file, args: shared[tool_call].get(args, {}), result: result, })为什么把补丁操作拆成子流程因为文件修改必须经过校验而校验失败时需要有恢复路径PatchRead读取目标文件全文暂存到shared[_patch_content]PatchValidate检查old_str是否存在于文件内容中。找不到时用difflib.get_close_matches做模糊匹配给出你是不是想改这段的提示若old_str出现多次则提示包含更多上下文nodes.py。校验失败时post返回error跳过 Apply 节点直接把错误结果写回历史PatchApply只有校验通过返回ok才会执行content.replace(old_str, new_str, 1)并写盘。这个模式的威力在于LLM 生成的old_str经常与真实代码存在细微出入模糊校验 明确的错误反馈能让模型在下一轮自纠而不是盲目写坏文件。这也是生产可用production-ready的关键细节。五、记忆与 Skills让 Agent 越用越聪明两个机制让 Agent 具备学习能力实现都围绕工作目录中的隐藏文件跨会话记忆.memory.mdload_memory在会话开始时读取save_memory在结束时用 LLM 把最近 5 条历史摘要成 2-3 条要点写入nodes.py。下次运行DecideAction会把这段记忆拼进 Prompt项目级 SkillsAGENTS.mdload_skills检查工作目录是否存在AGENTS.md存在则把其内容作为Project rules注入决策 Promptnodes.py。这与当前主流 Agent 约定项目根目录放AGENTS.md描述编码规范一致只要在test_project/里放一份AGENTS.md如所有函数需保留 docstring不得修改测试文件Agent 就会自动遵守。六、Agent 的考场迷你 SQL 引擎测试工程每次运行main.py都会调用 setup_test_project.py 在test_project/下重建一个完整的测试工程文件作用employees.csv/departments.csv/projects.csv三张 CSV 数据表员工、部门、项目database.pyCSV 加载器已实现、无 TODO负责表数据的读取与类型转换db.py迷你 SQL 引擎主体tokenizer已完整实现parser 与 executor 全是带详细注释的骨架函数test_tokenizer.py/test_parser.py/test_executor.py三份 pytest 测试覆盖词法、语法、执行三个层面这个设计非常聪明tokenizer 已完成约 68 个测试中 tokenizer 相关测试天然通过parser 与 executor 的每个骨架函数都带一段详尽的该返回什么结构、按什么顺序解析注释。Agent 的任务就是读懂注释、参考_parse_dotted_name、_resolve、_compare、_aggregate等已提供的辅助函数把parse_select、parse_condition、execute、execute_join、evaluate_condition等骨架逐一补全。例如parse_select的注释明确要求返回包含type/distinct/columns/from/alias/joins/where/group_by/having/order_by/limit的 AST dict并给出 12 步解析顺序evaluate_condition则要求处理comparison/and/or/not/is_null/in/between/like七类条件节点。测试文件如test_parser.py断言WHERE salary 80000解析为{type: comparison, left: salary, op: , right: 80000}即为最精确的需求规格书——这正是测试驱动的 Agent 任务设计范式测试即需求。运行结束时README 展示的典型输出如下节选 Coding Agent starting... Task: Implement the skeleton functions to make all tests pass. Working in: test_project/ [1] run_command — Run the tests to see whats failing ✅ 20 passed, 48 failed [2] read_file — Read db.py to understand the skeleton ✅ (file contents displayed) [3] patch_file — Implement parse_select() ✍️ Patched db.py ... [15] run_command — Run tests again ✅ 68 passed, 0 failed [16] done — All tests pass! Result: All 68 tests pass. Implemented parser and executor functions.可以看到 Agent 的完整行动轨迹先跑测试建立基线 → 读代码理解骨架 → 多次打补丁 → 复跑测试验证 → 判定完成。七、文件清单文件说明main.py入口——创建测试工程并运行 Agentflow.py定义 Agent 流程工具路由与 PatchFile 子流程nodes.py全部节点实现CompactHistory、DecideAction、工具节点、Patch 子流程utils/call_llm.pyLLM 封装自动探测 OpenAI / Gemini API Keysetup_test_project.py生成含骨架代码与测试的迷你 SQL 引擎工程requirements.txtPython 依赖其中 utils/call_llm.py 值得单独一提它默认使用gpt-4oOpenAI或gemini-2.0-flashGemini只要设置了OPENAI_API_KEY或GEMINI_API_KEY其中之一即可运行为没有 OpenAI 账号的用户提供了第二条路径。八、如何改造成你自己的 Coding Agent这个示例的架构刻意保持一工具一节点扩展成本很低新增工具实现一个继承ToolNode的节点如RunSQL、WriteTest把工具名加入TOOL_DESC与VALID_TOOLS再在 flow.py 加一行decide - 工具名 新节点() compact即可调整步数与压缩阈值修改 nodes.py 顶部的MAX_STEPS 50与COMPACT_AFTER 30换成真实项目把WORKDIR指向任意真实代码仓库用--参数传入新任务如--Fix the flaky test in auth/test_login.py加入项目规范在目标目录放置AGENTS.md写出编码约定Agent 会自动遵守替换 LLM改写 utils/call_llm.py 中的call_llm即可接入任意兼容 Chat API 的模型。从源码结构看patch_file的模糊匹配容错、history三元组格式、retry路由这些细节共同保证了 LLM 输出不完美时系统依然稳定——这正是把它从demo提升为生产级的关键也是将 pocketflow-coding-agent 作为模板复用到其他自动化任务重构、Bug 修复、代码评审时最值得保留的骨架。总结PocketFlow 的 pocketflow-coding-agent 完整展示了一种轻量而可靠的 Agent 构建范式用 Flow 表达循环与路由、用 Node 封装工具与决策、用 SubFlow 承载需要校验的原子操作。整个 Agent 的可运行代码不含测试工程仅三个文件却同时具备工具循环、上下文压缩、跨会话记忆、项目 Skills 与容错打补丁能力。如果你正打算基于 PocketFlow 自建编码智能体或类似的自动化工作流这个示例是极佳的起点先跑通默认的 SQL 引擎任务再逐步替换工具集与任务目标即可演进成属于自己的生产级 Agent。赞分享人工智能大模型AI Agent工作流自动化RAG【免费下载链接】PocketFlowPocket Flow: 100-line LLM framework. Let Agents build Agents!项目地址https://gitcode.com/gh_mirrors/poc/PocketFlow点击查看免费下载相关推荐NOFX Telegram Agent 重构实战单个 api_call 工具驱动的无界 LLM 循环架构NOFX Telegram Agent 重构实战单个 api_call 工具驱动的无界 LLM 循环架构 导读 本文以 NOFX 仓库内的重构实施计划 docAI Agent金融科技后端前端A-Evolve 完全指南用 LLM 驱动的演化循环自动优化 AI AgentA Evolve 完全指南用 LLM 驱动的演化循环自动优化 AI Agent A Evolve 是一套面向任意领域、任意演化算法的 AI Agent 自我进AI 技能人工智能大模型深度学习基于 TypeScript SDK 构建 LLM 驱动的 MCP 客户端stdio 连接与工具调用循环实战基于 TypeScript SDK 构建 LLM 驱动的 MCP 客户端stdio 连接与工具调用循环实战 导读本文围绕仓库 examples/client人工智能MCP 服务MCP Clients上一篇开源智能镜平台MagicMirror²如何把一面普通镜子变成家庭个人助理下一篇k6 负载测试指南5 行 JavaScript 跑通第一次接口压测可扩到分布式创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考