Pocketflow Framework

A Node is the fundamental building block of a flow in the Pocketflow framework. To create a custom node, extend BaseNode and optionally override the following three lifecycle methods to separate concerns:

1. prepAsync(sharedState): Use this for preprocessing data. It is a reliable step to query databases, read files, or serialize data. It returns a prepResult which is passed to the next step.
2. execAsync(prepResult): This is the main execution step (e.g., an LLM call or remote API request).
   • Important: Do not write to sharedState here. Instead, gather necessary data in prepAsync and pass it via prepResult.
   • If retries are enabled, ensure this implementation is idempotent.
   • Returns an execResult passed to postAsync.
3. postAsync(sharedState, prepResult, execResult): Use this to write results back to the sharedState and determine the next step in the flow. It must return a string representing the next Action (e.g., "default").

You do not need to implement all three; for example, a node that only prepares data might only implement prepAsync.

import { BaseNode } from "../src/pocket";

export class MyCustomNode extends BaseNode {
  public async prepAsync(sharedState: any): Promise<any> {
    // Prepare data
    return "prepared data";
  }

  public async execAsync(prepResult: any): Promise<any> {
    // Main logic (e.g. LLM call)
    return "execution result";
  }

  public async postAsync(sharedState: any, prepResult: any, execResult: any): Promise<string> {
    // Write to state and return next action
    sharedState.result = execResult;
    return "default";
  }
}

You can implement complex logic like branching and loops by returning specific action strings from a node's post() method and mapping those strings to successors using addSuccessor.

• Branching: Return different strings (e.g., "approved", "reject") based on logic in post() to route to different nodes.
• Cycles: Create a loop by adding a successor that points back to a previously visited node in the graph.

class QualityCheckNode extends BaseNode {
  async post(prepResult: any, execResult: any, sharedState: any) {
    if (execResult.score >= 0.8) return "approved";
    if (sharedState.retryCount < 3) return "retry";
    return "reject";
  }
}

const check = new QualityCheckNode();
const publish = new PublishNode();
const revise = new ReviseNode();
const reject = new RejectNode();

check.addSuccessor(publish, "approved");
check.addSuccessor(revise, "retry");
check.addSuccessor(reject, "reject");

// Create a cycle: revise feeds back into check
revise.addSuccessor(check, DEFAULT_ACTION);

The Map-Reduce paradigm allows you to process large inputs by splitting them into smaller chunks (the Map step) and then combining those results into a single output (the Reduce step).

To implement this, you typically create two specialized BaseNode subclasses:

1. Map Node:

   • prepAsync: Splits the large input (e.g., from sharedState) into an array of chunks.
   • execAsync: Processes each individual chunk (e.g., via an LLM call).
   • postAsync: Collects the results from all chunk executions and saves them into the sharedState for the next node.

2. Reduce Node:

   • prepAsync: Retrieves the array of processed chunks from the sharedState.
   • execAsync: Performs a final operation (e.g., merging or summarizing) on the entire collection of results.
   • postAsync: Saves the final consolidated result into the sharedState.

You connect them by calling mapNode.addSuccessor(reduceNode, DEFAULT_ACTION) and initializing a Flow with the map node.

import { BaseNode, Flow, DEFAULT_ACTION } from "../src/pocket";

// 1. Define the Map Node
export class MapSummaries extends BaseNode {
  public async prepAsync(sharedState: any): Promise<string[]> {
    // Logic to split sharedState.text into chunks
    return chunks;
  }

  public async execAsync(chunk: string): Promise<string> {
    // Logic to process a single chunk
    return result;
  }

  public async postAsync(sharedState: any, prepResult: string[], execResultList: string[]): Promise<string> {
    sharedState.summaries = execResultList;
    return DEFAULT_ACTION;
  }
}

// 2. Define the Reduce Node
export class ReduceSummaries extends BaseNode {
  public async prepAsync(sharedState: any): Promise<string[]> {
    return sharedState.summaries ?? [];
  }

  public async execAsync(summaries: string[]): Promise<string> {
    // Logic to combine all summaries
    return finalResult;
  }

  public async postAsync(sharedState: any, prepResult: string[], execResult: string): Promise<string> {
    sharedState.final_summary = execResult;
    return DEFAULT_ACTION;
  }
}

// 3. Wire them together
const mapNode = new MapSummaries();
const reduceNode = new ReduceSummaries();
mapNode.addSuccessor(reduceNode, DEFAULT_ACTION);

const flow = new Flow(mapNode);
await flow.runAsync({ text: "large input..." });