Metadata-Version: 2.4
Name: open_orbyt
Version: 0.1.1
Summary: A workflow framework for Python. Build AI agents, data pipelines, and orchestration flows with nodes, graphs, and a shared store.
Project-URL: Homepage, https://github.com/erickweyunga/orbyt
Project-URL: Repository, https://github.com/erickweyunga/orbyt
Project-URL: Documentation, https://github.com/erickweyunga/orbyt#readme
Project-URL: Issues, https://github.com/erickweyunga/orbyt/issues
Author-email: Maverick Weyunga <maverickweyunga@gmail.com>
License-Expression: MIT
Keywords: agent,ai,dag,flow,graph,orchestration,pipeline,workflow
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Scientific/Engineering :: Artificial Intelligence
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Python: >=3.11
Requires-Dist: pydantic>=2.0
Description-Content-Type: text/markdown

# orbyt

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

## Install

```bash
pip install open_orbyt
```

The PyPI distribution is `open_orbyt`; the import package is `orbyt`:

```python
import orbyt
```

## Quick Start

```python
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).

```python
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:

```python
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:

```python
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:

```python
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:

```python
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

```python
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.

```python
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

```python
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` |

## Contributing

Contributions are welcome — issues, ideas, and pull requests.

The repo is a [uv](https://docs.astral.sh/uv/) workspace (`packages/orbyt` is
the library; `examples/` and `docs/` are members). Set up and check out a copy:

```bash
git clone https://github.com/erickweyunga/orbyt
cd orbyt
uv sync --all-packages     # install the workspace + dev tools
uv run pytest              # run the test suite
uvx pyrefly check packages/orbyt   # type-check
```

Guidelines:

- Keep the library's single runtime dependency (pydantic) — no new ones in `packages/orbyt`.
- Match the surrounding style; every public API carries a docstring.
- Add tests for new behaviour and keep the suite green.
- Open the PR against `main`.

## License

MIT
