Pocket Flow Agentic Coding 指南：人类设计、AI 编码的 LLM 应用开发方法论

• 人工智能
• 大模型
• AI Agent
• 工作流自动化
• RAG

本指南翻译并深度解析 Pocket Flow 官方文档中的《Agentic Coding》章节（即 docs/guide.md）。它定义了一套"人类负责系统设计、AI 负责编码实现"的协作式开发流程，覆盖从需求澄清、Flow 设计、工具函数、数据契约到节点实现、优化与可靠性加固的完整生命周期。读完本文，你将掌握一套可复用的 LLM 应用开发套路，并能在 Pocket Flow 的图（Graph）+ 共享存储（Shared Store）抽象之上，把任何业务问题拆解为可交付给 AI Agent 编码的工程方案。

一、什么是 Agentic Coding：人类设计，AI 编码

Pocket Flow 是一个以"100 行核心代码"著称的极简 LLM 框架（核心实现见 pocketflow/init.py），其设计哲学之一就是Agentic Coding——让 LLM 应用开发变成"人类系统设计 + Agent 实现"的协作过程。

在传统开发中，工程师既要理解需求、又要设计架构、还要逐行实现。而 Agentic Coding 把这两类工作显式拆分：

• 人类负责：需求理解、高层架构设计、结果评估；
• AI Agent 负责：数据模型设计、节点实现、流程编排、测试与兜底。

文档中反复强调："If Humans can't specify the flow, AI Agents can't automate it!"（如果人类都无法描述流程，AI Agent 就无法自动化它）。因此这套方法论的核心前提是：在动手写代码之前，先把系统设计想清楚、写明白。

二、Agentic Coding 的八个步骤

官方文档用一张"人机分工表"概括了整个流程（摘自 docs/guide.md）：

下面逐一展开每一步的具体做法。

步骤 1：需求澄清（Requirements）

先明确项目需求，并评估"AI 系统是否适合解决这个问题"。文档特别提醒要理解 AI 系统的能力边界：

• 擅长：需要常识的常规任务（填表、回复邮件）；
• 擅长：输入定义清晰的创造性任务（生成幻灯片、写 SQL）；
• 不擅长：需要复杂决策的模糊问题（商业战略、创业规划）。

同时遵循两条原则：

• 以用户为中心：从用户视角描述"问题"，而不是罗列功能清单；
• 平衡复杂度与价值：优先用最小复杂度交付最高价值的特性。

步骤 2：Flow 设计（Flow Design）

在高层面上勾勒系统如何编排各个节点：

1. 识别适用的设计模式。Pocket Flow 官方文档提供了多种成熟模式，均可作为 Flow 设计的起点：
   • Map Reduce：大输入/大输出数据的分治处理；
   • Agent：节点基于上下文动态决策；
   • RAG：检索增强生成；
   • 更多模式见 docs/design_pattern/index.md。
2. 为 Flow 中每个节点写一句高层描述。
3. 针对具体模式补充细节：
   • Map Reduce：明确如何 map（切分什么）与如何 reduce（如何合并）；
   • Agent：明确输入（上下文 context）是什么、可选动作（action）有哪些；
   • RAG：明确要 embed 什么，通常同时存在离线（索引）与在线（检索）两条工作流。
4. 画出流程并用 mermaid 图表达。官方示例：

步骤 3：工具函数（Utilities）

把 AI 系统看作"大脑"，它需要"身体"——即一组外部工具函数——来与现实世界交互。典型类别包括：

• 读取输入（如拉取 Slack 消息、读取邮件）；
• 写入输出（如生成报告、发送邮件）；
• 使用外部工具（如调用 LLM、搜索网络）。

关键区分：基于 LLM 的任务（如摘要、情感分析）不是工具函数，而是 AI 系统内部的核心函数。

对每个工具函数，文档建议：实现它、写一个简单测试，并记录输入/输出及存在必要性，例如：

• name:get_embedding（utils/get_embedding.py）
• input:str
• output: 一个 3072 维浮点向量
• necessity: 供第二个节点对文本做向量化

