官方 docstring 用一个例子点明了核心行为（graph/message.py:101）：

# graph/message.py:101 —— docstring 里的"覆盖同 ID"示例
msgs1 = [HumanMessage(content="Hello", id="1")]
msgs2 = [HumanMessage(content="Hello again", id="1")]   # 同一个 id="1"
add_messages(msgs1, msgs2)
# 结果只有一条：[HumanMessage(content='Hello again', id='1')]  ← 被更新，不是追加

你可能见过两种写法：直接 Annotated[list, add_messages]，也见过 Annotated[list, add_messages(format="langchain-openai")]（带括号调用）。这个"既能当函数、又能先配置再当函数"的魔法，靠一个装饰器 _add_messages_wrapper（graph/message.py:41）：

# graph/message.py:41
def _add_messages_wrapper(func):
    def _add_messages(left=None, right=None, **kwargs):
        if left is not None and right is not None:
            return func(left, right, **kwargs)         # ① 正常调用：真的去合并
        elif left is not None or right is not None:
            msg = "Must specify non-null arguments for both 'left' and 'right'. ..."
            raise ValueError(msg)                       # ② 只给一半 → 报错
        else:
            return partial(func, **kwargs)              # ③ 一个都没给 → 返回"预配置好的函数"
    _add_messages.__doc__ = func.__doc__
    return cast(..., _add_messages)


@_add_messages_wrapper
def add_messages(left, right, *, format=None):
    ...

进入真身后，第一件事是把 left/right 各种松散写法统一成标准消息对象（graph/message.py:187）：

# graph/message.py:187
remove_all_idx = None
# ① 单条也裹成 list
if not isinstance(left, list):
    left = [left]
if not isinstance(right, list):
    right = [right]
# ② 各种写法统一成 BaseMessage（元组/字典/字符串 → 消息对象）
left = [message_chunk_to_message(m) for m in convert_to_messages(left)]
right = [message_chunk_to_message(m) for m in convert_to_messages(right)]
# ③ 补齐缺失的 id（用随机 uuid）
for m in left:
    if m.id is None:
        m.id = str(uuid.uuid4())
for idx, m in enumerate(right):
    if m.id is None:
        m.id = str(uuid.uuid4())
    if isinstance(m, RemoveMessage) and m.id == REMOVE_ALL_MESSAGES:   # 顺便记下"清空"标记的位置
        remove_all_idx = idx

归一化后，进入真正的合并循环（graph/message.py:216）。这是 add_messages 的心脏：

# graph/message.py:216
merged = left.copy()
merged_by_id = {m.id: i for i, m in enumerate(merged)}   # id → 在 merged 里的下标
ids_to_remove = set()
for m in right:
    if (existing_idx := merged_by_id.get(m.id)) is not None:   # 这个 id 已经存在
        if isinstance(m, RemoveMessage):
            ids_to_remove.add(m.id)                    # 是删除标记 → 记下待删
        else:
            ids_to_remove.discard(m.id)                # 之前可能标了删，现在又更新 → 取消删
            merged[existing_idx] = m                   # ★原地更新（覆盖同 ID）
    else:                                              # 这个 id 没见过
        if isinstance(m, RemoveMessage):
            raise ValueError(
                f"Attempting to delete a message with an ID that doesn't exist ('{m.id}')"
            )                                          # 删一个不存在的 → 报错
        merged_by_id[m.id] = len(merged)
        merged.append(m)                               # ★全新消息 → 追加
merged = [m for m in merged if m.id not in ids_to_remove]   # 最后统一剔除待删的

L04 已看到按 ID 删除单条（RemoveMessage(id="x")）。还有一种"一键清空"，靠常量 REMOVE_ALL_MESSAGES（graph/message.py:38）实现，处理逻辑在归一化之后、合并循环之前（graph/message.py:212）：

# graph/message.py:38
REMOVE_ALL_MESSAGES = "__remove_all__"

# graph/message.py:212 —— 如果 right 里出现了"清空"标记
if remove_all_idx is not None:
    return right[remove_all_idx + 1 :]      # ★丢掉 left 全部 + 清空标记之前的，只留其后的

99% 的对话图不用自己写 Annotated[list, add_messages]——直接继承现成的 MessagesState（graph/message.py:372）：

# graph/message.py:372
class MessagesState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]

源码里还有个实验性的批量版 _messages_delta_reducer（graph/message.py:247），它体现了一个更深的设计约束——批处理不变性（batching-invariant）：

# graph/message.py:247（docstring 节选）
"""**Experimental.** Batch reducer for use with `DeltaChannel`.

This reducer is batching-invariant, as required by `DeltaChannel`:
    reducer(reducer(state, xs), ys) == reducer(state, xs + ys)
"""
# graph/message.py:292 —— 用 dict 做索引，一趟扫描完成去重/更新/删除（tombstone）
index = {m.id: i for i, m in enumerate(state_msgs) if m.id is not None}
result = list(state_msgs)
for msg in msgs:
    mid = msg.id
    if mid is None:
        result.append(msg)
    elif isinstance(msg, RemoveMessage):
        if mid in index:
            result[index[mid]] = None      # 先置 None（墓碑），最后统一过滤
            del index[mid]
    elif mid in index:
        result[index[mid]] = msg           # 更新
    else:
        index[mid] = len(result); result.append(msg)   # 追加
return [m for m in result if m is not None]