PocketFlow：100 行代码实现智能体框架，开发者：终于不用在依赖地狱里挣扎了

写在前面：一个关于"做减法"的故事

在智能体框架里，大家可能有个现象特别有意思：框架越做越大，文档越写越厚,依赖包动辄几百 MB，但真正用到的核心功能可能就那么几个。很多开发者盯着 LangChain 这类框架的代码库,心里嘀咕：“搞这么复杂,真的有必要吗？”

PocketFlow 的作者也有同样的困惑。在与臃肿框架斗争了整整一年后，这位开发者做了个大胆的决定：把所有不必要的东西统统砍掉。最终的成果令人惊艳——一个仅用 100 行代码实现的极简 LLM 框架，却能完成主流框架的所有核心功能。

主流框架到底哪里出了问题？

先来看看开发者们的真实吐槽。Octomind 工程团队在博客里直言不讳：“刚开始用 LangChain 时确实挺顺手，但这些高度抽象很快就让代码变得难以理解,维护起来简直是噩梦。”

Reddit 上的讨论更加热闹。有人调侃道："就在你读完这句话的功夫，LangChain 已经废弃了 4 个类，而且文档还没更新。"另一位开发者的评价更直接：“极其不稳定，接口天天变，文档经常过时。”

这些问题归根结底就两个：过度抽象把简单功能埋在复杂性之下，实现混乱让开发者疲于应对依赖冲突和版本问题。

PocketFlow 的核心理念：LLM 应用本质就是有向图

在从零构建了多个 LLM 应用后，PocketFlow 的作者有了一个关键洞察：剥离所有花哨的外壳，LLM 系统的本质其实就是简单的有向图。基于这个发现，PocketFlow 应运而生——零臃肿、零依赖、无厂商锁定，核心代码仅 100 行。

AI 框架在抽象层级、应用封装、厂商封装及代码量方面的对比

三个核心概念，像管理厨房一样简单

PocketFlow 把 LLM 工作流建模成图 + 共享存储的结构，可以用厨房来类比：

1. 节点（Node）—— 不同的料理台

每个节点就像厨房里的一个工作台，专门负责某项任务。节点只做三件事：

• Prep：从共享台拿取原料
• Exec：进行专业加工
• Post：把结果放回共享台，并决定下一步去哪

class BaseNode:
    def __init__(self):
        self.params, self.successors = {}, {}
    
    def prep(self, shared): pass       # 准备工作
    def exec(self, prep_res): pass     # 执行任务
    def post(self, shared, prep_res, exec_res): pass  # 后处理
    
    def run(self, shared): 
        p = self.prep(shared)
        e = self.exec(p)
        return self.post(shared, p, e)

2. 流（Flow）—— 规定顺序的菜谱

Flow 决定了任务的执行顺序，就像菜谱规定先切菜、再烹饪、最后摆盘。

class Flow(BaseNode):
    def __init__(self, start): 
        super().__init__()
        self.start = start
    
    def orch(self, shared, params=None):  # 编排逻辑
        curr = copy.copy(self.start)
        while curr:
            action = curr.run(shared)
            curr = copy.copy(curr.successors.get(action or "default"))

3. 共享存储（Shared Store）—— 厨房的备料台

所有料理台都能看到备料台上的食材，这就是共享存储的作用。通常就是一个内存中的字典：

load_data_node = LoadDataNode()
summarize_node = SummarizeNode()
load_data_node >> summarize_node  # 定义流程

flow = Flow(start=load_data_node)
shared = {"file_name": "data.txt"}
flow.run(shared)

在这个"智能厨房"里，菜谱（Flow）根据条件动态调度：“菜切好了就去烹饪台”，“饭煮好了就去摆盘台”。整个过程清晰透明，易于扩展。

支持的设计模式：该有的都有

基于这套极简抽象，PocketFlow 能实现当前主流的所有 AI 设计模式：

• 智能体（Agent）：具备自主决策能力
• 工作流（Workflow）：将多个任务串联成流水线
• RAG（检索增强生成）：数据检索与内容生成无缝集成
• MapReduce：大规模数据处理的经典范式
• 结构化输出：确保输出格式一致性
• 多智能体协作：协调多个智能体共同工作