官方给出的工具函数实现示例：

# utils/call_llm.py from openai import OpenAI def call_llm(prompt): client = OpenAI(api_key="YOUR_API_KEY_HERE") r = client.chat.completions.create( model="gpt-4o", messages=[{"role": "user", "content": prompt}] ) return r.choices[0].message.content if __name__ == "__main__": prompt = "What is the meaning of life?" print(call_llm(prompt))

两条重要实践（来自文档警告框）：

• 有时先设计工具函数再设计 Flow：例如自动化遗留系统的 LLM 项目，瓶颈往往是该系统的可用接口。此时应优先设计最难对接的工具函数，再围绕它们构建 Flow；
• 工具函数内避免异常处理：如果工具函数被节点的exec()调用，不要在工具内部使用try...except，让节点内置的重试机制去处理失败。

步骤 4：数据设计（Data Design）

设计节点间通信的共享存储（Shared Store）。这是 Pocket Flow 的核心设计原则之一——用一份精心设计的共享存储作为所有节点约定存取数据的"数据契约"（详见 docs/core_abstraction/communication.md）：

• 简单系统：使用内存字典（in-memory dict）；
• 复杂系统或需要持久化时：使用数据库；
• 避免重复（Don't Repeat Yourself）：使用内存引用或外键。

官方给出的共享存储设计示例：

