PocketFlow：我试了试这个100行代码的"智能体构建智能体"框架，结果…

说实话，一开始我是拒绝的。100行代码能搞出什么花样？直到我真正动手跑起来…

一、为什么关注到PocketFlow？

今天早上刷GitHub Trending，看到一个叫PocketFlow的项目突然飙到前面。标题很嚣张：“让智能体构建智能体”，整个框架就100行代码。

我第一反应：又是那种"hello world"级别的玩具项目吧？

但点进去看了眼README，发现作者是个Google DeepMind的前研究员。这就有点意思了。一个在大厂搞过AI的人，为什么要写这么个极简框架？

我当时的想法很简单：如果100行代码真的能跑通，那我现在用的LangChain算什么？

二、动手前的预期管理

先看了一眼项目结构：

PocketFlow/
├── pocketflow.py          # 就这一个核心文件，100行
├── examples/
│   ├── chatbot.py
│   ├── rag_pipeline.py
│   └── multi_agent.py
└── README.md

我心里犯嘀咕：这也太简了吧？连个setup.py都没有。

但转念一想，也许这就是作者想表达的 —— AI框架不需要那么重。

三、第一次跑通的磕磕绊绊

坑1：import就报错

按README写的直接跑：

from pocketflow import Flow, Node

结果：

ModuleNotFoundError: No module named 'pocketflow'

我：？？？

后来发现作者根本没上传到PyPI，得直接下载文件用。README里写了，但我习惯性跳过那段…

wget https://raw.githubusercontent.com/The-Pocket/PocketFlow/main/pocketflow.py

坑2：理解这个"反向"设计理念

PocketFlow的核心概念就两个：

• Node（节点）：做一件具体的事
• Flow（流程）：把节点串起来

但和传统框架不一样的是，它把控制权完全交给开发者。没有预设的prompt模板，没有强制的输入输出格式。

我一开始觉得这样很反人类 —— 什么都要自己写？那用框架的意义在哪？

直到我写了第一个Node：

from pocketflow import Node
import openai

class GPTNode(Node):
    def prep(self, shared):
        # 准备prompt
        return shared.get("user_input", "")
    
    def exec(self, prep_res):
        # 调用GPT
        response = openai.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prep_res}]
        )
        return response.choices[0].message.content
    
    def post(self, shared, prep_res, exec_res):
        # 处理结果
        shared["gpt_response"] = exec_res
        return "default"  # 默认走向下一个节点

写完之后我突然懂了：这不是一个"框架"，而是一个"脚手架"。

它不提供轮子，但给你一个装轮子的标准接口。

四、尝试搭建一个多智能体工作流

既然号称"让智能体构建智能体"，那我就试试能不能让AI帮我写PocketFlow的代码。

我的实验场景：

• 输入：用户的需求描述
• 输出：一个能完成该需求的PocketFlow工作流代码

第一步：定义"需求分析智能体"

class RequirementAnalyzer(Node):
    """分析用户需求，拆解成任务步骤"""
    
    def prep(self, shared):
        return shared["user_requirement"]
    
    def exec(self, prep_res):
        prompt = f"""
        分析以下需求，拆解成3-5个执行步骤：
        {prep_res}
        
        输出格式：每行一个步骤
        """
        # 调用GPT...
        return gpt_call(prompt)
    
    def post(self, shared, prep_res, exec_res):
        shared["steps"] = exec_res.strip().split("\n")
        return "default"

第二步：定义"代码生成智能体"

class CodeGenerator(Node):
    """根据步骤生成PocketFlow代码"""
    
    def prep(self, shared):
        return shared["steps"]
    
    def exec(self, prep_res):
        prompt = f"""
        根据以下步骤，生成PocketFlow框架的Python代码：
        {chr(10).join(prep_res)}
        
        要求：
        - 使用pocketflow.py的Node和Flow
        - 每个步骤一个Node类
        - 包含简单的prep/exec/post方法
        """
        return gpt_call(prompt)
    
    def post(self, shared, prep_res, exec_res):
        shared["generated_code"] = exec_res
        return "default"

第三步：串起来

from pocketflow import Flow

# 创建节点
analyzer = RequirementAnalyzer()
generator = CodeGenerator()

