Skip to content

INVALID_CONCURRENT_GRAPH_UPDATE

一个 LangGraph StateGraph 从多个节点接收到了对其状态属性的并发更新，而这些属性并不支持这种更新方式。

这种情况发生的一种方式是，如果你在图中使用了扇出或其他并行执行，并定义了一个如下的图：

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_string_value" 覆盖状态属性 "some_key" 的值。 然而，如果多个节点在例如单步中的扇出返回了 "some_key" 的值，图将抛出此错误，因为不确定如何更新内部状态。

为了避免这种情况，你可以定义一个合并多个值的归约器：

import operator
from typing import Annotated

class State(TypedDict):
    # operator.add 归约函数使这个属性只能追加
    some_key: Annotated[list, operator.add]

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

故障排除

以下方法可能有助于解决此错误：

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

Comments