Processing channels with tactics

TacticProcessor runs a tactic over the events of one channel. For each pending event it calls the tactic with the event's payload, which the tactic validates against its input type, and appends the output to a sink channel. Each output points back at its input through parent_ids and carries the input's correlation_id, so you can chain processors into pipelines and still trace every result to where it came from.

inbox ──subscription──▶ TacticProcessor(tactic) ──append──▶ triaged

Argument	Default	Meaning
`tactic`	required	Any lllm `Tactic`: a subclass, `as_tactic(fn)`, a native tactic with agents, a `RemoteTactic`.
`store`	required	`LocalStore` or `RemoteStore` for `process()`; any store, including `AsyncRemoteStore`, for `aprocess()`.
`source`	required	Channel to read.
`sink`	`None`	Channel to append outputs to. Without one, outputs are returned instead.
`consumer`	`"<tactic name>-<source>-processor"`, or `"<tactic name>-<source>-<kind>-processor"` with a `kind`	Subscription id, and the `source` field of output events. See Resuming.
`kind`	`None`	Only process source events of this kind.
`output_kind`	`"result"`	`kind` of the output events.
`errors`	`"raise"`	`"raise"` stops on the first failure and retries that event on the next call; `"record"` writes an error event and continues. See Failures.

The examples on this page run top to bottom as one session.

One stage

import re

from pydantic import BaseModel

from lllm import Tactic
from lllm.network import Event, LocalStore, TacticProcessor

URGENT = re.compile(r"\b(down|outage|failing|broken)\b", re.IGNORECASE)


class Ticket(BaseModel):
    text: str


class Triage(BaseModel):
    priority: str
    summary: str


class TriageTicket(Tactic[Ticket, Triage]):
    name = "triage"
    input_type = Ticket
    output_type = Triage

    def _run(self, input_value, *, context=None):
        priority = "urgent" if URGENT.search(input_value.text) else "normal"
        return Triage(priority=priority, summary=input_value.text[:40])


store = LocalStore(".lllm/network")
store.create_channel({"name": "inbox"})
store.create_channel({"name": "triaged"})


def submit(text, ticket):
    store.append_event(
        Event(
            channel="inbox",
            source="web-form",
            kind="message",
            payload={"text": text},
            correlation_id=ticket,
        )
    )


submit("Checkout is down", "T-100")
submit("How do I export my data?", "T-101")

processor = TacticProcessor(TriageTicket(), store, source="inbox", sink="triaged")
for event in processor.process():
    print(event.source, event.kind, event.correlation_id, event.payload)

triage-inbox-processor result T-100 {'priority': 'urgent', 'summary': 'Checkout is down'}
triage-inbox-processor result T-101 {'priority': 'normal', 'summary': 'How do I export my data?'}

process() returns the sink events it wrote. Each one holds:

Field	Value
`channel`	The sink.
`source`	The consumer name.
`kind`	`output_kind` (`"result"`).
`payload`	The tactic output as JSON. Pydantic models are dumped with `model_dump(mode="json")`.
`parent_ids`	`(input_event.id,)`
`correlation_id`	The input's `correlation_id`, or the input's `id` when it has none.

Create the sink channel before processing. The processor does not create channels.

What the tactic sees

The tactic gets the event's payload as its input and a CallContext that describes the event:

Context field	Value
`caller`	The consumer name.
`tags`	`{"channel": <source>}`
`metadata["event_id"]`	The input event's id.
`metadata["event_kind"]`	The input event's kind.
`metadata["correlation_id"]`	The input's correlation id, or its id.

from lllm import as_tactic


def show_context(payload: dict, *, context=None) -> dict:
    return {
        "caller": context.caller,
        "tags": context.tags,
        "event_kind": context.metadata["event_kind"],
        "correlation_id": context.metadata["correlation_id"],
    }


peek = TacticProcessor(as_tactic(show_context), store, source="inbox", consumer="peek")
print(peek.process(limit=1))

[{'caller': 'peek', 'tags': {'channel': 'inbox'}, 'event_kind': 'message', 'correlation_id': 'T-100'}]

Resuming

The consumer name is the id of a subscription that lives in the store. The first process() call creates it; every later processor with the same consumer, in this process or another one, continues from its cursor:

submit("Search is broken", "T-102")

restarted = TacticProcessor(TriageTicket(), store, source="inbox", sink="triaged")
print([event.correlation_id for event in restarted.process()])
print(store.get_subscription("triage-inbox-processor").cursor)

['T-102']
5

