Agentsop State Reducer

State Reducer · Tool Skill

Scope: a single decision — for each state key, declare a reducer or guarantee single-writer. Out of scope: checkpointers, HITL, supervisor vs. swarm — see langgraph-sop for those.

1. 何时激活 (Activation Rules)

Activate when any trigger fires:

• The task involves a LangGraph StateGraph, TypedDict/BaseModel schema, or Annotated[..., <reducer>] typing.
• The graph has ≥1 of: parallel branches via static fan-out, Send API, multiple agents writing shared state, or a supervisor pattern where workers return concurrently.
• The run raises langgraph.errors.InvalidUpdateError — message looks like At key 'messages': Can receive only one value per step. Use an Annotated key to handle multiple values.
• The user pastes a state schema and asks "why does this crash on parallel?" or "do I need a reducer here?".
• The user is migrating a linear chain to a fan-out / map-reduce shape.

Do not activate if every key is written by exactly one node per superstep (see §6: over-reducing single-writer keys is an anti-pattern).

2. 核心心智模型 (Core Mental Model)

LangGraph state is either single-writer or has a reducer. Nothing in between.

When a node returns {"k": v}, LangGraph must decide how to merge v into existing state["k"]. There are exactly two legal regimes:

1. Single-writer / "set" semantics (default, no Annotated). At most one node writes the key per superstep. The new value replaces the old. Two concurrent writers → InvalidUpdateError.
2. Reducer / "merge" semantics (Annotated[T, reducer_fn]). Any number of writers may write per superstep; LangGraph folds them via reducer_fn(current, new).

The reducer is commutative-enough algebra that lets the engine schedule parallel writes without you reasoning about interleavings. Three canonical reducers cover ~90% of real graphs:

"Reducers are mandatory, not optional, for parallel execution." — [cheatsheet/gotchas]

The add_messages quirk that beginners miss: it is not plain append. If a new message shares an id with one in state, it overwrites in place. This is what makes HITL interrupt() + edit-state work — a human can edit the last AI turn and resume.

The decision is per-key, not per-schema. A schema can mix freely: one key single-writer, another with add_messages, another with operator.add.

3. SOP 工作流 (Agentic Protocol)

Run top-down for every new or modified state schema.

Step 1 · Enumerate writers per key

For each key k in the state schema, list every node that returns k in its update dict. Be paranoid: include nodes spawned via Send, subgraphs whose output schema overlaps the parent, and any conditional branches.

Step 2 · Classify each key

• Single-writer-per-superstep → leave un-annotated. Plain k: T.
• Multi-writer in the same superstep (parallel branches, fan-out workers, swarm handoffs) → must declare a reducer. Plain k: T will crash.
• Sequential multi-writer (different supersteps) — a reducer is optional: without one, each later write overwrites; with one, each write folds in. Choose based on intent.

Decision gate: if you cannot answer "which nodes write this key?" in one sentence per key, stop and redraw the graph before continuing.

Step 3 · Pick the reducer

Use the OP table in §4. Default ladder:

1. List of BaseMessage → add_messages.
2. Append-only list of anything else → operator.add.
3. Counter / accumulator → operator.add.
4. Anything weirder → write a custom reducer (§4 OP-5).

Step 4 · Probe-test the parallel update

Before shipping, write a unit test that invokes the graph along the parallel path with deterministic node outputs and asserts state matches expectation. If the reducer is wrong (e.g., operator.add on dicts), this is where you find out — not in production.

def test_parallel_merge():
    out = graph.invoke({"items": []})
    assert sorted(out["items"]) == ["a", "b"]  # both workers' contributions present

Step 5 · Document the contract

In a docstring on the TypedDict / Pydantic model, note for each non-default key why it has a reducer. Future maintainers will thank you when they refactor.

4. 操作模型 (Operation Models)

Each op: Trigger → Action → Output.

OP-1 · Declare add_messages for a chat history key

• Trigger: state has a messages: list[BaseMessage] field; ≥1 of (HITL edit-state, multi-agent that all append turns, ReAct loop).
• Action:

  from typing import Annotated, TypedDict
  from langchain_core.messages import AnyMessage
  from langgraph.graph.message import add_messages
  
  class S(TypedDict):
      messages: Annotated[list[AnyMessage], add_messages]

• Output: Concurrent message writes append; same-id writes overwrite in-place (enables HITL state edits).

OP-2 · Declare operator.add for an accumulating list

• Trigger: parallel workers each emit zero or more items into a shared list (e.g., one chunk per worker in a map step).
• Action:

  import operator
  from typing import Annotated, TypedDict
  
  class S(TypedDict):
      chunks: Annotated[list[str], operator.add]

• Output: Worker outputs concatenate in arbitrary order. Order is not guaranteed — if you need ordering, attach an index to each item and sort downstream.

