Metadata-Version: 2.4
Name: lambda-pipe
Version: 0.2.0
Summary: Compose deterministic pipelines for LLM tasks using typed combinators.
Project-URL: Repository, https://github.com/joey/lambda-pipe
License-Expression: MIT
License-File: LICENSE
Keywords: ai,combinators,llm,openrouter,pipeline
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Scientific/Engineering :: Artificial Intelligence
Requires-Python: >=3.11
Provides-Extra: anthropic
Requires-Dist: anthropic>=0.40; extra == 'anthropic'
Provides-Extra: google
Requires-Dist: google-genai>=1.0; extra == 'google'
Provides-Extra: openai
Requires-Dist: openai>=1.0; extra == 'openai'
Provides-Extra: openrouter
Requires-Dist: openrouter>=0.7.11; extra == 'openrouter'
Description-Content-Type: text/markdown

# lambda-pipe

Compose deterministic pipelines for LLM tasks using typed combinators.

```python
from lambda_pipe import split, map, reduce
from lambda_pipe.models import openrouter

summarize = (
    split(8)
    | map("Summarize this section:\n{chunk}")
    | reduce("Synthesize:\n{all}")
)

result = await summarize.run(document, openrouter("google/gemini-3-flash-preview"))
```

9 LLM calls, runs in parallel, done. You know the upper bound before you run it.

## Why

Most LLM frameworks let the model decide what to do next. That means you can't bound cost, you can't guarantee it finishes, and you need a big model to get good results.

lambda-pipe flips it: you write the control flow with combinators, the LLM only handles small bounded tasks. A [recent paper](https://arxiv.org/abs/2603.20105) showed this lets an 8B model match a 70B model.

## Install

```bash
uv add "lambda-pipe[openrouter]"
export OPENROUTER_API_KEY=your-key
```

Works with any model on [OpenRouter](https://openrouter.ai) (300+ models). Or use a provider directly:

```bash
uv add "lambda-pipe[openai]"      # OpenAI
uv add "lambda-pipe[anthropic]"   # Anthropic
uv add "lambda-pipe[google]"      # Google Gemini
```

## Combinators

There are 7. Pass a string to `map`, `reduce`, or `filter` to make an LLM call.

| Combinator | What it does | LLM? |
|---|---|---|
| `split(k)` | Chop text into k chunks | no |
| `peek(start, n)` | Grab a substring | no |
| `map(fn)` | Run fn on each chunk (parallel) | depends on fn |
| `filter(pred)` | Keep chunks where pred says "yes" | depends on pred |
| `reduce(fn)` | Combine everything into one | depends on fn |
| `concat()` | Join chunks back together | no |
| `cross()` | All pairs, cartesian product | no |

Compose them with `|`. That's the whole API.

## Patterns

**Summarize** — split, summarize each chunk in parallel, synthesize:

```python
split(8) | map("Summarize:\n{chunk}") | reduce("Synthesize:\n{all}")
```

**Search** — split, toss irrelevant chunks, extract from what's left:

```python
split(20)
| filter("Relevant to '{query}'? yes/no\n{chunk}")
| map("Extract answer:\n{chunk}")
| reduce("Best answer:\n{all}")
```

**Find contradictions** — extract claims, pair them all up (free), check each pair:

```python
split(10)
| map("Extract claims:\n{chunk}")
| cross()                                  # 45 pairs, zero LLM calls
| map("Do these contradict?\n{chunk}")
| reduce("Summarize:\n{all}")
```

## Planning

Call `.plan()` before `.run()` to get a static cost model. No API calls, just math.

For straight-line pipelines, the call count is exact. If you use model-gated branching like `filter()`, the plan is a worst-case upper bound.

```python
plan = pipeline.plan(text, model="google/gemini-3-flash-preview")
print(plan)
# Plan(
#   llm_calls=9
#   estimated_cost=$0.0009
#   recursion_depth=0
# )
```

## Recursion

Got a 200-page PDF? `.recursive()` auto-subdivides until chunks fit the model's context window.

```python
big = (
    split(8)
    | map("Summarize:\n{chunk}")
    | reduce("Synthesize:\n{all}")
).recursive(tau="auto")
```

`tau="auto"` picks the threshold from the model name. Built-in providers set it for you. For a custom callable, tag it first:

```python
from lambda_pipe import named_model

async def local_model(prompt: str) -> str:
    ...

result = await big.run(text, named_model("openai/o4-mini", local_model))
```

Or pass a token count directly.

## Examples

```bash
# summarize text
uv run python examples/summarize.py

# search a chemistry textbook
uv run python examples/search.py "What is nuclear fusion?"

# compare 5 models on the same query
uv run python examples/compare.py

# compare one-shot prompting vs a chunked pipeline
uv run python examples/pipeline_vs_one_shot.py
```

## Paper

Inspired by [The Y-Combinator for LLMs](https://arxiv.org/abs/2603.20105) (Roy et al., 2026). This is a small combinator runtime and planner, not a reproduction of the paper's full system.
