thinkagain

# ThinkAgain

[![PyPI version](https://img.shields.io/pypi/v/thinkagain.svg)](https://pypi.org/project/thinkagain/)
[![License](https://img.shields.io/pypi/l/thinkagain.svg)](https://github.com/BDHU/thinkagain/blob/main/LICENSE)
[![GitHub Workflow Status](https://img.shields.io/github/actions/workflow/status/BDHU/thinkagain/test.yml?branch=main)](https://github.com/BDHU/thinkagain/actions/workflows/test.yml)
[![uv](https://img.shields.io/endpoint?url=https://raw.githubusercontent.com/astral-sh/uv/main/assets/badge/v0.json)](https://github.com/astral-sh/uv)

A minimal framework for building distributed AI pipelines. Write async functions, compose them with `.go()`, and scale them across clusters.

## Features

- **Dynamic execution** – Submit tasks with `.go()` and let the scheduler handle parallelism automatically
- **Transparent distribution** – `@service` marks classes for distributed execution
- **Automatic parallelism** – Independent tasks run in parallel without extra code
- **Built-in profiling** – Automatic dependency tracking and performance metrics
- **Minimal** – Clean API with no complex schedulers or DSLs

## Install

```bash
pip install thinkagain
# or with uv
uv add thinkagain
```

### Basic Usage

```python
import thinkagain as ta

@ta.op
async def add_one(x: int) -> int:
    return x + 1

@ta.op
async def multiply(x: int, factor: int) -> int:
    return x * factor

async def pipeline(x: int) -> int:
    # Submit tasks immediately (non-blocking)
    x_ref = add_one.go(x)
    result_ref = multiply.go(x_ref, 3)
    
    # Wait for result
    return await result_ref

result = await pipeline(5)  # Returns 18
```

### Distributed Execution

```python
import thinkagain as ta

# CPU-only service
@ta.service()
class Retriever:
    async def retrieve(self, query: str) -> list[str]:
        return ["doc1", "doc2", "doc3"]

# GPU service
@ta.service(gpus=4)
class Generator:
    def __init__(self):
        self.model = load_llm()

    async def generate(self, que