From df11b2b5260b687f021c183ebfd6b3a11401adc8 Mon Sep 17 00:00:00 2001 From: Eric Wang Date: Sat, 25 Jul 2026 22:59:04 +0800 Subject: [PATCH] Add example demonstrating parallel nodes and reducers in LangGraph, including a valid graph with a custom reducer and an invalid graph triggering an update conflict. The script showcases the execution flow and error handling for concurrent updates. --- 10_parallel_and_reducers.py | 236 ++++++++++++++++++++++++++++++++++++ 1 file changed, 236 insertions(+) create mode 100644 10_parallel_and_reducers.py diff --git a/10_parallel_and_reducers.py b/10_parallel_and_reducers.py new file mode 100644 index 0000000..4146ad4 --- /dev/null +++ b/10_parallel_and_reducers.py @@ -0,0 +1,236 @@ +"""LangGraph 并行节点与 Reducer 示例。 + +本例包含两个图： + +1. good_graph：两个搜索节点并行执行，并用 reducer 合并结果。 +2. bad_graph：两个并行节点同时覆盖同一字段，触发 InvalidUpdateError。 + +主图结构： + + ┌─> search_web ─────┐ + START -> prepare ──┤ ├─> summarize -> END + └─> search_database ┘ + +运行： + python 10_parallel_and_reducers.py +""" + +import operator +import time +from typing import Annotated, TypedDict + +from langgraph.errors import InvalidUpdateError +from langgraph.graph import END, START, StateGraph + + +# ----------------------------------------------------------------------------- +# 1. Reducer +# ----------------------------------------------------------------------------- + + +def merge_unique(left: list[str], right: list[str]) -> list[str]: + """合并两个列表并去重，同时保留第一次出现的顺序。""" + return list(dict.fromkeys(left + right)) + + +class ResearchState(TypedDict): + topic: str + + # operator.add：把各节点返回的列表连接起来。 + results: Annotated[list[str], operator.add] + + # 自定义 reducer：合并日志并去除重复内容。 + logs: Annotated[list[str], merge_unique] + + # 没有 reducer：使用默认覆盖策略。 + # 这里只有 summarize 节点更新它，因此不会发生并行冲突。 + summary: str + + +# ----------------------------------------------------------------------------- +# 2. 主图节点 +# ----------------------------------------------------------------------------- + + +def prepare(state: ResearchState) -> dict: + """准备搜索；完成后两个下游节点会在同一 super-step 并行执行。""" + print(f"[prepare] 准备研究：{state['topic']}") + return {"logs": ["准备完成"]} + + +def search_web(state: ResearchState) -> dict: + """模拟耗时 1 秒的网页搜索。""" + print("[search_web] 开始") + time.sleep(1) + print("[search_web] 完成") + return { + "results": [ + f"网页结果：{state['topic']} 支持有状态工作流", + "共同结果：LangGraph 使用 StateGraph 编排节点", + ], + "logs": ["网页搜索完成", "搜索阶段完成"], + } + + +def search_database(state: ResearchState) -> dict: + """模拟耗时 1 秒的数据库搜索。""" + print("[search_database] 开始") + time.sleep(1) + print("[search_database] 完成") + return { + "results": [ + f"数据库结果：{state['topic']} 支持持久化和恢复", + "共同结果：LangGraph 使用 StateGraph 编排节点", + ], + "logs": ["数据库搜索完成", "搜索阶段完成"], + } + + +def summarize(state: ResearchState) -> dict: + """汇合节点：两个搜索分支都结束后才会执行。""" + print(f"[summarize] 收到 {len(state['results'])} 条结果") + + lines = "\n".join( + f"{index}. {result}" + for index, result in enumerate(state["results"], start=1) + ) + return { + "summary": f"主题：{state['topic']}\n{lines}", + "logs": ["摘要生成完成"], + } + + +# ----------------------------------------------------------------------------- +# 3. 构建正确的并行图 +# ----------------------------------------------------------------------------- + + +good_builder = StateGraph(ResearchState) +good_builder.add_node("prepare", prepare) +good_builder.add_node("search_web", search_web) +good_builder.add_node("search_database", search_database) +good_builder.add_node("summarize", summarize) + +good_builder.add_edge(START, "prepare") + +# 一个节点拥有两条普通出边：下一 super-step 中两个目标节点并行执行。 +good_builder.add_edge("prepare", "search_web") +good_builder.add_edge("prepare", "search_database") + +# 两个分支都指向 summarize。LangGraph 会等待同一轮的两个分支完成后汇合。 +good_builder.add_edge("search_web", "summarize") +good_builder.add_edge("search_database", "summarize") +good_builder.add_edge("summarize", END) + +good_graph = good_builder.compile() + + +# ----------------------------------------------------------------------------- +# 4. 错误示例：并行节点覆盖同一个无 reducer 字段 +# ----------------------------------------------------------------------------- + + +class BadState(TypedDict): + value: str # 没有 reducer，默认策略是覆盖 + + +def left_branch(state: BadState) -> dict: + return {"value": "来自左分支"} + + +def right_branch(state: BadState) -> dict: + return {"value": "来自右分支"} + + +bad_builder = StateGraph(BadState) +bad_builder.add_node("left", left_branch) +bad_builder.add_node("right", right_branch) +bad_builder.add_edge(START, "left") +bad_builder.add_edge(START, "right") +bad_builder.add_edge("left", END) +bad_builder.add_edge("right", END) +bad_graph = bad_builder.compile() + + +# ----------------------------------------------------------------------------- +# 5. 演示函数 +# ----------------------------------------------------------------------------- + + +def demo_good_graph() -> None: + print("=== 正确示例：并行 + Reducer + 汇合 ===") + + initial_state: ResearchState = { + "topic": "LangGraph", + "results": [], + "logs": [], + "summary": "", + } + + started = time.perf_counter() + result = good_graph.invoke(initial_state) + elapsed = time.perf_counter() - started + + print(f"\n耗时：{elapsed:.2f} 秒") + print("两个搜索节点各等待 1 秒；总耗时约 1 秒，说明它们并行执行。") + + print("\nresults（operator.add，保留重复项）：") + for item in result["results"]: + print(f"- {item}") + + print("\nlogs（merge_unique，重复日志只保留一次）：") + for item in result["logs"]: + print(f"- {item}") + + print("\nsummary（默认覆盖，由 summarize 单独写入）：") + print(result["summary"]) + + +def demo_stream_updates() -> None: + print("\n=== 使用 updates 观察并行 super-step ===") + + initial_state: ResearchState = { + "topic": "LangGraph", + "results": [], + "logs": [], + "summary": "", + } + + for part in good_graph.stream( + initial_state, + stream_mode="updates", + version="v2", + ): + if part["type"] != "updates": + continue + + node_names = ", ".join(part["data"].keys()) + print(f"节点更新：{node_names}") + + +def demo_conflict() -> None: + print("\n=== 错误示例：并行更新冲突 ===") + try: + bad_graph.invoke({"value": "初始值"}) + except InvalidUpdateError as exc: + print(f"捕获到：{type(exc).__name__}") + print("原因：left 和 right 在同一个 super-step 中都写入 value，") + print("而 value 没有 reducer，LangGraph 不知道应该保留哪一个值。") + print(f"原始错误：{exc}") + + +def main() -> None: + demo_good_graph() + demo_stream_updates() + demo_conflict() + + print("\n=== 学习结论 ===") + print("1. 多条普通出边会让目标节点在同一个 super-step 中并行运行。") + print("2. 并行分支汇合时，下游节点等待本轮所有上游分支完成。") + print("3. 无 reducer 的字段采用覆盖策略，同一步只能接收一个更新。") + print("4. operator.add 可以连接列表，但不会去重。") + print("5. 自定义 reducer 可以实现去重、排序或其他合并规则。") + + +if __name__ == "__main__": + main()