How To Annotated Reducers

How-To: TypedDict Annotated Channel Reducers

Channel reducers specify how state values are aggregated when written to by multiple nodes or across graph cycles.

---

1. List Concatenation (operator.add)

import operator
from typing import TypedDict, Annotated, List
from langgraph.graph import StateGraph
import agentmesh_adapter

class State(TypedDict):
    messages: Annotated[List[str], operator.add]

def agent_a(s): return {"messages": ["Message from A"]}
def agent_b(s): return {"messages": ["Message from B"]}

builder = StateGraph(State)
builder.add_node("a", agent_a)
builder.add_node("b", agent_b)
builder.add_edge("__start__", "a")
builder.add_edge("a", "b")
builder.add_edge("b", "__end__")

app = agentmesh_adapter.compile(builder)
result = app.invoke({"messages": ["Initial"]})
assert result["messages"] == ["Initial", "Message from A", "Message from B"]

---

2. Custom Binary Reducers

You can provide custom aggregation logic:

def merge_dicts(old: dict, new: dict) -> dict:
    merged = old.copy()
    merged.update(new)
    return merged

class State(TypedDict):
    metrics: Annotated[dict, merge_dicts]
    counter: Annotated[int, lambda a, b: a + b]

---

3. How C++ Handles Reducers Under the Hood

When a wave finishes execution, the C++ engine iterates through modified keys and invokes the registered reducer pointer (PyObject*):

auto redIt = reducers_.find(keyStr);
if (redIt != reducers_.end() && redIt->second != nullptr) {
    PyObject* redArgs = PyTuple_Pack(2, existingVal, val);
    PyObject* reducedVal = PyObject_CallObject(redIt->second, redArgs);
    Py_DECREF(redArgs);
    PyDict_SetItem(stateObj, key, reducedVal);
    Py_DECREF(reducedVal);
}
```\n