shared = { "user": { "id": "user123", "context": { # 嵌套 dict "weather": {"temp": 72, "condition": "sunny"}, "location": "San Francisco" } }, "results": {} # 空 dict 用于存放输出 }

从实现层面看，pocketflow/init.py 中BaseNode._run()把shared作为参数贯穿prep -> exec -> post全流程，而Flow._orch()则用copy.copy复制节点后依次执行，确保共享存储是节点间唯一的"全局内存"。

步骤 5：节点设计（Node Design）

规划每个节点如何读写数据、使用哪些工具函数。对每个 Node（核心抽象详见 docs/core_abstraction/node.md），用"具体但无代码"的方式描述：

• type：Regular（普通）、Batch（批处理）或 Async（异步）；
• prep：从共享存储读取 "text"；
• exec：调用 embedding 工具函数。避免在此做异常处理，交给节点重试机制；
• post：将 "embedding" 写回共享存储。

步骤 6：实现（Implementation）

到这里，人类已完成设计，"Agentic Coding"正式开始。实现阶段的纪律：

• Keep it simple, stupid!避免复杂特性和全量类型检查；
• FAIL FAST！充分利用 Node 内置的重试与回退机制优雅地处理失败，快速暴露系统薄弱点；
• 全程添加日志便于调试。

步骤 7：优化（Optimization）

• 直觉优先：初评阶段，人的直觉往往是很好的起点；
• 重新设计 Flow（回到步骤 3）：考虑进一步拆解任务、引入 Agent 式决策、或更好地管理输入上下文；
• 如果 Flow 设计已稳固，再做微观优化：
  • Prompt 工程：使用清晰、具体、带示例的指令降低歧义；
  • 上下文内学习（In-Context Learning）：对难以用指令描述的任务提供高质量示例。

文档特别提醒：你大概率会迭代很多次——步骤 3~6 可能重复数百次。

步骤 8：可靠性（Reliability）

• 节点重试：在节点exec中增加输出质量检查，并考虑调大max_retries与wait；
• 日志与可视化：保留所有尝试的日志、可视化节点结果，便于调试；
• 自评估：对不确定的结果，增加一个由 LLM 驱动的独立节点来复核输出。

三、配套工程文件结构

文档给出了一套标准的 LLM 项目文件组织方式（这也是 cookbook 下所有示例项目共同遵循的骨架）：

my_project/ ├── main.py ├── nodes.py ├── flow.py ├── utils/ │ ├── __init__.py │ ├── call_llm.py │ └── search_web.py ├── requirements.txt └── docs/ └── design.md

requirements.txt

声明 Python 依赖，最少只需两行：

PyYAML pocketflow

docs/design.md

包含上述每一步的项目文档，要求高层、无代码。官方模板（要点摘录）：

# Design Doc: Your Project Name > Please DON'T remove notes for AI ## Requirements > Notes for AI: Keep it simple and clear. > If the requirements are abstract, write concrete user stories ## Flow Design > Notes for AI: > 1. Consider the design patterns of agent, map-reduce, rag, and workflow. Apply them if they fit. > 2. Present a concise, high-level description of the workflow. ### Applicable Design Pattern: 1. Map the file summary into chunks, then reduce these chunks into a final summary. 2. Agentic file finder - Context: The entire summary of the file - Action: Find the file ### Flow high-level Design: 1. **First Node**: This node is for ... 2. **Second Node**: This node is for ... 3. **Third Node**: This node is for ... ## Utility Functions > Notes for AI: > 1. Understand the utility function definition thoroughly by reviewing the doc. > 2. Include only the necessary utility functions, based on nodes in the flow. 1. **Call LLM** (`utils/call_llm.py`) - Input: prompt (str) - Output: response (str) - Generally used by most nodes for LLM tasks 2. **Embedding** (`utils/get_embedding.py`) - Input: str - Output: a vector of 3072 floats - Used by the second node to embed text ## Node Design ### Shared Store > Notes for AI: Try to minimize data redundancy shared = { "key": "value" } ### Node Steps > Notes for AI: Carefully decide whether to use Batch/Async Node/Flow. 1. First Node - Purpose: Provide a short explanation of the node's function - Type: Decide between Regular, Batch, or Async - Steps: - prep: Read "key" from the shared store - exec: Call the utility function - post: Write "key" to the shared store

utils/：工具函数目录

• 建议每个 API 调用一个 Python 文件，如call_llm.py、search_web.py；
• 每个文件都应包含一个main()入口用于单独测试该 API 调用，例如使用 Google GenAI 的实现：

from google import genai import os def call_llm(prompt: str) -> str: client = genai.Client( api_key=os.getenv("GEMINI_API_KEY", ""), ) model = os.getenv("GEMINI_MODEL", "gemini-2.5-flash") response = client.models.generate_content(model=model, contents=[prompt]) return response.text if __name__ == "__main__": test_prompt = "Hello, how are you?" # First call - should hit the API print("Making call...") response1 = call_llm(test_prompt, use_cache=False) print(f"Response: {response1}")

nodes.py：节点定义

# nodes.py from pocketflow import Node from utils.call_llm import call_llm class GetQuestionNode(Node): def exec(self, _): # Get question directly from user input user_question = input("Enter your question: ") return user_question def post(self, shared, prep_res, exec_res): # Store the user's question shared["question"] = exec_res return "default" # Go to the next node class AnswerNode(Node): def prep(self, shared): # Read question from shared return shared["question"] def exec(self, question): # Call LLM to get the answer return call_llm(question) def post(self, shared, prep_res, exec_res): # Store the answer in shared shared["answer"] = exec_res

flow.py：Flow 组装

# flow.py from pocketflow import Flow from nodes import GetQuestionNode, AnswerNode def create_qa_flow(): """Create and return a question-answering flow.""" # Create nodes get_question_node = GetQuestionNode() answer_node = AnswerNode() # Connect nodes in sequence get_question_node >> answer_node # Create flow starting with input node return Flow(start=get_question_node)

main.py：程序入口

# main.py from flow import create_qa_flow # Example main function # Please replace this with your own main function def main(): shared = { "question": None, # Will be populated by GetQuestionNode from user input "answer": None # Will be populated by AnswerNode } # Create the flow and run it qa_flow = create_qa_flow() qa_flow.run(shared) print(f"Question: {shared['question']}") print(f"Answer: {shared['answer']}") if __name__ == "__main__": main()

四、Flow 与 Node 的底层原理解读

要让 Agentic Coding 的设计文档真正落地，需要理解 Pocket Flow 的运行时行为。以下是结合源码（pocketflow/init.py）与官方核心抽象文档（docs/core_abstraction/index.md）的解读。

1. Node 的三段式生命周期

每个 Node 执行prep -> exec -> post三步（详见 docs/core_abstraction/node.md）：

• prep(shared)：从共享存储读取并预处理数据，返回prep_res；
• exec(prep_res)：执行计算逻辑（主要是 LLM 调用、远程 API、工具使用），不应访问shared，返回exec_res；
• post(shared, prep_res, exec_res)：写回共享存储，并返回一个动作字符串决定下一步（不返回时等价于"default"）。

源码中BaseNode._run()正是依次调用这三个方法：

def _run(self, shared): p = self.prep(shared) e = self._exec(p) return self.post(shared, p, e)

为什么是三段式？文档给出的理由是关注点分离：数据存储（读写 shared）与数据处理（LLM 计算）分开操作，且所有步骤都可选。

2. Flow 的 Action 驱动转移

Flow 用Action（带标签的边）连接节点：

• node_a >> node_b：默认转移，等价于node_a - "default" >> node_b；
• node_a - "action_name" >> node_b：命名动作转移。

Flow 从start节点开始，执行节点、读取post()返回的 Action、沿对应边前进，直到没有后继节点为止。源码中的get_next_node()负责查表并发出警告：

def get_next_node(self, curr, action): nxt = curr.successors.get(action or "default") if not nxt and curr.successors: warnings.warn(f"Flow ends: '{action}' not found in {list(curr.successors)}") return nxt

这一行为在测试 tests/test_flow_basic.py 中有完整覆盖：包括start()链式初始化、>>串联、positive/negative条件分支、以及check -> subtract -> check的循环直到满足终止条件，还有"Action 未找到导致 Flow 结束并告警"的用例。

3. 重试与优雅回退

Node 构造时可传入两个参数（pocketflow/init.py 中Node.__init__(self, max_retries=1, wait=0)）：

• max_retries（int）：exec()最大执行次数，默认1（不重试）；
• wait（int）：下次重试前等待的秒数，默认0。遇到 LLM 提供商的限流/配额错误时，wait非常有用。

my_node = SummarizeFile(max_retries=3, wait=10)

exec()抛异常时，Node 自动重试直到成功，或重试满max_retries - 1次后最后一次失败。可用self.cur_retry获取当前重试次数（0 起）：

class RetryNode(Node): def exec(self, prep_res): print(f"Retry {self.cur_retry} times") raise Exception("Failed")

重试耗尽后，可重写exec_fallback优雅兜底而不是继续抛异常：

def exec_fallback(self, prep_res, exc): raise exc # 默认行为：直接重新抛出

若返回一个结果，它会作为exec_res传给post()。完整示例见 docs/core_abstraction/node.md 的SummarizeFile，而 tests/test_fall_back.py 则用unittest验证了"成功时不调用 fallback"、"重试耗尽后调用 fallback"、"默认 fallback 重新抛异常"以及 Flow 内 fallback 的传播行为。

4. 共享存储与 Params 的取舍

节点间通信有两种方式（docs/core_abstraction/communication.md）：

• Shared Store（绝大多数情况）：全局数据结构（常为内存 dict），所有节点通过prep()读、post()写。适合结果数据、大内容等需要多节点共享的数据；
• Params（仅用于 Batch）：父 Flow 传入的节点本地临时paramsdict，用作任务标识（如文件名、数字 ID），键值不可变。

用内存管理来类比：Shared Store 像堆（所有函数调用共享），Params 像栈（由调用方分配）。实践中建议"Shared Store 用于几乎所有场景，Params 是 Batch 的语法糖"。

5. Batch、Async 与 Parallel：数据密集型与 I/O 密集型扩展

Agentic Coding 的步骤 5 要求节点设计时选择类型（Regular/Batch/Async），这三类扩展能力如下：

• BatchNode/BatchFlow（docs/core_abstraction/batch.md）：BatchNode的prep()返回可迭代对象、exec()对每个 item 执行一次、post()收到结果列表；BatchFlow则用不同params反复重放子 Flow。支持嵌套多级 Batch。测试见 tests/test_batch_node.py（数组分块求和、Map-Reduce 全流程、空数组边界等）；
• AsyncNode/AsyncFlow（docs/core_abstraction/async.md）：实现prep_async/exec_async/exec_fallback_async/post_async，适合异步文件读取、异步 LLM 调用、等待用户反馈或多 Agent 协调；AsyncNode必须包在AsyncFlow中运行；
• Parallel（docs/core_abstraction/parallel.md）：AsyncParallelBatchNode用asyncio.gather并发执行exec_async()，AsyncParallelBatchFlow让子 Flow 各次迭代并发运行。注意：受 Python GIL 限制，并行对 CPU 密集无效，只擅长重叠 LLM 调用、DB 查询、API 请求、文件 I/O 等 I/O 密集任务；同时要警惕限流，可能需要信号量或休眠节流。

五、从设计模式到实战：把八步法用起来

Agentic Coding 的 Flow 设计步骤（步骤 2）要求识别设计模式。官方文档在 docs/design_pattern/index.md 汇总了可直接套用的模式，这里给出两个与八步法强相关的典型落地：

模式一：Map Reduce 文档摘要

当输入数据大（多文件）或输出数据大（多表单）且能逻辑拆分为独立子任务时，用 Map Reduce：Map 阶段用BatchNode拆分处理，Reduce 阶段聚合。官方示例中SummarizeAllFiles作为 BatchNode 逐文件摘要，CombineSummaries再合并为最终摘要，二者通过batch_node >> combine_node串联成Flow(start=batch_node)。

模式二：RAG 检索增强生成

RAG 是典型的两阶段架构：

• 离线阶段：ChunkDocs（切块）→EmbedDocs（向量化）→StoreIndex（写入向量库），三个节点顺序串联成OfflineFlow；
• 在线阶段：EmbedQuery（问题向量化）→RetrieveDocs（检索最相关块）→GenerateAnswer（LLM 生成答案），组成OnlineFlow。

它天然对应八步法中的 Data 设计（共享存储里放all_chunks、all_embeds、index）与 Node 设计（每个节点明确读什么、写什么、调用哪个工具函数）。

模式三：Agent 决策循环

Agent 模式通过 Action 分支实现动态决策：DecideAction节点用 LLM 决定"search"还是"answer"，search节点执行搜索后通过search - "decide" >> decide循环回去，直到上下文足够再进入DirectAnswer。这正是步骤 2 中"Agent：明确输入上下文与可选动作"的落地形态。

六、总结：一套可复用的开发纪律

Agentic Coding 本质上是一套开发纪律，而不是某个具体 API：

1. 先想清楚再编码：需求、Flow、工具、数据、节点设计前置，并以docs/design.md形式沉淀为"给 AI 的实现说明书"；
2. 人类与 AI 各司其职：人类解决"做什么、为什么"，AI 解决"怎么写、怎么测"；
3. 小步快跑、快速失败：保持简单，依赖 Node 内置的max_retries/wait/exec_fallback容错，让失败尽早暴露；
4. 迭代优化：直觉评估 → 重设 Flow → Prompt/上下文优化，循环往复；
5. 可靠性收尾：节点输出校验、日志可视化、LLM 自评估节点三管齐下。

掌握这套方法后，你可以在 Pocket Flow 之上按图索骥地开发各类 LLM 应用。仓库中的 cookbook 目录收录了 40+ 个完整示例（涵盖聊天、RAG、多 Agent、代码生成、批处理、流式输出、人机协同等场景），每个示例都遵循本文介绍的项目骨架（main.py/nodes.py/flow.py/utils//docs/design.md），可以作为你实践 Agentic Coding 的现成参照。

• 人工智能
• 大模型
• AI Agent
• 工作流自动化
• RAG