从 JSON 到可执行工作流：图构建器与节点适配器的通用设计

在低代码平台、AI Agent、审批系统和数据处理平台中，用户通常通过前端画布配置工作流，系统再把这份配置转换成真正可执行的流程。

表面上看，这只是“解析 JSON、执行节点”，但要让系统具备可扩展性、可维护性和框架兼容性，通常需要解决两个核心问题：

1. 如何把前端传来的节点和连线转换成一张可执行的流程图？
2. 如何把不同业务节点接入统一的工作流引擎？

一种通用的解决方案是：

使用 GraphBuilder 负责构建流程图，使用 NodeAdapter 负责连接业务节点和工作流引擎。

本文以 LangGraph4j 一类的状态图框架为例，介绍这套设计思路。

一、整体架构

系统可以拆成三层：

前端流程 JSON
        ↓
GraphBuilder
        ↓
工作流引擎 StateGraph
        ↓
NodeAdapter
        ↓
NodeExecutor
        ↓
具体业务节点

每一层职责不同：

• 前端 JSON：描述节点和节点之间的连线
• GraphBuilder：把 JSON 翻译成工作流引擎能够理解的图结构
• 工作流引擎：负责节点调度、状态传递和流程执行
• NodeAdapter：将业务节点适配成引擎要求的统一接口
• NodeExecutor：执行具体业务逻辑

这种分层的好处是，业务节点不需要感知工作流框架，工作流框架也不需要理解每一种业务节点的细节。

二、前端配置如何表示一张图

前端一般会提交类似这样的 JSON：

{
  "nodes": [
    {
      "id": "input-1",
      "type": "input",
      "data": {
        "label": "输入"
      }
    },
    {
      "id": "llm-1",
      "type": "llm",
      "data": {
        "configId": 1,
        "prompt": "{{question}}"
      }
    },
    {
      "id": "output-1",
      "type": "output",
      "data": {
        "responseContent": "{{llm-1.response}}"
      }
    }
  ],
  "edges": [
    {
      "source": "input-1",
      "target": "llm-1"
    },
    {
      "source": "llm-1",
      "target": "output-1"
    }
  ]
}

其中：

• nodes 描述流程中的节点
• edges 描述节点之间的连接关系
• id 是节点唯一标识
• type 决定节点使用哪一种执行器
• data 保存节点自己的配置

前端不需要关心 Java 类，也不需要了解工作流引擎的 API，只需要表达“有哪些节点”和“它们如何连接”。

三、GraphBuilder：把 JSON 翻译成工作流

GraphBuilder 的职责不是执行节点，而是构建图。

它通常包含以下几个步骤：

1. 解析节点列表
2. 解析边列表
3. 为每个节点创建对应的适配器
4. 将节点注册到 StateGraph
5. 将边注册到 StateGraph
6. 推断入口节点和出口节点
7. 返回构建完成的工作流

1. 自动识别入口节点

入口节点的判断逻辑很简单：

如果一个节点没有任何入边，那么它就是入口节点。

例如：

input-1 → llm-1 → output-1

input-1 没有其他节点指向它，因此它是入口。

这种方式不要求前端额外配置 start: true，减少了前后端约定，也避免出现“标记和实际连线不一致”的问题。

伪代码如下：

Set<String> targets = edges.stream()
    .map(Edge::getTarget)
    .collect(Collectors.toSet());

List<Node> entryNodes = nodes.stream()
    .filter(node -> !targets.contains(node.getId()))
    .toList();

2. 自动识别出口节点

出口节点的判断方式相反：

如果一个节点没有任何出边，那么它就是出口节点。

伪代码如下：

Set<String> sources = edges.stream()
    .map(Edge::getSource)
    .collect(Collectors.toSet());

List<Node> exitNodes = nodes.stream()
    .filter(node -> !sources.contains(node.getId()))
    .toList();

对于简单线性流程，这种推断非常自然。对于复杂流程，还需要额外校验：

• 是否存在多个入口
• 是否存在多个出口
• 是否存在孤立节点
• 是否存在环
• 是否存在无法到达的节点

GraphBuilder 最好在构建阶段就完成这些校验，而不是等到运行时才暴露问题。

四、NodeAdapter：连接框架和业务

大多数工作流引擎都会要求节点实现统一接口。例如：

