Skip to content
little big braindocs

Run a worker in Python (experimental)

View Markdown

A workflow worker runs your code for the messages of a workflow. Each message is one turn. A turn’s named steps save their outputs in LBB. After a crash, a turn runs again: it replays the saved steps and does each effect once.

lbb.workflows is the worker of the Python SDK. It follows the same rules as the TypeScript WorkflowWorker (Workflows). The server needs the workflow engine (--workflows, or a hosted stack).

Terminal window
pip install littlebigbrain

Start from a template

Section titled “Start from a template”

npm create @littlebigbrain creates a project from a workflow template. With --lang python, the project is the template’s Python port:

Terminal window
npm create @littlebigbrain -- --template values-extract --lang python my-values
cd my-values
python3 -m venv .venv
source .venv/bin/activate
pip install -e ".[test]"
pytest

pytest runs the workflow on the template’s samples, with mock models and no server. python -m values_extract try puts the samples on a graph of your stack (try-values-extract) and creates the workflow there. Then LBB_GRAPH=try-values-extract python -m values_extract start serves it. Both commands read LBB_URL and LBB_API_KEY.

Write a workflow

Section titled “Write a workflow”

A workflow is an async handler with a name and a version. It gets the turn’s context, the saved state and the message. It returns the new state and a reply.

from lbb.workflows import WorkflowContext, WorkflowResult, workflow




@workflow(name="counter", version="v1", initial_state={"count": 0})
async def counter(ctx: WorkflowContext, state: dict, message: dict) -> WorkflowResult[dict]:
    amount = await ctx.step("add", lambda s: message["add"])
    total = state["count"] + amount
    return WorkflowResult(state={"count": total}, result=total)

Start an instance of the workflow and send it a message. Reuse the message id when you send the same message again: the server admits it once.

import asyncio
import os


from lbb import AsyncLbbClient




def connect() -> AsyncLbbClient:
    return AsyncLbbClient(
        os.environ.get("LBB_URL", "http://127.0.0.1:7400"),
        api_key=os.environ.get("LBB_API_KEY"),
        graph="main",
    )




async def send() -> None:
    async with connect() as client:
        handle = await counter.start(client, "counter-1")
        await handle.send({"add": 2}, id="request-1")




asyncio.run(send())

Then run the worker. It claims the turns of the workflows you register and runs them.

from lbb.workflows import WorkflowWorker


client = connect()
WorkflowWorker(client, [counter], worker="counter-1").run()

run() serves until the process gets SIGTERM or Ctrl-C.

Steps and effect ids

Section titled “Steps and effect ids”

Put I/O, clock reads and randomness inside ctx.step(key, fn). The step’s output is saved as JSON before the turn goes on. Keys are unique in a turn. Await the steps one after the other.

An async step function runs on the event loop. A sync one runs in a thread. So a blocking call there (requests, the sync LbbClient) holds no other turn. It can report progress from that thread.

A step’s effect_id is <effect prefix>/<key>. It stays the same for the instance, the message and the step on every attempt. Pass it to an external service as its idempotency key.

The effect can succeed before its step is saved. The next attempt then sends it again, with the same id.

The model activities take the effect id. LBB keeps the first answer under it, and a repeat costs nothing.

from lbb import models
from lbb.workflows import lbb_activities


lbb = lbb_activities(client)
SCHEMA = {"type": "object", "properties": {"code": {"type": "string"}}}




@workflow(name="define-codes", version="v1", initial_state={})
async def define_codes(ctx: WorkflowContext, state: dict, message: dict) -> WorkflowResult[dict]:
    answer = await ctx.step(
        "define",
        lambda s: lbb.reason(message["prompt"], SCHEMA, effect_id=s.effect_id, label="code"),
        model=models.ReasonActivityResponse,
    )
    return WorkflowResult(state={"code": answer.value["code"]})

A step returns JSON. A step that returns a Pydantic model needs model=. The step then reads its saved JSON back as that model, also on a replay.

