并发
并发
英语为 Concurrency
指的是在同一时间段内，多个任务交替进行。这些任务没有真正同时运行，而是通过任务切换来营造“同时进行”的效果。
类比：一个服务员同时负责 3 桌客人，他会先给 A 桌上菜，再去 B 桌点单，然后回到 C 桌加水……看起来好像在同时照顾三桌，其实是快速切换任务。
并发的特点：一个执行单元（单线程）通过任务调度来处理多个任务。
并行
英语为 Parallelism
指的是在同一时刻，多个任务真正同时运行。这通常依赖于多核 CPU 或多台机器的同时执行。
类比：3 个服务员分别负责 3 桌客人，大家同时干活，互不干扰，这就是真正的同时进行。
并行的特点：需要多个执行单元（多线程、多进程、多核硬件）共同工作。
[!tip]
在 LangGraph 中，采用的是并发的方式执行多个任务。
扇出和扇入
扇出
英语为 Fan-out
指的是一个节点添加多个出边，将其同时连接到下游节点。当 LangGraph 执行到该节点时，会并发的触发所有出边。
总结起来，就是一个节点连接多个下游节点。
扇入
英语为 Fan-in
指多个分支汇聚到同一个下游节点的操作。实现扇入时，需要为目标节点添加多条入边，使其能够接收多个上游节点的输入。
总结起来，就是一个节点连接多个上游节点。
超步
英语为 super-step
中文：超级步骤、超步
图执行过程中的一个“批次轮次”，也就是一轮统一调度的执行阶段。
在 LangGraph 中，超步的设计思想借鉴自 Google 的分布式计算框架 Pregel，以及其他分布式图计算模型。它的核心目的，是让多个节点能够在同一轮中并发执行，并在每一轮结束时进行全局同步。
LangGraph 中超步特点：
- 并发性：同一个超步内，多个节点可以并发执行，从而提升一个执行效率。
- 同步性：每个超步结束后，会有一个全局的同步点，同步全局状态，同步完成之后，才进入下一个超步。
- 迭代性：图执行的过程其实就是超步的迭代过程。每一个超步都是在前一个超步的状态基础上进行计算和更新的。
每个超步的执行流程，大致分为三步：
- 就绪节点并发执行
- 统一同步结果
- 解锁下一轮节点
import { END, START, StateGraph } from "@langchain/langgraph";
import { registry } from "@langchain/langgraph/zod";
import z from "zod";const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));const fmt = (t0?: number) => `${t0 !== undefined ? Date.now() - t0 : 0}ms`;const Schema = z.object({a: z.string().optional(),b: z.string().optional(),c: z.string().optional(),t0: z.number().optional(),logs: z.array(z.string()).register(registry, {reducer: {fn: (oldVal, newVal) => [...oldVal, ...newVal],},default: () => [],}),
});type TState = z.infer<typeof Schema>;const graph = new StateGraph(Schema).addNode("A", () => {console.log("运行A节点");return {a: "A节点执行后的结果",t0: Date.now(),logs: ["【A节点】开始(+0ms)", "【A节点】结束(+0ms)"],};}).addNode("B", async (state) => {console.log("运行B节点");const start = fmt(state.t0); // 得到一个相对于A节点的时间await sleep(2500);return {b: "b节点执行后的结果",logs: [`【B节点】开始(+${start})`, `【B节点】结束(+${fmt(state.t0)})`],};}).addNode("C", async (state) => {console.log("运行C节点");const start = fmt(state.t0); // 得到一个相对于A节点的时间await sleep(1000);return {c: "c节点执行后的结果",logs: [`【C节点】开始(+${start})`, `【C节点】结束(+${fmt(state.t0)})`],};}).addNode("D", async (state) => {console.log("运行D节点");const summary = `D节点已运行，b=${state.b ?? "none"}，c=${state.c ?? "none"}`;return {logs: [`${summary}`, `【D节点】结束(+${fmt(state.t0)})`, `整个流程结束`],};}).addEdge(START, "A").addEdge("A", "B").addEdge("A", "C").addEdge("B", "D").addEdge("C", "D").addEdge("D", END).compile();const result = await graph.invoke({});
console.log(result);