Coverage for core/langgraph/_reducers.py: 100%

1""" 

2_reducers.py — Core-owned reducer layer shared by channel reducers and the 

3aggregate step mechanism (core/docs/builtins/aggregate.md). 

4 

5Two surfaces, one implementation: 

6 

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. 

10 

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 

17 

18from typing import Any, Callable 

19 

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}) 

35 

36 

37# --------------------------------------------------------------------------- 

38# Binary primitives (relocated from runner.py; behaviour unchanged) 

39# --------------------------------------------------------------------------- 

40 

41def _reduce_last(left: Any, right: Any) -> Any: 

42 """Replace: the newest write wins.""" 

43 return right 

44 

45 

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] 

51 

52 

53def _dedupe_first_occurrence(values: list, key: Callable[[Any], Any] | None = None) -> list: 

54 """Drop later duplicates, preserving first-occurrence order. 

55 

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 

68 

69 

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 

81 

82 

83def _reduce_dict_merge(left: dict, right: dict) -> dict: 

84 """Shallow merge: later keys win.""" 

85 return {**left, **right} 

86 

87 

88def _is_collection_map(value: Any) -> bool: 

89 """True for a non-empty dict whose every value is a list or set.""" 

90 return ( 

91 isinstance(value, dict) 

92 and bool(value) 

93 and all(isinstance(item, (list, set)) for item in value.values()) 

94 ) 

95 

96 

97def _require_mapping(right: Any) -> None: 

98 if not isinstance(right, dict): 

99 raise ValueError( 

100 "BinaryOperatorAggregate reducer selected a dict reducer but the " 

101 f"incoming write is {type(right).__name__}, not a dict" 

102 ) 

103 

104 

105# --------------------------------------------------------------------------- 

106# List folds (aggregate operator catalogue) 

107# --------------------------------------------------------------------------- 

108 

109def _require_numeric(value: Any) -> int | float: 

110 """Numeric means int/float; bool is deliberately excluded.""" 

111 if isinstance(value, bool) or not isinstance(value, (int, float)): 

112 raise ValueError( 

113 "aggregate numeric operator received a non-numeric value: " 

114 f"{type(value).__name__}" 

115 ) 

116 return value 

117 

118 

119def _as_list(value: Any) -> list: 

120 return list(value) if value is not None else [] 

121 

122 

123def _fold_union(items: list, acc: Any, key: Callable[[Any], Any] | None = None) -> list: 

124 return _dedupe_first_occurrence(_as_list(acc) + list(items), key) 

125 

126 

127def _fold_concat(items: list, acc: Any, key: Callable[[Any], Any] | None = None) -> list: 

128 return _reduce_add_or_extend(_as_list(acc), list(items)) 

129 

130 

131def _fold_merge(items: list, acc: Any, key: Callable[[Any], Any] | None = None) -> dict: 

132 merged = dict(acc) if acc is not None else {} 

133 for item in items: 

134 merged = _reduce_dict_merge(merged, item) 

135 return merged 

136 

137 

138def _fold_last(items: list, acc: Any, key: Callable[[Any], Any] | None = None) -> Any: 

139 return _reduce_last(acc, items[-1]) if items else acc 

140 

141 

142def _fold_first(items: list, acc: Any, key: Callable[[Any], Any] | None = None) -> Any: 

143 if acc is not None: 

144 return acc 

145 return items[0] if items else None 

146 

147 

148def _fold_sum(items: list, acc: Any, key: Callable[[Any], Any] | None = None) -> int | float: 

149 total = 0 if acc is None else acc 

150 for item in items: 

151 total += _require_numeric(item) 

152 return total 

153 

154 

155def _fold_avg(items: list, acc: Any, key: Callable[[Any], Any] | None = None) -> dict | None: 

156 total, count = (acc["sum"], acc["count"]) if acc is not None else (0, 0) 

157 for item in items: 

158 total += _require_numeric(item) 

159 count += 1 

160 if count == 0: 

161 return None 

162 return {"sum": total, "count": count} 

163 

164 

165def _fold_min(items: list, acc: Any, key: Callable[[Any], Any] | None = None) -> int | float | None: 

166 minimum = acc 

167 for item in items: 

168 value = _require_numeric(item) 

169 if minimum is None or value < minimum: 

170 minimum = value 

171 return minimum 

172 

173 

174def _fold_max(items: list, acc: Any, key: Callable[[Any], Any] | None = None) -> int | float | None: 

175 maximum = acc 

176 for item in items: 

177 value = _require_numeric(item) 

178 if maximum is None or value > maximum: 

179 maximum = value 

180 return maximum 

181 

182 

183def _fold_count(items: list, acc: Any, key: Callable[[Any], Any] | None = None) -> int: 

184 return (0 if acc is None else acc) + len(items) 

185 

186 

187def _fold_group(items: list, acc: Any, key: Callable[[Any], Any] | None = None) -> dict: 

188 buckets = {bucket: list(values) for bucket, values in acc.items()} if acc else {} 

189 for item in items: 

190 bucket = item if key is None else key(item) 

191 buckets.setdefault(bucket, []).append(item) 

192 return buckets 

193 

194 

195def _fold_any(items: list, acc: Any, key: Callable[[Any], Any] | None = None) -> bool: 

196 if acc: 

197 return True 

198 return any(items) 

199 

200 

201def _fold_all(items: list, acc: Any, key: Callable[[Any], Any] | None = None) -> bool: 

202 if acc is not None and not acc: 

203 return False 

204 return all(items) 

205 

206 

207_OP_REDUCERS: dict[str, Callable[..., Any]] = { 

208 "union": _fold_union, 

209 "concat": _fold_concat, 

210 "merge": _fold_merge, 

211 "last": _fold_last, 

212 "first": _fold_first, 

213 "sum": _fold_sum, 

214 "avg": _fold_avg, 

215 "min": _fold_min, 

216 "max": _fold_max, 

217 "count": _fold_count, 

218 "group": _fold_group, 

219 "any": _fold_any, 

220 "all": _fold_all, 

221} 

222 

223 

224def apply_op( 

225 op_name: str, 

226 items: Any, 

227 acc: Any = None, 

228 key: Callable[[Any], Any] | None = None, 

229) -> Any: 

230 """Fold ``items`` into ``acc`` with the named aggregate operator. 

231 

232 The only dispatch point the aggregate builtin uses. ``key`` is the 

233 already-resolved identity/grouping callable for ``union`` and ``group``; 

234 path resolution is the caller's concern. ``acc`` is never mutated. 

235 """ 

236 try: 

237 reducer = _OP_REDUCERS[op_name] 

238 except KeyError: 

239 raise ValueError(f"unknown aggregate operator: {op_name!r}") from None 

240 return reducer(_as_list(items), acc, key)