Coverage for core/langgraph/_reducers.py: 100%
117 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-25 03:42 +0000
1"""
2_reducers.py — Core-owned reducer layer shared by channel reducers and the
3aggregate step mechanism (core/docs/builtins/aggregate.md).
5Two surfaces, one implementation:
7* binary primitives (``_reduce_*``) consumed by ``_channels.py``'s
8 ``select_reducer`` ``BinaryOperatorAggregate`` dispatch;
9* list folds (``_fold_*``) plus ``apply_op`` consumed by the aggregate builtin.
11Stdlib only. This module must not import ``runner`` or ``_builtins``: runner
12imports this module, never the reverse. Non-numeric input to a numeric
13operator always raises ``ValueError``; a later phase maps that to the engine's
14hard failure (ADR-0012, no silent fallbacks).
15"""
16from __future__ import annotations
18from typing import Any, Callable
20VALID_OPERATORS = frozenset({
21 "union",
22 "concat",
23 "merge",
24 "last",
25 "first",
26 "sum",
27 "avg",
28 "min",
29 "max",
30 "count",
31 "group",
32 "any",
33 "all",
34})
37# ---------------------------------------------------------------------------
38# Binary primitives (relocated from runner.py; behaviour unchanged)
39# ---------------------------------------------------------------------------
41def _reduce_last(left: Any, right: Any) -> Any:
42 """Replace: the newest write wins."""
43 return right
46def _reduce_add_or_extend(left: list, right: Any) -> list:
47 """Append: concatenate list writes, wrap a scalar write in a list."""
48 if isinstance(right, list):
49 return left + right
50 return left + [right]
53def _dedupe_first_occurrence(values: list, key: Callable[[Any], Any] | None = None) -> list:
54 """Drop later duplicates, preserving first-occurrence order.
56 ``key`` derives the identity of an item; without it the whole value is the
57 identity, so the channel path's behaviour is unchanged.
58 """
59 unique: list = []
60 seen: list = []
61 for value in values:
62 identity = value if key is None else key(value)
63 if identity in seen:
64 continue
65 seen.append(identity)
66 unique.append(value)
67 return unique
70def _reduce_union_by_key(left: dict, right: dict, key: Callable[[Any], Any] | None = None) -> dict:
71 """First-wins key union; list values concatenate and de-duplicate."""
72 merged = dict(left)
73 for right_key, right_value in right.items():
74 if right_key not in merged:
75 merged[right_key] = right_value
76 continue
77 left_value = merged[right_key]
78 if isinstance(left_value, list) and isinstance(right_value, list):
79 merged[right_key] = _dedupe_first_occurrence(left_value + right_value, key)
80 return merged
83def _reduce_dict_merge(left: dict, right: dict) -> dict:
84 """Shallow merge: later keys win."""
85 return {**left, **right}
88def accumulate_map_reduce(left: Any, right: Any) -> dict:
89 """Fold per-step ``{index: item_result}`` maps from concurrent ``Send`` workers.
91 ``left`` and ``right`` are ``{step_id: {index: item_result}}``. A ``None``
92 slot clears that step's entry (the join consumes it); a later write for the
93 same step id merges by index. Item order is recovered by the join's sort,
94 never by completion order.
95 """
96 merged: dict = {
97 step_id: dict(slot)
98 for step_id, slot in (left or {}).items()
99 }
100 for step_id, slot in (right or {}).items():
101 if slot is None:
102 merged.pop(step_id, None)
103 continue
104 bucket = dict(merged.get(step_id) or {})
105 bucket.update(slot)
106 merged[step_id] = bucket
107 return merged
110def _is_collection_map(value: Any) -> bool:
111 """True for a non-empty dict whose every value is a list or set."""
112 return (
113 isinstance(value, dict)
114 and bool(value)
115 and all(isinstance(item, (list, set)) for item in value.values())
116 )
119def _require_mapping(right: Any) -> None:
120 if not isinstance(right, dict):
121 raise ValueError(
122 "BinaryOperatorAggregate reducer selected a dict reducer but the "
123 f"incoming write is {type(right).__name__}, not a dict"
124 )
127# ---------------------------------------------------------------------------
128# List folds (aggregate operator catalogue)
129# ---------------------------------------------------------------------------
131def _require_numeric(value: Any) -> int | float:
132 """Numeric means int/float; bool is deliberately excluded."""
133 if isinstance(value, bool) or not isinstance(value, (int, float)):
134 raise ValueError(
135 "aggregate numeric operator received a non-numeric value: "
136 f"{type(value).__name__}"
137 )
138 return value
141def _as_list(value: Any) -> list:
142 return list(value) if value is not None else []
145def _fold_union(items: list, acc: Any, key: Callable[[Any], Any] | None = None) -> list:
146 return _dedupe_first_occurrence(_as_list(acc) + list(items), key)
149def _fold_concat(items: list, acc: Any, key: Callable[[Any], Any] | None = None) -> list:
150 return _reduce_add_or_extend(_as_list(acc), list(items))
153def _fold_merge(items: list, acc: Any, key: Callable[[Any], Any] | None = None) -> dict:
154 merged = dict(acc) if acc is not None else {}
155 for item in items:
156 merged = _reduce_dict_merge(merged, item)
157 return merged
160def _fold_last(items: list, acc: Any, key: Callable[[Any], Any] | None = None) -> Any:
161 return _reduce_last(acc, items[-1]) if items else acc
164def _fold_first(items: list, acc: Any, key: Callable[[Any], Any] | None = None) -> Any:
165 if acc is not None:
166 return acc
167 return items[0] if items else None
170def _fold_sum(items: list, acc: Any, key: Callable[[Any], Any] | None = None) -> int | float:
171 total = 0 if acc is None else acc
172 for item in items:
173 total += _require_numeric(item)
174 return total
177def _fold_avg(items: list, acc: Any, key: Callable[[Any], Any] | None = None) -> dict | None:
178 total, count = (acc["sum"], acc["count"]) if acc is not None else (0, 0)
179 for item in items:
180 total += _require_numeric(item)
181 count += 1
182 if count == 0:
183 return None
184 return {"sum": total, "count": count}
187def _fold_min(items: list, acc: Any, key: Callable[[Any], Any] | None = None) -> int | float | None:
188 minimum = acc
189 for item in items:
190 value = _require_numeric(item)
191 if minimum is None or value < minimum:
192 minimum = value
193 return minimum
196def _fold_max(items: list, acc: Any, key: Callable[[Any], Any] | None = None) -> int | float | None:
197 maximum = acc
198 for item in items:
199 value = _require_numeric(item)
200 if maximum is None or value > maximum:
201 maximum = value
202 return maximum
205def _fold_count(items: list, acc: Any, key: Callable[[Any], Any] | None = None) -> int:
206 return (0 if acc is None else acc) + len(items)
209def _fold_group(items: list, acc: Any, key: Callable[[Any], Any] | None = None) -> dict:
210 buckets = {bucket: list(values) for bucket, values in acc.items()} if acc else {}
211 for item in items:
212 bucket = item if key is None else key(item)
213 buckets.setdefault(bucket, []).append(item)
214 return buckets
217def _fold_any(items: list, acc: Any, key: Callable[[Any], Any] | None = None) -> bool:
218 if acc:
219 return True
220 return any(items)
223def _fold_all(items: list, acc: Any, key: Callable[[Any], Any] | None = None) -> bool:
224 if acc is not None and not acc:
225 return False
226 return all(items)
229_OP_REDUCERS: dict[str, Callable[..., Any]] = {
230 "union": _fold_union,
231 "concat": _fold_concat,
232 "merge": _fold_merge,
233 "last": _fold_last,
234 "first": _fold_first,
235 "sum": _fold_sum,
236 "avg": _fold_avg,
237 "min": _fold_min,
238 "max": _fold_max,
239 "count": _fold_count,
240 "group": _fold_group,
241 "any": _fold_any,
242 "all": _fold_all,
243}
246def apply_op(
247 op_name: str,
248 items: Any,
249 acc: Any = None,
250 key: Callable[[Any], Any] | None = None,
251) -> Any:
252 """Fold ``items`` into ``acc`` with the named aggregate operator.
254 The only dispatch point the aggregate builtin uses. ``key`` is the
255 already-resolved identity/grouping callable for ``union`` and ``group``;
256 path resolution is the caller's concern. ``acc`` is never mutated.
257 """
258 try:
259 reducer = _OP_REDUCERS[op_name]
260 except KeyError:
261 raise ValueError(f"unknown aggregate operator: {op_name!r}") from None
262 return reducer(_as_list(items), acc, key)