“Pocket Flow，一个仅用 100 行代码实现的 LLM 框架”

PocketFlow介绍

PocketFlow是我最近在探索的一个LLM 框架，我觉得很有意思，因此推荐给大家。
这个框架最大的特点就是：“Pocket Flow，一个仅用 100 行代码实现的 LLM 框架”。
我很好奇，一个框架只有100行代码是怎么做到的，它又有什么魅力呢？
正如作者所言现在的LLM框架过于臃肿了！
在使用各种框架的过程中，你可能会有如下的感觉：

• 臃肿的抽象：正如 Octomind 的工程团队所解释的：“LangChain 在最初对我们简单的功能需求与它的使用假设相匹配时很有帮助。但其高级抽象很快使我们的代码更难以理解并令人沮丧地难以维护。”这些框架通过不必要的复杂性隐藏了简单功能。
  
• 实现噩梦：除了抽象之外，这些框架还给开发者带来了依赖项臃肿、版本冲突和不断变化的接口的负担。开发者经常抱怨：“它不稳定，接口不断变化，文档经常过时。”另一个开发者开玩笑说：“在读这句话的时间内，LangChain 已经弃用了 4 个类而没有更新文档。”
  

PocketFlow作者开始思考这个问题：“我们真的需要这么多的包装器吗？如果我们去掉一切会怎样？什么是真正最小且可行的？”
PocketFlow作者在过去一年从零开始构建 LLM 应用程序后，有了一个顿悟：在所有复杂性之下，LLM 系统本质上只是简单的有向图。通过去除不必要的层，他创建了 Pocket Flow——一个没有任何冗余、没有任何依赖、没有任何供应商锁定的框架，全部代码仅 100 行。

AI 系统框架的抽象、应用特定封装、供应商特定封装、代码行数和大小的比较。
来源：https://pocketflow.substack.com/p/i-built-an-llm-framework-in-just
GitHub地址：https://github.com/The-Pocket/PocketFlow

PocketFlow的构建块

flowchart TD    id1[PocketFlow] -->b[Node] & c[Flow] & D[Shared Store]

理解PocketFlow需要理解Node、Flow与Shared Store这三个基本的概念。
想象 Pocket Flow 就像一个井然有序的厨房：

• Node就像烹饪站（切菜、烹饪、摆盘）
  
• Flow就像食谱，指示下一步访问哪个站台。
  
• Shared Store是所有工作站都能看到原料的台面。
  

在我们的厨房（代理系统），每个站点（Node）执行三个简单的操作：

• Prep: 从共享存储中获取你需要的东西（收集原料）
  
• Exec: 执行你的专门任务（烹饪原料）
  
• Post: 将结果返回到共享存储并确定下一步行动（上菜并决定下一步做什么）
  

flowchart TD    id1[Node] -->b[Prep] & c[Exec] & D[Post]

食谱 (Flow) 依据条件 (Orch) 指导执行：

• “如果蔬菜被切碎，前往烹饪站”
  
• “如果饭菜做好了，移到装盘站“
  

PocketFlow还支持批处理、异步执行和并行处理，适用于节点和流程。就是这样！这就是构建LLM应用程序所需的一切。没有不必要的抽象，没有复杂的架构——只有简单的构建块，可以组合成强大的系统。

Pocket Flow 核心图抽象
来源：https://pocketflow.substack.com/p/i-built-an-llm-framework-in-just
PocketFlow作者介绍

Zachary Huang：即将加入微软研究院AI前沿研究。目前从事大规模语言模型代理和系统的研究。喜欢构建、写作和制作梗图。之前经历：哥伦比亚大学博士，微软Gray Systems Lab，Databricks，2023年谷歌博士奖学金。

大佬不仅代码厉害还喜欢写通俗易懂的文章，最近看完了大佬的所有文章，感谢大佬的贡献，感兴趣的朋友也可以去看看。

地址：https://pocketflow.substack.com
PocketFlow实践

直接上手学习大佬提供的cookbook即可。
这里就演示一下几个入门的demo。
pocketflow-hello-world

定义Node与Flow：

1. from pocketflow import Node, Flow
2. from utils.call_llm import call_llm
4. # An example node and flow
5. # Please replace this with your own node and flow
6. class AnswerNode(Node):
7.     def prep(self, shared):
8.         # Read question from shared
9.         return shared["question"]
10.    
11.     def exec(self, question):
12.         return call_llm(question)
13.    
14.     def post(self, shared, prep_res, exec_res):
15.         # Store the answer in shared
16.         shared["answer"] = exec_res
18. answer_node = AnswerNode()
19. qa_flow = Flow(start=answer_node)

