PocketFlow Source Walkthrough: Understanding Agent Flow Abstractions in 100 Lines

What's PocketFlow?

The questions in this article can also be addressedMulti-modular Agent AI Overview: Model, Learning Mechanisms and Applications、From the cognitive structure of an intelligent body to the framework of an intelligent bodyHow the concept of a relatively close read together is developed in different contexts.

PocketFlow is one 100 Line CodeThe very simple LLM framework.

It is completely light, with zero swelling, zero dependence, zero supplier locking.
New abstract models are used, but many of the work streams that have been proposed can be easily achieved.
Better for Argentina Coding. Because it's very simple enough, LLM can easily understand all the documents and core codes.

Why would there be PocketFlow? After a year of struggle with a swollen framework, the author decided to remove all unnecessary things, meaning meaningless wrapper. The result is PocketFlow: A very simple LLM framework with a core of 100 lines.

Existing frameworks such as Langchain are helpful when simple demands are consistent with their use assumptions, but too many abstract layers make code difficult to understand and to maintain. These frameworks also raise the issue of dependence, version conflict and changing interfaces for developers.

The judgement behind PocketFlow is that the LLM system can be seen as simple and oriented in essence. By stripping off unnecessary layers, a framework of zero redundancy, zero dependence and zero supplier lock-in can be obtained.

The relevant entrances are available. PocketFlow Document、Community warehouse and Go Language Version。

Charts, Nodes and Shared Storage

PocketFlow modeling LLM workflows as maps and sharing storage:

Node handles simple LLM tasks.
Flow connects nodes by operation, i.e., by the side with the label.
Store allows nodes in the process to communicate with each other.

On the Agent system, Node performs three simple operations:

Prep: Retrieval of required content from shared storage.
Exec: Performing professional tasks.
Post: returns the result to the shared storage and determines the next operation.

Flow is executed according to conditions, that is, Orch, as stated in the PocketFlow document.

It also supports bulk processing, step execution and parallel processing of nodes and processes:

Catch Node/ Flow processing data-intensive tasks.
Async Node/ Flow awaits a walk-in.
Parallel Node/ Flow handles I/O intensive tasks.

PocketFlow specifically avoids binding a supplier-specific API, so the original code does not contain unnecessary wrapper. If API wrapper is needed, it can be prepared by itself at any time, or it can be used by any company.

Document Home and Design Mode

The following figure shows the core abstraction of PocketFlow.

From here on, it's more direct to achieve common sense of the pactor.

PocketFlow does not provide built-in examples of use, but rather examples:

The tools needed to build freely, based on the examples, preserve the bottom-up optimization space and facilitate the construction of various transmission structures.

Source Structure Overview

The core code for PocketFlow requires four Python standard library packages.

import asyncio, warnings, copy, time

They are used for the stept, warning information, object copying and time-related functions.

The main abstractions in the source code include:

BaseNode: all node and flow base protocols.
Node: implementable nodes, add retry and fallback.
Flow: Process organizer, also special Node.
BatchNode / BatchFlow: Batch execution.
AsyncNode / AsyncFlow: Step execution.
AsyncParallelBatchNode / AsyncParallelBatchFlow: Parallel batching.

BaseNode: Node Agreement and Transfer Relationship

BaseNode is the basis of all Node and Flows. In the abstract structure of PocketFlow, Flow is also a special Node, thus supporting the mix of Node and Flow and the Flow-to-Flow nesting.

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")

Construct Functions for All BaseNode And the following inheritance categories provide two basic attributes:params and successorsI'm sorry. The former describes node parameters, while the latter describes the node's subsequent relationship.

def __init__(self): self.params,self.successors={},{}

The method for setting the parameters is straightforward.

def set_params(self,params): self.params=params

To construct a link between nodes, you need to define the next node.next Accept action and its proxies, modify successorsand return node Self-in order to achieve a chain call. Repeats the settings by giving the waning to avoid unexpectedly overwrite the already completed node transfer.

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

The logic of the specific nodes is carried out by three placeholder methods. They are the core of the need to rewrite when building a node.

def prep(self,shared): pass
def exec(self,prep_res): pass
def post(self,shared,prep_res,exec_res): pass

Node execution is broken _exec、_run and run Three._exec Only call. execI'm sorry. This internal approach is needed to freely re-write the implementation logic in the following subcategory without modifying the developers' own definition. exec。_run Implementation process in-house.run is the open interface to initiate the implementation logic, and also to preserve space for a single node, and to warn when a subsequent node exists.

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.")

Syntax: Construct Flow with an operator

PocketFlow builds Flow with the following syntax sugar.

node >> next_node  # 设置一个节点的默认后继节点，即 default action 下的 next node
node - "action" >> next_node  # 设置一个节点在某个 action 下的 next node