lbb_activities(client) has reason, decide and embed, on the generated models of lbb.models. A call names the turn’s run and the step that makes it (workflow.step, the step’s key). Its usage goes with the turn’s completion.

A turn replays its saved steps by key and kind, in order. A change of the step order fails the turn for good. Change the order only while no turn of the version runs.

A step that reported run results saves them beside its value. An SDK from before run results does not take them out again. On a replay, it gives your code the saved wrapper, not the step’s value. So roll workers forward only. Do not go back to an earlier SDK while newer turns can still run.

Waits, signals and continuations

Section titled “Waits, signals and continuations”

ctx.wait_for_signal waits for an approval or a callback without holding a worker. ctx.sleep saves a timer and releases the worker until it is due.

@workflow(name="review-draft", version="v1", initial_state={"sent": 0})
async def review_draft(ctx: WorkflowContext, state: dict, message: dict) -> WorkflowResult[dict]:
    draft = await ctx.step("draft", lambda s: f"Draft for {message['task']}")
    review = await ctx.wait_for_signal("review", name="approval", timeout_ms=86_400_000)
    if not review.received:
        return WorkflowResult(state=state, result="approval timed out")
    if not review.value["approved"]:
        return WorkflowResult(state=state, result="rejected")
    await ctx.sleep("cool-down", 1_000)
    return WorkflowResult(state={"sent": state["sent"] + 1}, result=draft)

Send the signal to the turn from your app. The same event id is delivered once.

from lbb.workflows import WorkflowHandle




async def approve() -> None:
    async with connect() as client:
        handle = WorkflowHandle(client, "review-1")
        await handle.signal("approval", {"approved": True}, turn=1, id="review-event-1")




asyncio.run(approve())

A turn holds at most 64 steps. For a large message, return ctx.continue_(state, next_message). The state and the next message are saved in one write. The next message joins the end of the inbox.

@workflow(name="pages", version="v1", initial_state={"read": 0})
async def pages(ctx: WorkflowContext, state: dict, message: dict) -> WorkflowResult[dict]:
    batch, rest = message["pages"][:20], message["pages"][20:]
    read = await ctx.step("read", lambda s: len(batch))
    after = {"read": state["read"] + read}
    if rest:
        return ctx.continue_(after, {"pages": rest})
    return WorkflowResult(state=after)

Ask a person

Section titled “Ask a person”

lbb.wait_for_review asks a person and waits for the answer. The worker is free while the turn waits. The item is a value, a choice, a grade or an answer.

@workflow(name="rooms", version="v1", initial_state={})
async def rooms(ctx: WorkflowContext, state: dict, message: dict) -> WorkflowResult[dict]:
    room = message["room"]
    asked = await lbb.wait_for_review(
        ctx,
        "floor",
        {
            "kind": "choice",
            "key": f"floor:{room}",
            "title": f"Which floor covering does {room} have?",
            "question": {
                "options": [
                    {"id": "parquet", "label": "Parquet"},
                    {"id": "tiles", "label": "Tiles"},
                ]
            },
        },
        timeout_ms=7 * 24 * 3_600_000,
    )
    floor = asked.answer.option if asked.answer else None
    return WorkflowResult(state={"floor": floor})

The step floor.ask files the item, with this run as the one that waits. The turn then waits for the answer in floor.answer. An answer that arrives before the wait is kept, and the wait returns at once.

When the wait times out, the step floor.read reads the item again. An answer whose signal did not arrive still counts.

A trigger shard serves many inputs, and a wait would hold them all. So wait_for_review refuses to run there. File the item with lbb.review, and take the answer with a review trigger. See Review in workflows.

Long steps

Section titled “Long steps”

A long step reports progress and saves a resume cursor with heartbeat. The next attempt of the step reads the cursor as checkpoint. progress reports what the step does now; the console draws it as a bar.

from lbb.workflows import WorkflowStepContext




async def crawl(s: WorkflowStepContext) -> dict:
    start = s.checkpoint or 0
    for page in range(start, 3):
        s.progress("reading pages", done=page, total=3, unit="pages")
        await s.heartbeat({"page": page}, checkpoint=page + 1)
    return {"pages": 3}




