人工智能大模型AI Agent工作流自动化RAG【免费下载链接】PocketFlowPocket Flow: 100-line LLM framework. Let Agents build Agents!项目地址https://gitcode.com/gh_mirrors/poc/PocketFlow点击查看免费下载本指南翻译并深度解析 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 cant specify the flow, AI Agents cant automate it!如果人类都无法描述流程AI Agent 就无法自动化它。因此这套方法论的核心前提是在动手写代码之前先把系统设计想清楚、写明白。二、Agentic Coding 的八个步骤官方文档用一张人机分工表概括了整个流程摘自 docs/guide.md步骤人类AI说明1. Requirements★★★ 高★☆☆ 低人类理解需求与上下文2. Flow★★☆ 中★★☆ 中人类给出高层设计AI 填充细节3. Utilities★★☆ 中★★☆ 中人类提供可用的外部 API 与集成AI 协助实现4. Data★☆☆ 低★★★ 高AI 设计数据模型人类负责验证5. Node★☆☆ 低★★★ 高AI 基于 Flow 设计节点6. Implementation★☆☆ 低★★★ 高AI 基于设计实现 Flow7. Optimization★★☆ 中★★☆ 中人类评估结果AI 协助优化8. Reliability★☆☆ 低★★★ 高AI 编写测试用例、处理边界情况下面逐一展开每一步的具体做法。步骤 1需求澄清Requirements先明确项目需求并评估AI 系统是否适合解决这个问题。文档特别提醒要理解 AI 系统的能力边界擅长需要常识的常规任务填表、回复邮件擅长输入定义清晰的创造性任务生成幻灯片、写 SQL不擅长需要复杂决策的模糊问题商业战略、创业规划。同时遵循两条原则以用户为中心从用户视角描述问题而不是罗列功能清单平衡复杂度与价值优先用最小复杂度交付最高价值的特性。步骤 2Flow 设计Flow Design在高层面上勾勒系统如何编排各个节点识别适用的设计模式。Pocket Flow 官方文档提供了多种成熟模式均可作为 Flow 设计的起点Map Reduce大输入/大输出数据的分治处理Agent节点基于上下文动态决策RAG检索增强生成更多模式见 docs/design_pattern/index.md。为 Flow 中每个节点写一句高层描述。针对具体模式补充细节Map Reduce明确如何 map切分什么与如何 reduce如何合并Agent明确输入上下文 context是什么、可选动作action有哪些RAG明确要 embed 什么通常同时存在离线索引与在线检索两条工作流。画出流程并用 mermaid 图表达。官方示例步骤 3工具函数Utilities把 AI 系统看作大脑它需要身体——即一组外部工具函数——来与现实世界交互。典型类别包括读取输入如拉取 Slack 消息、读取邮件写入输出如生成报告、发送邮件使用外部工具如调用 LLM、搜索网络。关键区分基于 LLM 的任务如摘要、情感分析不是工具函数而是 AI 系统内部的核心函数。对每个工具函数文档建议实现它、写一个简单测试并记录输入/输出及存在必要性例如name:get_embeddingutils/get_embedding.pyinput:stroutput: 一个 3072 维浮点向量necessity: 供第二个节点对文本做向量化官方给出的工具函数实现示例# utils/call_llm.py from openai import OpenAI def call_llm(prompt): client OpenAI(api_keyYOUR_API_KEY_HERE) r client.chat.completions.create( modelgpt-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复杂系统或需要持久化时使用数据库避免重复Dont 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用具体但无代码的方式描述typeRegular普通、Batch批处理或 Async异步prep从共享存储读取 textexec调用 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.mdrequirements.txt声明 Python 依赖最少只需两行PyYAML pocketflowdocs/design.md包含上述每一步的项目文档要求高层、无代码。官方模板要点摘录# Design Doc: Your Project Name Please DONT 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 nodes 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 storeutils/工具函数目录建议每个 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_keyos.getenv(GEMINI_API_KEY, ), ) model os.getenv(GEMINI_MODEL, gemini-2.5-flash) response client.models.generate_content(modelmodel, 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_cacheFalse) print(fResponse: {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 users 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_resflow.pyFlow 组装# 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(startget_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(fQuestion: {shared[question]}) print(fAnswer: {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.mdprep(shared)从共享存储读取并预处理数据返回prep_resexec(prep_res)执行计算逻辑主要是 LLM 调用、远程 API、工具使用不应访问shared返回exec_respost(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_bnode_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(fFlow 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_retries1, wait0)max_retriesintexec()最大执行次数默认1不重试waitint下次重试前等待的秒数默认0。遇到 LLM 提供商的限流/配额错误时wait非常有用。my_node SummarizeFile(max_retries3, wait10)exec()抛异常时Node 自动重试直到成功或重试满max_retries - 1次后最后一次失败。可用self.cur_retry获取当前重试次数0 起class RetryNode(Node): def exec(self, prep_res): print(fRetry {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.mdShared 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/BatchFlowdocs/core_abstraction/batch.mdBatchNode的prep()返回可迭代对象、exec()对每个 item 执行一次、post()收到结果列表BatchFlow则用不同params反复重放子 Flow。支持嵌套多级 Batch。测试见 tests/test_batch_node.py数组分块求和、Map-Reduce 全流程、空数组边界等AsyncNode/AsyncFlowdocs/core_abstraction/async.md实现prep_async/exec_async/exec_fallback_async/post_async适合异步文件读取、异步 LLM 调用、等待用户反馈或多 Agent 协调AsyncNode必须包在AsyncFlow中运行Paralleldocs/core_abstraction/parallel.mdAsyncParallelBatchNode用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 ReduceMap 阶段用BatchNode拆分处理Reduce 阶段聚合。官方示例中SummarizeAllFiles作为 BatchNode 逐文件摘要CombineSummaries再合并为最终摘要二者通过batch_node combine_node串联成Flow(startbatch_node)。模式二RAG 检索增强生成RAG 是典型的两阶段架构离线阶段ChunkDocs切块→EmbedDocs向量化→StoreIndex写入向量库三个节点顺序串联成OfflineFlow在线阶段EmbedQuery问题向量化→RetrieveDocs检索最相关块→GenerateAnswerLLM 生成答案组成OnlineFlow。它天然对应八步法中的 Data 设计共享存储里放all_chunks、all_embeds、index与 Node 设计每个节点明确读什么、写什么、调用哪个工具函数。模式三Agent 决策循环Agent 模式通过 Action 分支实现动态决策DecideAction节点用 LLM 决定search还是answersearch节点执行搜索后通过search - decide decide循环回去直到上下文足够再进入DirectAnswer。这正是步骤 2 中Agent明确输入上下文与可选动作的落地形态。六、总结一套可复用的开发纪律Agentic Coding 本质上是一套开发纪律而不是某个具体 API先想清楚再编码需求、Flow、工具、数据、节点设计前置并以docs/design.md形式沉淀为给 AI 的实现说明书人类与 AI 各司其职人类解决做什么、为什么AI 解决怎么写、怎么测小步快跑、快速失败保持简单依赖 Node 内置的max_retries/wait/exec_fallback容错让失败尽早暴露迭代优化直觉评估 → 重设 Flow → Prompt/上下文优化循环往复可靠性收尾节点输出校验、日志可视化、LLM 自评估节点三管齐下。掌握这套方法后你可以在 Pocket Flow 之上按图索骥地开发各类 LLM 应用。仓库中的 cookbook 目录收录了 40 个完整示例涵盖聊天、RAG、多 Agent、代码生成、批处理、流式输出、人机协同等场景每个示例都遵循本文介绍的项目骨架main.py/nodes.py/flow.py/utils//docs/design.md可以作为你实践 Agentic Coding 的现成参照。赞分享人工智能大模型AI Agent工作流自动化RAG【免费下载链接】PocketFlowPocket Flow: 100-line LLM framework. Let Agents build Agents!项目地址https://gitcode.com/gh_mirrors/poc/PocketFlow点击查看免费下载相关推荐Agentic Actions AuditorGitHub Actions 中 AI 编码 Agent 集成的静态安全审计方法论Agentic Actions AuditorGitHub Actions 中 AI 编码 Agent 集成的静态安全审计方法论 本篇技术指南围绕开源仓库 .人工智能计算机视觉深度学习模型评测Pocket Flow用 100 行代码构建的极简 LLM 框架Pocket Flow用 100 行代码构建的极简 LLM 框架 本文以 Pocket Flow 项目的中文 README 为核心系统讲解这个只有约 100人工智能大模型AI Agent工作流自动化RAGFlow AI Evals 完全指南用 SWE-bench 风格评测集度量 LLM 的 Flow 类型系统编码能力Flow AI Evals 完全指南用 SWE bench 风格评测集度量 LLM 的 Flow 类型系统编码能力 Flow AI Evals 是 Flow开发工具静态分析代码质量创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考