The government is not going to be able to do this.BaseNode It's overloading. __rshift__ Operator, that's... >>and __sub__ Operator, that's... -I'm sorry. The former is called. next way to set the next node. The latter returns one after checking the validity of the string _ConditionalTransition Internal category, to facilitate next use.

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")

Supporting Internal Classes _ConditionalTransition For temporary reloading __sub__ As a result, more complex grammar sugars are achieved.

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

Node: retry with fallback

Node It is the basic implementable node. It needs to include automatic retry, avoid. exec ; The requested LLM output is unreliable; it also requires a fallback, and do not cause the entire process to collapse by error.

As the core definition,Node Inherits the parent parameter and adds two new parameters.super().__init__() Executes a parent construction function that ensures reliable initialization, and then loads the external parameters into the class parameters.

class Node(BaseNode):
    def __init__(self,max_retries=1,wait=0): super().__init__(); self.max_retries,self.wait=max_retries,wait

The backup method is used to retreat after an error. Default realization is just a throwout error.

def exec_fallback(self,prep_res,exc): raise exc

As a true enforceable class,Node Rewriting of the implementation process in conjunction with the retry and retreat mechanisms. Here you see the advantages of the logical separation of the execution and function: the developer achieves exec,Framework Rewrite _exec。_exec Automatically retry and record the number of re-tests, and call Fallback after too many errors, not cyclical, and not directly exit the program.

    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)

Flow: organiser and parameter dissemination

Flow It's the whole process controller. From the abstract design point of view, Flow is the collection of Node; from the code design point of view, Flow is the entry point for the implementation of a series of nodes. It's inherited from... BaseNode, therefore supports complex nesting. To show the special, Flow has increased. start_node properties, and the function to set this parameter.

class Flow(BaseNode):
    def __init__(self,start=None): super().__init__(); self.start_node=start
    def start(self,start): self.start_node=start; return start

To get Flow running, it needs to know who the next node is. And so... get_next_node It's based on the current node. curr Yes. actionYes. successors ; if a non-terminated node jumps out of the process, give the waning. If no follow-up node is available, this method returns None。

    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

PocketFlow provides the logical implementation of the whole Flow _orch Alternatives _execI'm sorry. Although Flow also inherited from BaseNodeHowever, its implementation logic and individual nodes are inconsistent and therefore require separate process organization.

    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

This code can be broken down into steps:

curr, p, last_action = ... Initializes three variables.
curr is the current node to be implemented.copy.copy(self.start_node) Creates a copy of the starting node to avoid interference with the state of the node while running the process repeatedly.
p is the parameter for the current node. It's coming in. params With Process itself self.params .
last_action Records the action returned from the previous node, initially as None。
while curr: Means that if there is a next node, it's always circular.
curr.set_params(p) Sets the parameters for the current node.
last_action = curr._run(shared) Runs the current node and saves the next action that you want to return.
curr = copy.copy(self.get_next_node(curr, last_action)) Finds the next node according to the current node and returns action and creates a copy.
return last_action Returns the results of the last node after the end of the cycle.

Collapse the whole Flow and find it modified the logic of the portal. _run Methodology post Methodology to be implemented _exec For implementation read implementation _orch, and set the whole Flow return value to the last node after it is executed.

    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

params Yes. BaseNode One of the basic attributes. It provides an independent shared Dictionary, a layer of parameters that can be accessed by nodes and solidified during running. If you need to use params, the parameters access is considered when handwritten Node.

Because (params or {**self.params}) , for a Flow, the external input parameter has higher priority and will overwrite the Flow internal parameter.Flow Classes are also reserved for rewriting. prep and post, preserve space for subsequent Flow nesting and special needs.

BatchNode and BatchFlow: BatchFlow

BatchNode Succession Node, for processing large amounts of duplicate data on a case-by-case basis. It naturally got it. Node . Retry with fallback capability. Because of the manual distribution of the area in front of you. exec, logic of implementation of internal nodes _execOverall logic _run And start the interface. run, lot processing only requires rewrite the logic of the internal nodes _exec。

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

This is what we're asking. BatchNode It's manual. prep Step to generate an iterative object without modification of the manual execI'm sorry. Of concern are:post Step needs to be addressed. BatchNode list of the items. New realization by list-based extrapolation and parent _exec (c) The implementation of bulk processing to address data-intensive tasks.

BatchFlow The volume execution structure is allowed to be fully consistent but with different content. Different times paramsI'm sorry. It can be understood as a cycle: it runs the Flow over each parameter set. All right. shared The changes to the dictionary need to be made in Node, in principle BatchFlow Just a dispatcher.