AsyncNodeAction<AgentState>

节点接收一个状态对象，并返回一个异步结果：

CompletableFuture<Map<String, Object>>

但业务节点的实现方式各不相同：

• 大模型节点需要调用模型服务
• 输入节点需要读取用户输入
• 输出节点需要解析模板
• HTTP 节点需要调用外部接口
• 条件节点需要计算分支条件

如果让每个业务节点都直接实现工作流引擎接口，业务代码就会和框架绑定在一起。

NodeAdapter 的作用，就是把这些业务节点包装成框架能够识别的统一节点。

它的执行流程通常是：

读取当前状态
    ↓
提取 currentInput
    ↓
根据节点类型查找 NodeExecutor
    ↓
执行节点业务逻辑
    ↓
复制原状态
    ↓
写回 currentInput
    ↓
保存到 nodeOutputs
    ↓
返回新状态

五、NodeExecutor：只关心输入和输出

一个业务执行器可以设计成类似这样：

public interface NodeExecutor {

    NodeResult execute(
        Object input,
        Map<String, Object> config
    );
}

它只关心两件事：

• 输入是什么
• 输出是什么

例如，LLM 执行器可以这样理解：

public class LlmNodeExecutor implements NodeExecutor {

    @Override
    public NodeResult execute(
        Object input,
        Map<String, Object> config
    ) {
        String prompt = buildPrompt(input, config);
        String response = llmClient.chat(prompt);

        return NodeResult.success(
            Map.of("response", response)
        );
    }
}

这个类不需要知道：

• LangGraph4j 如何调度节点
• AgentState 内部如何存储数据
• 当前节点前面还有哪些节点
• 下一个节点是谁

这些事情都由适配层和工作流引擎负责。

六、NodeAdapter 的核心实现

NodeAdapter 负责完成状态转换：

public class NodeAdapter
        implements AsyncNodeAction<AgentState> {

    private final NodeConfig nodeConfig;
    private final NodeExecutorFactory executorFactory;

    @Override
    public CompletableFuture<Map<String, Object>> apply(
            AgentState state) {

        Object currentInput =
            state.data().get("currentInput");

        NodeExecutor executor =
            executorFactory.get(nodeConfig.getType());

        NodeResult result =
            executor.execute(
                currentInput,
                nodeConfig.getData()
            );

        Map<String, Object> newData =
            new HashMap<>(state.data());

        newData.put(
            "currentInput",
            result.getValue()
        );

        Map<String, Object> nodeOutputs =
            new HashMap<>(
                (Map<String, Object>)
                newData.getOrDefault(
                    "nodeOutputs",
                    new HashMap<>()
                )
            );

        nodeOutputs.put(
            nodeConfig.getId(),
            result.getValue()
        );

        newData.put("nodeOutputs", nodeOutputs);

        return CompletableFuture.completedFuture(newData);
    }
}

这段代码体现了三个重要设计。

1. 不可变状态更新

每次执行节点时，先复制一份状态：

Map<String, Object> newData =
    new HashMap<>(state.data());

然后只修改新状态，不直接修改原状态。

这样可以避免：

• 节点之间互相污染数据
• 执行失败后状态处于半修改状态
• 并发执行时出现隐蔽的数据竞争
• 调试时难以还原每一步的状态

从设计角度看，每个节点都是：

旧状态 → 新状态

而不是：

多个节点共同修改同一个对象

2. currentInput：形成默认数据管道

节点执行完成后，把结果写回 currentInput：

newData.put("currentInput", result);

下一个节点再从 currentInput 读取数据。

于是，线性流程就形成了一个隐式的数据管道：

输入节点输出
    → 下一个节点输入
    → 下一个节点输出
    → 再传给后续节点

这种方式实现简单，适合问答链、文本处理链等线性工作流。

不过，在复杂分支场景中，单一的 currentInput 可能不够用。此时可以将状态设计得更明确，例如：

{
  "question": "...",
  "userProfile": {},
  "documents": [],
  "llmResponse": {},
  "currentInput": "..."
}

也就是说：

• 线性流程可以依赖 currentInput
• 复杂流程更适合使用具名状态字段

3. nodeOutputs：保存完整执行记录

除了传递当前输入，还应该保存每个节点的输出：