The default consumer joins the tactic's name (tactic.tactic_name, which is the function name for as_tactic(fn)), the source channel, and the kind filter if there is one: TriageTicket() reading inbox is triage-inbox-processor, and with kind="message" it would be triage-inbox-message-processor. Rules that follow from this:

Name consumers explicitly when two processors would share the default. The sink is not part of it, so a dry run without a sink and the real processor need different names (the dry run below uses consumer="dry-run").
If the default would contain whitespace, ?, or #, the constructor raises ValueError; pass consumer=.
To process a channel again from the start, use a new consumer name.
A subscription keeps the channel and kind filter it was created with. Reusing a consumer name with another channel or kind makes process() raise ValueError. With the default name, a new kind gives a new consumer, which starts at the beginning of the channel.
Run one worker per consumer name. Two workers sharing a consumer can process the same event twice, and with errors="raise" they can lose one.

Failures

process() handles one event at a time. It reads the next event at the subscription cursor, runs the tactic, appends the output to the sink, and only then acknowledges the event, which moves the cursor past it. Delivery is therefore at least once: an event that was not acknowledged, because processing raised or the worker stopped, runs again on the next call. A tactic can see the same event twice, and the sink can then hold more than one output for it, so tactics should tolerate repeats. errors decides what a failure does.

With errors="raise" (the default), the exception propagates and the failing event stays pending. The next call runs it again, before any event behind it:

from lllm import SchemaError

store.append_event(
    Event(channel="inbox", kind="message", payload={"subject": "no text field"}, correlation_id="T-103")
)
submit("The API is down", "T-104")

for attempt in (1, 2):
    try:
        processor.process()
    except SchemaError as exc:
        print(f"attempt {attempt} failed:", type(exc).__name__)

attempt 1 failed: SchemaError
attempt 2 failed: SchemaError

T-103 fails input validation on every call, and T-104 waits behind it. With errors="record", a failure is written to the sink as a kind="error" event with the same lineage fields, the event is acknowledged, and processing continues. This processor reads the same triage-inbox-processor subscription, so it starts at T-103:

store.append_event(
    Event(channel="inbox", kind="message", payload={"subject": "still no text"}, correlation_id="T-105")
)

recording = TacticProcessor(
    TriageTicket(), store, source="inbox", sink="triaged", errors="record"
)
results = recording.process()
print([(event.kind, event.correlation_id) for event in results])
print(results[0].payload["type"])

[('error', 'T-103'), ('result', 'T-104'), ('error', 'T-105')]
SchemaError

The error payload is {"error": str(exc), "type": type(exc).__name__}. Use "record" in long-running workers, so one bad event cannot block the subscription and failures stay visible in the store; read them back with store.query_events("triaged", kind="error"). "record" needs a sink; without one, it acts like "raise".

Any exception counts as a failure: payloads that do not match the tactic's input type raise SchemaError, the tactic's own exceptions pass through unchanged, and so do errors from writing the output, such as the InvalidPayloadError for an output the store cannot encode as JSON. If the error event cannot be written either, as when the sink channel does not exist, process() raises and the event stays pending. Create channels up front.

Without a sink

With no sink, process() returns the tactic outputs themselves. That is useful for dry runs, or when the results go somewhere other than a channel. limit caps how many results one call returns:

dry_run = TacticProcessor(TriageTicket(), store, source="inbox", consumer="dry-run")
print(dry_run.process(limit=2))

[Triage(priority='urgent', summary='Checkout is down'), Triage(priority='normal', summary='How do I export my data?')]

limit counts outputs and recorded errors alike. process() with no limit keeps going until the subscription has nothing pending.

Remote and async stores

The processor only calls store methods, so a store served over HTTP works the same way. With lllm network serve --port 7700 running on this store:

from lllm.network import RemoteStore

remote = RemoteStore("http://127.0.0.1:7700")
remote.append_event(
    {"channel": "inbox", "kind": "message", "payload": {"text": "Exports time out"}, "correlation_id": "T-106"}
)

worker = TacticProcessor(TriageTicket(), remote, source="inbox", sink="triaged")
print([event.correlation_id for event in worker.process()])

['T-106']

The remote worker continued the triage-inbox-processor subscription where the local processors left it.

aprocess() is the async variant. It calls tactic.arun() and awaits store calls that return coroutines, so it works with AsyncRemoteStore as well as the sync stores. process() needs a sync store and raises TypeError when given an async one. This example talks to the store server in-process through httpx.ASGITransport, which is also how to test a worker without a running server; in production, pass only the URL.

import asyncio

import httpx

