The-Pocket/PocketFlow · 上手攻略

• 仓库：The-Pocket/PocketFlow
• 链接：https://github.com/The-Pocket/PocketFlow
• 分类：AI · LLM 应用开发框架 · Agent 编排
• 作者：Tom
• 更新：2026-07-27

是什么

PocketFlow 是一个仅用 100 行 Python 代码实现的轻量级 LLM 框架，核心思想是将 LLM 应用建模为有向图（Graph）。框架本身零依赖（仅需 asyncio），安装包仅 56KB，却能表达 Agent、Workflow、RAG、Multi-Agent 等常见设计模式。

作者的核心论点是：市面上的主流 LLM 框架（LangChain、LangGraph、CrewAI、AutoGen 等）都过度封装、过度依赖，把简单的事情搞复杂了。真正的核心抽象只有一样——图（Graph），用 100 行代码就能抓住本质。

原文："Current LLM frameworks are bloated... You only need 100 lines for LLM Framework!"

官方还提供了多种语言实现：TypeScript、Java、C++、Go、Rust、PHP。

解决什么问题

当你想快速构建一个 LLM 应用时，LangChain/LangGraph 的学习曲线陡峭、版本迭代快导致代码易过时、动不动就引入几十个依赖。PocketFlow 的出现是为了：

1. 降低门槛：100 行源码直接读懂，不依赖任何外部库
2. 避免 vendor lock-in：不绑定 OpenAI/Anthropic，随便换底座模型
3. 快速原型：用图的思想搭流程，改起来像搭积木
4. 可读可维护：没有黑盒魔法，debug 直接看源码

快速安装

方式一：pip 安装（推荐）

pip install pocketflow

方式二：直接复制源码

框架核心只有一个文件 pocketflow/__init__.py，100 行出头，直接复制到项目里即可，零依赖。

安装 LLM 依赖（可选）

# OpenAI 模型
pip install openai

# Anthropic 模型（如 Claude）
pip install anthropic

# Google Gemini
pip install google-genai

# 如果用到 Agent 教程里的网络搜索
pip install duckduckgo-search

核心概念与用法

三大核心类

基础节点定义

from pocketflow import Node

class MyNode(Node):
    def prep(self, shared):
        # 准备阶段：从 shared 读数据、做预处理
        return shared.get("input")

    def exec(self, prep_res):
        # 执行阶段：核心逻辑，返回值决定下一跳
        return f"processed: {prep_res}"

    def post(self, shared, prep_res, exec_res):
        # 后处理阶段：将结果写回 shared
        shared["result"] = exec_res
        return "default"  # 返回动作名，决定走哪条边

节点连接与流程编排

from pocketflow import Flow

# 定义节点
node_a = MyNode()
node_b = AnotherNode()

# 用 >> 连接节点（默认 "default" 边）
node_a >> node_b

# 用 -"action">> 定义条件分支
node_a -"search" >> search_node
node_a -"answer" >> answer_node

# 包装成 Flow
flow = Flow(start=node_a)

# 执行，shared 是节点间共享的字典
shared = {"input": "hello world"}
result = flow.run(shared)
print(shared["result"])

完整示例：简单聊天机器人

# main.py
import os
from pocketflow import Node, Flow

os.environ.setdefault("OPENAI_API_KEY", os.getenv("OPENAI_API_KEY", ""))

def llm(messages):
    from openai import OpenAI
    client = OpenAI()
    return client.chat.completions.create(
        model="gpt-4o",
        messages=messages
    ).choices[0].message.content

class ChatNode(Node):
    def prep(self, shared):
        return shared.get("history", [])

    def exec(self, history):
        user_input = input("You: ")
        if user_input.lower() in ("exit", "quit"):
            return None, True  # (result, should_stop)
        history.append({"role": "user", "content": user_input})
        response = llm(history)
        history.append({"role": "assistant", "content": response})
        print(f"Bot: {response}")
        return history, False

    def post(self, shared, prep_res, exec_res):
        history, should_stop = exec_res
        shared["history"] = history
        return None if should_stop else "default"

