PocketFlow

Embedding is considered a micro-optimization compared to Flow Design. It is recommended to start with the most convenient embedding provider and optimize later. Below is a comparison of available embedding APIs, their free tiers, and pricing models.

| **API**              | **Free Tier**                           | **Pricing Model**                   | **Docs**                                                                                                                  |
| -------------------- | --------------------------------------- | ----------------------------------- | ------------------------------------------------------------------------------------------------------------------------- |
| **OpenAI**           | ~$5 credit                              | ~$0.0001/1K tokens                  | [OpenAI Embeddings](https://platform.openai.com/docs/api-reference/embeddings)                                            |
| **Azure OpenAI**     | $200 credit                             | Same as OpenAI (~$0.0001/1K tokens) | [Azure OpenAI Embeddings](https://learn.microsoft.com/azure/cognitive-services/openai/how-to/create-resource?tabs=portal) |
| **Google Vertex AI** | $300 credit                             | ~$0.025 / million chars             | [Vertex AI Embeddings](https://cloud.google.com/vertex-ai/docs/generative-ai/embeddings/get-text-embeddings)              |
| **AWS Bedrock**      | No free tier, but AWS credits may apply | ~$0.00002/1K tokens (Titan V2)      | [Amazon Bedrock](https://docs.aws.amazon.com/bedrock/)                                                                   |
| **Cohere**           | Limited free tier                       | ~$0.0001/1K tokens                  | [Cohere Embeddings](https://docs.cohere.com/docs/cohere-embed)                                                               |
| **Hugging Face**     | ~$0.10 free compute monthly             | Pay per second of compute           | [HF Inference API](https://huggingface.co/docs/api-inference)                                                             |
| **Jina**             | 1M tokens free                          | Pay per token after                 | [Jina Embeddings](https://jina.ai/embeddings/)                                                                            |

A BatchFlow is a mechanism to rerun an entire Flow multiple times, each time with different parameters. It acts as a loop that replays a sub-flow for every parameter set generated by its prep() method.

How it works under the hood:

1. prep(shared) returns a list of parameter objects (e.g., [{id: 1}, {id: 2}]).
2. The BatchFlow iterates through these objects.
3. For each object, it merges it with the BatchFlow's own params.
4. It then calls flow.run(mergedParams) for the sub-flow.

Use a BatchFlow when you want to execute a complete sequence of nodes (a Flow) repeatedly for different inputs (e.g., running a full analysis pipeline for every file in a list).

type SharedStorage = {
  files: string[];
};

type FileParams = {
  filename: string;
};

class SummarizeAllFiles extends BatchFlow<SharedStorage> {
  async prep(shared: SharedStorage): Promise<FileParams[]> {
    return shared.files.map((filename) => ({ filename }));
  }
}

// Create a per-file summarization flow
const summarizeFile = new SummarizeFile();
const summarizeAllFiles = new SummarizeAllFiles(summarizeFile);

await summarizeAllFiles.run({ files: ["file1.txt", "file2.txt"] });

A BatchNode is a specialized type of Node designed for chunk-based or iterative processing within a single node execution. Instead of processing one input, it splits an input into multiple items, processes them individually, and then aggregates the results.

Use a BatchNode when you need to:

• Chunk-based processing: e.g., splitting a large text into smaller pieces for LLM processing.
• Iterative processing: e.g., looping over a list of URLs or files within a single step of a Flow.

class MyBatchNode extends BatchNode<SharedStorage> {
  async prep(shared: SharedStorage): Promise<string[]> {
    // Return an array of items to process
    return ['item1', 'item2'];
  }

  async exec(item: string): Promise<string> {
    // Called once per item in the array returned by prep()
    return `processed ${item}`;
  }

  async post(shared: SharedStorage, prepRes: string[], execResList: string[]): Promise<string> {
    // Called once after all items are processed
    // execResList contains the results from all exec() calls
    return 'default';
  }
}

In PocketFlow, communication between Nodes and Flows occurs through two primary mechanisms: Shared Store and Params.

1. Shared Store (Recommended)

This is a global data structure (typically an in-memory dictionary) that all nodes can access.

• Reading: Nodes access data during the prep() phase.
• Writing: Nodes write data during the post() phase.
• Best Practice: Use the Shared Store for almost all data exchange. This separates your Data Schema from your Compute Logic, making your code more maintainable and flexible. It is ideal for large content or data that multiple nodes need to access.

2. Params (Batch Mode Only)

Params are local, ephemeral dictionaries passed from a parent Flow to a Node.

• Usage: Primarily used in Batch mode as identifiers (e.g., filenames, numeric IDs).
• Immutability: Parameter keys and values must be immutable during a Node's run cycle (prep -> exec -> post).
• Lifecycle: Params are set via setParams(), cleared, and updated each time a parent Flow calls them.

/* Conceptual comparison of Shared Store vs Params */
// Shared Store: Global/Heap-like
// Params: Local/Stack-like

The Offline Indexing stage builds a searchable index from raw files. It uses a SharedStore to pass data between nodes.

Key components:

• ChunkDocs (BatchNode): Reads files, splits text into chunks, and saves them to shared.allChunks.
• EmbedDocs (BatchNode): Takes chunks from shared.allChunks, generates embeddings, and saves them to shared.allEmbeds.
• StoreIndex (Node): Takes all embeddings, creates a vector index, and saves it to shared.index.

Nodes are connected via .next() and wrapped in a Flow.

const chunkNode = new ChunkDocs();
const embedNode = new EmbedDocs();
const storeNode = new StoreIndex();

chunkNode.next(embedNode).next(storeNode);
const offlineFlow = new Flow(chunkNode);

type SharedStore = {
  files?: string[];
  allChunks?: string[];
  allEmbeds?: number[][];
  index?: any;
};

class ChunkDocs extends BatchNode<SharedStore> {
  async prep(shared: SharedStore): Promise<string[]> {
    return shared.files || [];
  }

  async exec(filepath: string): Promise<string[]> {
    const text = fs.readFileSync(filepath, "utf-8");
    const chunks: string[] = [];
    const size = 100;
    for (let i = 0; i < text.length; i += size) {
      chunks.push(text.substring(i, i + size));
    }
    return chunks;
  }

  async post(
    shared: SharedStore,
    _: string[],
    chunks: string[][]
  ): Promise<undefined> {
    shared.allChunks = chunks.flat();
    return undefined;
  }
}

class EmbedDocs extends BatchNode<SharedStore> {
  async prep(shared: SharedStore): Promise<string[]> {
    return shared.allChunks || [];
  }

  async exec(chunk: string): Promise<number[]> {
    return await getEmbedding(chunk);
  }

  async post(
    shared: SharedStore,
    _: string[],
    embeddings: number[][]
  ): Promise<undefined> {
    shared.allEmbeds = embeddings;
    return undefined;
  }
}

class StoreIndex extends Node<SharedStore> {
  async prep(shared: SharedStore): Promise<number[][]> {
    return shared.allEmbeds || [];
  }

  async exec(allEmbeds: number[][]): Promise<unknown> {
    return await createIndex(allEmbeds);
  }

  async post(
    shared: SharedStore,
    _: number[][],
    index: unknown
  ): Promise<undefined> {
    shared.index = index;
    return undefined;
  }
}

const chunkNode = new ChunkDocs();
const embedNode = new EmbedDocs();
const storeNode = new StoreIndex();

chunkNode.next(embedNode).next(storeNode);
const offlineFlow = new Flow(chunkNode);

PocketFlow relies on a shared store for communication between nodes. When designing nodes, plan how they will read from and write to this central state.

Shared Store Strategies

• Simple Systems: Use an in-memory object.
• Complex/Persistent Systems: Use a database.
• Principle: Use in-memory references or foreign keys to avoid data duplication.

Node Specification Template

For each node, define:

• type: (e.g., Node, BatchNode, or ParallelBatchNode)
• prep: How it reads data from the shared store.
• exec: The core logic or utility function it calls.
• post: How it writes the result back to the shared store.

interface SharedStore {
  user: {
    id: string;
    context: {
      weather: { temp: number; condition: string };
      location: string;
    };
  };
  results: Record<string, unknown>;
}

const shared: SharedStore = {
  user: {
    id: "user123",
    context: {
      weather: { temp: 72, condition: "sunny" },
      location: "San Francisco",
    },
  },
  results: {}, // Empty object to store outputs
};

A Flow can act as a Node within another Flow. This allows you to compose complex pipelines from smaller, reusable sub-flows.

Composition Rules:

• Execution: When a parent Flow runs a sub-flow, it executes the sub-flow's nodes. Once the sub-flow completes, the parent Flow continues to the next node in its own sequence.
• Logic: A Flow acting as a Node will run its own prep() and post() methods, but it will not run exec(), as its purpose is orchestration.
• Data: In a Flow's post() method, execRes is always undefined. Instead, use the shared store to access execution results.
• Parameters: A Node's params will be a merge of all its parent Flows' params.

// Create a sub-flow
nodeA.next(nodeB);
const subflow = new Flow(nodeA);

// Connect the sub-flow to another node in a parent flow
subflow.next(nodeC);

// Create the parent flow
const parentFlow = new Flow(subflow);