Coverage for core/langgraph/tools/aggregate.py: 95%
193 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-28 00:19 +0000
1"""Per-tool module for the `aggregate` deterministic builtin.
3Owns the tool's model-facing definition, its instruction surface (none), the
4aggregate-only helpers (dot-path resolution, operator/error/dedupe resolution,
5the field fold) and the fold execution itself (core/docs/builtins/aggregate.md,
6ADR-0012/ADR-0016/ADR-0018). The definition and the execution here are the
7single source; the legacy builtin site delegates to the pure `aggregate` fold.
8"""
9from __future__ import annotations
11import json
12from typing import Any
14from .contract import ToolContext, ToolResult
16try:
17 from _reducers import VALID_OPERATORS, apply_op
18except ImportError: # pragma: no cover — main.py run (qualified imports)
19 from langgraph._reducers import VALID_OPERATORS, apply_op
21try:
22 from _files_output import FilesOutputError
23except ImportError: # pragma: no cover — main.py run (qualified imports)
24 from langgraph._files_output import FilesOutputError
26AGGREGATE_TOOL_NAME = "aggregate"
28_DEFAULT_OPERATOR = "union"
29_DEFAULT_ON_ERROR = "keep"
30_ON_ERROR_VALUES = ("error", "skip", "keep")
31_ON_ERROR_POLICIES = frozenset(_ON_ERROR_VALUES)
32_RESERVED_OUTPUT_KEYS = frozenset({"acc"})
33_OPERATOR_VALUES: list[str] = sorted(VALID_OPERATORS)
34_MISSING_PATH = object()
36AGGREGATE_DEFINITION: dict[str, Any] = {
37 "type": "function",
38 "function": {
39 "name": AGGREGATE_TOOL_NAME,
40 "description": (
41 "Fold the collected per-iteration results deterministically with "
42 "`aggregate` instead of recomputing them yourself. Provide "
43 "`results` (or its `sources` alias), a `fields` list of "
44 "`{target, source, op, dedupe_key/dedupe_keys, on_error}` entries, "
45 "and optionally the `op`/`on_error` defaults and the `acc` "
46 "loop-carry; each `target` is written to the result."
47 ),
48 "parameters": {
49 "type": "object",
50 "properties": {
51 "results": {
52 "type": "array",
53 "description": (
54 "This iteration's result list to fold. Required unless "
55 "`sources` is given."
56 ),
57 },
58 "sources": {
59 "type": "array",
60 "description": (
61 "Alias for `results`; when both are given they must be "
62 "equal, otherwise the call is a hard error."
63 ),
64 },
65 "fields": {
66 "type": "array",
67 "description": "The fields to fold, one entry per target.",
68 "items": {
69 "type": "object",
70 "properties": {
71 "target": {
72 "type": "string",
73 "description": (
74 "Output key written to the step result."
75 ),
76 },
77 "source": {
78 "type": ["string", "null"],
79 "description": (
80 "Dot-path plucked from each result element; "
81 "null or empty means the element itself is "
82 "the item list."
83 ),
84 },
85 "op": {
86 "type": "string",
87 "enum": _OPERATOR_VALUES,
88 "description": (
89 "Field operator; overrides the top-level "
90 "`op`."
91 ),
92 },
93 "dedupe_key": {
94 "type": ["string", "null"],
95 "description": (
96 "Scalar dot-path (or scalar-item value) "
97 "used as the dedupe key."
98 ),
99 },
100 "dedupe_keys": {
101 "type": ["array", "null"],
102 "items": {"type": "string"},
103 "description": (
104 "List of dot-paths forming a composite "
105 "key; wins over `dedupe_key`."
106 ),
107 },
108 "on_error": {
109 "type": "string",
110 "enum": list(_ON_ERROR_VALUES),
111 "description": "Per-field error policy.",
112 },
113 },
114 "required": ["target"],
115 "additionalProperties": False,
116 },
117 },
118 "acc": {
119 "type": "object",
120 "description": (
121 "Optional loop-carry {target: value} map; read-only, "
122 "never mutated."
123 ),
124 },
125 "op": {
126 "type": "string",
127 "enum": _OPERATOR_VALUES,
128 "description": (
129 "Default operator for fields without their own `op`; "
130 "defaults to `union`."
131 ),
132 },
133 "on_error": {
134 "type": "string",
135 "enum": list(_ON_ERROR_VALUES),
136 "description": (
137 "Default error policy (error | skip | keep); defaults "
138 "to `keep`."
139 ),
140 },
141 },
142 "required": ["fields"],
143 "anyOf": [{"required": ["results"]}, {"required": ["sources"]}],
144 "additionalProperties": False,
145 },
146 },
147}
150def _hard_failure_type() -> type[Exception]:
151 """The runner's hard-failure type, resolved without an import cycle."""
152 try:
153 from runner import RunHardFailure
154 except ImportError: # pragma: no cover — main.py run (qualified imports)
155 from langgraph.runner import RunHardFailure
156 return RunHardFailure
159def _hard_failure(code: str, message: str, details: dict | None = None) -> Exception:
160 return _hard_failure_type()(code, message, details)
163def _resolve_dot_path(obj: Any, path: str) -> Any:
164 """Canonical dot-path resolution; ``_MISSING_PATH`` when unresolved.
166 Delegates to the runner's strict resolver so an unresolved path is
167 distinguishable from a path that resolves to ``None`` — the on_error
168 triggers depend on that difference (ADR-0012).
169 """
170 try:
171 from runner import _dot_get_strict
172 except ImportError: # pragma: no cover — main.py run (qualified imports)
173 from langgraph.runner import _dot_get_strict
174 try:
175 return _dot_get_strict(obj, path)
176 except FilesOutputError:
177 raise
178 except ValueError:
179 return _MISSING_PATH
182def _aggregate_failure(message: str, step_id: str, **details: Any) -> Exception:
183 return _hard_failure(
184 "VALIDATION_ERROR",
185 f"aggregate: {message} (ADR-0012)",
186 {"step_id": step_id, **details},
187 )
190def _results_param(input_spec: dict, step_id: str) -> Any:
191 """Resolve the required ``results``/``sources`` alias pair (ADR-0012)."""
192 results = input_spec.get("results")
193 sources = input_spec.get("sources")
194 if results is None and sources is None:
195 raise _aggregate_failure("missing required 'results'/'sources'", step_id)
196 if results is not None and sources is not None and results != sources:
197 raise _aggregate_failure("'results' and 'sources' disagree", step_id)
198 return results if results is not None else sources
201def _operator(field: dict, top_op: Any, step_id: str, target: str) -> str:
202 op = field.get("op") or top_op or _DEFAULT_OPERATOR
203 if op not in VALID_OPERATORS:
204 raise _aggregate_failure(
205 f"unknown operator {op!r}", step_id, target=target, operator=op,
206 )
207 return op
210def _error_policy(field: dict, top_on_error: Any, step_id: str, target: str) -> str:
211 policy = field.get("on_error") or top_on_error or _DEFAULT_ON_ERROR
212 if policy not in _ON_ERROR_POLICIES:
213 raise _aggregate_failure(
214 f"unknown on_error {policy!r}", step_id, target=target, on_error=policy,
215 )
216 return policy
219def _dedupe_paths(field: dict, step_id: str, target: str) -> list[str] | None:
220 """Resolve the field's key paths; ``dedupe_keys`` wins over ``dedupe_key``.
222 Presence of ``dedupe_keys`` takes precedence unconditionally — a present
223 but empty list is invalid input, never a silent fall-through (ADR-0012).
224 """
225 if "dedupe_keys" in field:
226 dedupe_keys = field["dedupe_keys"]
227 if not isinstance(dedupe_keys, list):
228 raise _aggregate_failure(
229 "'dedupe_keys' must be a list", step_id, target=target,
230 )
231 if not dedupe_keys:
232 raise _aggregate_failure(
233 "'dedupe_keys' must not be empty", step_id, target=target,
234 )
235 if any(str(path) == "" for path in dedupe_keys):
236 raise _aggregate_failure(
237 "'dedupe_keys' entries must not be empty", step_id, target=target,
238 )
239 return [str(path) for path in dedupe_keys]
240 dedupe_key = field.get("dedupe_key")
241 if dedupe_key is None:
242 return None
243 if dedupe_key == "":
244 raise _aggregate_failure(
245 "'dedupe_key' must not be empty", step_id, target=target,
246 )
247 return [dedupe_key]
250def _resolve_item_key(item: Any, paths: list[str]) -> tuple[bool, Any]:
251 """Resolve one item's dedupe/group key; scalar items key by their value."""
252 if not isinstance(item, dict):
253 return True, item
254 if len(paths) == 1:
255 value = _resolve_dot_path(item, paths[0])
256 return (value is not _MISSING_PATH), (None if value is _MISSING_PATH else value)
257 values = []
258 for path in paths:
259 value = _resolve_dot_path(item, path)
260 if value is _MISSING_PATH:
261 return False, None
262 values.append(value)
263 return True, tuple(values)
266def _field_key(paths: list[str], op: str, on_error: str, step_id: str, target: str):
267 """Build the reducer key callable, applying the field's error policy.
269 ``keep`` includes an item with an unresolved key without ever deduping it
270 (a fresh identity); ``group`` has no dedupe, so it buckets under ``None``.
271 """
272 def key(item: Any) -> Any:
273 found, value = _resolve_item_key(item, paths)
274 if found:
275 return value
276 if on_error == "error":
277 raise _aggregate_failure(
278 f"key {paths!r} did not resolve on an item",
279 step_id, target=target,
280 )
281 if op == "group":
282 return None
283 return object()
284 return key
287def _pluck_items(element: Any, source: Any) -> Any:
288 if source is None or source == "":
289 return element
290 return _resolve_dot_path(element, source)
293def _collect_items(results: list, source: Any, on_error: str, step_id: str, target: str) -> list:
294 """Flatten the field's source list across every result element.
296 An unresolved ``source`` triggers the error policy; a source that resolves
297 to ``None`` simply contributes no items (a resolved null is not a miss).
298 """
299 items: list = []
300 for element in results:
301 plucked = _pluck_items(element, source)
302 if plucked is _MISSING_PATH:
303 _on_source_miss(on_error, source, step_id, target)
304 continue
305 if plucked is None:
306 continue
307 if not isinstance(plucked, (list, tuple)):
308 _on_source_miss(on_error, source, step_id, target)
309 continue
310 items.extend(plucked)
311 return items
314def _on_source_miss(on_error: str, source: Any, step_id: str, target: str) -> None:
315 if on_error == "error":
316 raise _aggregate_failure(
317 f"source {source!r} did not resolve to an item list",
318 step_id, target=target,
319 )
322def _drop_unresolved_keys(items: list, paths: list[str]) -> list:
323 return [item for item in items if _resolve_item_key(item, paths)[0]]
326def _finalize_avg(accumulator: Any) -> Any:
327 if accumulator is None:
328 return None
329 return accumulator["sum"] / accumulator["count"]
332def _fold_field(
333 field: dict,
334 results: list,
335 acc_value: Any,
336 top_op: Any,
337 top_on_error: Any,
338 step_id: str,
339) -> tuple[Any, Any]:
340 target = field.get("target")
341 if not target:
342 raise _aggregate_failure("field is missing 'target'", step_id)
343 if target in _RESERVED_OUTPUT_KEYS:
344 raise _aggregate_failure(
345 f"target {target!r} is a reserved output key", step_id, target=target,
346 )
347 op = _operator(field, top_op, step_id, target)
348 on_error = _error_policy(field, top_on_error, step_id, target)
349 paths = _dedupe_paths(field, step_id, target)
350 items = _collect_items(results, field.get("source"), on_error, step_id, target)
351 if paths is not None and on_error == "skip":
352 items = _drop_unresolved_keys(items, paths)
353 if op == "merge" and on_error != "error":
354 items = [item for item in items if isinstance(item, dict)]
355 key = _field_key(paths, op, on_error, step_id, target) if paths else None
356 try:
357 accumulator = apply_op(op, items, acc_value, key)
358 except (ValueError, TypeError) as exc:
359 raise _aggregate_failure(
360 f"{op} failed: {exc}", step_id, target=target,
361 ) from exc
362 if op == "avg":
363 result = _finalize_avg(accumulator)
364 accumulator = {"sum": 0, "count": 0} if accumulator is None else accumulator
365 else:
366 result = accumulator
367 return result, accumulator
370def aggregate(step: dict, state: dict) -> dict:
371 """Generic deterministic fold of per-iteration `results` (aggregate.md).
373 input:
374 results: list — this iteration's result list (required; `sources` alias).
375 fields: [{target, source?, op?, dedupe_key?, dedupe_keys?, on_error?}]
376 acc: optional {target: value} loop-carry; read-only, never mutated.
377 op: optional default operator; per-field `op` wins (default union).
378 on_error: optional default policy: error | skip | keep (default keep).
380 output (state[step.id]):
381 {target: <op result>} for each field; when `acc` is supplied as input a
382 reserved top-level `acc` map `{target: <post-fold accumulator>}` is
383 emitted alongside the targets.
384 """
385 step_id = step.get("id", "aggregate")
386 input_spec = step.get("input") or {}
387 results = _results_param(input_spec, step_id)
388 if not isinstance(results, (list, tuple)):
389 raise _aggregate_failure("'results' must be a list", step_id)
390 fields = input_spec.get("fields")
391 if not isinstance(fields, list):
392 raise _aggregate_failure("'fields' must be a list", step_id)
393 acc_supplied = "acc" in input_spec
394 acc = input_spec.get("acc")
395 if acc is not None and not isinstance(acc, dict):
396 raise _aggregate_failure("'acc' must be a target map", step_id)
397 acc = acc or {}
398 top_op = input_spec.get("op")
399 top_on_error = input_spec.get("on_error")
400 out: dict = {}
401 acc_out: dict = {}
402 for field in fields:
403 if not isinstance(field, dict):
404 raise _aggregate_failure("each field must be a mapping", step_id)
405 target = field.get("target")
406 result, accumulator = _fold_field(
407 field, results, acc.get(target), top_op, top_on_error, step_id,
408 )
409 out[target] = result
410 acc_out[target] = accumulator
411 if acc_supplied:
412 out["acc"] = acc_out
413 return {**state, step_id: out}
416def _success_message(output: dict[str, Any]) -> str:
417 return json.dumps(
418 {"name": AGGREGATE_TOOL_NAME, "ok": True, "output": output},
419 ensure_ascii=False,
420 )
423def _failure_message(error: str) -> str:
424 return json.dumps(
425 {"name": AGGREGATE_TOOL_NAME, "ok": False, "error": error},
426 ensure_ascii=False,
427 )
430class _AggregateTool:
431 name = AGGREGATE_TOOL_NAME
433 def definition(self) -> dict[str, Any]:
434 return AGGREGATE_DEFINITION
436 def instruction(self) -> str | None:
437 return None
439 def params(self) -> list[dict[str, Any]] | None:
440 return None
442 def execute(self, context: ToolContext) -> ToolResult:
443 step_id = (context.step or {}).get("id", AGGREGATE_TOOL_NAME)
444 synthetic_step = {"id": step_id, "input": dict(context.args or {})}
445 try:
446 new_state = aggregate(synthetic_step, context.state or {})
447 except _hard_failure_type() as exc:
448 return ToolResult(
449 ok=False,
450 content=_failure_message(exc.message),
451 error=exc.message,
452 )
453 output = new_state[step_id]
454 return ToolResult(
455 ok=True,
456 content=_success_message(output),
457 state_patch={step_id: output},
458 )
461TOOL = _AggregateTool()