Metadata-Version: 2.4
Name: tk-llmbda
Version: 0.2.0
Summary: Skill composition for LLMs — like lambda calculus, but the functions talk back
Project-URL: Repository, https://github.com/tkukurin/llmbda
Project-URL: Issues, https://github.com/tkukurin/llmbda/issues
Author: Toni Kukurin
License-Expression: MIT
License-File: LICENSE
Keywords: agents,composition,llm,pipeline,skill
Classifier: Development Status :: 3 - Alpha
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: Typing :: Typed
Requires-Python: >=3.11
Provides-Extra: dev
Requires-Dist: ipykernel; extra == 'dev'
Requires-Dist: pytest-cov>=5.0; extra == 'dev'
Requires-Dist: pytest>=8.0; extra == 'dev'
Requires-Dist: ruff>=0.11; extra == 'dev'
Description-Content-Type: text/markdown

# tk.llmbda

Skill composition for LLM pipelines. Chain deterministic and LLM-powered
steps into a skill; the runtime walks them in order until one resolves.

## Deterministic skill

```python
from tk.llmbda import Skill, SkillContext, StepResult, run_skill

def greet(ctx: SkillContext) -> StepResult:
    name = ctx.entry.get("name", "world")
    return StepResult(value=f"hello, {name}")

skill = Skill(name="greeter", steps=[Skill("greet", fn=greet)])
result = run_skill(skill, {"name": "λ"})
# SkillResult(skill="greeter", resolved_by=("greet",), value="hello, λ", ...)
```

## LLM skill

Self-contained with a fake model so the snippet runs as-is:

```python
from tk.llmbda import Skill, SkillContext, StepResult, lm, run_skill

def fake_model(*, messages, **_):
    return "2025-01-15"

@lm(fake_model, system_prompt="Extract a date. Return ISO format.")
def extract_date(ctx: SkillContext, call) -> StepResult:
    """Extract a date from natural language."""
    raw = call(messages=[{"role": "user", "content": ctx.entry["text"]}])
    return StepResult(value=raw)

skill = Skill(name="dates", steps=[Skill("extract", fn=extract_date)])
result = run_skill(skill, {"text": "let's meet on the 15th of January 2025"})
# SkillResult(skill="dates", resolved_by=("extract",), value="2025-01-15", ...)
```

## Multi-step with `ctx.prev`

Each step can access the previous step's result via `ctx.prev`:

```python
from tk.llmbda import Skill, SkillContext, StepResult, run_skill

def step_a(ctx: SkillContext) -> StepResult:
    return StepResult(value=ctx.entry["x"] * 2)

def step_b(ctx: SkillContext) -> StepResult:
    return StepResult(value=ctx.prev.value + 10)

skill = Skill(name="math", steps=[Skill("a", fn=step_a), Skill("b", fn=step_b)])
result = run_skill(skill, {"x": 5})
# result.value == 20
```

Before any step runs, `ctx.prev` is the `ROOT` sentinel (`ROOT.value is None`).
Use `ctx.prev is ROOT` to check if you're the first step.

## Short-circuit with `resolved=True`

Steps fall through by default (`resolved=False`). Set `resolved=True` to
stop early — remaining steps are skipped:

```python
def try_cache(ctx: SkillContext) -> StepResult:
    if ctx.entry.get("key") in CACHE:
        return StepResult(value=CACHE[ctx.entry["key"]], resolved=True)
    return StepResult(value=None)

def expensive(ctx: SkillContext) -> StepResult:
    return StepResult(value=compute(ctx.entry))

skill = Skill(name="s", steps=[Skill("cache", fn=try_cache), Skill("compute", fn=expensive)])
```

## Nested composition

`Skill` is the single composition primitive. Leaf (has `fn`), composite
(has `steps`), or orchestrator (has both):

```python
skill = Skill(
    name="analyzer",
    steps=[
        Skill("preprocess", steps=[
            Skill("normalize", fn=normalize),
            Skill("count_words", fn=count_words),
        ]),
        Skill("classify", steps=[
            Skill("tag_length", fn=tag_length),
        ]),
    ],
)
```

The runtime walks composites via DFS. All leaves share a single `ctx.trace`.

## Orchestrator: `fn` + `steps`

A skill with both `fn` and `steps` is an orchestrator — `fn` receives the
children as an explicit second argument and controls how they execute:

```python
def retry(ctx: SkillContext, steps: list[Skill]) -> StepResult:
    """Run children up to 3 times until valid."""
    inner = Skill(name="inner", steps=steps)
    for attempt in range(1, 4):
        r = run_skill(inner, ctx.entry)
        if r.metadata.get("valid"):
            return StepResult(value=r.value, metadata={"attempts": attempt}, resolved_by=r.resolved_by)
    return StepResult(value=r.value, metadata={"valid": False}, resolved_by=r.resolved_by)

skill = Skill(
    name="retry",
    fn=retry,
    steps=[
        Skill("ψ::extract", fn=extract_date_llm),
        Skill("ψ::verify", fn=verify_date),
    ],
)
```

- Leaf fns are `(ctx) -> StepResult`. Orchestrator fns are `(ctx, steps) -> StepResult`.
- Children run in a fresh `SkillContext` (via `run_skill`), so they don't
  see the outer trace. Pass data through `entry` if needed.

## Static validation

`check_skill` catches trace-key typos and forward references at definition
time via AST analysis:

```python
from tk.llmbda import check_skill

issues = check_skill(skill)
# ["'bad_step' references undeclared trace key 'typo'"]
```

- Validates orchestrator children as a separate scope.
- Checks `ctx.trace["key"]` and `ctx.trace.get("key")` patterns.

## Concepts

- **`Skill`** — recursive composition primitive. Leaf (`fn`), composite (`steps`), or orchestrator (`fn` + `steps`).
- **`StepResult.resolved`** — defaults to `False`; steps fall through. Set `True` to short-circuit.
- **`StepResult.resolved_by`** — inner resolution path as `tuple[str, ...]`. Orchestrators propagate it from nested `run_skill` calls; `SkillResult.resolved_by` prepends the step name, building a hierarchical path like `("orchestrator", "inner_step")`.
- **`ctx.prev`** — most recently executed step's `StepResult`. Starts as `ROOT` (`value=None`).
- **`ctx.trace`** — dict of all prior results keyed by step name. Raises informative `KeyError` on miss; use `.get()` for optional lookups.
- **`ctx.entry`** — the original input passed to `run_skill`.
- **`@lm(model, system_prompt=...)`** — binds model at decoration time. Decorated fn is `(ctx, call)` for leaves or `(ctx, steps, call)` for orchestrators; `call` prepends `system_prompt`.
- **`Skill.description`** — human-readable summary; falls back to fn docstring. Separate from `@lm` system prompts (read those via `fn.lm_system_prompt`).
- **`StepResult.metadata`** — auxiliary context: reasons, raw provider output, confidence.
- **`iter_skill`** — same as `run_skill` but yields `(name, result)` per step for streaming or early exit.
- **`check_skill`** — static validation of trace-key references. Catches typos and forward refs.
- **Test re-binding** — `lm(fake)(my_step.__wrapped__)` re-decorates with a different model.

## Examples

```bash
# deterministic + LLM date extraction with retry
OPENAI_API_KEY=sk-... uv run examples/readme.py

# calendar booking: regex parsers + LLM verifier
uv run examples/calendar_booking.py

# support triage: extraction, classification, validation loop
uv run examples/support_triage.py

# all 20 use cases in one file (no external deps)
uv run examples/showcase.py
```
