Skip to content

INVALID_CONCURRENT_GRAPH_UPDATE

一个 LangGraph StateGraph 收到来自多个节点对某个不支持并发更新的状态属性的状态的并发更新。

这种情况可能发生在您使用图中的 fanout 或其他并行执行时，并且您定义了如下图：

class State(TypedDict):
    some_key: str

def node(state: State):
    return {"some_key": "some_string_value"}

def other_node(state: State):
    return {"some_key": "some_string_value"}


builder = StateGraph(State)
builder.add_node(node)
builder.add_node(other_node)
builder.add_edge(START, "node")
builder.add_edge(START, "other_node")
graph = builder.compile()

如果上述图中的一个节点返回 { "some_key": "some_string_value" }，这将覆盖 "some_key" 的状态值为 "some_string_value"。然而，如果在例如单个步骤中的 fanout 内部的多个节点返回 "some_key" 的值，则图将抛出此错误，因为对如何更新内部状态存在不确定性。

为了解决这个问题，可以定义一个合并多个值的 reducer：

import operator
from typing import Annotated

class State(TypedDict):
    # operator.add reducer 函数使其仅支持追加
    some_key: Annotated[list, operator.add]

这将允许您定义逻辑来处理从并行执行的多个节点返回的相同键。

排查故障

以下内容可能有助于解决此错误：

  • 如果您的图以并行方式执行节点，请确保您已使用 reducer 定义了相关状态键。