PocketFlow 极简Agent框架教程

引言

PocketFlow 是一个极简的 LLM（大型语言模型）框架，仅用 100 行代码实现，专为 AI 代理（Agents）设计。它允许 AI 代理“自我编程”，即通过 LLM 动态构建和执行工作流，而无需复杂的 boilerplate 代码。该框架的核心目标是让开发者（或 AI 本身）快速构建模块化的 LLM 应用，支持从简单问答到复杂代理协作的场景。

PocketFlow 不是一个传统的深度学习工具（如早期的 PocketFlow 用于 TensorFlow 的模型压缩），而是一个现代的代理式编码框架，由 The-Pocket 团队开发。你可以通过 pip install pocketflow 安装，或直接复制其源代码（位于 GitHub: https://github.com/The-Pocket/PocketFlow）。它依赖 Python 3.x 和 LLM API（如 OpenAI 或 Gemini），适用于构建 AI 驱动的自动化工具。

本教程将详细解释 PocketFlow 的核心抽象理念、设计模式、使用场景，并通过案例分析和示例代码演示其用法。教程基于官方文档和示例，旨在帮助你从零起步。

核心抽象理念

PocketFlow 的设计哲学是“最小主义 + 模块化”，它将 LLM 应用抽象为三个核心概念：

1. Node（节点）：表示工作流中的一个原子操作单元。每个 Node 负责一个具体任务，如调用 LLM、处理输入或存储数据。Node 是可复用的“积木块”，允许你将复杂逻辑拆分成小块。

2. Flow（流程）：一个由多个 Node 连接而成的有向图（DAG），定义了数据和控制流的路径。Flow 像一个“管道”，从起始 Node 开始执行，直到结束，支持分支、循环等高级结构。

3. Shared Store（共享存储）：一个全局的数据契约（data contract），所有 Node 共享一个字典-like 的存储空间，用于读写数据。这解耦了数据结构与计算逻辑，避免了传统框架中繁琐的状态管理。Shared Store 确保节点间通信高效、类型安全，且易于 AI 代理修改。

这些抽象的精髓在于：让 LLM 能“编程”自身。传统 LLM 框架（如 LangChain）需要大量配置，而 PocketFlow 只需定义 Node 和连接 Flow，LLM 代理就能通过代码生成新 Node 或调整 Flow，实现自适应行为。

设计模式

PocketFlow 借鉴了软件工程中的经典设计模式，结合 AI 特性，形成独特的“代理式编码”范式：

1. 工厂模式（Factory Pattern）：用于创建 Node 和 Flow。开发者定义一个工厂函数（如 create_qa_flow()），返回预配置的 Flow。这允许动态生成实例，支持 AI 代理根据输入参数“工厂化”生产不同 Flow。

2. 责任链模式（Chain of Responsibility）：Node 通过 >> 操作符连接成链，每个 Node 处理后将控制权传递给下一个。这类似于责任链，但更灵活，支持条件分支（通过 post 方法返回不同标签）。

3. 观察者/发布-订阅模式（Observer/Pub-Sub）：Shared Store 充当“发布者”，Node 通过 prep（准备）、exec（执行）和 post（后置）方法“订阅”数据变化。这确保了松耦合：Node 只关心自己的数据键，不需知道上游/下游细节。

4. 策略模式（Strategy Pattern）：每个 Node 的 exec 方法可以注入不同策略（如不同 LLM 调用），允许运行时切换行为。

这些模式确保框架简洁（仅 100 行），却支持复杂扩展：AI 代理可以生成 Python 代码来创建新 Node，实现“代理构建代理”。

使用场景

PocketFlow 适用于需要 LLM 动态决策的场景，尤其适合 AI 代理系统：

1. 问答系统：构建简单 QA 管道，用户输入问题 → LLM 生成答案。
2. 代理协作：多个代理分工，如一个代理搜索信息，另一个总结输出。适用于聊天机器人、代码助手。
3. 自动化工作流：如代码生成、文档分析。AI 可以根据上下文动态添加 Node（e.g., 如果问题复杂，插入“搜索 Node”）。
4. 自适应 AI 工具：LLM 代理使用 PocketFlow 框架“编程”自身，例如生成一个用于股票分析的 Flow。
5. 边缘场景：资源受限环境（如移动设备），因其轻量级。

不适合纯数值计算或非 LLM 任务（如传统 ML 训练）。

案例分析

