open-orbyt 0.1.0

A workflow framework for Python. Build AI agents, data pipelines, and orchestration flows with nodes, graphs, and a shared store.

orbyt

A workflow framework for Python. Build AI agents, data pipelines, and orchestration flows with nodes, graphs, and a shared store.

Install

pip install orbyt

Quick Start

from orbyt import Flow, SharedStore, Result, new_node, run, DEFAULT_ACTION

# Create nodes
fetch = (
    new_node()
    .with_name("fetch")
    .with_exec_func(lambda r: Result({"users": ["Alice", "Bob", "Charlie"]}))
    .with_post_func(lambda shared, p, e: shared.set("data", e.value) or DEFAULT_ACTION)
)

transform = (
    new_node()
    .with_name("transform")
    .with_prep_func(lambda shared: Result(shared.get_dict("data")))
    .with_exec_func(lambda r: Result([name.upper() for name in r.must_dict()["users"]]))
    .with_post_func(lambda shared, p, e: shared.set("result", e.value) or DEFAULT_ACTION)
)

# Connect into a flow
flow = Flow(fetch)
flow.connect(fetch, DEFAULT_ACTION, transform)

# Run
shared = SharedStore()
flow.run_flow(shared)
print(shared.get_list("result"))  # ['ALICE', 'BOB', 'CHARLIE']

Core Concepts

Nodes

Every node runs three phases: prep (read inputs) → exec (do the work, retryable) → post (write results, return an action).

node = (
    new_node()
    .with_max_retries(3)
    .with_wait(1.0)
    .with_prep_func(lambda shared: Result(shared.get_string("input")))
    .with_exec_func(lambda r: Result(r.must_string().upper()))
    .with_post_func(lambda shared, p, e: shared.set("output", e.value) or DEFAULT_ACTION)
)

Or subclass for stateful nodes:

class MyNode(BaseNode):
    def __init__(self):
        super().__init__(max_retries=3, wait=0.5)

    def prep(self, shared):
        return shared.get_string("input")

    def exec(self, prep_result):
        return prep_result.upper()

    def post(self, shared, prep_result, exec_result):
        shared.set("output", exec_result)
        return DEFAULT_ACTION

Flows

Connect nodes with action-based routing to build directed graphs:

flow = Flow(start_node)
flow.connect(start_node, "success", next_node)
flow.connect(start_node, "error", error_node)
flow.connect(next_node, DEFAULT_ACTION, final_node)
flow.run_flow(shared)

Flows implement the Node interface - nest them inside other flows.

SharedStore

Thread-safe key-value store with typed getters:

shared = SharedStore()
shared.set("count", 42)
shared.get_int("count")          # 42
shared.get_int_or("missing", -1) # -1
shared.bind("user", UserModel)   # pydantic or dataclass

Batch Processing

Process lists of items with configurable concurrency:

from orbyt import new_batch_node

batch = (
    new_batch_node()
    .with_batch_concurrency(5)
    .with_batch_error_handling(True)
    .with_prep_func(lambda shared: [Result(url) for url in shared.get_list("urls")])
    .with_exec_func(lambda r: Result(fetch(r.must_string())))
    .with_post_func(lambda shared, items, results: ...)
)

Retry + Fallback

node = (
    new_node()
    .with_max_retries(3)
    .with_wait(1.0)
    .with_exec_func(call_flaky_api)
    .with_exec_fallback_func(lambda prep, err: {"status": "cached"})
)

Async

Every primitive has an async counterpart. Phase functions may be sync or async, and async flows can contain plain sync nodes — each phase is awaited only when it returns a coroutine.

import asyncio
from orbyt import AsyncFlow, new_async_node, run_async, Result, SharedStore, DEFAULT_ACTION

fetch = (
    new_async_node()
    .with_name("fetch")
    .with_exec_func(async_fetch_users)  # a coroutine function
    .with_post_func(lambda s, p, e: s.set("data", e.value) or DEFAULT_ACTION)
)

flow = AsyncFlow(fetch, max_steps=50)  # max_steps guards against runaway cycles
flow.connect(fetch, DEFAULT_ACTION, transform)

await flow.run_flow(SharedStore())

run_async mirrors run (retries sleep via asyncio.sleep, same fallback behavior). new_async_batch_node() runs items concurrently with an asyncio.Semaphore instead of a thread pool — the right model for I/O-bound work. AsyncFlow.max_steps turns infinite loops into a loud RuntimeError.

Mermaid Graph Export

flow = Flow(node_a, name="My Flow")
flow.connect(node_a, DEFAULT_ACTION, node_b)
print(flow.to_mermaid())

API Reference

Core Types

Type	Description
Node	Abstract base class - implement prep, exec, post
BaseNode	Default implementation with retry/batch config
CustomNode	Node with function fields
NodeBuilder	Fluent builder from new_node()
Flow	Directed graph of nodes (also a Node)
SharedStore	Thread-safe key-value store
Result	Value wrapper with typed accessors
BatchNode	Node for processing lists
WorkerPool	Fixed-size thread pool
AsyncNode	Abstract async node - async prep/exec/post
AsyncBaseNode	Default async node with retry/batch config
AsyncFlow	Async directed graph (also an AsyncNode); max_steps loop guard
AsyncBatchNode	Async list processing, concurrency via asyncio.Semaphore

Factory Functions

Function	Description
new_node()	Create a NodeBuilder
new_batch_node()	Create a BatchNodeBuilder
run(node, shared)	Execute a node through prep→exec→post
new_async_node()	Create an AsyncNodeBuilder
new_async_batch_node()	Create an AsyncBatchNodeBuilder
run_async(node, shared)	Await a node through prep→exec→post (sync nodes too)
to_slice(v)	Convert any value to list

License

MIT