复制代码

主脚本写了Shared Store：

1. from flow import qa_flow
3. # Example main function
4. # Please replace this with your own main function
5. def main():
6.     shared = {
7.         "question": "你是谁？",
8.         "answer": None
9.     }
11.     qa_flow.run(shared)
12.     print("Question:", shared["question"])
13.     print("Answer:", shared["answer"])
15. if __name__ == "__main__":
16.     main()

复制代码

call_llm可以改成这样：

1. from openai import OpenAI
2. import os
4. def call_llm(prompt):
5.     client = OpenAI(api_key="your api key",
6.                     base_url="https://api.siliconflow.cn/v1")
7.    
8.     response = client.chat.completions.create(
9.         model="Qwen/Qwen2.5-72B-Instruct",
10.         messages=[{"role": "user", "content": prompt}],
11.         temperature=0.7
12.     )
13.    
14.     return response.choices[0].message.content
16. if __name__ == "__main__":
17.     # Test the LLM call
18.     messages = [{"role": "user", "content": "In a few words, what's the meaning of life?"}]
19.     response = call_llm(messages)
20.     print(f"Prompt: {messages[0]['content']}")
21.     print(f"Response: {response}")

复制代码

效果：

pocketflow-chat

flowchart LR    chat[ChatNode] -->|continue| chat

1. from pocketflow import Node, Flow
2. from utils import call_llm
4. class ChatNode(Node):
5.     def prep(self, shared):
6.         # Initialize messages if this is the first run
7.         if "messages" not in shared:
8.             shared["messages"] = []
9.             print("Welcome to the chat! Type 'exit' to end the conversation.")
10.         
11.         # Get user input
12.         user_input = input("\nYou: ")
13.         
14.         # Check if user wants to exit
15.         if user_input.lower() == 'exit':
16.             return None
17.         
18.         # Add user message to history
19.         shared["messages"].append({"role": "user", "content": user_input})
20.         
21.         # Return all messages for the LLM
22.         return shared["messages"]
24.     def exec(self, messages):
25.         if messages is None:
26.             return None
27.         
28.         # Call LLM with the entire conversation history
29.         response = call_llm(messages)
30.         return response
32.     def post(self, shared, prep_res, exec_res):
33.         if prep_res is None or exec_res is None:
34.             print("\nGoodbye!")
35.             return None  # End the conversation
36.         
37.         # Print the assistant's response
38.         print(f"\nAssistant: {exec_res}")
39.         
40.         # Add assistant message to history
41.         shared["messages"].append({"role": "assistant", "content": exec_res})
42.         
43.         # Loop back to continue the conversation
44.         return "continue"
46. # Create the flow with self-loop
47. chat_node = ChatNode()
48. chat_node - "continue" >> chat_node  # Loop back to continue conversation
50. flow = Flow(start=chat_node)
52. # Start the chat
53. if __name__ == "__main__":
54.     shared = {}
55.     flow.run(shared)

复制代码

效果：

pocketflow-chat-guardrail

flowchart LR    user[UserInputNode] -->|validate| guardrail[GuardrailNode]    guardrail -->|retry| user    guardrail -->|process| llm[LLMNode]    llm -->|continue| user

• 一个 UserInputNode，其 exec 方法收集用户输入
  
• 一个GuardrailNode，用于验证查询是否与旅行相关，使用：
  
  
  
  • 基本验证检查（空输入、过短）
    
  • 基于 LLM 的验证，以确定查询是否与旅行相关
    
  
  
• 一个 LLMNode，使用带旅行顾问系统提示的LLM处理有效的旅行查询
  
• 流连接，在处理之前通过验证路由输入，并处理与旅行无关查询的重复尝试
  

