Coverage for core/langgraph/_reducers.py: 100%
122 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-10-04 15:34 +0000
1"""
2_reducers.py — Core-owned reducer layer shared by channel reducers and the
3aggregate step mechanism (core/docs/builtins/aggregate.md).
5Two independent surfaces live here:
7* the live binary primitives (``_reduce_last``, ``_reduce_add_or_extend``) that
8 ``_channels.py`` binds to the built-in ``LastValue`` and ``Topic`` channel
9 types — those channels ignore the manifest ``operator``;
10* the list folds (``_fold_*``) plus ``apply_op`` that back the aggregate
11 builtin's operator catalogue.
13``BinaryOperatorAggregate`` is *not* dispatched through this module's shape
14primitives: its mandatory manifest ``operator`` is resolved by name through the
15name-keyed registry and selector in ``_channel_reducers.py``.
17Stdlib only. This module must not import ``runner`` or ``_builtins``: runner
18imports this module, never the reverse. Malformed input raises the typed
19``ReducerTypeError`` — a ``ValueError`` subclass, never a raw ``TypeError`` —
20and reducer errors propagate as typed ``ValueError`` rather than being mapped
21to any engine hard failure (ADR-0012, no silent fallbacks).
22"""
23from __future__ import annotations
25from typing import Any, Callable
28class ReducerTypeError(ValueError):
29 """A reducer received a value of the wrong type (still a ``ValueError``)."""
32VALID_OPERATORS = frozenset({
33 "union",
34 "concat",
35 "merge",
36 "last",
37 "first",
38 "sum",
39 "avg",
40 "min",
41 "max",
42 "count",
43 "group",
44 "any",
45 "all",
46})
49# ---------------------------------------------------------------------------
50# Binary primitives (relocated from runner.py; behaviour unchanged)
51# ---------------------------------------------------------------------------
53def _reduce_last(left: Any, right: Any) -> Any:
54 """Replace: the newest write wins."""
55 return right
58def _reduce_add_or_extend(left: list, right: Any) -> list:
59 """Append: concatenate list writes, wrap a scalar write in a list."""
60 _require_list(left)
61 if isinstance(right, list):
62 return left + right
63 return left + [right]
66def _dedupe_first_occurrence(values: list, key: Callable[[Any], Any] | None = None) -> list:
67 """Drop later duplicates, preserving first-occurrence order.
69 ``key`` derives the identity of an item; without it the whole value is the
70 identity, so the channel path's behaviour is unchanged.
71 """
72 unique: list = []
73 seen: list = []
74 for value in values:
75 identity = value if key is None else key(value)
76 if identity in seen:
77 continue
78 seen.append(identity)
79 unique.append(value)
80 return unique
83def _reduce_union_by_key(left: dict, right: dict, key: Callable[[Any], Any] | None = None) -> dict:
84 """First-wins key union; list values concatenate and de-duplicate."""
85 merged = dict(left)
86 for right_key, right_value in right.items():
87 if right_key not in merged:
88 merged[right_key] = right_value
89 continue
90 left_value = merged[right_key]
91 if isinstance(left_value, list) and isinstance(right_value, list):
92 merged[right_key] = _dedupe_first_occurrence(left_value + right_value, key)
93 return merged
96def _reduce_dict_merge(left: dict, right: dict) -> dict:
97 """Shallow merge: later keys win."""
98 return {**left, **right}
101def accumulate_map_reduce(left: Any, right: Any) -> dict:
102 """Fold per-step ``{index: item_result}`` maps from concurrent ``Send`` workers.
104 ``left`` and ``right`` are ``{step_id: {index: item_result}}``. A ``None``
105 slot clears that step's entry (the join consumes it); a later write for the
106 same step id merges by index. Item order is recovered by the join's sort,
107 never by completion order.
108 """
109 merged: dict = {
110 step_id: dict(slot)
111 for step_id, slot in (left or {}).items()
112 }
113 for step_id, slot in (right or {}).items():
114 if slot is None:
115 merged.pop(step_id, None)
116 continue
117 bucket = dict(merged.get(step_id) or {})
118 bucket.update(slot)
119 merged[step_id] = bucket
120 return merged
123def _is_collection_map(value: Any) -> bool:
124 """True for a non-empty dict whose every value is a list or set."""
125 return (
126 isinstance(value, dict)
127 and bool(value)
128 and all(isinstance(item, (list, set)) for item in value.values())
129 )
132def _require_mapping(value: Any) -> None:
133 if not isinstance(value, dict):
134 raise ReducerTypeError(
135 "dict reducer expected a mapping but received "
136 f"{type(value).__name__}"
137 )
140def _require_list(value: Any) -> None:
141 if not isinstance(value, list):
142 raise ReducerTypeError(
143 "list reducer expected a list but received "
144 f"{type(value).__name__}"
145 )
148# ---------------------------------------------------------------------------
149# List folds (aggregate operator catalogue)
150# ---------------------------------------------------------------------------
152def _require_numeric(value: Any) -> int | float:
153 """Numeric means int/float; bool is deliberately excluded."""
154 if isinstance(value, bool) or not isinstance(value, (int, float)):
155 raise ValueError(
156 "aggregate numeric operator received a non-numeric value: "
157 f"{type(value).__name__}"
158 )
159 return value
162def _as_list(value: Any) -> list:
163 return list(value) if value is not None else []
166def _fold_union(items: list, acc: Any, key: Callable[[Any], Any] | None = None) -> list:
167 return _dedupe_first_occurrence(_as_list(acc) + list(items), key)
170def _fold_concat(items: list, acc: Any, key: Callable[[Any], Any] | None = None) -> list:
171 return _reduce_add_or_extend(_as_list(acc), list(items))
174def _fold_merge(items: list, acc: Any, key: Callable[[Any], Any] | None = None) -> dict:
175 merged = dict(acc) if acc is not None else {}
176 for item in items:
177 merged = _reduce_dict_merge(merged, item)
178 return merged
181def _fold_last(items: list, acc: Any, key: Callable[[Any], Any] | None = None) -> Any:
182 return _reduce_last(acc, items[-1]) if items else acc
185def _fold_first(items: list, acc: Any, key: Callable[[Any], Any] | None = None) -> Any:
186 if acc is not None:
187 return acc
188 return items[0] if items else None
191def _fold_sum(items: list, acc: Any, key: Callable[[Any], Any] | None = None) -> int | float:
192 total = 0 if acc is None else acc
193 for item in items:
194 total += _require_numeric(item)
195 return total
198def _fold_avg(items: list, acc: Any, key: Callable[[Any], Any] | None = None) -> dict | None:
199 total, count = (acc["sum"], acc["count"]) if acc is not None else (0, 0)
200 for item in items:
201 total += _require_numeric(item)
202 count += 1
203 if count == 0:
204 return None
205 return {"sum": total, "count": count}
208def _fold_min(items: list, acc: Any, key: Callable[[Any], Any] | None = None) -> int | float | None:
209 minimum = acc
210 for item in items:
211 value = _require_numeric(item)
212 if minimum is None or value < minimum:
213 minimum = value
214 return minimum
217def _fold_max(items: list, acc: Any, key: Callable[[Any], Any] | None = None) -> int | float | None:
218 maximum = acc
219 for item in items:
220 value = _require_numeric(item)
221 if maximum is None or value > maximum:
222 maximum = value
223 return maximum
226def _fold_count(items: list, acc: Any, key: Callable[[Any], Any] | None = None) -> int:
227 return (0 if acc is None else acc) + len(items)
230def _fold_group(items: list, acc: Any, key: Callable[[Any], Any] | None = None) -> dict:
231 buckets = {bucket: list(values) for bucket, values in acc.items()} if acc else {}
232 for item in items:
233 bucket = item if key is None else key(item)
234 buckets.setdefault(bucket, []).append(item)
235 return buckets
238def _fold_any(items: list, acc: Any, key: Callable[[Any], Any] | None = None) -> bool:
239 if acc:
240 return True
241 return any(items)
244def _fold_all(items: list, acc: Any, key: Callable[[Any], Any] | None = None) -> bool:
245 if acc is not None and not acc:
246 return False
247 return all(items)
250_OP_REDUCERS: dict[str, Callable[..., Any]] = {
251 "union": _fold_union,
252 "concat": _fold_concat,
253 "merge": _fold_merge,
254 "last": _fold_last,
255 "first": _fold_first,
256 "sum": _fold_sum,
257 "avg": _fold_avg,
258 "min": _fold_min,
259 "max": _fold_max,
260 "count": _fold_count,
261 "group": _fold_group,
262 "any": _fold_any,
263 "all": _fold_all,
264}
267def apply_op(
268 op_name: str,
269 items: Any,
270 acc: Any = None,
271 key: Callable[[Any], Any] | None = None,
272) -> Any:
273 """Fold ``items`` into ``acc`` with the named aggregate operator.
275 The only dispatch point the aggregate builtin uses. ``key`` is the
276 already-resolved identity/grouping callable for ``union`` and ``group``;
277 path resolution is the caller's concern. ``acc`` is never mutated.
278 """
279 try:
280 reducer = _OP_REDUCERS[op_name]
281 except KeyError:
282 raise ValueError(f"unknown aggregate operator: {op_name!r}") from None
283 return reducer(_as_list(items), acc, key)