OP-3 · Declare operator.add for a counter

• Trigger: tracking retries, tool calls made, tokens spent across parallel branches.
• Action: retries: Annotated[int, operator.add]. Each node returns the delta ({"retries": 1}), not the new total.
• Output: Summed across writers.
• Watch out: returning the new total ({"retries": state["retries"]+1}) with operator.add double-counts. Always return deltas under operator.add.

OP-4 · Single-writer key (no reducer)

• Trigger: exactly one node — usually the supervisor or a planner — owns the key.
• Action: leave it un-annotated: current_task: str.
• Output: Last (only) writer's value wins. If a second writer ever appears in the same superstep, the graph crashes loudly — which is the desired safety signal.

OP-5 · Custom reducer — keep latest by timestamp

• Trigger: two parallel agents both write a summary field; you want whichever was generated more recently to win.
• Action:

  from datetime import datetime
  
  Summary = dict  # {"text": str, "ts": datetime}
  
  def keep_latest(curr: Summary | None, new: Summary) -> Summary:
      if curr is None or new["ts"] >= curr["ts"]:
          return new
      return curr
  
  class S(TypedDict):
      summary: Annotated[Summary, keep_latest]

• Output: Newer timestamp survives regardless of which branch finished first.

OP-6 · Custom reducer — dedupe by id

• Trigger: parallel retrievers each return a list of {"id": ..., ...} documents and the union should be de-duplicated.
• Action:

  def dedupe_by_id(curr: list[dict], new: list[dict]) -> list[dict]:
      seen = {d["id"]: d for d in (curr or [])}
      for d in new:
          seen[d["id"]] = d   # later wins on collision
      return list(seen.values())
  
  class S(TypedDict):
      docs: Annotated[list[dict], dedupe_by_id]

• Output: Union without duplicates; collision policy is explicit and testable.

OP-7 · Fix InvalidUpdateError post-mortem

• Trigger: error message InvalidUpdateError: At key '<k>': Can receive only one value per step. Use an Annotated key to handle multiple values.
• Action: locate the key <k> in the schema, identify the concurrent writers, then pick OP-1/2/3/5/6 based on key semantics.
• Output: Crash gone; merges behave as designed (verify with the §3 Step 4 probe test).

5. 困境决策案例 (Dilemma Cases)

Case 1 · "Two agents write messages — what reducer?"

• 困境: Supervisor pattern. Both the retrieval agent and the calculator agent emit {"messages": [AIMessage(...)]} in the same superstep when the supervisor fans out. Without a reducer → InvalidUpdateError. Naive operator.add works but loses HITL edit-in-place semantics.
• 约束:
  • Need both concurrent appends and later HITL edits.
  • Message order across branches doesn't matter (each carries its own timestamp), but duplicates from retries must collapse.
• 决策步骤:
  1. Reject operator.add — it concatenates blindly; retried messages with the same id would appear twice and break the chat UI.
  2. Use add_messages — explicitly designed for this. Append by default, overwrite in place when id matches [lc-docs/messages].
  3. Generate stable ids in each agent (e.g., uuid4()) so dedupe is deterministic. If an agent retries, reuse the previous attempt's id.
• 结果: Concurrent appends work, HITL "edit the last AI message and resume" works, retried turns dedupe themselves.
• 可提取的操作: OP-1. Whenever the key is a message list, add_messages is almost always the right answer — operator.add is a code smell on message lists.

Case 2 · "Two agents write summary — append-merge-or-replace?"

• 困境: A research agent and a critic agent both produce a summary string for the same document, in parallel. With no reducer → crash. With operator.add on strings → concatenation glues them edge-to-edge ("A.B." not "A. B."). Neither is semantically right.
• 约束:
  • Output must be a single coherent paragraph downstream.
  • Both agents have signal; neither is strictly authoritative.
  • Cost-sensitive: an extra LLM call is acceptable only once per document.
• 决策步骤:
  1. Identify the semantic merge: not "concat", not "pick one", but "synthesise into one paragraph".
  2. Write a custom reducer that buffers contributions, then a downstream single-writer node calls the LLM once to merge:

     def collect(curr: list[str] | None, new: str) -> list[str]:
         return (curr or []) + [new]
     
     class S(TypedDict):
         summary_drafts: Annotated[list[str], collect]
         summary: str   # single-writer, set by merge_node

  3. After the parallel branches join, a merge_node reads state["summary_drafts"], runs one LLM call, writes {"summary": "..."} (single-writer, no reducer needed).
• 结果: Crash gone; merge logic is explicit, testable, costs exactly one extra LLM call.
• 可提取的操作: When parallel writers don't produce algebraically mergeable values, don't force a reducer — collect into a list with operator.add / a custom collector, then merge in a single-writer node downstream. Reducers are for trivial folds; LLM merges belong in nodes.