{
  "nodeOutputs": {
    "input-1": {
      "question": "介绍一下状态机"
    },
    "llm-1": {
      "response": "状态机是一种..."
    }
  }
}

这样，后续节点就可以通过模板引用任意历史结果：

{{llm-1.response}}

currentInput 解决的是“下一个节点拿什么数据”，而 nodeOutputs 解决的是“整个流程中曾经产生过什么数据”。

两者的用途不同，不能互相替代。

七、工厂模式与适配器模式配合

在节点类型较多时，通常会同时使用工厂模式和适配器模式。

public class NodeExecutorFactory {

    private final Map<String, NodeExecutor> executors;

    public NodeExecutor get(String type) {
        NodeExecutor executor = executors.get(type);

        if (executor == null) {
            throw new IllegalArgumentException(
                "Unsupported node type: " + type
            );
        }

        return executor;
    }
}

新增节点时，只需要：

1. 实现一个新的 NodeExecutor
2. 注册到工厂
3. 在前端增加对应的节点类型

例如新增一个 HTTP 节点：

public class HttpNodeExecutor
        implements NodeExecutor {
    // 调用远程 HTTP 服务
}

NodeAdapter 不需要修改，工作流引擎也不需要修改。

这就是扩展性的重要来源：

新增业务能力，主要通过增加执行器完成，而不是修改核心流程代码。

八、这套设计的真正价值

这套架构并不只是为了“把代码拆成几个类”，它解决的是系统演进过程中的耦合问题。

业务与框架解耦

业务执行器不依赖具体工作流框架，便于测试、复用和迁移。

节点类型易于扩展

新增 Qwen、OpenAI、HTTP、数据库等节点时，不需要修改 GraphBuilder 和 NodeAdapter 的核心逻辑。

状态流转统一

所有节点都通过统一的状态模型读写数据，便于调试、持久化和恢复。

前端配置与后端执行分离

前端只描述图结构，后端负责将图结构转换成可执行流程。

更容易支持异步执行

NodeAdapter 返回 CompletableFuture，可以自然接入模型调用、HTTP 请求等异步操作。

九、实际开发中的几个注意点

对图结构进行完整校验

除了入口和出口，还应校验：

• 节点 ID 是否重复
• 边是否引用不存在的节点
• 是否存在环
• 是否存在孤立节点
• 是否存在多个无法合并的分支

明确定义状态结构

不要让所有数据都堆在一个无约束的 Map 中。可以使用常量、枚举或类型化对象，减少字段拼写错误。

统一异常处理

NodeExecutor 抛出的异常应该由 NodeAdapter 统一包装，记录节点 ID、节点类型和输入摘要，方便定位问题。

控制 nodeOutputs 的大小

如果每个节点都保存完整响应，长流程可能导致状态越来越大。实际系统可以：

• 只保存必要字段
• 对大文本做摘要
• 将完整结果存到外部存储
• 在状态中只保留引用地址

处理分支和并行

currentInput 很适合线性流程，但对于并行分支，需要明确：

• 分支之间如何共享状态
• 多个结果如何合并
• 合并冲突如何处理
• 哪些字段允许覆盖

十、总结

一个可扩展的工作流系统，通常可以用下面这句话概括：

GraphBuilder 负责把配置转换成图，工作流引擎负责调度，NodeAdapter 负责状态适配，NodeExecutor 负责执行业务。

其中：

• GraphBuilder 解决“流程如何搭建”
• NodeAdapter 解决“框架如何接入”
• NodeExecutor 解决“节点具体做什么”
• currentInput 解决“数据如何向后传递”
• nodeOutputs 解决“历史结果如何被引用”

如果在面试中被问到“业务节点是如何接入 LangGraph4j 的”，可以这样回答：

我们采用适配器模式，将业务节点执行逻辑与工作流引擎的状态模型解耦。NodeExecutor 只负责处理业务输入和输出，不依赖 LangGraph4j；NodeAdapter 实现引擎要求的统一节点接口，负责从状态中提取输入、调用对应执行器，并将结果写回新状态。同时，我们通过 GraphBuilder 将前端 JSON 配置转换成 StateGraph，实现流程结构与节点执行逻辑的分离。

这套思路并不局限于 AI Agent。只要系统需要“前端配置流程、后端动态执行节点”，都可以使用类似的设计。