案例 1: 简单问答系统（QA Flow）
场景：用户输入问题，系统调用 LLM 生成答案。
分析：这是一个线性链（Chain），使用 Shared Store 传递问题和答案。优势：模块化，便于扩展（e.g., 添加缓存 Node）。如果问题涉及实时数据，可插入搜索 Node。PocketFlow 的最小主义让 AI 代理轻松修改 Flow，例如让 LLM 生成“如果问题是数学题，添加计算 Node”。

案例 2: 代理式代码生成
场景：AI 代理根据用户需求构建一个新 Flow，例如“创建一个翻译 + 总结的工具”。
分析：使用工厂模式，AI 生成 Node 代码，然后实例化 Flow。Shared Store 确保数据一致性。优势：支持“元编程”，LLM 可以输出 Python 代码片段，PocketFlow 执行它。这在 DevOps 或内容创作中强大，但需注意安全（e.g., 沙箱执行生成代码）。

案例 3: 多代理协作
场景：一个代理检索网页，另一个分析情感。
分析：分支 Flow（通过 post 返回标签如 “analyze” 或 “end”）。设计模式确保可扩展：新代理只需新 Node。场景如客服系统：如果情感负面，路由到“安慰 Node”。

这些案例展示了 PocketFlow 的灵活性：从静态管道到动态自组装。

示例与 Demo 代码

以下是官方 QA 示例的完整实现。假设你有 utils/call_llm.py 文件，用于调用 LLM（例如使用 OpenAI API）。

步骤 1: 安装与准备

pip install pocketflow openai  # 假设使用 OpenAI

在 utils/call_llm.py 中定义 LLM 调用：

# utils/call_llm.py
import openai

def call_llm(prompt):
    response = openai.ChatCompletion.create(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": prompt}]
    )
    return response.choices[0].message.content

步骤 2: 定义 Node（nodes.py）

# nodes.py
from pocketflow import Node

class GetQuestionNode(Node):
    def exec(self, _):
        # 从用户输入获取问题
        user_question = input("Enter your question: ")
        return user_question

    def post(self, shared, prep_res, exec_res):
        # 存储问题到 Shared Store
        shared["question"] = exec_res
        return "default"  # 继续到下一个 Node

class AnswerNode(Node):
    def prep(self, shared):
        # 从 Shared Store 读取问题
        return shared["question"]

    def exec(self, question):
        # 调用 LLM 生成答案
        return call_llm(question)

    def post(self, shared, prep_res, exec_res):
        # 存储答案到 Shared Store
        shared["answer"] = exec_res
        return "end"  # 结束 Flow

步骤 3: 定义 Flow（flow.py）

# flow.py
from pocketflow import Flow
from nodes import GetQuestionNode, AnswerNode

def create_qa_flow():
    """工厂函数：创建 QA Flow"""
    # 创建 Node
    get_question_node = GetQuestionNode()
    answer_node = AnswerNode()
    
    # 连接 Node（责任链）
    get_question_node >> answer_node
    
    # 创建 Flow，从起始 Node 开始
    return Flow(start=get_question_node)

# 运行 Demo
if __name__ == "__main__":
    qa_flow = create_qa_flow()
    shared = {}  # 初始化 Shared Store
    qa_flow.run(shared)  # 执行 Flow
    print(f"Answer: {shared['answer']}")  # 输出结果

步骤 4: 运行 Demo

python flow.py

输入：What is Python?
输出：LLM 生成的答案（如“Python 是一种高级编程语言…”），存储在 shared["answer"] 中。

扩展 Demo: 添加分支 Node
为案例 2 添加一个条件 Node（e.g., 如果问题含“搜索”，路由到搜索）：

# 新 Node: SearchNode
class SearchNode(Node):
    def prep(self, shared):
        return shared["question"]

    def exec(self, question):
        # 模拟搜索
        return f"Search results for '{question}': ..."

    def post(self, shared, prep_res, exec_res):
        shared["search_results"] = exec_res
        return "answer"  # 自定义标签，继续到 AnswerNode

# 在 Flow 中连接：get_question >> search_node (条件) >> answer
# 使用 post 返回标签实现分支

结语

PocketFlow 通过简洁的抽象和模式，让 LLM 应用开发像搭积木一样高效。起步时，从 QA 示例开始，逐步添加 Node 探索动态代理。更多资源：GitHub 仓库（https://github.com/The-Pocket/PocketFlow）和文档（https://the-pocket.github.io/PocketFlow/）。如果需要自定义 LLM 或高级分支，参考源代码（仅 100 行，易读）。欢迎实验，让 AI 构建你的下一个 Flow！