A minimal RAG/agent orchestration framework

ThinkAgain

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

pip install thinkagain
# or with uv
uv add thinkagain

Basic Usage

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

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