weftai

Composable AI workflows. A model emits a declarative plan of named steps; your application validates and executes it; results flow between steps by $name and never travel through the model.

One tool call answers a question that would otherwise take several, and the model never copies an identifier from one result into the next request.

pip install weftai

Python 3.12 or newer; pure Python, tested on Linux and Windows. The same library is on npm as weftai, with the same plan format and the same model-facing text.

Quick start

Define a collection and the operations that return it, then execute a plan:

import asyncio
from dataclasses import dataclass

from weftai import (
    BaseModel,
    FieldSpec,
    collection,
    create_registry,
    create_runtime,
    define_operation,
    standard_operations,
)
from weftai.schema.spec import object_schema


class Ticket(BaseModel):
    id: str
    title: str
    status: str


Tickets = collection(
    "tickets",
    Ticket,
    label=lambda ticket: ticket.title,
    key=lambda ticket: ticket.id,
    fields=lambda _ctx: [FieldSpec("status", lambda ticket: ticket.status)],
)

find = define_operation(
    {
        "name": "tickets.find",
        "description": "Every ticket. Start here.",
        "input": object_schema({}),
        "output": Tickets,
        "run": lambda run: run.ctx.tickets,
    }
)

runtime = create_runtime(
    {"registry": create_registry({"operations": [find, *standard_operations(Tickets)]})}
)


@dataclass
class Context:
    tickets: list[Ticket]


tickets = [
    Ticket(id="t1", title="Login page times out", status="open"),
    Ticket(id="t2", title="Export drops the last row", status="open"),
    Ticket(id="t3", title="Typo on the pricing page", status="closed"),
]


async def main() -> None:
    result = await runtime.execute(
        {
            "steps": [
                {"id": "all", "op": "tickets.find"},
                {
                    "id": "open",
                    "op": "tickets.filter",
                    "input": {"from": "$all", "filters": [{"field": "status", "value": "open"}]},
                },
            ]
        },
        {"ctx": Context(tickets), "session": {"id": "conversation-1"}},
    )
    print(result["text"])


asyncio.run(main())

result["text"] is what the model reads:

all (tickets): 3 matched
  1. Login page times out
  2. Export drops the last row
  3. Typo on the pricing page
  [intermediate step - preview only; reference $all to use the full set]
open (tickets): 2 matched
  1. Login page times out
  2. Export drops the last row

result["steps"] has each step's status, data, notices and timing, and result["trace"] is JSON for the CLI or an inspector. A later plan in the same session can use $open or $open[2]: results are kept in a session-scoped store, and positions always index the full set.

The plan a model writes

A plan is a list of steps. Each has an id, an operation op, and an input; a string such as "$all" in a field that takes a reference is the result of an earlier step, resolved by the runtime rather than copied by the model:

{ "steps": [
  { "id": "all",  "op": "tickets.find" },
  { "id": "open", "op": "tickets.filter", "input": { "from": "$all", "filters": [{ "field": "status", "value": "open" }] } }
] }

What you get from one definition

• Validation with actionable errors. An unknown operation suggests the nearest name, an unknown field lists the available ones, and a reference to the wrong kind of result says which kind was expected. A wrong question is an error, never an empty result.
• Typed references. ref(Tickets) renders as a $stepId string for the model and arrives in the handler as a collection. A whole reference written into a field that does not take one is refused before anything runs.
• Execution in dependency order. Independent steps run together, a failed step skips only its dependents, and timeouts and cancellation are built in.
• Token-budgeted formatting that never truncates silently: exact counts, showing 30 of 35, and a notice for every cap that applied.
• Standard operations for every collection: filter, count, countBy, distinct, mostCommon, first, pick and details.
• Decisions. A Decider answers short, typed questions (a choice, a score, a yes or no) with probabilities; NullDecider abstains, and weftai-providers has an HTTP one.

Packages

pip install "weftai[all]" installs the first four. All five are released together and share a version. The concepts, plan format and formatting rules are described in the weftai guides, written for the npm package; the Python API spells each name in snake_case.

License

MIT