所有这些模式都遵循同一套规则：在 100 行核心代码基础上，编写几百行业务逻辑就能实现。开发者不需要翻遍大型框架的成千上万个文件，而是从底层原理出发构建自己的理解。

实战案例：搭建一个网页搜索智能体

来看一个具体例子——搭建类似 Perplexity AI 的搜索智能体，能联网搜索并回答问题。

流程设计

整个智能体的行为可以建模为这样的图结构：

# 初始化节点
decide = DecideAction()   # 决策节点
search = SearchWeb()      # 搜索节点
answer = AnswerQuestion() # 回答节点

# 定义连接逻辑
decide - "search" >> search
decide - "answer" >> answer
search - "decide" >> decide

# 启动流程
flow = Flow(start=decide)

节点职责分工

DecideAction（决策节点）：判断当前应该去搜索网页，还是已有信息足够回答问题。

SearchWeb（搜索节点）：调用搜索引擎 API 抓取信息，提炼关键内容后存入上下文。

AnswerQuestion（回答节点）：汇总所有搜索信息，生成最终答案。

整个过程动态透明，随时可以更换 LLM 模型或搜索引擎，而无需改动核心逻辑。

RAG 系统实现示例

再看一个 RAG 系统的实现，代码结构同样清晰：

from pocketflow import Node, Flow, BatchNode

# 离线流程：文档处理
class ChunkDocumentsNode(BatchNode):
    def exec(self, text):
        """将单个文本切分成小块"""
        return fixed_size_chunk(text)
    
    def post(self, shared, prep_res, exec_res_list):
        """存储切分后的文本"""
        all_chunks = [chunk for chunks in exec_res_list for chunk in chunks]
        shared["texts"] = all_chunks
        print(f"✅ 从 {len(prep_res)} 个文档创建了 {len(all_chunks)} 个片段")

class EmbedDocumentsNode(BatchNode):
    def exec(self, text):
        """对单个文本进行向量化"""
        return get_embedding(text)

# 在线流程：查询处理
class RetrieveDocumentNode(Node):
    def exec(self, inputs):
        """在索引中搜索相似文档"""
        query_embedding, index, texts = inputs
        distances, indices = index.search(query_embedding, k=1)
        best_idx = indices[0][0]
        return {"text": texts[best_idx], "index": best_idx}

为什么不内置 API 封装？刻意的设计选择

细心的开发者会发现，PocketFlow 刻意避开了绑定特定厂商的 API。这个决定背后有深思熟虑的考量：

摆脱依赖地狱：主流框架动辄几百 MB 的依赖包，PocketFlow 则是零依赖，让项目保持轻量灵活。

避免厂商锁定：可以自由使用任何模型，包括 OpenLLaMA 这样的本地模型，无需修改核心架构。

完全的自定义控制：想实现 Prompt 缓存、批处理或流式输出？直接根据需求构建，不受预设抽象的限制。

框架作者的观点很明确：API 易变、需求多样、性能优化各有侧重。如果需要 API 封装，完全可以让 ChatGPT 即时生成一个，通常也就 10 行代码。这比维护那些很快就会过时的内置库要灵活得多。

智能体化编程：未来的开发范式

PocketFlow 最具革命性的地方在于开启了**智能体化编程（Agentic Coding）**的可能。

这是一种由 AI 助手辅助构建和修改 AI 应用的新范式。打个比方：

• 开发者是建筑师，负责高层设计和战略决策
• AI 助手是施工队，处理细节实现
• 开发者通过评审和微调引导整个过程

这种模式能带来 10 倍的生产力提升，让开发者从枯燥的模板代码中解放出来。

文档即代码的理念

以往的框架试图为每类应用硬编码封装，结果让开发者和 AI 助手都感到困惑。PocketFlow 的解决方案是：把文档当作第二代码库。