BatchFlow Request rewrite prep Step, and let prep method returns a list of parameters, i.e. a list of dictionaries. Each element is a set of parameters that run the process.BatchFlow Run once for each group of parameters _orch, the running time parameter is the combination of the process's own parameters and this group of specific parameters. Just change. _run The way we do it. BatchFlow。

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)

One. BatchFlow Or you can embed it in another. BatchFlow Medium. Because BatchFlow special design, which consolidates all the parameters in the BatchFlow layer and then passes them to the innermost node. When actually implemented, the first external parameter will be used to run through all the parameters of the inner layer and then the most basic Flow-by-case.BatchFlow You can embed a single node internally or a multiple node-based Flow.

Use BatchNode and BatchFlow The first question is: What parameters does Node have to recycle, not fix Node only to change? shared Data.

AsyncNode and AsyncFlow: Step execution

Next, the code goes into the world of the opposite. The core difference is use async / await Keywords.

async def Defines the co-ordinate function, which is a stept function. It may suspend its execution and give way to control.
await Only async def function is used internally, meaning that you wait for the walk-in to be completed. During the waiting period, the procedure may perform other tasks.

In the process of the different steps,await The real meaning is to suspend the current mission, hand over CPU control, and let the other missions run first. Notifys that the current task continues down after I/O operations have been completed. This way, the whole program can be avoided by some I.O. Cardon.

In conducting the async programming, the following rules need to be noted:

Use in definition async defI'm sorry. Any function or method, if internal use is made await, the definition must be def Add async。
Use when calling awaitI'm sorry. Call one. async def , the function must be used await Keywords.
Transmissible. If the function A Internal await Another function B, then function A It must be defined as well. async defI'm sorry. This rule will be passed up until the top caller.

Rewrite as each method is different when using the step nodes and stepts Flow prep、exec、post , all the positions that I/O waiting should be increased awaitI'm sorry. Use to create functions that contain aniso nodes and arcae async Keywords, and yes. flow.run_async Use awaitI'm sorry. If you want to start an aniso-function from the sync function, you need to use:

# asyncio.run 是连接同步世界和异步世界的桥梁
asyncio.run(main())

AsyncNode Succession from the ordinary nodes and rewrite all the methods associated with the walk.

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

These methods are reserved for rewrite business logic. The method names were modified accordingly to avoid confusion with the synchronized version. When rewriting these methods, care needs to be taken to use them while waiting for a mission awaitJean. AsyncNode Reads data more efficiently, calls LLM, waits for user feedback or coordinates multiple Agents.

The logic of retrying and fallback is unchanged, but is used extensively because of the contagious nature of the arctic function await。asyncio.sleep It's a cosmobilized hibernation function, which does not block the whole program.

    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)

I'm gonna need to get it. run The stale version. The logic itself is unchanged, but it needs to be introduced because of the insularity of the infection. await, and limit the user 's need to pass run_async Starts the node. If using previous sync method, throw directly RuntimeError。

    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.")

Run only one AsyncNode, and when no other parallel task is added, the async does not cause the async to be swallowed up, but only the unblocked effect. Run to corresponding AsyncNode , the program does not block the I/O while waiting, thus leaving the CPU for other tasks.

AsyncFlow Multiple inheritance, existing Flow And the ability to organize, and the ability to organize, AsyncNode The stept properties.

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

It's almost identical to the sync version, but it just increased. awaitand supports the mixing of synchronous nodes and hexeronodes in an all-step Flow.

isinstance(curr, AsyncNode) Checks whether the current node is an anecdotal.
If it's a stale node, it's a stale node. await curr._run_async(shared);, if not, call directly curr._run(shared)。
This allows for the use of a hybrid of synchronous and heteronodes in a walk process.

AsyncBatch and AsyncParalBattch: Batch in sequence and in parallel

The staggered capacity also uses multiple inheritances, combining Node with Flow.

AsyncBatchNode The text reads as follows:

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

It's a succession. AsyncNode and BatchNodeI'm sorry. It's... _exec Method through List, for each item await Parent _exec, therefore is sequentially executed.

AsyncBatchFlow The text reads as follows:

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)

This is a bulk process in a stept version and will also be the case for multiple processes in sequence.

AsyncParallelBatchNode Use asyncio.gather Perform parallel batching.

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

The key here is... asyncio.gather(...)I'm sorry. It receives a list of the courses, while initiating them and awaiting completion of all the courses.(... for i in items) Is the generator expression,* It's going to be extended into multiple parameters, which is equivalent to asyncio.gather(coro1, coro2, coro3, ...)。

AsyncParallelBatchFlow Several examples of processes are initiated in parallel in the same way.

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)

This is the parallel batch process: use asyncio.gather Several examples of processes are initiated simultaneously.