from lllm.network import AsyncRemoteStore, create_app


async def main():
    remote = AsyncRemoteStore(
        "http://store.local", transport=httpx.ASGITransport(app=create_app(store))
    )
    await remote.append_event(
        {"channel": "inbox", "kind": "message", "payload": {"text": "Uploads are failing"}, "correlation_id": "T-107"}
    )
    worker = TacticProcessor(TriageTicket(), remote, source="inbox", sink="triaged")
    return [event.correlation_id for event in await worker.aprocess()]


print(asyncio.run(main()))

['T-107']

Pipelines

A pipeline is processors whose sink is the next one's source. Add a stage that drafts a reply for every triage result. The kind="result" filter skips the error events the first stage recorded:

class Reply(BaseModel):
    priority: str
    body: str


def draft_reply(result: Triage) -> Reply:
    opening = "We are on it." if result.priority == "urgent" else "Thanks for writing in."
    return Reply(priority=result.priority, body=f"{opening} Re: {result.summary}")


store.create_channel({"name": "replies"})

stages = [
    TacticProcessor(TriageTicket(), store, source="inbox", sink="triaged", errors="record"),
    TacticProcessor(
        as_tactic(draft_reply),
        store,
        source="triaged",
        sink="replies",
        consumer="draft-reply",
        kind="result",
    ),
]


def drain(processors):
    """Run every stage until none of them finds new events."""
    while any([processor.process() for processor in processors]):
        pass


submit("Refunds are failing for every customer", "T-108")
drain(stages)
print([(reply.correlation_id, reply.payload["priority"]) for reply in store.query_events("replies")])

[('T-100', 'urgent'), ('T-101', 'normal'), ('T-102', 'urgent'), ('T-104', 'urgent'), ('T-106', 'normal'), ('T-107', 'urgent'), ('T-108', 'urgent')]

The new ticket went through both stages in one drain; the earlier results reached the second stage because its draft-reply subscription started at the beginning of triaged.

Tracing a result

parent_ids lead from any output back to the event that started the chain:

def lineage(store, event):
    """Follow parent_ids back to the event that started the chain."""
    chain = [event]
    while chain[-1].parent_ids:
        chain.append(store.get_event(chain[-1].parent_ids[0]))
    return chain


for event in lineage(store, store.query_events("replies")[-1]):
    print(f"{event.channel:8} {event.source:23} {event.payload}")

replies  draft-reply             {'body': 'We are on it. Re: Refunds are failing for every customer', 'priority': 'urgent'}
triaged  triage-inbox-processor  {'priority': 'urgent', 'summary': 'Refunds are failing for every customer'}
inbox    web-form                {'text': 'Refunds are failing for every customer'}

correlation_id groups the whole thread across channels without walking links:

print([
    (event.channel, event.kind)
    for channel in ("inbox", "triaged", "replies")
    for event in store.query_events(channel)
    if event.correlation_id == "T-108"
])

[('inbox', 'message'), ('triaged', 'result'), ('replies', 'result')]

Fan-out

Each consumer has its own cursor, so several processors can read the same source and each sees every event:

def word_count(payload: dict) -> dict:
    return {"words": len(payload.get("text", "").split())}


store.create_channel({"name": "stats"})
stats = TacticProcessor(
    as_tactic(word_count), store, source="inbox", sink="stats", consumer="word-count"
)
print(len(stats.process()), "of", len(store.query_events("inbox")), "inbox events counted")

9 of 9 inbox events counted

Latest state

To give readers the newest result without replaying a channel, write a snapshot that points at the event it came from:

latest = store.query_events("replies")[-1]
store.put_snapshot(
    {"name": "latest-reply", "channel": "replies", "value": latest.payload, "source_event_id": latest.id}
)
print(store.get_snapshot("latest-reply").value["body"])

We are on it. Re: Refunds are failing for every customer

Running a worker

A worker is a loop around process() that sleeps when there is nothing to do:

import time


def run_worker(processor, *, idle_seconds=1.0):
    while True:
        if not processor.process(limit=100):
            time.sleep(idle_seconds)

The async version, for an AsyncRemoteStore or an event loop you already run:

import asyncio


async def run_async_worker(processor, *, idle_seconds=1.0):
    while True:
        if not await processor.aprocess(limit=100):
            await asyncio.sleep(idle_seconds)

Give each worker its own consumer name and errors="record", and stop it at any point: the next start resumes from the subscription cursor, and an event that was running when the worker stopped runs again. To serve the store these workers share, see Serving; to put workers, tactics, and channels behind one API, see AAAX.