LangGraph MessagesState Append vs Replace Pitfalls

from langchain_core.messages import SystemMessage, HumanMessage, AIMessage
from langgraph.graph import StateGraph, START, END, MessagesState


def responder(state: MessagesState) -> dict:
    reply = llm.invoke(state["messages"])
    return {"messages": [reply]}


graph = StateGraph(MessagesState)
graph.add_node("responder", responder)
graph.add_edge(START, "responder")
graph.add_edge("responder", END)
app = graph.compile(checkpointer=memory)


# ✅ System prompt is provided ONCE, as part of initial invocation
config = {"configurable": {"thread_id": "chat-1"}}
result = app.invoke(
    {"messages": [
        SystemMessage(content="You are a concise expert on LangGraph."),
        HumanMessage(content="What is a StateGraph?"),
    ]},
    config,
)

# On subsequent turns, ONLY append the new user message.
# The SystemMessage is already in state from turn 1.
result = app.invoke(
    {"messages": [HumanMessage(content="And what is a reducer?")]},
    config,
)


# ✅ ALTERNATIVE — inject with stable ID so dedupe works
def responder_with_system(state):
    system = SystemMessage(content="You are a concise expert.", id="sys-1")
    # add_messages will dedupe on id="sys-1" — appears once even if returned each turn
    return {"messages": [system, llm.invoke([system] + state["messages"])]}


# ❌ ANTI-PATTERN — no ID, duplicates every turn
def bad(state):
    return {"messages": [
        SystemMessage(content="You are..."),   # new UUID every call
        llm.invoke(state["messages"]),
    ]}

from langchain_core.messages import (
    SystemMessage, HumanMessage, AIMessage, trim_messages,
)
from langgraph.graph import StateGraph, START, END, MessagesState
from langchain_anthropic import ChatAnthropic


llm = ChatAnthropic(model="claude-opus-4-7", temperature=0.2)


def responder(state: MessagesState) -> dict:
    # trim to fit — SystemMessage kept, oldest human/ai messages dropped first
    trimmed = trim_messages(
        state["messages"],
        max_tokens=180_000,          # leave headroom for the response
        token_counter=llm,
        strategy="last",             # keep the most recent messages
        include_system=True,         # never drop the SystemMessage
        allow_partial=False,         # never split a message in half
        start_on="human",            # first non-system message must be human
    )
    reply = llm.invoke(trimmed)
    return {"messages": [reply]}    # append the reply to FULL history


# state["messages"] still contains everything — the LLM just sees the trimmed slice
# This gives you: full audit trail in state + safe LLM input + no 400s


# ✅ ALTERNATIVE — bounded history in state (aggressive)
from langchain_core.messages import RemoveMessage

def prune_old(state):
    """Keep only the last 40 messages IN STATE."""
    msgs = state["messages"]
    if len(msgs) <= 40:
        return {}
    # Tombstone the oldest overflow via RemoveMessage
    overflow = msgs[:-40]
    return {"messages": [RemoveMessage(id=m.id) for m in overflow]}

# Wire it as a node that runs after the responder
graph.add_node("prune", prune_old)
graph.add_edge("responder", "prune")
graph.add_edge("prune", END)

from langchain_core.messages import AIMessage, RemoveMessage
from langgraph.graph import MessagesState


# ❌ BUG — mutation isn't persisted through the reducer
def bad_edit(state):
    state["messages"][-1].content = "corrected content"
    return {}   # nothing in the update = reducer sees nothing = checkpoint keeps the original


# ✅ Return a replacement message with the SAME ID
def edit_last(state: MessagesState) -> dict:
    prior = state["messages"][-1]
    replacement = AIMessage(
        content="corrected content",
        id=prior.id,                          # <-- key: matching id triggers replace
    )
    return {"messages": [replacement]}        # add_messages replaces in place


# ✅ Delete a message via tombstone
def delete_message_by_id(state, target_id: str) -> dict:
    return {"messages": [RemoveMessage(id=target_id)]}


# ✅ Redact PII in the last user message (real production pattern)
import re
def redact_last(state):
    prior = state["messages"][-1]
    if prior.type != "human":
        return {}
    redacted = re.sub(r"\b\d{3}-\d{2}-\d{4}\b", "[SSN]", prior.content)
    if redacted == prior.content:
        return {}   # no change, don't touch state
    return {"messages": [HumanMessage(content=redacted, id=prior.id)]}


# ✅ Batch edit — multiple replacements in one return
def rewrite_all_ai(state):
    updates = []
    for m in state["messages"]:
        if isinstance(m, AIMessage) and "TODO" in m.content:
            updates.append(AIMessage(
                content=m.content.replace("TODO", "[in progress]"),
                id=m.id,
            ))
    return {"messages": updates} if updates else {}