如何在LangGraph中并行节点且无需等待非必需慢节点完成？

你的问题核心是LangChain StateGraph默认会等待所有入边节点完成才执行下一个节点，要实现"提前触发+可选终止非必要节点"，可以分两种场景处理：

场景1：仅需提前执行assign_city（允许耗时节点继续运行） #

这种方案无需修改为异步，只需调整图的流程结构，通过条件分支让assign_city在满足条件时直接触发，不用等待get_city_preference：

import time
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

class State(TypedDict):
    random_number: int
    city: str
    city_preference: str

def set_random_number(state):
    random_number = 1  # 测试用固定值
    print(f"SET RANDOM NUMBER: {random_number}")
    return {"random_number": random_number}

def get_city_preference(state):
    time.sleep(4)  # 模拟耗时操作
    city_preference = "Philadelphia"
    print(f"GOT CITY PREFERENCE: {city_preference}")
    return {"city_preference": city_preference}

def assign_city(state):
    city = "NYC" if state["random_number"] <= 50 else state["city_preference"]
    print(f"ASSIGNED CITY: {city}")
    return {"city": city}

# 新增判断节点，决定是否需要等待偏好结果
def decide_next_step(state):
    return "assign_city" if state["random_number"] <= 50 else "wait_for_preference"

graph: StateGraph = StateGraph(state_schema=State)

# 添加所有节点
graph.add_node("set_random_number", set_random_number)
graph.add_node("get_city_preference", get_city_preference)
graph.add_node("decide_next_step", decide_next_step)
graph.add_node("assign_city", assign_city)

# 并行启动初始节点
graph.add_edge(START, "set_random_number")
graph.add_edge(START, "get_city_preference")

# 随机数生成后进入判断逻辑
graph.add_edge("set_random_number", "decide_next_step")

# 条件分支：满足条件直接分配城市，否则等待偏好节点完成
graph.add_conditional_edges(
    "decide_next_step",
    decide_next_step,
    {
        "assign_city": "assign_city",
        "wait_for_preference": "get_city_preference"
    }
)

# 偏好节点完成后进入分配步骤
graph.add_edge("get_city_preference", "assign_city")
graph.add_edge("assign_city", END)

# 编译并运行
graph_compiled = graph.compile(checkpointer=MemorySaver())
input = {"random_number": 0, "city": "Nowhere", "city_preference": "N/A"}
config = {"configurable": {"thread_id": "test"}, "recursion_limit": 50}
state = graph_compiled.invoke(input=input, config=config)

效果：当random_number≤50时，assign_city会在set_random_number完成后立即执行，get_city_preference会继续运行但不会阻塞流程。

场景2：提前执行并终止耗时节点（节省资源） #

如果需要彻底终止不需要的get_city_preference任务，需使用异步版本的AsyncStateGraph，通过保存异步任务引用来实现取消：

import asyncio
from typing import TypedDict
from langgraph.graph import AsyncStateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

class State(TypedDict):
    random_number: int
    city: str
    city_preference: str
    preference_task: asyncio.Task | None  # 存储异步任务引用

async def set_random_number(state):
    random_number = 1  # 测试用固定值
    print(f"SET RANDOM NUMBER: {random_number}")
    return {"random_number": random_number}

async def get_city_preference(state):
    await asyncio.sleep(4)  # 模拟异步耗时操作
    city_preference = "Philadelphia"
    print(f"GOT CITY PREFERENCE: {city_preference}")
    return {"city_preference": city_preference}

async def assign_city(state):
    city = "NYC" if state["random_number"] <= 50 else state["city_preference"]
    print(f"ASSIGNED CITY: {city}")
    # 取消不需要的偏好任务
    if state["random_number"] <= 50 and state.get("preference_task"):
        state["preference_task"].cancel()
        print("CANCELED CITY PREFERENCE TASK")
    return {"city": city}

async def decide_next_step(state):
    # 若需要偏好结果，先等待任务完成
    if state["random_number"] > 50 and state.get("preference_task"):
        await state["preference_task"]
    return "assign_city"

# 自定义并行启动节点，同时启动两个任务并保存偏好任务引用
async def start_parallel(state):
    preference_task = asyncio.create_task(get_city_preference(state))
    random_result = await set_random_number(state)
    return {**random_result, "preference_task": preference_task}

graph: AsyncStateGraph = AsyncStateGraph(state_schema=State)

graph.add_node("start_parallel", start_parallel)
graph.add_node("decide_next_step", decide_next_step)
graph.add_node("assign_city", assign_city)

graph.add_edge(START, "start_parallel")
graph.add_edge("start_parallel", "decide_next_step")
graph.add_edge("decide_next_step", "assign_city")
graph.add_edge("assign_city", END)

# 编译并异步运行
graph_compiled = graph.compile(checkpointer=MemorySaver())
input = {"random_number": 0, "city": "Nowhere", "city_preference": "N/A", "preference_task": None}
config = {"configurable": {"thread_id": "test"}, "recursion_limit": 50}
state = await graph_compiled.invoke(input=input, config=config)

效果：当random_number≤50时，assign_city执行的同时会取消get_city_preference任务，避免不必要的资源占用。

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