LangGraph Reducer Functions & add_messages Update Errors

from typing import Annotated, TypedDict
from langchain_core.messages import AnyMessage, AIMessage, HumanMessage, SystemMessage
from langgraph.graph.message import add_messages
from langgraph.graph import StateGraph, START, END


class State(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]     # ← THE fix


def node_a(state):
    # Use real message classes (they auto-assign IDs — critical for dedup)
    return {"messages": [AIMessage(content="hello from a")]}


def node_b(state):
    # add_messages merges by id — no double-append on retry
    return {"messages": [AIMessage(content="hello from b")]}


graph = StateGraph(State)
graph.add_node("a", node_a)
graph.add_node("b", node_b)
graph.add_edge(START, "a")
graph.add_edge("a", "b")
graph.add_edge("b", END)
app = graph.compile()

result = app.invoke({"messages": [SystemMessage(content="You are helpful.")]})
# result["messages"] contains ALL 3 messages, in order:
#   SystemMessage, AIMessage("hello from a"), AIMessage("hello from b")


# ✅ Updating a prior message by ID — add_messages will REPLACE it
def edit_node(state):
    prior = state["messages"][-1]
    replacement = AIMessage(content="edited content", id=prior.id)   # same id
    return {"messages": [replacement]}   # replaces, not appends


# ✅ Deleting a message via tombstone (LangGraph 0.2+)
from langchain_core.messages import RemoveMessage
def delete_last(state):
    target_id = state["messages"][-1].id
    return {"messages": [RemoveMessage(id=target_id)]}

from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END


# ❌ BUG — crashes when current is None
def bad_merge(current: list, update: list) -> list:
    return current + update       # TypeError if current is None


# ✅ Pattern 1: default the current side inside the reducer
def safe_merge(current: list | None, update: list | None) -> list:
    return (current or []) + (update or [])


# ✅ Pattern 2: strict dict merge that preserves prior on empty update
def merge_meta(current: dict | None, update: dict | None) -> dict:
    if not update:
        return current or {}
    return {**(current or {}), **update}


# ✅ Pattern 3: bounded accumulator — cap the list at N items
def last_n(cap: int):
    def reducer(current: list | None, update: list | None) -> list:
        merged = (current or []) + (update or [])
        return merged[-cap:]
    return reducer


class State(TypedDict):
    citations: Annotated[list, safe_merge]
    metadata: Annotated[dict, merge_meta]
    recent_events: Annotated[list, last_n(20)]     # only keep last 20


# All three reducers now survive an empty initial state
initial = {}                       # nothing set
app.invoke(initial)                # ← works; reducers use their None-defaults

from langchain_core.messages import (
    AIMessage, HumanMessage, SystemMessage, ToolMessage, AnyMessage
)
from typing import Annotated, TypedDict
from langgraph.graph.message import add_messages


class State(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]


# ❌ Raw tuple — no ID, add_messages can't dedup
def bad_node(state):
    return {"messages": [("assistant", "hi")]}   # tuple has no .id


# ❌ Plain dict — no ID field
def also_bad(state):
    return {"messages": [{"role": "assistant", "content": "hi"}]}


# ✅ LangChain message class — auto-assigned UUID as id
def good_node(state):
    msg = AIMessage(content="hi")
    print(msg.id)   # e.g. "run-abc123..."
    return {"messages": [msg]}


# ✅ Explicit ID (useful for editing / replacing a specific message)
def edit_previous(state):
    prior = state["messages"][-1]
    return {"messages": [AIMessage(content="corrected", id=prior.id)]}


# ✅ Tool response — always use ToolMessage with the tool_call_id
def handle_tool_result(state, result):
    tool_call = state["messages"][-1].tool_calls[0]
    return {"messages": [ToolMessage(
        content=result,
        tool_call_id=tool_call["id"],   # links response to the request
    )]}


# When migrating from tuples: convert once at the boundary
def normalize_messages(msgs):
    normalized = []
    for m in msgs:
        if isinstance(m, tuple):
            role, content = m
            cls = {"system": SystemMessage, "user": HumanMessage,
                   "assistant": AIMessage}[role]
            normalized.append(cls(content=content))
        else:
            normalized.append(m)
    return normalized