State and reducers

State schemas, channels, and the built-in reducers. See the State and reducers guide for the concepts.

agentflow.state

State model: channels, reducers, and update merging.

A workflow's state is a set of named channels. Each channel has a reducer: a pure function (current, update) -> new that folds a node's partial update into the current value. Reducers are what make concurrent node updates in a single super-step merge deterministically instead of racing on last-writer-wins.

Schemas are declared as a TypedDict subclass of :class:State, with each field optionally annotated with its reducer::

from typing import Annotated
from agentflow.state import State, append, add, last

class ChatState(State):
    messages: Annotated[list, append]   # concatenate
    tokens: Annotated[int, add]         # sum
    best: Annotated[float, last]        # overwrite (the default)

A field without an explicit reducer uses :func:last (last-value-wins).

This module is pure data: no async, no I/O, no engine imports.

Reducer module-attribute

Reducer = Callable[[Any, Any], Any]

State

Bases: TypedDict

Base class for a workflow state schema.

Subclass with Annotated[type, reducer] fields. Being total=False means every channel is optional at the type level — nodes emit partial updates, and a channel simply has no value until first written.

Source code in agentflow/state.py

class State(TypedDict, total=False):
    """Base class for a workflow state schema.

    Subclass with ``Annotated[type, reducer]`` fields. Being ``total=False``
    means every channel is optional at the type level — nodes emit partial
    updates, and a channel simply has no value until first written.
    """

Channel

Bases: NamedTuple

A resolved channel: its name and the reducer that folds its updates.

Source code in agentflow/state.py

class Channel(typing.NamedTuple):
    """A resolved channel: its name and the reducer that folds its updates."""

    name: str
    reducer: Reducer

last

last(current: Any, update: Any) -> Any

Last-value-wins. The default reducer.

Source code in agentflow/state.py

def last(current: Any, update: Any) -> Any:
    """Last-value-wins. The default reducer."""
    return update

append

append(current: Any, update: Any) -> list

Concatenate list updates onto a list channel.

current of None is treated as an empty list. A non-list update is appended as a single element; a list update is extended in order.

Source code in agentflow/state.py

def append(current: Any, update: Any) -> list:
    """Concatenate list updates onto a list channel.

    ``current`` of ``None`` is treated as an empty list. A non-list ``update``
    is appended as a single element; a list ``update`` is extended in order.
    """
    base = list(current) if current is not None else []
    if isinstance(update, (list, tuple)):
        base.extend(update)
    else:
        base.append(update)
    return base

add

add(current: Any, update: Any) -> Any

Numeric/additive reducer using +. None current starts at 0.

Source code in agentflow/state.py

def add(current: Any, update: Any) -> Any:
    """Numeric/additive reducer using ``+``. ``None`` current starts at 0."""
    if current is None:
        return update
    return current + update

merge

merge(current: Any, update: Any) -> dict

Shallow dict merge; keys in update win. None current is {}.

Source code in agentflow/state.py

def merge(current: Any, update: Any) -> dict:
    """Shallow dict merge; keys in ``update`` win. ``None`` current is ``{}``."""
    base = dict(current) if current is not None else {}
    if update:
        base.update(update)
    return base

by_key

by_key(policy: Mapping[str, Reducer]) -> Reducer

Build a reducer for a dict channel that folds each key with its own reducer.

Where :func:merge applies one policy (last-wins) to a whole dict, this lets a single structured channel hold keys that accumulate differently — for example an intents list that grows with :func:append beside an events list that is replaced with :func:last and an incumbent that is overwritten whole::

search = by_key({
    "intents":   append,   # grow
    "candidates": append,  # grow (concurrent workers each add one)
    "verified":  append,   # grow
    "events":    last,     # replace (planner clears what it consumed)
    "incumbent": last,     # overwrite the current best, whole
})

class OptState(State):
    search: Annotated[dict, search]

Folding rules, given current (None is treated as {}) and an update dict:

• Only keys present in update are folded; a key absent from update keeps its current value untouched.
• Each folded key k is combined as policy[k](current.get(k), update[k]) — so {"events": []} genuinely clears events (that is a last of an empty list, not a no-op), while {} or None is a no-op returning a copy of current.
• A key in update that is not in policy raises :class:KeyError. Writing an undeclared field is a bug (usually a typo), mirroring how the engine rejects writes to undeclared top-level channels.
• A key in current that is not in policy is passed through unchanged; the policy only governs keys an update actually writes.

The value under each key is combined whole by that key's reducer; this is one level of structure, not a recursive deep-merge. It is for distinct keys that accumulate differently, not for concurrent partial patches of the same nested object.

Source code in agentflow/state.py

def by_key(policy: Mapping[str, Reducer]) -> Reducer:
    """Build a reducer for a dict channel that folds each key with its own reducer.

    Where :func:`merge` applies one policy (last-wins) to a whole dict, this
    lets a single structured channel hold keys that accumulate differently —
    for example an ``intents`` list that grows with :func:`append` beside an
    ``events`` list that is replaced with :func:`last` and an ``incumbent`` that
    is overwritten whole::

        search = by_key({
            "intents":   append,   # grow
            "candidates": append,  # grow (concurrent workers each add one)
            "verified":  append,   # grow
            "events":    last,     # replace (planner clears what it consumed)
            "incumbent": last,     # overwrite the current best, whole
        })

        class OptState(State):
            search: Annotated[dict, search]

    Folding rules, given ``current`` (``None`` is treated as ``{}``) and an
    ``update`` dict:

    - Only keys present in ``update`` are folded; a key absent from ``update``
      keeps its ``current`` value untouched.
    - Each folded key ``k`` is combined as ``policy[k](current.get(k),
      update[k])`` — so ``{"events": []}`` genuinely clears ``events`` (that is
      a ``last`` of an empty list, not a no-op), while ``{}`` or ``None`` is a
      no-op returning a copy of ``current``.
    - A key in ``update`` that is not in ``policy`` raises :class:`KeyError`.
      Writing an undeclared field is a bug (usually a typo), mirroring how the
      engine rejects writes to undeclared top-level channels.
    - A key in ``current`` that is not in ``policy`` is passed through
      unchanged; the policy only governs keys an update actually writes.

    The value under each key is combined *whole* by that key's reducer; this is
    one level of structure, not a recursive deep-merge. It is for distinct keys
    that accumulate differently, not for concurrent partial patches of the same
    nested object.
    """
    policy = dict(policy)

    def _by_key(current: Any, update: Any) -> dict:
        base = dict(current) if current is not None else {}
        if not update:
            return base
        for key, value in update.items():
            reducer = policy.get(key)
            if reducer is None:
                raise KeyError(
                    f"by_key update wrote field {key!r} with no reducer in policy; "
                    f"declared fields: {sorted(policy)}"
                )
            base[key] = reducer(base.get(key), value)
        return base

    return _by_key

union

union(current: Any, update: Any) -> set

Set union. None current is the empty set.

Source code in agentflow/state.py

def union(current: Any, update: Any) -> set:
    """Set union. ``None`` current is the empty set."""
    base = set(current) if current is not None else set()
    base.update(update if isinstance(update, (set, frozenset, list, tuple)) else {update})
    return base