框架提供极简的基础模块，辅以清晰的"教学文档"，告诉 AI 如何组合这些模块。这些说明直接作为规则文件（如 Cursor 的 .cursorrules）喂给 AI 助手，使其具备灵活构建复杂系统的知识，而不是死记硬背框架接口。

能力边界：不仅仅是搜索智能体

PocketFlow 的潜力远不止于此。开发者可以用同样优雅简洁的方式构建：

• 多智能体协作系统：多个智能体分工合作解决复杂问题
• RAG 检索系统：结合知识库的智能问答
• MapReduce 计算：处理大规模数据任务
• 模型上下文协议（MCP）：标准化的模型交互接口

更复杂的应用可能需要 5-15 个工具调用，结合 Web 搜索、内部工具（如 Google Drive、Gmail、Slack）来综合处理信息。PocketFlow 让开发者能够灵活编排这些工具，而不是被框架预设的模式所束缚。

开发者反馈：终于不用跳水了

从社区反馈来看，PocketFlow 戳中了很多开发者的痛点：

“终于有个框架不需要先读几天文档才能上手了。”

“100 行核心代码意味着出了问题能自己调试，不用在框架的黑盒里瞎猜。”

“最喜欢的是没有依赖冲突，项目部署简直太省心了。”

技术细节：支持批处理、异步和并行

虽然核心只有 100 行，但 PocketFlow 该有的高级特性一个不少：

批处理（Batch）：支持节点或流处理大规模数据密集型任务。

异步（Async）：支持节点或流等待异步任务执行。

并行（Parallel）：专门优化 I/O 密集型任务的性能。

这些能力让 PocketFlow 能够应对从简单脚本到复杂生产环境的各种场景。

快速开始：三步上手

想要尝试 PocketFlow？过程非常简单：

1. 克隆仓库：git clone https://github.com/the-pocket/PocketFlow
2. 阅读文档：官方文档 提供了详细的教程和示例
3. 开始构建：从示例代码开始，逐步构建自己的应用

还有 TypeScript 版本 可供选择，满足不同技术栈的需求。

写在最后：大道至简

PocketFlow 的出现提醒开发者：复杂的问题不一定需要复杂的解决方案。通过将 LLM 应用建模为简单的有向图，剔除所有冗余，PocketFlow 实现了透明的逻辑和完全的掌控。

如果已经厌倦了在复杂框架中"跳水"，想从零开始真正掌握 AI 应用的构建，PocketFlow 的极简主义或许正是通往智能体革命的门票。

立即体验：

• GitHub 仓库 https://github.com/the-pocket/PocketFlow
• 官方文档 https://the-pocket.github.io/PocketFlow/

代码

PocketFlow核心代码：

import asyncio, warnings, copy, time

class BaseNode:
    def __init__(self): self.params,self.successors={},{}
    def set_params(self,params): self.params=params
    def next(self,node,action="default"):
        if action in self.successors: warnings.warn(f"Overwriting successor for action '{action}'")
        self.successors[action]=node; return node
    def prep(self,shared): pass
    def exec(self,prep_res): pass
    def post(self,shared,prep_res,exec_res): pass
    def _exec(self,prep_res): return self.exec(prep_res)
    def _run(self,shared): p=self.prep(shared); e=self._exec(p); return self.post(shared,p,e)
    def run(self,shared): 
        if self.successors: warnings.warn("Node won't run successors. Use Flow.")  
        return self._run(shared)
    def __rshift__(self,other): return self.next(other)
    def __sub__(self,action):
        if isinstance(action,str): return _ConditionalTransition(self,action)
        raise TypeError("Action must be a string")

class _ConditionalTransition:
    def __init__(self,src,action): self.src,self.action=src,action
    def __rshift__(self,tgt): return self.src.next(tgt,self.action)

class Node(BaseNode):
    def __init__(self,max_retries=1,wait=0): super().__init__(); self.max_retries,self.wait=max_retries,wait
    def exec_fallback(self,prep_res,exc): raise exc
    def _exec(self,prep_res):
        for self.cur_retry in range(self.max_retries):
            try: return self.exec(prep_res)
            except Exception as e:
                if self.cur_retry==self.max_retries-1: return self.exec_fallback(prep_res,e)
                if self.wait>0: time.sleep(self.wait)