@workflow(name="crawler", version="v1", initial_state={})
async def crawler(ctx: WorkflowContext, state: dict, message: dict) -> WorkflowResult[dict]:
    read = await ctx.step("crawl", crawl, kind="tool", heartbeat_timeout_ms=30_000)
    return WorkflowResult(state=read)

A step has a timeout for each attempt, 5 minutes by default (timeout_ms). With heartbeat_timeout_ms, a step that reports no progress that long ends its attempt.

Work on the instances of a class

Section titled “Work on the instances of a class”

A triggered workflow watches a class and sends its new and changed instances to your workflow. triggered_workflow defines that workflow, and write_owned writes what it owns on the instances.

from lbb.workflows import OwnedWrite, entity_ref, triggered_workflow, write_owned


QUERY = """SELECT ?s ?text WHERE {{
  VALUES ?s {{ {} }} ?s <https://littlebigbrain.com/p/text> ?text }}"""




@triggered_workflow(name="word-count", version="v1")
async def word_count(ctx: WorkflowContext, message: models.WorkflowTriggerMessage, state: dict):
    async def count(s: WorkflowStepContext) -> dict:
        query = QUERY.format(" ".join(f"<{iri}>" for iri in message.instances))
        rows = await lbb.sparql(query, as_of_commit_seq=message.as_of_commit_seq)
        return {row["s"]: len(row["text"].split()) for row in rows.rows()}


    counts = await ctx.step("count", count)
    write = OwnedWrite(
        subjects=[entity_ref(iri, "Note") for iri in message.instances],
        properties=["word_count"],
        entity_properties=[
            {**entity_ref(iri, "Note"), "properties": {"word_count": words}}
            for iri, words in counts.items()
        ],
    )
    await write_owned(client, ctx, "write", write, graph=message.graph)

write_owned commits inside a step, with the step’s effect id as the commit’s idempotency key. A turn that runs again sends the same key. The server answers with the first commit.

The commit also carries the turn’s fence. A worker whose lease ended gets 409 workflow_fence_lost and writes nothing.

Create the triggered workflow on the graph. It names your workflow type and the properties it owns.

async def watch() -> None:
    async with connect() as client:
        await client.use_workflow_starter(
            "workflow",
            name="notes",
            class_="Note",
            fields=["text"],
            workflow={"workflow_type": "word-count", "version": "v1"},
            owns={"properties": ["word_count"]},
        )




asyncio.run(watch())

An instance the turn cannot do goes back to the watch with requeue_failed. It is sent again after 1, 4, then 16 minutes, at most 3 times.

Test a workflow before it writes

Section titled “Test a workflow before it writes”

capture_activities gives activities that write nothing. The models and the reads run, and they cost what they cost live. Each write_owned is checked with the commit’s dry run and kept. Each review item is kept and filed nowhere.

Make the workflow a function of its activities. A test then runs it on chosen instances, under its own instance and version. No live turn runs.

from lbb.workflows import LbbActivities, TriggeredWorkflowDefinition, capture_activities




def word_counter(lbb: LbbActivities, version: str = "v1") -> TriggeredWorkflowDefinition:
    @triggered_workflow(name="word-count", version=version)
    async def count(ctx: WorkflowContext, message: models.WorkflowTriggerMessage, state: dict):
        write = OwnedWrite(
            subjects=[entity_ref(iri, "Note") for iri in message.instances],
            properties=["word_count"],
            entity_properties=[
                {**entity_ref(iri, "Note"), "properties": {"word_count": 3}}
                for iri in message.instances
            ],
        )
        await lbb.write_owned(ctx, "write", write, graph=message.graph)


    return count




