[langgraph]: <The code example in the LangGraph interrupts wiki page is not correct>

[wanjundegithub]

Type of issue

issue / bug

Language

Python

Description

import sqlite3
from typing import Literal, Optional, TypedDict

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command, interrupt


class ApprovalState(TypedDict):
    action_details: str
    status: Optional[Literal["pending", "approved", "rejected"]]


def approval_node(state: ApprovalState) -> Command[Literal["proceed", "cancel"]]:
    # Expose details so the caller can render them in a UI
    decision = interrupt({
        "question": "Approve this action?",
        "details": state["action_details"],
    })

    # Route to the appropriate node after resume
    return Command(goto="proceed" if decision else "cancel")


def proceed_node(state: ApprovalState):
    return {"status": "approved"}


def cancel_node(state: ApprovalState):
    return {"status": "rejected"}


builder = StateGraph(ApprovalState)
builder.add_node("approval", approval_node)
builder.add_node("proceed", proceed_node)
builder.add_node("cancel", cancel_node)
builder.add_edge(START, "approval")
builder.add_edge("approval", "proceed")
builder.add_edge("approval", "cancel")
builder.add_edge("proceed", END)
builder.add_edge("cancel", END)

# Use a more durable checkpointer in production
checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "approval-123"}}
initial = graph.invoke(
    {"action_details": "Transfer $500", "status": "pending"},
    config=config,
)
print(initial["__interrupt__"])  # -> [Interrupt(value={'question': ..., 'details': ...})]

# Resume with the decision; True routes to proceed, False to cancel
resumed = graph.invoke(Command(resume=True), config=config)
print(resumed["status"])  # -> "approved"

error:
langgraph.errors.InvalidUpdateError: At key 'status': Can receive only one value per step. Use an Annotated key to handle multiple values.

For troubleshooting, visit: https://python.langchain.com/docs/troubleshooting/errors/INVALID_CONCURRENT_GRAPH_UPDATE

you need remove the ambigious edge：
builder.add_edge("approval", "proceed")
builder.add_edge("approval", "cancel")

[sarathak]

Kenta Murata (@mrkn)

This is expected behavior when nodes run in parallel.

The Issue

     START
      / \
   node  other_node  ← Both update "status" at same time = ERROR
      \ /
     END

Quick Fix

Add a reducer to handle concurrent updates:

import operator
from typing import Annotated

class State(TypedDict):
    status: Annotated[list, operator.add]  # Combines values into a list

Alternative

Run nodes sequentially instead:

builder.add_edge(START, "node")
builder.add_edge("node", "other_node")  # Sequential, not parallel

START → node → other_node → END  ← No conflict

TL;DR: Parallel execution + same state key = need a reducer

Let me know if this helps!

[casparb]

Hi wanjundegithub, closing as this is expected behavior. See the comment above for a good explanation - thank you Sarath ak (@sarathak).

[wanjundegithub]

Hi Sarath ak (@sarathak) Caspar Broekhuizen (@casparb), thank you for sharing the correct implementation. I actually identified the cause and resolved it earlier, but I think it would be helpful to update the wiki page on LangGraph (https://docs.langchain.com/oss/python/langgraph/interrupts#full-example
), since the current example doesn’t work as expected and might cause confusion for others.

[casparb]

Thank you wanjundegithub, good callout

Transferred to docs repo and adding context here: the docs define static parallel edges:

builder.add_edge("approval", "proceed")
builder.add_edge("approval", "cancel")

But these edges are conditional, defined implicitly here:

def approval_node(state: ApprovalState) -> Command[Literal["proceed", "cancel"]]:

They should be removed