class BatchNode(Node):
    def _exec(self,items): return [super(BatchNode,self)._exec(i) for i in (items or [])]

class Flow(BaseNode):
    def __init__(self,start=None): super().__init__(); self.start_node=start
    def start(self,start): self.start_node=start; return start
    def get_next_node(self,curr,action):
        nxt=curr.successors.get(action or "default")
        if not nxt and curr.successors: warnings.warn(f"Flow ends: '{action}' not found in {list(curr.successors)}")
        return nxt
    def _orch(self,shared,params=None):
        curr,p,last_action =copy.copy(self.start_node),(params or {**self.params}),None
        while curr: curr.set_params(p); last_action=curr._run(shared); curr=copy.copy(self.get_next_node(curr,last_action))
        return last_action
    def _run(self,shared): p=self.prep(shared); o=self._orch(shared); return self.post(shared,p,o)
    def post(self,shared,prep_res,exec_res): return exec_res

class BatchFlow(Flow):
    def _run(self,shared):
        pr=self.prep(shared) or []
        for bp in pr: self._orch(shared,{**self.params,**bp})
        return self.post(shared,pr,None)

class AsyncNode(Node):
    async def prep_async(self,shared): pass
    async def exec_async(self,prep_res): pass
    async def exec_fallback_async(self,prep_res,exc): raise exc
    async def post_async(self,shared,prep_res,exec_res): pass
    async def _exec(self,prep_res): 
        for self.cur_retry in range(self.max_retries):
            try: return await self.exec_async(prep_res)
            except Exception as e:
                if self.cur_retry==self.max_retries-1: return await self.exec_fallback_async(prep_res,e)
                if self.wait>0: await asyncio.sleep(self.wait)
    async def run_async(self,shared): 
        if self.successors: warnings.warn("Node won't run successors. Use AsyncFlow.")  
        return await self._run_async(shared)
    async def _run_async(self,shared): p=await self.prep_async(shared); e=await self._exec(p); return await self.post_async(shared,p,e)
    def _run(self,shared): raise RuntimeError("Use run_async.")

class AsyncBatchNode(AsyncNode,BatchNode):
    async def _exec(self,items): return [await super(AsyncBatchNode,self)._exec(i) for i in items]

class AsyncParallelBatchNode(AsyncNode,BatchNode):
    async def _exec(self,items): return await asyncio.gather(*(super(AsyncParallelBatchNode,self)._exec(i) for i in items))

class AsyncFlow(Flow,AsyncNode):
    async def _orch_async(self,shared,params=None):
        curr,p,last_action =copy.copy(self.start_node),(params or {**self.params}),None
        while curr: curr.set_params(p); last_action=await curr._run_async(shared) if isinstance(curr,AsyncNode) else curr._run(shared); curr=copy.copy(self.get_next_node(curr,last_action))
        return last_action
    async def _run_async(self,shared): p=await self.prep_async(shared); o=await self._orch_async(shared); return await self.post_async(shared,p,o)
    async def post_async(self,shared,prep_res,exec_res): return exec_res

class AsyncBatchFlow(AsyncFlow,BatchFlow):
    async def _run_async(self,shared):
        pr=await self.prep_async(shared) or []
        for bp in pr: await self._orch_async(shared,{**self.params,**bp})
        return await self.post_async(shared,pr,None)

class AsyncParallelBatchFlow(AsyncFlow,BatchFlow):
    async def _run_async(self,shared): 
        pr=await self.prep_async(shared) or []
        await asyncio.gather(*(self._orch_async(shared,{**self.params,**bp}) for bp in pr))
        return await self.post_async(shared,pr,None)

RAG示例完整代码如下：

from pocketflow import Node, Flow, BatchNode
import numpy as np
import faiss
from utils import call_llm, get_embedding, fixed_size_chunk