shared = {"history": []}
chat_node = ChatNode()
chat_node >> chat_node  # self-loop
flow = Flow(start=chat_node)
flow.run(shared)

完整示例：研究 Agent（含条件分支）

来自 cookbook/pocketflow-agent，核心流程图：

DecideAction --"search"--> SearchWeb --"decide"--> DecideAction
DecideAction --"answer"--> AnswerQuestion

# nodes.py（简化）
class DecideAction(Node):
    def exec(self, prep_res):
        action = llm_decide(prep_res["question"], prep_res.get("context",""))
        return action  # 返回 "search" 或 "answer"

class SearchWeb(Node):
    def exec(self, prep_res):
        results = duckduckgo_search(prep_res["question"])
        return results

class AnswerQuestion(Node):
    def exec(self, prep_res):
        answer = llm_answer(prep_res["question"], prep_res.get("search_results",""))
        return answer

# flow.py
decide = DecideAction()
search = SearchWeb()
answer = AnswerQuestion()

decide -"search" >> search
decide -"answer" >> answer
search -"decide" >> decide  # 搜完继续判断

flow = Flow(start=decide)
result = flow.run({"question": "2024年诺贝尔物理学奖得主是谁？"})

带重试的节点

class UnreliableNode(Node):
    def __init__(self):
        super().__init__(max_retries=3, wait=1)  # 最多重试3次，间隔1秒

    def exec(self, prep_res):
        # 可能会失败的逻辑
        return risky_call(prep_res)

    def exec_fallback(self, prep_res, exc):
        print(f"全部重试失败: {exc}")
        return "fallback_result"

异步节点

from pocketflow import AsyncNode, AsyncFlow

class AsyncLLMNode(AsyncNode):
    async def exec_async(self, prep_res):
        response = await openai_client.chat.completions.create(
            model="gpt-4o",
            messages=prep_res["messages"]
        )
        return response

async def main():
    node = AsyncLLMNode()
    flow = AsyncFlow(start=node)
    result = await flow.run_async({"messages": [{"role": "user", "content": "hello"}]})

批量处理（Map-Reduce）

from pocketflow import BatchFlow

class MapNode(Node):
    def prep(self, shared):
        return shared["items"]  # 返回列表

    def exec(self, item):
        return process(item)

class ReduceNode(Node):
    def exec(self, results):
        return summarize(results)

map_node = MapNode()
reduce_node = ReduceNode()

map_node >> reduce_node
flow = BatchFlow(start=map_node)
flow.run({"items": [1, 2, 3, 4, 5]})

典型适用场景

坑与注意

1. shared 字典是唯一的状态传递方式。节点间通过读写 shared 字典通信，没有隐式状态，习惯就好，但也意味着要规划好数据结构的 schema。

2. 节点的 prep/exec/post 返回值决定流向。如果忘记返回值，Flow 会按 "default" 边走；但显式返回动作名（如 "search"）才能触发条件分支。

3. run vs run_async 不能混用。在 AsyncFlow 里放同步 Node 会正常执行，但在同步 Flow 里放 AsyncNode 会直接抛出 RuntimeError: Use run_async。

4. 生产级项目仍需补充：框架本身只有图编排，Agent 工具调用、记忆管理、向量检索等都需要自己搭或引入其他库。官方 cookbook 是参考，不是 production-ready 的库。

5. 版本稳定性：仓库相对活跃，但核心 API（Node/Flow）已相当稳定，breaking change 风险较低。

6. pip 安装后记得设置 API Key：环境变量 OPENAI_API_KEY 或各 provider 对应的 KEY，不设置会直接报 AuthenticationError。

与同类对比

一句话总结：如果你需要的是一个能读懂源码的轻量框架来快速搭原型，PocketFlow 非常合适；如果需要开箱即用的工具调用、向量存储、生产级监控，还是用 LangChain/LangGraph 更省力。

一句话推荐结论

想用一张图理解所有 LLM 应用编排？先花 10 分钟读 PocketFlow 的 100 行源码，保证比看 LangChain 文档效率高 10 倍——上手简单、源码可读、换模型零成本，是 LLM 框架学习者和快速原型选手的最佳起点。