Pocket Flow 批量翻译实战：用 100 行 LLM 框架让 Agent 并行完成多语言文档翻译

人工智能大模型AI Agent工作流自动化RAG【免费下载链接】PocketFlowPocket Flow: 100-line LLM framework. Let Agents build Agents!项目地址https://gitcode.com/gh_mirrors/poc/PocketFlow点击查看免费下载Pocket Flow 是一个仅 100 行的极简 LLM 框架其官方德语版文档cookbook/pocketflow-batch/translations/README_GERMAN.md完整介绍了框架的设计理念与入门路径而其中最典型的实战示例就是 Batch 批量翻译一个TranslateTextNode即可把 Markdown 文档同时翻译成中文、西班牙语、日语、德语、俄语、葡萄牙语、法语、韩语 8 种语言并落盘保存。读完本文你将掌握 Pocket Flow 的核心抽象Graph、BatchNode/BatchFlow的批量处理机制并能够复现、扩展这套多语言文档翻译流水线。Pocket Flow 是什么一个 100 行的 LLM 框架德语版 README 开篇即点明项目定位Pocket Flow 是一个 100 行 的极简 LLM 框架100-zeiliges minimalistisches LLM-Framework并归纳了三个核心卖点轻量Leichtgewichtig整个框架只有 100 行代码无冗余依赖Kein Ballast、无供应商绑定keine Anbieterbindung。表达力强Ausdrucksstark框架虽小却足以支撑主流 LLM 应用模式——Agent、Multi-Agent、Workflow、RAG 等。Agent 化编程Agenten-basiertes Programmieren让 AI 编程助手如 Cursor基于这套框架让 Agent 构建 Agent官方称其为 LLM 应用开发最快的范式。安装方式德语 README 提供了两种安装路径二选一即可# 方式一通过 pip 安装 pip install pocketflow # 方式二直接把 pocketflow/__init__.py 复制进项目仅 100 行两种方式的效果等价——框架的全部实现都收敛在 pocketflow/init.py 这一个文件里。仓库中的 setup.py 与 tests 目录则分别负责打包发布与功能验证。与其他框架的对比官方 README 中的数据德语 README 附有一张对比表列出 LangChain、CrewAI、SmolAgent、LangGraph、AutoGen 与 PocketFlow 在抽象层级、Wrapper 数量、代码行数与体积上的差异。该表为项目官方文档自述此处仅如实转述关键结论这些主流框架往往同时提供应用级与供应商级 Wrapper代码量从 7K 到 405K 行不等、体积从 26MB 到 198MB而 PocketFlow 只有 Graph 这一种抽象不提供任何应用级或供应商级 Wrapper代码 100 行、体积约 56KB。这组数字的意图是强调极简抽象 由你自由组合的设计哲学而非宣称某个方案绝对更优——取舍取决于具体场景。核心抽象一切 LLM 应用都是一个 Graph德语 README 用一句话概括了框架的底层逻辑那 100 行代码 抓住了 LLM 框架的核⼼抽象——Graph。从 Graph 出发Agent、Workflow、RAG 等设计模式都只是图上的不同编排方式。三个基类与三阶段生命周期打开 pocketflow/init.py 可以看到整套抽象BaseNode所有节点的基类持有params参数与successors后继节点映射两个属性并提供prep/exec/post三个可覆写方法以及默认连线与-条件连线两个运算符重载。Node在BaseNode基础上增加max_retries最大重试次数默认 1与wait重试等待秒数默认 0exec抛异常时自动按max_retries重试最终仍失败则交给exec_fallback处理。Flow图的编排器。它持有start_node在_orch中循环执行取后继节点 →set_params→ 运行节点 → 用返回的 Action 查询下一跳直到没有后继节点为止。每个节点的一次运行都遵循三阶段生命周期def _run(self, shared): p self.prep(shared) # 1. 从共享存储中准备输入 e self._exec(p) # 2. 执行核心逻辑含重试 return self.post(shared, p, e) # 3. 处理结果并返回 Actionshared是一个跨节点共享的字典shared store节点之间通过它传递数据params则通过set_params注入用于区分节点在不同上下文中的行为。条件分支通过node - action next_node声明Flow.get_next_node依据节点返回的 Action 字符串查找对应的后继。Flow 与 Batch 的关系当节点没有后继时直接node.run(shared)即可当存在多条连线时必须用Flow驱动源码中Node.run检测到存在 successors 会发出 Node wont run successors. Use Flow. 警告。这一设计保证了单节点独立执行与整图编排执行两种模式的清晰边界也让批量处理有了两种完全不同的实现层次——这正是下文 Batch 翻译示例的关键。Batch 教程把 Markdown 文档翻译成 8 种语言德语 README 的入门教程表中Batch 被标记为初级☆☆☆示例Ein Batch-Prozessor, der Markdown-Inhalte in mehrere Sprachen übersetzt一个将 Markdown 内容翻译成多种语言的批处理器。项目英文原版 cookbook/pocketflow-batch/README.md 给出了完整功能描述并行地把 Markdown 内容翻译成多种语言把翻译结果保存到指定输出目录保持原始 Markdown 格式链接、代码块不被破坏。快速开始三步在cookbook/pocketflow-batch目录下依次执行# 1. 安装依赖pocketflow、anthropic、pyyaml pip install -r requirements.txt # 2. 配置 Anthropic API Key export ANTHROPIC_API_KEYyour-api-key-here # 3. 运行翻译流程 python main.pyrequirements.txt 只有三个依赖pocketflow0.0.1、anthropic0.15.0、pyyaml6.0与框架无冗余依赖的定位一致。TranslateTextNode整个批处理的唯一节点main.py 中整个批量翻译只定义了一个继承自BatchNode的节点TranslateTextNode并通过三个方法完成全部工作class TranslateTextNode(BatchNode): def prep(self, shared): text shared.get(text, (No text provided)) languages shared.get(languages, [Chinese, Spanish, Japanese, German, Russian, Portuguese, French, Korean]) # 为每种语言构造一个 (text, lang) 批次 return [(text, lang) for lang in languages] def exec(self, data_tuple): text, language data_tuple prompt f Please translate the following markdown file into {language}. But keep the original markdown format, links and code blocks. Directly return the translated text, without any other text or comments. Original: {text} Translated: result call_llm(prompt) print(fTranslated {language} text) return {language: language, translation: result} def post(self, shared, prep_res, exec_res_list): output_dir shared.get(output_dir, translations) os.makedirs(output_dir, exist_okTrue) for result in exec_res_list: language, translation result[language], result[translation] filename os.path.join(output_dir, fREADME_{language.upper()}.md) with open(filename, w, encodingutf-8) as f: f.write(translation) print(fSaved translation to {filename})这三个方法精确对应BatchNode的语义prep(shared)从共享存储读取text与languages返回一个可迭代对象——这里是 8 个(text, lang)元组组成的列表exec(item)对每个元组调用一次 LLM。Prompt 中显式要求保留原始 Markdown 格式、链接和代码块直接返回译文这是保持文档结构完整的关键设计post(shared, prep_res, exec_res_list)接收全部 8 个执行结果组成的列表exec_res_list用os.makedirs(..., exist_okTrue)创建输出目录再按README_{语言大写}.md命名规则逐文件写盘。主程序从仓库根目录的 README.md 读取源文本构造共享存储并用两行代码完成整个图的装配与执行translate_node TranslateTextNode(max_retries3) flow Flow(starttranslate_node) flow.run(shared)注意max_retries3对应 pocketflow/init.py 中Node._exec的重试循环——LLM 调用偶发失败时同一翻译任务会自动重试最多 3 次无需手写任何异常处理。call_llm极简的模型调用封装utils.py 中的call_llm是唯一的 LLM 封装固定使用claude-3-7-sonnet-20250219模型并通过 Anthropic 的thinking参数开启扩展思考budget_tokens16000、max_tokens20000def call_llm(prompt): client Anthropic(api_keyos.environ.get(ANTHROPIC_API_KEY, your-api-key)) response client.messages.create( modelclaude-3-7-sonnet-20250219, max_tokens20000, thinking{type: enabled, budget_tokens: 16000}, messages[{role: user, content: prompt}] ) return response.content[1].text返回值取response.content[1].text索引 1 对应 thinking 模式下的正文内容。main.py中还使用time.perf_counter()对 8 种语言的顺序翻译进行了耗时统计方便与并行版本做基准对比。运行输出示例README.md记录了典型输出形态——每完成一种语言打印一行最终汇总到translations目录Translated Chinese text Translated Spanish text ... Saved translation to translations/README_CHINESE.md Saved translation to translations/README_JAPANESE.md ... Translation Complete Translations saved to: translations 仓库中 translations 目录下的 8 份README_*.md含本文所依据的德语版 README_GERMAN.md就是该流程的真实产物验证了输出文件按目标语言命名的约定。深入 BatchNode一次 prep多次 exec要真正理解上面的翻译节点需要回到框架源码与官方文档层面。项目文档 docs/core_abstraction/batch.md 对BatchNode的定义是它继承Node但改变了prep与exec的契约——prep(shared)返回一个可迭代对象列表、生成器等exec(item)对该可迭代对象中的每个元素调用一次post(shared, prep_res, exec_res_list)在所有元素处理完毕后接收结果列表并返回 Action。源码层面的实现只有一行却点明了批处理的本质class BatchNode(Node): def _exec(self, items): return [super(BatchNode, self)._exec(i) for i in (items or [])]即prep产出的列表被逐个送入exec每个元素都独立享有Node的重试能力max_retries/wait最终汇总为结果列表交给post。这正是翻译示例中一次 prep 产出 8 个任务、exec 并行处理、post 批量写盘的底层机制。docs/core_abstraction/batch.md同时强调BatchNode与BatchFlow是两个不同层次的批处理。BatchFlow用于用不同参数反复重放整个子 Flow它的prep返回的不是待处理数据而是参数字典列表如[{filename: a.txt}, ...]子 Flow 中的节点通过self.params读取这些参数且子节点可以是普通Node而无需是BatchNode。测试用例 tests/test_batch_flow.py 中的NestedBatchFlow外层按 group、内层按 item 双层迭代则演示了 BatchFlow 嵌套合并参数的完整行为。翻译场景因为每种语言只需一次 LLM 调用、无需子流程重放所以用BatchNode更直接若任务升级为每个文件都要跑一段多步翻译流程则应切换到BatchFlow。从顺序到并行AsyncParallelBatchNode德语 README 的教程表中还列出了一个关联示例 ParallelEine parallele Ausführungsdemo, die 3-fache Beschleunigung zeigt展示 3 倍加速的并行执行演示。该示例与批量翻译同源——cookbook/pocketflow-parallel-batch/main.py 用AsyncParallelBatchNode实现了同样的 8 语言翻译只是把exec换成exec_async、把Flow换成AsyncFlow并在post_async中优先用aiofiles异步写盘缺失时回退同步写。框架源码为其提供了直接支撑——并行的关键在 pocketflow/init.py 的这一行class AsyncParallelBatchNode(AsyncNode, BatchNode): async def _exec(self, items): return await asyncio.gather(*(super(AsyncParallelBatchNode, self)._exec(i) for i in items))asyncio.gather让所有批内任务的exec_async并发执行因此8 种语言翻译在 I/O 密集的 LLM 调用场景下可从顺序版的逐次等待变为并发完成。两个示例在各自main.py中都打印了time.perf_counter()耗时可以直接对照体验顺序BatchNode与并行AsyncParallelBatchNode两种模式的实际差异。此外框架还提供AsyncBatchNode异步但顺序与AsyncBatchFlow/AsyncParallelBatchFlow分别对应异步世界里的顺序批处理与并行批处理语义与同步版本一一对应。教程生态与延伸阅读德语 README 用一张难度表列出了完整的入门教程体系本文只挑其中与本主题密切相关的列出其余可浏览 cookbook 目录教程难度主题Batch 批量翻译☆☆☆ 入门将 Markdown 内容翻译为多种语言Map-Reduce★☆☆ 进阶用 Map-Reduce 模式批量评估简历Parallel 并行★☆☆ 进阶展示 3 倍加速的并行执行Parallel Flow★☆☆ 进阶多滤镜并行图片处理展示 8 倍加速Agent☆☆☆ 入门可联网搜索并作答的研究型 AgentRAG☆☆☆ 入门简单的检索增强生成流程Workflow☆☆☆ 入门列提纲、写内容、套格式的写作工作流若想从用起来走向造出来建议按以下顺序深入仓库通读核心实现 pocketflow/init.py100 行建议逐行阅读阅读核心抽象文档 docs/core_abstraction/batch.md、docs/core_abstraction/flow.md、docs/core_abstraction/node.md运行测试验证行为tests/test_batch_node.py数组分块求和、map-reduce 流水线、空数组边界与 tests/test_batch_flow.py空输入、单元素、异常传播、嵌套 BatchFlow、自定义参数。另外德语 README 提到 Pocket Flow 原生是 Python 实现社区还维护了 TypeScript、Java、C、Go 等语言移植版本这些仓库不在当前仓库内仅作了解。总结从德语版 README 出发这条技术路径非常清晰Pocket Flow 用 100 行代码定义了节点 图的最小抽象BatchNode在单节点内实现一次 prep、逐项 exec、汇总 post的批处理能力cookbook/pocketflow-batch把这一能力落地为一个可运行的 8 语言 Markdown 翻译工具AsyncParallelBatchNode又将其升级为并发执行。当你需要处理大批量同构任务多文件翻译、多文档摘要、多轮评估时这条定义 prep 产出任务列表 → exec 处理单个任务 → post 汇总写盘的套路可以直接复用而一旦任务内部还有多步流程BatchFlow及其嵌套能力就是下一步的自然延伸。赞分享人工智能大模型AI Agent工作流自动化RAG【免费下载链接】PocketFlowPocket Flow: 100-line LLM framework. Let Agents build Agents!项目地址https://gitcode.com/gh_mirrors/poc/PocketFlow点击查看免费下载相关推荐Pocket Flow 极简 LLM 框架实战用 100 行核心代码构建 Agent、Workflow 与批量翻译流水线Pocket Flow 极简 LLM 框架实战用 100 行核心代码构建 Agent、Workflow 与批量翻译流水线 Pocket Flow 是一个只有人工智能大模型AI Agent工作流自动化RAGCANN/cannbot-skillsHost-Device函数绑定关系Host Device 函数绑定关系 1. 概述 在 HACC 异构模型中Device Kernel 的执行需要 Host 端提供多种辅助函数如 Tilin人工智能大模型AI Agent工作流自动化RAGPocketFlow 并行批量翻译实战用 AsyncFlow 与 AsyncParallelBatchNode 将多语言翻译耗时从 1136 秒压到 209 秒PocketFlow 并行批量翻译实战用 AsyncFlow 与 AsyncParallelBatchNode 将多语言翻译耗时从 1136 秒压到 209人工智能大模型AI Agent工作流自动化RAG上一篇Obtainium项目APK包ID获取失败问题分析与解决方案下一篇InternLM2.5-20B-Chat部署实战LMDeploy与vLLM高效部署方案创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考