# Nodes for the offline flow
class ChunkDocumentsNode(BatchNode):
    def prep(self, shared):
        """Read texts from shared store"""
        return shared["texts"]
    
    def exec(self, text):
        """Chunk a single text into smaller pieces"""
        return fixed_size_chunk(text)
    
    def post(self, shared, prep_res, exec_res_list):
        """Store chunked texts in the shared store"""
        # Flatten the list of lists into a single list of chunks
        all_chunks = []
        for chunks in exec_res_list:
            all_chunks.extend(chunks)
        
        # Replace the original texts with the flat list of chunks
        shared["texts"] = all_chunks
        
        print(f"✅ Created {len(all_chunks)} chunks from {len(prep_res)} documents")
        return "default"
    
class EmbedDocumentsNode(BatchNode):
    def prep(self, shared):
        """Read texts from shared store and return as an iterable"""
        return shared["texts"]
    
    def exec(self, text):
        """Embed a single text"""
        return get_embedding(text)
    
    def post(self, shared, prep_res, exec_res_list):
        """Store embeddings in the shared store"""
        embeddings = np.array(exec_res_list, dtype=np.float32)
        shared["embeddings"] = embeddings
        print(f"✅ Created {len(embeddings)} document embeddings")
        return "default"

class CreateIndexNode(Node):
    def prep(self, shared):
        """Get embeddings from shared store"""
        return shared["embeddings"]
    
    def exec(self, embeddings):
        """Create FAISS index and add embeddings"""
        print("🔍 Creating search index...")
        dimension = embeddings.shape[1]
        
        # Create a flat L2 index
        index = faiss.IndexFlatL2(dimension)
        
        # Add the embeddings to the index
        index.add(embeddings)
        
        return index
    
    def post(self, shared, prep_res, exec_res):
        """Store the index in shared store"""
        shared["index"] = exec_res
        print(f"✅ Index created with {exec_res.ntotal} vectors")
        return "default"

# Nodes for the online flow
class EmbedQueryNode(Node):
    def prep(self, shared):
        """Get query from shared store"""
        return shared["query"]
    
    def exec(self, query):
        """Embed the query"""
        print(f"🔍 Embedding query: {query}")
        query_embedding = get_embedding(query)
        return np.array([query_embedding], dtype=np.float32)
    
    def post(self, shared, prep_res, exec_res):
        """Store query embedding in shared store"""
        shared["query_embedding"] = exec_res
        return "default"

class RetrieveDocumentNode(Node):
    def prep(self, shared):
        """Get query embedding, index, and texts from shared store"""
        return shared["query_embedding"], shared["index"], shared["texts"]
    
    def exec(self, inputs):
        """Search the index for similar documents"""
        print("🔎 Searching for relevant documents...")
        query_embedding, index, texts = inputs
        
        # Search for the most similar document
        distances, indices = index.search(query_embedding, k=1)
        
        # Get the index of the most similar document
        best_idx = indices[0][0]
        distance = distances[0][0]
        
        # Get the corresponding text
        most_relevant_text = texts[best_idx]
        
        return {
            "text": most_relevant_text,
            "index": best_idx,
            "distance": distance
        }
    
    def post(self, shared, prep_res, exec_res):
        """Store retrieved document in shared store"""
        shared["retrieved_document"] = exec_res
        print(f"📄 Retrieved document (index: {exec_res['index']}, distance: {exec_res['distance']:.4f})")
        print(f"📄 Most relevant text: \"{exec_res['text']}\"")
        return "default"
    
class GenerateAnswerNode(Node):
    def prep(self, shared):
        """Get query, retrieved document, and any other context needed"""
        return shared["query"], shared["retrieved_document"]
    
    def exec(self, inputs):
        """Generate an answer using the LLM"""
        query, retrieved_doc = inputs
        
        prompt = f"""
Briefly answer the following question based on the context provided:
Question: {query}
Context: {retrieved_doc['text']}
Answer:
"""
        
        answer = call_llm(prompt)
        return answer
    
    def post(self, shared, prep_res, exec_res):
        """Store generated answer in shared store"""
        shared["generated_answer"] = exec_res
        print("\n🤖 Generated Answer:")
        print(exec_res)
        return "default"