Node

A Node is the smallest building block. Each Node has 3 steps prep -> exec -> post:

1. async prep(shared)

   • Read and preprocess data from shared store.

   • Examples: query DB, read files, or serialize data into a string.

   • Return prep_res, which is used by exec() and post().

2. async exec(prep_res)

   • Execute compute logic, with optional retries and error handling (below).

   • Examples: (mostly) LLM calls, remote APIs, tool use.

   • ⚠️ This shall be only for compute and NOT access shared.

   • ⚠️ If retries enabled, ensure idempotent implementation.

   • Return exec_res, which is passed to post().

3. async post(shared, prep_res, exec_res)

   • Postprocess and write data back to shared.

   • Examples: update DB, change states, log results.

   • Decide the next action by returning a string (action = "default" if None).

Why 3 steps? To enforce the principle of separation of concerns. The data storage and data processing are operated separately.

All steps are optional. E.g., you can only implement prep and post if you just need to process data.

Fault Tolerance & Retries

You can retry exec() if it raises an exception via two parameters when defining the Node:

• max_retries (int): Max times to run exec(). The default is 1 (no retry).

• wait (int): The time to wait (in seconds) before next retry. By default, wait=0 (no waiting).wait is helpful when you encounter rate-limits or quota errors from your LLM provider and need to back off.

Python

my_node = SummarizeFile(max_retries=3, wait=10)

TypeScript

const myNode = new SummarizeFile({ maxRetries: 3, wait: 10 })

When an exception occurs in exec(), the Node automatically retries until:

• It either succeeds, or

• The Node has retried max_retries - 1 times already and fails on the last attempt.

You can get the current retry times (0-based) from cur_retry.

Python

class RetryNode(Node):
    async def exec(self, prep_res):
        print(f"Retry {self.cur_retry} times")
        raise Exception("Failed")

TypeScript

class RetryNode extends Node {
  async exec(prepRes: any): any {
    console.log(`Retry ${this.curRetry} times`)
    throw new Error('Failed')
  }
}

Graceful Fallback

To gracefully handle the exception (after all retries) rather than raising it, override:

Python

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

TypeScript

async execFallback(prepRes: any, exc: Error): any {
  throw exc;
}

By default, it just re-raises exception. But you can return a fallback result instead, which becomes the exec_res passed to post().

Example: Summarize File

Python

class SummarizeFile(Node):
    async def prep(self, shared):
        return shared["data"]

    async def exec(self, prep_res):
        if not prep_res:
            return "Empty file content"
        prompt = f"Summarize this text in 10 words: {prep_res}"
        summary = call_llm(prompt)  # might fail
        return summary

    async def exec_fallback(self, prep_res, exc):
        # Provide a simple fallback instead of crashing
        return "There was an error processing your request."

    async def post(self, shared, prep_res, exec_res):
        shared["summary"] = exec_res
        # Return "default" by not returning

summarize_node = SummarizeFile(max_retries=3)

async def main():
    # node.run() calls prep->exec->post
    # If exec() fails, it retries up to 3 times before calling exec_fallback()
    action_result = await summarize_node.run(shared)
    print("Action returned:", action_result)  # "default"
    print("Summary stored:", shared["summary"])

asyncio.run(main())

TypeScript

class SummarizeFile extends Node {
  async prep(shared: any): Promise<any> {
    return shared['data']
  }

  async exec(prepRes: any): Promise<string> {
    if (!prepRes) {
      return 'Empty file content'
    }
    const prompt = `Summarize this text in 10 words: ${prepRes}`
    const summary = await callLLM(prompt) // might fail
    return summary
  }

  async execFallback(prepRes: any, exc: Error): Promise<string> {
    // Provide a simple fallback instead of crashing
    return 'There was an error processing your request.'
  }

  async post(shared: any, prepRes: any, execRes: any): Promise<string> {
    shared['summary'] = execRes
    // Return "default" by not returning
  }
}

const summarizeNode = new SummarizeFile({ maxRetries: 3 })

// node.run() calls prep->exec->post
// If exec() fails, it retries up to 3 times before calling execFallback()
const actionResult = await summarizeNode.run(shared)

console.log('Action returned:', actionResult) // "default"
console.log('Summary stored:', shared['summary'])