# 连接：analyzer -> generator
analyzer >> generator

# 创建流程
flow = Flow(analyzer)

# 运行
shared = {"user_requirement": "帮我写一个能查询天气并发送邮件的机器人"}
flow.run(shared)

print(shared["generated_code"])

五、真实运行结果（有惊喜也有坑）

跑第一遍的时候，生成的代码居然能直接跑通！虽然很简单，但确实是个能工作的PocketFlow流程。

但我发现了一个问题：上下文传递不够智能。

比如我在RequirementAnalyzer里生成的步骤列表，传到CodeGenerator时，GPT有时候理解不了前面的意图，生成的代码和预期有偏差。

我试着在中间加了一个"上下文整理"节点：

class ContextRefiner(Node):
    """整理和规范化上下文"""
    
    def exec(self, prep_res):
        # 把前面的结果格式化得更清晰
        return f"任务列表：\n" + "\n".join([f"{i+1}. {step}" for i, step in enumerate(prep_res)])

加了这个之后，稳定性提升了不少。

六、性能对比：PocketFlow vs LangChain

我做了个不严谨的对比测试：

指标	LangChain	PocketFlow	备注
核心代码行数	~50000	~100	PocketFlow确实极简
学习曲线	陡峭	平缓	LC概念太多，PF就两个
灵活性	中等	极高	PF几乎无约束
生产就绪	✅	⚠️	PF缺少监控、错误处理
社区生态	丰富	几乎没有	这很致命

我的感受是：PocketFlow像是一个"原型验证神器"，快速验证想法特别好用。但真要上生产，还得自己补很多基础设施。

七、我踩过的坑（血泪史）

坑3：并发问题

当我试图并行运行多个Node时，发现PocketFlow默认是顺序执行的。想要并行得自己写：

import asyncio

async def run_parallel(nodes, shared):
    tasks = [node.run(shared) for node in nodes]
    return await asyncio.gather(*tasks)

文档里没提这茬，我翻了源码才发现…

坑4：错误处理太原始

如果某个Node抛异常，整个Flow就挂了。没有重试、没有降级、没有日志。

我不得不在每个Node的exec方法外包try-catch：

def exec(self, prep_res):
    try:
        return self._do_work(prep_res)
    except Exception as e:
        shared["error"] = str(e)
        return "error"  # 跳转到错误处理分支

坑5：共享状态的坑

所有Node共享一个shared字典，一开始我觉得很方便，但后来发现：一个Node改了数据结构，后面的Node可能崩。

比如我在Node A里把shared["data"]从list改成dict，Node B还当list用，直接报错。

我的 workaround 是给每个Node定义明确的输入输出schema，但这样就失去了"极简"的意义…

八、我的结论（不太成熟）

折腾了一下午，我对PocketFlow的感受很复杂：

喜欢的点：

• 真的极简，100行代码我30分钟就看完了
• 概念清晰，就Node和Flow两个东西
• 没有魔法，每一步都在掌控中

犹豫的点：

• 生态几乎为零，啥都要自己造
• 缺少生产级功能（监控、错误处理、持久化）
• 文档太简陋，很多行为得读源码

适用场景（我猜的）：

• ✅ 快速原型验证
• ✅ 教育/学习目的（源码太好读了）
• ✅ 高度定制化的工作流
• ❌ 直接上生产环境
• ❌ 需要丰富生态的项目

九、我还在想的问题

1. 如果智能体真的要构建智能体，上下文怎么设计才合理？ 我现在的方案感觉还是太笨重了。

2. PocketFlow的"极简"是不是太极端了？ 100行代码确实很酷，但实际用起来发现缺的东西太多。

3. 有没有可能基于PocketFlow封装一个"半 opinionated"的版本？ 保留简洁的同时，提供一些常用组件（重试、日志、并行）。

参考链接

• PocketFlow GitHub：https://github.com/The-Pocket/PocketFlow
• 我的实验代码：GitHub仓库 https://github.com/YaBoom/pocketflow-zyt

---

这只是我一下午的折腾记录，肯定有很多理解不到位的地方。如果你也试过PocketFlow，欢迎指点！

P.S. 文中代码有删减，完整版（包括我的各种试错版本）在GitHub仓库里。