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_string_value"
覆盖 "some_key"
的状态值。但是,如果在单个步骤中,例如一个扇出(fanout)内的多个节点都为 "some_key"
返回了值,图就会抛出这个错误,因为不确定如何更新内部状态。
要解决这个问题,你可以定义一个合并多个值的 reducer:
import operator
from typing import Annotated
class State(TypedDict):
# The operator.add reducer fn makes this append-only
some_key: Annotated[list, operator.add]
这将允许你定义逻辑来处理从并行执行的多个节点返回的同一个键。
故障排除¶
以下建议可能有助于解决此错误:
- 如果你的图并行执行节点,请确保你已经为相关的状态键定义了 reducer。