async def try_word_count(iris: list[str]) -> None:
    async with connect() as client:
        capture = capture_activities(lbb_activities(client), workflow="notes")
        test = word_counter(capture.activities, version="test")
        handle = await test.start(client, "word-count-test-1")
        await handle.send(
            {
                "type": "instances",
                "workflow": "notes",
                "graph": "main",
                "as_of_commit_seq": 1,
                "reason": "refresh",
                "instances": iris,
                "removed": [],
            },
            id="word-count-test-1",
        )
        await WorkflowWorker(client, [test], worker="test").run_once()
        for write in capture.result().writes:
            print(write.step_key, write.check, write.errors or "passes")




asyncio.run(try_word_count(["https://littlebigbrain.com/e/0123456789abcdef0123456789abcdef"]))

The model calls of a capture name the test (workflow="notes"). The call log shows them under tests/notes, and model checks leave them out.

A wait_for_review ends at once, unanswered. On a trigger shard it fails, as the live call fails. On an RDF-native graph, the dry run refuses the write, as the commit would.

Failures and waits

Section titled “Failures and waits”

An exception fails the attempt. The turn runs again, up to its max_attempts (3 by default). A WorkflowError fails the turn for good. A WorkflowWait waits until a time and counts no attempt. A step that meets 429 workflow_budget_exceeded waits that way by itself.

import time


from lbb.workflows import WorkflowError, WorkflowWait




@workflow(name="invoice", version="v1", initial_state={})
async def invoice(ctx: WorkflowContext, state: dict, message: dict) -> WorkflowResult[dict]:
    if "order" not in message:
        raise WorkflowError("an invoice message names its order")
    ready = await ctx.step("ready", lambda s: message.get("paid", False))
    if not ready:
        raise WorkflowWait(retry_at_ms=int(time.time() * 1000) + 60_000)
    return WorkflowResult(state={"last": message["order"]})

A turn that failed for good holds the inbox until a person retries or skips it. Start the instance with on_exhausted="skip" to keep the failed turn in the history and go on.

Run, scale and stop

Section titled “Run, scale and stop”

A worker runs at most concurrency turns at once (default 1). Each turn runs in its own asyncio task. Run more worker processes for more.

On SIGTERM the worker stops claiming. Its held turns finish within drain_seconds (default 30). Then it releases the rest.

A released turn fails with retry_at_ms. It counts no attempt and keeps its saved steps. Another worker claims it a second later. Ctrl-C (SIGINT), or a second SIGTERM, releases at once. A second Ctrl-C ends the process.

A claim can wait on the server when the stop comes. The worker lets it answer, and gives a turn it claimed back at once. The signal handlers need the main thread. Off it, run() installs none: call stop() from another thread.

run_async() serves inside your own event loop, for example beside a web app. stop() ends it in the same way.

async def serve() -> None:
    async with connect() as client:
        worker = WorkflowWorker(client, [counter, review_draft], worker="api-1", concurrency=4)
        serving = asyncio.create_task(worker.run_async())
        await asyncio.sleep(1)
        worker.stop()
        await serving




asyncio.run(serve())

A claim that meets a network failure, a 429 or a 5xx waits and claims again. Any other refusal of a claim ends the loop with its error.

The rules of a worker

Section titled “The rules of a worker”

The TypeScript and Python workers follow the same rules. A suite of scripted server exchanges in the SDKs’ repositories (conformance/workflow-worker) runs against both.

Request Rule
turns/claim Sent once. A failed claim is not sent again.
turns/heartbeat Every third of the lease. A failed heartbeat ends the attempt in this worker.
turns/checkpoint, turns/complete Retried. A retried completion sends the same body.
turns/fail Sent once. A wait sends retry_at_ms, at least a second ahead.
A 409 or 404 on a turn The turn stops. The worker sends nothing more for it.
409 workflow_fence_lost The turn stops. The turn’s next holder writes.

Try the template on a local stack

Section titled “Try the template on a local stack”

values-extract fills properties of a class from the text of its instances. Its Python port is in examples/values_extract of the Python SDK. One command starts a local stack with mock models and the Python worker:

Terminal window
npm run dev:stack -- --worker python:values-extract

The worker reads the six sample records and writes their values, each with its quote. It gives a person the values it cannot confirm, such as a size in another unit. The workflow templates guide describes the template.