1. from pocketflow import Node, Flow
2. from utils import call_llm
4. class UserInputNode(Node):
5.     def prep(self, shared):
6.         # Initialize messages if this is the first run
7.         if "messages" not in shared:
8.             shared["messages"] = []
9.             print("Welcome to the Travel Advisor Chat! Type 'exit' to end the conversation.")
10.         
11.         return None
13.     def exec(self, _):
14.         # Get user input
15.         user_input = input("\nYou: ")
16.         return user_input
18.     def post(self, shared, prep_res, exec_res):
19.         user_input = exec_res
20.         
21.         # Check if user wants to exit
22.         if user_input and user_input.lower() == 'exit':
23.             print("\nGoodbye! Safe travels!")
24.             return None  # End the conversation
25.         
26.         # Store user input in shared
27.         shared["user_input"] = user_input
28.         
29.         # Move to guardrail validation
30.         return "validate"
32. class GuardrailNode(Node):
33.     def prep(self, shared):
34.         # Get the user input from shared data
35.         user_input = shared.get("user_input", "")
36.         return user_input
37.    
38.     def exec(self, user_input):
39.         # Basic validation checks
40.         if not user_input or user_input.strip() == "":
41.             return False, "Your query is empty. Please provide a travel-related question."
42.         
43.         if len(user_input.strip()) < 3:
44.             return False, "Your query is too short. Please provide more details about your travel question."
45.         
46.         # LLM-based validation for travel topics
47.         prompt = f"""
48. Evaluate if the following user query is related to travel advice, destinations, planning, or other travel topics.
49. The chat should ONLY answer travel-related questions and reject any off-topic, harmful, or inappropriate queries.
50. User query: {user_input}
51. Return your evaluation in YAML format:
52. ```yaml
53. valid: true/false
54. reason: [Explain why the query is valid or invalid]
55. ```"""
56.         
57.         # Call LLM with the validation prompt
58.         messages = [{"role": "user", "content": prompt}]
59.         response = call_llm(messages)
60.         
61.         # Extract YAML content
62.         yaml_content = response.split("```yaml")[1].split("```")[0].strip() if "```yaml" in response else response
63.         
64.         import yaml
65.         result = yaml.safe_load(yaml_content)
66.         assert result is not None, "Error: Invalid YAML format"
67.         assert "valid" in result and "reason" in result, "Error: Invalid YAML format"
68.         is_valid = result.get("valid", False)
69.         reason = result.get("reason", "Missing reason in YAML response")
70.         
71.         return is_valid, reason
72.    
73.     def post(self, shared, prep_res, exec_res):
74.         is_valid, message = exec_res
75.         
76.         if not is_valid:
77.             # Display error message to user
78.             print(f"\nTravel Advisor: {message}")
79.             # Skip LLM call and go back to user input
80.             return "retry"
81.         
82.         # Valid input, add to message history
83.         shared["messages"].append({"role": "user", "content": shared["user_input"]})
84.         # Proceed to LLM processing
85.         return "process"
87. class LLMNode(Node):
88.     def prep(self, shared):
89.         # Add system message if not present
90.         if not any(msg.get("role") == "system" for msg in shared["messages"]):
91.             shared["messages"].insert(0, {
92.                 "role": "system",
93.                 "content": "You are a helpful travel advisor that provides information about destinations, travel planning, accommodations, transportation, activities, and other travel-related topics. Only respond to travel-related queries and keep responses informative and friendly. Your response are concise in 100 words."
94.             })
95.         
96.         # Return all messages for the LLM
97.         return shared["messages"]
99.     def exec(self, messages):
100.         # Call LLM with the entire conversation history
101.         response = call_llm(messages)
102.         return response
104.     def post(self, shared, prep_res, exec_res):
105.         # Print the assistant's response
106.         print(f"\nTravel Advisor: {exec_res}")
107.         
108.         # Add assistant message to history
109.         shared["messages"].append({"role": "assistant", "content": exec_res})
110.         
111.         # Loop back to continue the conversation
112.         return "continue"
114. # Create the flow with nodes and connections
115. user_input_node = UserInputNode()
116. guardrail_node = GuardrailNode()
117. llm_node = LLMNode()
119. # Create flow connections
120. user_input_node - "validate" >> guardrail_node
121. guardrail_node - "retry" >> user_input_node  # Loop back if input is invalid
122. guardrail_node - "process" >> llm_node
123. llm_node - "continue" >> user_input_node     # Continue conversation
125. flow = Flow(start=user_input_node)
127. # Start the chat
128. if __name__ == "__main__":
129.     shared = {}
130.     flow.run(shared)

复制代码

效果：

最后

PocketFlow还有很多有趣的例子，感兴趣的朋友可以自己去试试！！
但是说实话PocketFlow的“易用性”还是不足的，没法像很多框架那样开箱即用，还是需要自己写很多代码的，但也就是它的小巧给了它很大的灵活性，开发者可以根据自己的想法灵活地去写程序。

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！