如何阻止LangGraph中条件边触发reduce函数重复执行？

作为LangGraph初学者，我在构建计算斐波那契数列的图时遇到了问题。程序包含两个节点：n1负责更新斐波那契数列，另一个条件节点控制循环。完整代码如下：

from langchain_core.runnables import Runnable
from collections.abc import Iterable
from langgraph.graph import StateGraph
from pydantic import BaseModel
from typing import Any, Dict, Annotated
from langgraph.constants import END

def append_to_list(a : list, b) -> list:
    print(f"append_to_list:{a}, {b}")
    if isinstance(b, Iterable) and not isinstance(b, str):
        a.extend(b)
    else:
        a.append(b)
    return a

class Node(Runnable):
    def __init__(self, name : str):
        self.name = name
    def invoke(self, input: Any, config: Dict | None = None) -> Any:
        c = input.a + input.b
        print(f"{self.name}:{input}")
        return {"a":input.b, "b":c, "fibonacci_list":c}

class ConditionalNode(Runnable):
    def __init__(self, name : str, threshold: int):
        self.threshold = threshold
        self.name = name

    def invoke(self, input: Any, config) -> Any:
        print(f"{self.name}:{input}")
        return {"path" : input.b >= self.threshold}

class State(BaseModel):
    a : int
    b : int
    fibonacci_list : Annotated[list[int], append_to_list]

graph = StateGraph(State)
graph.add_node("n1", Node(name = "n1"))
graph.add_conditional_edges("n1", ConditionalNode(name = "CN", threshold = 10), {True : END, False: "n1"})
graph.set_entry_point("n1")
g = graph.compile()
s = g.invoke({"a":1, "b":1, "fibonacci_list":[1, 1]})
print(s)

n1节点负责计算斐波那契数列的最新值并返回状态更新，append_to_list作为reduce函数，用于将新生成的数值添加到fibonacci_list中。我原本预期该函数仅在n1节点执行后调用一次，最终得到完整的斐波那契数列（如[1,1,2,3,5,8,13]），但实际每次条件节点执行完成后，append_to_list都会被触发，且传入参数与n1执行后一致，导致数列出现重复元素。

请问如何阻止条件边触发reduce函数？

解决方案 #

问题核心在于：你把条件判断逻辑封装成了Runnable节点，而LangGraph中所有Runnable节点的输出都会和当前状态进行合并，即使节点没有返回某个字段，LangGraph也会尝试用该字段的当前值触发reduce函数，最终导致重复调用。

最简单的解决方法是将条件判断逻辑改为普通路由函数，而非Runnable节点，因为add_conditional_edges的第二个参数可以直接接收一个路由函数，该函数只负责返回路径选择，不会触发状态合并操作。

修改后的代码如下：

from langchain_core.runnables import Runnable
from collections.abc import Iterable
from langgraph.graph import StateGraph
from pydantic import BaseModel
from typing import Any, Dict, Annotated
from langgraph.constants import END

def append_to_list(a : list, b) -> list:
    print(f"append_to_list:{a}, {b}")
    if isinstance(b, Iterable) and not isinstance(b, str):
        a.extend(b)
    else:
        a.append(b)
    return a

class Node(Runnable):
    def __init__(self, name : str):
        self.name = name
    def invoke(self, input: Any, config: Dict | None = None) -> Any:
        c = input.a + input.b
        print(f"{self.name}:{input}")
        return {"a":input.b, "b":c, "fibonacci_list":c}

# 替换ConditionalNode为普通路由函数
def should_end(state: State) -> str:
    print(f"should_end:{state}")
    return END if state.b >= 10 else "n1"

class State(BaseModel):
    a : int
    b : int
    fibonacci_list : Annotated[list[int], append_to_list]

graph = StateGraph(State)
graph.add_node("n1", Node(name = "n1"))
# 使用路由函数替代ConditionalNode
graph.add_conditional_edges("n1", should_end)
graph.set_entry_point("n1")
g = graph.compile()
s = g.invoke({"a":1, "b":1, "fibonacci_list":[1, 1]})
print(s)

修改后，路由函数should_end仅负责判断路径走向，不会触发状态合并，因此append_to_list只会在n1节点执行后被调用一次，最终得到正确的斐波那契数列。

若你坚持要使用Runnable作为条件节点，需确保其返回内容不包含任何状态字段，仅返回路径标识，但这种方式不如路由函数简洁可靠。

内容的提问来源于stack exchange，提问作者lei hu