Java Ai ：SpringAiAlibaba graph人工介入基础入门及详细解释【含完整代码】
·
前言
在企业级AI应用中，完全自动化的流程往往难以满足复杂业务需求。例如请假审批系统，AI可以帮我们校验信息、计算余额，但最终的决策权必须交给人。这就需要 人工介入（Human-in-the-Loop） 机制。
本文将带你从零开始，使用 Spring AI Alibaba Graph 框架，构建一个完整的请假审批工作流，深入讲解人工介入的实现原理。
代码仓库
实现效果
一、基础概念：先搞懂 人工介入 在AI工作流中的定位
1.1 什么是AI工作流中的人工介入？
人工介入，本质是在AI自动化流程中设置 中断点 ，当流程执行到该节点时，暂停AI的自动执行，将关键信息推送至人工端，由人工完成判断、确认、审批等操作后，再将人工操作结果反馈给工作流，恢复流程继续执行。
简单来说：AI负责“重复性、规则明确”的操作（如信息校验、数据计算），人工负责“不确定性、高风险、需主观判断”的操作（如信息确认、审批决策），二者协同完成完整业务流程。
1.2 人工介入的核心场景
以本文的员工请假流程为例，核心人工介入场景有2个，也是实际业务中最常见的场景：
- 场景1：AI校验后的 信息确认 — AI完成请假信息（天数、类型、原因）的基础校验后，需要人工确认信息是否准确、是否存在遗漏，避免AI误判。
- 场景2：流程中的 主管审批 — AI计算完员工假期余额后，需要主管人工审批是否同意请假，审批结果直接决定流程继续执行还是终止。
1.3 核心技术框架说明
本文案例基于Spring Cloud AI Graph 框架开发，该框架是Spring生态中用于构建AI工作流的核心框架，支持节点定义、流程编排、中断控制、状态管理，非常适合实现“AI+人工”混合工作流。
核心优势：无需手动编写复杂的流程控制逻辑，通过注解和API即可完成节点配置、分支判断、中断与恢复，大幅降低人工介入流程的开发成本。
二、核心组件
在开始解析代码前，我们先梳理案例中涉及的核心组件，明确每个组件的作用，避免后续看代码时“一头雾水”。所有组件均对应本文提供的代码，一一对应，方便大家对照查看。
2.1 核心组件分类及作用
| 组件类型 | 具体类名 | 核心作用 |
|---|---|---|
| 工作流配置类 | GraphHumanConfiguration | 定义工作流的节点、分支、状态字段，是整个工作流的“总配置” |
| 节点类（Node | LeaveInfoCheckNode、LeaveInfoConfirmNode等 | 工作流的最小执行单元，负责具体业务逻辑（AI校验、人工确认、余额计算等） |
| 调度器类（Dispatcher） | LeaveInfoDispatcher、ManagerApproveDispatcher | 负责分支判断，根据节点执行结果，决定流程下一步走向 |
| 流程处理类 | GraphProcess | 处理工作流的输出流，将节点结果、中断信息推送至前端（供人工操作） |
| 控制器类 | GraphHumanController | 提供接口供前端调用（提交申请、人工确认、主管审批、查询中断状态） |
| 状态相关 | KeyStrategyFactory、StateGraph | 管理工作流的状态字段（如user_id、leave_days），支持状态持久化与更新 |
2.2 核心概念补充
- StateGraph（状态图）：整个工作流的载体，包含所有节点、分支、状态信息，负责流程的启动、执行、中断与恢复。
- 中断点（Interrupt）：人工介入的核心触发点，通过 interruptAfter（节点执行后中断）、interruptBefore（节点执行前中断）定义。
- StateSnapshot（状态快照）：保存工作流中断时的所有状态信息，用于人工操作后恢复流程时，获取中断前的上下文。
- Stream（流）：工作流的输出方式，通过SSE（Server-Sent Events）推送给前端，实现实时反馈（如节点执行状态、中断提示）。
三、业务流程拆解
结合本文案例的员工请假流程，我们先梳理完整的业务流程，明确AI操作和人工操作的边界、流程走向，再通过流程图和时序图直观呈现，最后对应代码解析。
3.1 完整业务流程
整个请假流程分为 AI自动执行 和 人工介入 两个阶段，共7个核心步骤，其中2个人工介入步骤，具体如下：
- 员工提交请假申请：前端调用接口，传入user_id、user_name、leave_type（请假类型）、leave_days（请假天数）、leave_reason（请假原因）等参数。
- AI智能校验（自动）：工作流启动，执行AI校验节点（LeaveInfoCheckNode），校验请假信息是否完整（天数>0、类型和原因非空），返回校验结果（ok/error）。
- 人工信息确认（介入）：AI校验完成后，工作流中断（interruptAfter("ai_check")），将校验后的请假信息推送给人工，人工确认信息是否准确（确认/取消）。
- 分支判断（自动）：根据人工确认结果，调度器（LeaveInfoDispatcher）决定流程走向 — 确认通过则进入下一步，取消则流程终止。
- AI计算假期余额（自动）：确认通过后，执行余额计算节点（LeaveCalcBalanceNode），计算员工的年假、病假、事假余额。
- 主管审批（介入）：余额计算完成后，工作流再次中断（interruptBefore("manager_approve")），将员工请假信息、假期余额推送给主管，主管审批（同意/拒绝）。
- 流程收尾（自动）：根据主管审批结果，调度器（ManagerApproveDispatcher）决定流程走向 — 同意则发送通知（LeaveNotifyNode），流程结束；拒绝则流程终止。
3.2 工作流流程图
以下流程图直观呈现整个请假流程的节点、分支和人工介入点，清晰区分AI自动节点和人工节点：
3.3 人工介入时序图
时序图主要呈现 前端-后端-工作流-人工 的交互过程，重点体现人工介入时的中断、操作、恢复逻辑：
四、代码逐模块解析
接下来，我们对照业务流程，逐模块解析代码，重点关注 人工介入相关逻辑 （中断配置、中断信息推送、人工操作接口、流程恢复），其他基础代码会简要说明，确保大家能看懂每一行代码的作用。
4.1 核心依赖
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com.elvis</groupId>
<artifactId>human-in-the-loop-leave</artifactId>
<version>1.0</version>
<packaging>jar</packaging>
<name>human-in-the-loop-leave</name>
<url>http://maven.apache.org</url>
<properties>
<java.version>17</java.version>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
<spring-boot.version>3.5.4</spring-boot.version>
<spring-ai-alibaba.version>1.1.2.0</spring-ai-alibaba.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.ai</groupId>
<artifactId>spring-ai-autoconfigure-model-chat-client</artifactId>
<version>1.1.2</version>
<scope>compile</scope>
</dependency>
<!-- Spring AI Alibaba Agent Framework -->
<dependency>
<groupId>com.alibaba.cloud.ai</groupId>
<artifactId>spring-ai-alibaba-agent-framework</artifactId>
<version>${spring-ai-alibaba.version}</version>
</dependency>
<!-- DashScope ChatModel -->
<dependency>
<groupId>com.alibaba.cloud.ai</groupId>
<artifactId>spring-ai-alibaba-starter-dashscope</artifactId>
<version>${spring-ai-alibaba.version}</version>
</dependency>
<!--graph 核心-->
<dependency>
<groupId>com.alibaba.cloud.ai</groupId>
<artifactId>spring-ai-alibaba-graph-core</artifactId>
<version>${spring-ai-alibaba.version}</version>
</dependency>
</dependencies>
<dependencyManagement>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-dependencies</artifactId>
<version>${spring-boot.version}</version>
<type>pom</type>
<scope>import</scope>
</dependency>
</dependencies>
</dependencyManagement>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<version>3.8.1</version>
<configuration>
<source>17</source>
<target>17</target>
<encoding>UTF-8</encoding>
<compilerArgs>
<arg>-parameters</arg>
</compilerArgs>
</configuration>
</plugin>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
<version>${spring-boot.version}</version>
<configuration>
<mainClass>com.elvis.HumanInTheLoopLeaveApplication</mainClass>
<skip>true</skip>
</configuration>
<executions>
<execution>
<id>repackage</id>
<goals>
<goal>repackage</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</build>
</project>
4.2 主配置文件
server:
port: 8081
spring:
ai:
dashscope:
api-key: ${ALI_API_KEY}
chat:
options:
model: qwen-plus # 模型名称
temperature: 0.5 # 温度
max-tokens: 2048 # 最大令牌数
4.3 节点类：人工介入节点的核心逻辑
Graph中的节点类负责执行具体的业务逻辑，接收当前流程状态并返回要更新的数据，是工作流中实际完成AI计算、数据校验、人工交互等核心操作的执行单元。
4.3.1 人工信息确认节点：LeaveInfoConfirmNode
package com.elvis.node;
import com.alibaba.cloud.ai.graph.OverAllState;
import com.alibaba.cloud.ai.graph.StateGraph;
import com.alibaba.cloud.ai.graph.action.NodeAction;
import lombok.extern.slf4j.Slf4j;
import java.util.Map;
@Slf4j
public class LeaveInfoConfirmNode implements NodeAction {
@Override
public Map<String, Object> apply(OverAllState state) {
log.info("=== 人工信息确认节点 ===");
// 1. 获取人工确认结果（从状态中获取，人工操作后由接口更新）
boolean infoConfirm = state.value("info_confirm", false);
// 2. 根据确认结果，决定下一步走向：确认通过→计算余额，取消→结束
String nextStep = infoConfirm ? "calc_balance" : StateGraph.END;
log.info("信息确认结果：{}，下一步：{}", infoConfirm, nextStep);
// 3. 返回下一步走向，供调度器使用
return Map.of("next_after_info", nextStep);
}
}
4.4.2 主管审批节点：ManagerApproveNode
package com.elvis.node;
import com.alibaba.cloud.ai.graph.OverAllState;
import com.alibaba.cloud.ai.graph.StateGraph;
import com.alibaba.cloud.ai.graph.action.NodeAction;
import lombok.extern.slf4j.Slf4j;
import java.util.Map;
@Slf4j
public class ManagerApproveNode implements NodeAction {
@Override
public Map<String, Object> apply(OverAllState state) {
log.info("=== 主管审批节点 ===");
// 1. 获取主管审批结果（人工操作后由接口更新）
boolean approve = state.value("manager_approve", false);
// 2. 根据审批结果，决定下一步走向：同意→通知，拒绝→结束
String next = approve ? "notify" : StateGraph.END;
// 3. 返回下一步走向，供调度器使用
return Map.of("next_after_manager", next);
}
}
4.3.3 假期余额计算节点：LeaveCalcBalanceNode
从流程状态中获取用户信息，模拟计算并返回固定的年假、病假、事假余额数据。
package com.elvis.node;
import com.alibaba.cloud.ai.graph.OverAllState;
import com.alibaba.cloud.ai.graph.action.NodeAction;
import lombok.extern.slf4j.Slf4j;
import java.util.Map;
@Slf4j
public class LeaveCalcBalanceNode implements NodeAction {
@Override
public Map<String, Object> apply(OverAllState state) {
log.info("=== AI 计算假期余额 ===");
// 从 state 中获取用户信息（用于日志）
String userId = state.value("user_id", "");
String userName = state.value("user_name", "");
log.info("为用户 {} ({}) 计算假期余额", userId, userName);
// 只返回余额计算结果，用户信息会从原有 state 中保留
return Map.of(
"balance_annual", 5,
"balance_sick", 10,
"balance_personal", 3
);
}
}
4.3.4 请假信息 AI 校验节点：LeaveInfoCheckNode
从流程状态中读取请假天数、类型、原因，判断信息是否完整，并返回校验结果与提示信息。
package com.elvis.node;
import com.alibaba.cloud.ai.graph.OverAllState;
import com.alibaba.cloud.ai.graph.StateGraph;
import com.alibaba.cloud.ai.graph.action.NodeAction;
import lombok.extern.slf4j.Slf4j;
import org.springframework.ai.chat.client.ChatClient;
import java.util.Map;
@Slf4j
public class LeaveInfoCheckNode implements NodeAction {
public LeaveInfoCheckNode(ChatClient.Builder builder) {}
@Override
public Map<String, Object> apply(OverAllState state) {
log.info("=== AI 正在校验请假信息 ===");
int days = state.value("leave_days", 0);
String reason = state.value("leave_reason", "");
String type = state.value("leave_type", "");
boolean valid = days > 0 && !reason.isBlank() && !type.isBlank();
return Map.of(
"ai_check_result", valid ? "ok" : "error",
"ai_check_msg", valid ? "信息完整" : "信息不完整：天数/类型/原因不能为空"
);
}
}
4.3.5 请假流程通知节点：LeaveNotifyNode
在审批通过后执行，输出通知日志并返回包含员工、主管、HR 三方的通知结果。
package com.elvis.node;
import com.alibaba.cloud.ai.graph.OverAllState;
import com.alibaba.cloud.ai.graph.action.NodeAction;
import lombok.extern.slf4j.Slf4j;
import java.util.Map;
@Slf4j
public class LeaveNotifyNode implements NodeAction {
@Override
public Map<String, Object> apply(OverAllState state) {
log.info("=== 审批通过 → 已发送请假成功通知 ===");
return Map.of("notify_result", "通知成功：员工+主管+HR");
}
}
关键解析：
人工介入节点本身不直接和人工交互，而是通过 状态（OverAllState） 获取人工操作结果 — 人工操作（确认/审批）会通过接口更新状态中的对应字段（info_confirm、manager_approve），节点执行时读取该字段，决定下一步走向。
4.4 工作流配置类：GraphHumanConfiguration（核心配置）
该类是整个工作流的核心配置，负责定义 状态字段、节点、流程分支 ，也是人工介入节点的“注册中心”。
package com.elvis.config;
import com.alibaba.cloud.ai.graph.*;
import com.alibaba.cloud.ai.graph.action.AsyncEdgeAction;
import com.alibaba.cloud.ai.graph.action.AsyncNodeAction;
import com.alibaba.cloud.ai.graph.exception.GraphStateException;
import com.alibaba.cloud.ai.graph.state.strategy.ReplaceStrategy;
import com.elvis.dispatcher.LeaveInfoDispatcher;
import com.elvis.dispatcher.ManagerApproveDispatcher;
import com.elvis.node.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.ai.chat.client.ChatClient;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.Map;
@Configuration
@Slf4j
public class GraphHumanConfiguration {
@Bean
public StateGraph leaveGraph(ChatClient.Builder chatClientBuilder) throws GraphStateException {
// 1. 定义工作流的状态字段（所有需要传递的参数，如请假信息、校验结果、人工操作结果）
KeyStrategyFactory keyStrategyFactory = new KeyStrategyFactoryBuilder()
.addPatternStrategy("thread_id", new ReplaceStrategy()) // 会话ID，用于关联流程实例
.addPatternStrategy("user_id", new ReplaceStrategy()) // 员工ID
.addPatternStrategy("user_name", new ReplaceStrategy()) // 员工姓名
.addPatternStrategy("leave_type", new ReplaceStrategy())// 请假类型
.addPatternStrategy("leave_days", new ReplaceStrategy())// 请假天数
.addPatternStrategy("leave_reason", new ReplaceStrategy())// 请假原因
.addPatternStrategy("ai_check_result", new ReplaceStrategy())// AI校验结果
.addPatternStrategy("ai_check_msg", new ReplaceStrategy())// AI校验提示
.addPatternStrategy("info_confirm", new ReplaceStrategy())// 人工信息确认结果（true/false）
.addPatternStrategy("next_after_info", new ReplaceStrategy())// 信息确认后的下一步
.addPatternStrategy("balance_annual", new ReplaceStrategy())// 年假余额
.addPatternStrategy("balance_sick", new ReplaceStrategy())// 病假余额
.addPatternStrategy("balance_personal", new ReplaceStrategy())// 事假余额
.addPatternStrategy("manager_approve", new ReplaceStrategy())// 主管审批结果（true/false）
.addPatternStrategy("next_after_manager", new ReplaceStrategy())// 审批后的下一步
.build();
// 2. 构建状态图（工作流载体），配置节点和分支
StateGraph stateGraph = new StateGraph(keyStrategyFactory)
// 节点1：AI智能校验请假信息（自动节点）
.addNode("ai_check", AsyncNodeAction.node_async(new LeaveInfoCheckNode(chatClientBuilder)))
// 节点2：人工信息确认节点（人工介入节点，核心）
.addNode("info_check", AsyncNodeAction.node_async(new LeaveInfoConfirmNode()))
// 节点3：AI计算假期余额（自动节点）
.addNode("calc_balance", AsyncNodeAction.node_async(new LeaveCalcBalanceNode()))
// 节点4：主管审批节点（人工介入节点，核心）
.addNode("manager_approve", AsyncNodeAction.node_async(new ManagerApproveNode()))
// 节点5：执行通知（自动节点）
.addNode("notify", AsyncNodeAction.node_async(new LeaveNotifyNode()))
// 流程走向：开始 → AI校验
.addEdge(StateGraph.START, "ai_check")
// 流程走向：AI校验 → 人工信息确认
.addEdge("ai_check", "info_check")
// 条件分支：人工信息确认后的走向（核心分支）
.addConditionalEdges(
"info_check", // 分支节点：人工信息确认节点
AsyncEdgeAction.edge_async(new LeaveInfoDispatcher()), // 调度器：判断下一步
Map.of(
"calc_balance", "calc_balance", // 确认通过 → 计算余额
StateGraph.END, StateGraph.END // 确认取消 → 流程结束
)
)
// 流程走向：计算余额 → 主管审批
.addEdge("calc_balance", "manager_approve")
// 条件分支：主管审批后的走向（核心分支）
.addConditionalEdges(
"manager_approve", // 分支节点：主管审批节点
AsyncEdgeAction.edge_async(new ManagerApproveDispatcher()), // 调度器：判断下一步
Map.of(
"notify", "notify", // 审批通过 → 发送通知
StateGraph.END, StateGraph.END // 审批拒绝 → 流程结束
)
)
// 流程走向：通知完成 → 流程结束
.addEdge("notify", StateGraph.END);
// 打印流程图（方便调试，可在日志中查看）
GraphRepresentation representation = stateGraph.getGraph(GraphRepresentation.Type.PLANTUML, "员工请假流程");
log.info("\n==================================");
log.info("请假流程 UML:\n" + representation.content());
log.info("==================================\n");
return stateGraph;
}
}
关键解析：
- 状态字段配置：所有和人工介入相关的字段（如info_confirm、manager_approve）都需要注册到KeyStrategyFactory，确保流程中断后，人工操作结果能被正确保存和传递。
- 人工节点定义：info_check（人工信息确认）和manager_approve（主管审批）是两个核心人工介入节点，通过AsyncNodeAction.node_async定义为异步节点，支持流程中断。
- 分支条件：通过addConditionalEdges定义人工操作后的流程走向，由调度器（Dispatcher）根据人工操作结果（如info_confirm为true/false）决定下一步。
4.5 流程处理类：GraphProcess（中断信息推送）
该类负责处理工作流的输出流，当流程执行到中断点时，将中断信息（需要人工操作的内容）通过SSE推送给前端，供人工查看和操作，是 人工介入的交互桥梁 。
package com.elvis.controller.GraphProcess;
import com.alibaba.cloud.ai.graph.CompiledGraph;
import com.alibaba.cloud.ai.graph.NodeOutput;
import com.alibaba.cloud.ai.graph.streaming.StreamingOutput;
import com.fasterxml.jackson.annotation.JsonProperty;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.codec.ServerSentEvent;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Sinks;
import java.util.Map;
@Slf4j
public class GraphProcess {
private CompiledGraph compiledGraph;
public GraphProcess(CompiledGraph compiledGraph) {
this.compiledGraph = compiledGraph;
}
// 处理工作流输出流，推送中断信息和节点执行结果
public void processStream(Flux<NodeOutput> nodeOutputFlux, Sinks.Many<ServerSentEvent<ChatMessage>> sink) {
nodeOutputFlux
.doOnNext(output -> {
log.info("output = {}", output);
String nodeName = output.node();
ChatMessage chatMessage = null;
// 处理流式输出（本文暂不涉及，可忽略）
if (output instanceof StreamingOutput<?> streamingOutput) {
String chunk = streamingOutput.chunk();
if (chunk != null && !chunk.isEmpty()) {
chatMessage = new ChatMessage(nodeName, chunk, null, null);
}
}
else {
// 获取当前工作流的所有状态数据
Map<String, Object> stateData = output.state().data();
String interruptType = null; // 中断类型（info_confirm/manager_approve）
Map<String, Object> interruptInfo = null; // 中断信息（需人工操作的内容）
// 1. 当AI校验节点执行完成后，触发人工信息确认中断
if ("ai_check".equals(nodeName)) {
interruptType = "info_confirm";
interruptInfo = Map.of(
"message", "请确认请假信息",
"fields", Map.of(
"user_id", stateData.get("user_id"),
"user_name", stateData.get("user_name"),
"leave_type", stateData.get("leave_type"),
"leave_days", stateData.get("leave_days"),
"leave_reason", stateData.get("leave_reason"),
"ai_check_result", stateData.get("ai_check_result"),
"ai_check_msg", stateData.get("ai_check_msg")
)
);
}
// 2. 当余额计算节点执行完成后，触发主管审批中断（关键修复：传递完整用户信息）
else if ("calc_balance".equals(nodeName)) {
interruptType = "manager_approve";
interruptInfo = Map.of(
"message", "请主管审批",
"fields", Map.of(
"user_id", stateData.get("user_id"),
"user_name", stateData.get("user_name"),
"leave_type", stateData.get("leave_type"),
"leave_days", stateData.get("leave_days"),
"leave_reason", stateData.get("leave_reason"),
"balance_annual", stateData.get("balance_annual"),
"balance_sick", stateData.get("balance_sick"),
"balance_personal", stateData.get("balance_personal")
)
);
}
// 构建聊天消息（中断信息），推送给前端
chatMessage = new ChatMessage(nodeName, stateData, interruptType, interruptInfo);
}
// 将消息推送给前端（SSE方式，实时反馈）
if (chatMessage != null) {
sink.tryEmitNext(ServerSentEvent.builder(chatMessage).build());
}
})
.doOnComplete(() -> {
log.info("工作流执行完成");
sink.tryEmitComplete();
})
.doOnError(e -> {
log.error("在传输过程中出现错误", e);
sink.tryEmitError(e);
})
.subscribe();
}
// 定义推送的消息格式（前端可根据该格式解析中断信息）
public record ChatMessage(
@JsonProperty("node_name") String nodeName,
@JsonProperty("type") Object data,
@JsonProperty("interrupt_type") String interruptType,
@JsonProperty("interrupt_info") Object interruptInfo
) {
}
}
关键解析：
- 中断信息构建：当流程执行到ai_check（AI校验）和calc_balance（余额计算）节点后，会构建对应的中断信息（interruptType和interruptInfo），包含人工需要查看的所有内容（请假信息、校验结果、余额等）。
- SSE推送：通过Sinks和ServerSentEvent实现实时推送，前端接收后，可展示中断提示和相关信息，供人工操作。
- 消息格式：ChatMessage类定义了推送消息的格式，前端可根据interrupt_type判断是“信息确认”还是“主管审批”，根据interrupt_info展示具体内容。
4.6 控制器类：GraphHumanController
该类提供前端可调用的接口，负责 提交请假申请、人工确认、主管审批、查询中断状态 ，是人工操作与工作流交互的入口，核心接口对应人工介入的两个场景。
package com.elvis.controller;
import com.alibaba.cloud.ai.graph.*;
import com.alibaba.cloud.ai.graph.checkpoint.config.SaverConfig;
import com.alibaba.cloud.ai.graph.checkpoint.savers.MemorySaver;
import com.alibaba.cloud.ai.graph.exception.GraphRunnerException;
import com.alibaba.cloud.ai.graph.exception.GraphStateException;
import com.alibaba.cloud.ai.graph.state.StateSnapshot;
import com.elvis.controller.GraphProcess.GraphProcess;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.http.MediaType;
import org.springframework.http.codec.ServerSentEvent;
import org.springframework.web.bind.annotation.*;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Sinks;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
@Slf4j
@RestController
@RequestMapping("/leave")
@CrossOrigin
public class GraphHumanController {
private final CompiledGraph compiledGraph;
// 初始化工作流，配置中断点（核心：人工介入的触发配置）
@Autowired
public GraphHumanController(@Qualifier("leaveGraph") StateGraph stateGraph) throws GraphStateException {
SaverConfig saverConfig = SaverConfig.builder()
.register(new MemorySaver()) // 内存存储状态快照（生产环境可替换为Redis等）
.build();
this.compiledGraph = stateGraph.compile(
CompileConfig.builder()
.saverConfig(saverConfig)
.interruptAfter("ai_check") // 中断点1：AI校验完成后暂停，触发人工信息确认
.interruptBefore("manager_approve")// 中断点2：主管审批之前暂停，触发主管审批
.build()
);
}
// 接口1：提交请假申请（触发工作流启动）
@GetMapping(value = "/submit", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<ServerSentEvent<GraphProcess.ChatMessage>> submit(
@RequestParam String user_id,
@RequestParam String user_name,
@RequestParam String leave_type,
@RequestParam Integer leave_days,
@RequestParam String leave_reason,
@RequestParam(defaultValue = "leave_001") String thread_id
) {
RunnableConfig runnableConfig = RunnableConfig.builder().threadId(thread_id).build();
// 封装请假参数
Map<String, Object> params = new HashMap<>();
params.put("user_id", user_id);
params.put("user_name", user_name);
params.put("leave_type", leave_type);
params.put("leave_days", leave_days);
params.put("leave_reason", leave_reason);
// 处理工作流输出，推送中断信息
GraphProcess process = new GraphProcess(compiledGraph);
Sinks.Many<ServerSentEvent<GraphProcess.ChatMessage>> sink = Sinks.many().unicast().onBackpressureBuffer();
Flux<NodeOutput> flux = compiledGraph.stream(params, runnableConfig);
process.processStream(flux, sink);
return sink.asFlux()
.doOnCancel(() -> log.info("客户端断开连接"))
.doOnError(e -> log.error("流错误", e));
}
// 接口2：获取当前中断状态（前端可调用该接口，查看当前是否处于中断状态、中断类型）
@GetMapping("/interrupt-state")
public Map<String, Object> getInterruptState(@RequestParam String thread_id) throws GraphRunnerException {
RunnableConfig config = RunnableConfig.builder().threadId(thread_id).build();
Optional<StateSnapshot> snapshot = compiledGraph.stateOf(config);
if (snapshot.isEmpty()) {
return Map.of("error", "未找到会话：" + thread_id);
}
StateSnapshot state = snapshot.get();
Map<String, Object> currentState = state.state().data();
// 判断当前处于哪个中断点
String interruptPoint = null;
Map<String, Object> interruptData = new HashMap<>();
// 中断点1：AI校验后，等待人工信息确认
if (currentState.containsKey("ai_check_result") && !currentState.containsKey("info_confirm")) {
interruptPoint = "info_confirm";
interruptData.put("message", "请确认以下请假信息");
interruptData.put("leave_type", currentState.get("leave_type"));
interruptData.put("leave_days", currentState.get("leave_days"));
interruptData.put("leave_reason", currentState.get("leave_reason"));
interruptData.put("ai_check_result", currentState.get("ai_check_result"));
interruptData.put("ai_check_msg", currentState.get("ai_check_msg"));
}
// 中断点2：余额计算后，等待主管审批
else if (currentState.containsKey("balance_annual") && !currentState.containsKey("manager_approve")) {
interruptPoint = "manager_approve";
interruptData.put("message", "请主管审批");
interruptData.put("balance_annual", currentState.get("balance_annual"));
interruptData.put("balance_sick", currentState.get("balance_sick"));
interruptData.put("balance_personal", currentState.get("balance_personal"));
interruptData.put("user_name", currentState.get("user_name"));
interruptData.put("leave_type", currentState.get("leave_type"));
interruptData.put("leave_days", currentState.get("leave_days"));
interruptData.put("leave_reason", currentState.get("leave_reason"));
}
return Map.of(
"thread_id", thread_id,
"interrupt_point", interruptPoint,
"interrupt_data", interruptData,
"current_state", currentState
);
}
// 接口3：人工信息确认（人工操作后，调用该接口更新状态，恢复流程）
@GetMapping(value = "/confirm-info", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<ServerSentEvent<GraphProcess.ChatMessage>> confirmInfo(
@RequestParam String thread_id,
@RequestParam boolean info_confirm
) throws GraphRunnerException {
return resume(thread_id, Map.of("info_confirm", info_confirm));
}
// 接口4：主管审批（人工操作后，调用该接口更新状态，恢复流程）
@GetMapping(value = "/approve", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<ServerSentEvent<GraphProcess.ChatMessage>> approve(
@RequestParam String thread_id,
@RequestParam boolean manager_approve
) throws GraphRunnerException {
return resume(thread_id, Map.of("manager_approve", manager_approve));
}
// 通用恢复方法：更新人工操作结果，恢复工作流执行
private Flux<ServerSentEvent<GraphProcess.ChatMessage>> resume(String threadId, Map<String, Object> data) throws GraphRunnerException {
RunnableConfig config = RunnableConfig.builder().threadId(threadId).build();
Optional<StateSnapshot> snapshot = compiledGraph.stateOf(config);
return snapshot.map(state -> {
try {
// 1. 更新人工操作的状态（如info_confirm、manager_approve）
RunnableConfig newConfig = compiledGraph.updateState(config, data, null);
// 2. 恢复工作流执行
GraphProcess process = new GraphProcess(compiledGraph);
Sinks.Many<ServerSentEvent<GraphProcess.ChatMessage>> sink = Sinks.many().unicast().onBackpressureBuffer();
Flux<NodeOutput> flux = compiledGraph.stream(null, newConfig);
process.processStream(flux, sink);
return sink.asFlux();
} catch (Exception e) {
throw new RuntimeException(e);
}
}).orElseThrow(() -> new GraphRunnerException("未找到会话：" + threadId));
}
}
关键解析：
- 中断点配置：在构造方法中，通过interruptAfter("ai_check")和interruptBefore("manager_approve")定义两个中断点，这是人工介入的“触发开关”。
- 人工操作接口：/confirm-info（信息确认）和/approve（主管审批）接收人工操作结果，调用resume方法更新状态、恢复流程。
- 状态恢复：resume方法是核心，通过compiledGraph.updateState更新人工操作结果（如info_confirm=true），再通过compiledGraph.stream恢复流程执行。
- 中断状态查询：/interrupt-state接口供前端查询当前流程是否处于中断状态、中断类型和相关信息，方便前端展示对应的人工操作界面。
4.7 调度器类：Dispatcher（分支判断）
调度器负责根据人工操作结果，决定流程的下一步走向，本质是“读取状态中的下一步标识，返回对应的节点名称”。
4.7.1 人工信息确认调度器：LeaveInfoDispatcher
package com.elvis.dispatcher;
import com.alibaba.cloud.ai.graph.OverAllState;
import com.alibaba.cloud.ai.graph.StateGraph;
import com.alibaba.cloud.ai.graph.action.EdgeAction;
public class LeaveInfoDispatcher implements EdgeAction {
@Override
public String apply(OverAllState state) {
// 读取人工信息确认节点返回的下一步标识，决定流程走向
return state.value("next_after_info", StateGraph.END);
}
}
4.7.2 主管审批调度器：ManagerApproveDispatcher
package com.elvis.dispatcher;
import com.alibaba.cloud.ai.graph.OverAllState;
import com.alibaba.cloud.ai.graph.StateGraph;
import com.alibaba.cloud.ai.graph.action.EdgeAction;
public class ManagerApproveDispatcher implements EdgeAction {
@Override
public String apply(OverAllState state) {
// 读取主管审批节点返回的下一步标识，决定流程走向
return state.value("next_after_manager", StateGraph.END);
}
}
## 五、前端代码
该页面基于 Vue3+Element Plus 开发，采用左侧配置表单、右侧聊天窗口的布局，集成 SSE 流式通信对接后端请假审批工作流；前端封装消息渲染、自动滚动、节点文案转换通用逻辑，通过表单录入请假基础信息发起流程，监听后端 SSE 推送实时展示 AI 校验、余额计算等流程节点消息，遇到信息确认、主管审批等人工介入节点时自动弹出模态弹窗等待操作，确认后继续调用接口推进流程，完整实现 AI 自动审核 + 人工节点干预的请假审批可视化交互。
```vue
<template>
<div class="ai-chat-page">
<div class="sidebar">
<div class="sidebar-header">请假申请配置</div>
<el-card class="config-card" shadow="hover">
<el-form :model="form" label-width="100px" size="small">
<el-form-item label="会话ID">
<el-input v-model="form.threadId" style="width: 100%" />
</el-form-item>
<el-form-item label="员工ID">
<el-input v-model="form.userId" style="width: 100%" />
</el-form-item>
<el-form-item label="员工姓名">
<el-input v-model="form.userName" style="width: 100%" />
</el-form-item>
<el-form-item label="请假类型">
<el-select v-model="form.leaveType" style="width: 100%">
<el-option label="年假" value="年假" />
<el-option label="病假" value="病假" />
<el-option label="事假" value="事假" />
</el-select>
</el-form-item>
<el-form-item label="请假天数">
<el-input-number v-model="form.leaveDays" :min="1" :max="30" style="width: 100%" />
</el-form-item>
<el-form-item label="请假原因">
<el-input v-model="form.leaveReason" type="textarea" rows="3" style="width: 100%" />
</el-form-item>
<el-form-item>
<el-button type="primary" :loading="loading" @click="startLeave" style="width: 100%" round>
提交请假申请
</el-button>
</el-form-item>
</el-form>
</el-card>
</div>
<div class="chat-container">
<div class="chat-header">
<span>🤖 员工请假审批助手</span>
<el-button v-if="messages.length" text type="primary" size="small" @click="clearMessages">清空对话</el-button>
</div>
<div class="chat-content-box" v-loading="loading" ref="chatContainer">
<div v-if="messages.length === 0" class="empty-chat">
<div class="empty-icon">💬</div>
<div>填写左侧信息，点击「提交请假申请」开始</div>
</div>
<div v-for="(msg, idx) in messages" :key="idx" class="chat-message" :class="msg.role">
<div class="avatar">{{ msg.role === 'user' ? '👤' : '🤖' }}</div>
<div class="message-bubble">
<div class="message-time">{{ msg.time }}</div>
<div class="message-text" v-html="msg.content"></div>
</div>
</div>
</div>
<!-- 信息确认对话框 -->
<el-dialog v-model="showInfoConfirmDialog" title="📋 请假信息确认" width="500px" :close-on-click-modal="false">
<div class="confirm-info-content">
<el-descriptions :column="1" border>
<el-descriptions-item label="申请人">{{ interruptInfo?.fields?.user_name || form.userName }}</el-descriptions-item>
<el-descriptions-item label="请假类型">
<el-tag size="small">{{ interruptInfo?.fields?.leave_type || form.leaveType }}</el-tag>
</el-descriptions-item>
<el-descriptions-item label="请假天数">{{ interruptInfo?.fields?.leave_days || form.leaveDays }} 天</el-descriptions-item>
<el-descriptions-item label="请假原因">{{ interruptInfo?.fields?.leave_reason || form.leaveReason }}</el-descriptions-item>
<el-descriptions-item label="AI校验">
<el-tag :type="interruptInfo?.fields?.ai_check_result === 'ok' ? 'success' : 'danger'">
{{ interruptInfo?.fields?.ai_check_msg }}
</el-tag>
</el-descriptions-item>
</el-descriptions>
</div>
<template #footer>
<el-button @click="confirmInfo(false)">❌ 信息有误，结束流程</el-button>
<el-button type="primary" @click="confirmInfo(true)">✅ 信息无误，继续审批</el-button>
</template>
</el-dialog>
<!-- 主管审批对话框 -->
<el-dialog v-model="showApproveDialog" title="👔 主管审批" width="550px" :close-on-click-modal="false">
<div class="approve-content">
<el-descriptions :column="2" border>
<el-descriptions-item label="申请人" :span="2">
<strong>{{ interruptInfo?.fields?.user_name || form.userName }}</strong>
</el-descriptions-item>
<el-descriptions-item label="请假类型">{{ interruptInfo?.fields?.leave_type || form.leaveType }}</el-descriptions-item>
<el-descriptions-item label="请假天数">{{ interruptInfo?.fields?.leave_days || form.leaveDays }} 天</el-descriptions-item>
<el-descriptions-item label="请假原因" :span="2">{{ interruptInfo?.fields?.leave_reason || form.leaveReason }}</el-descriptions-item>
</el-descriptions>
<div class="balance-info">
<div class="balance-title">📊 剩余假期余额</div>
<div class="balance-items">
<div class="balance-item">
<span class="label">年假剩余</span>
<span class="value">{{ interruptInfo?.fields?.balance_annual ?? '--' }} 天</span>
</div>
<div class="balance-item">
<span class="label">病假剩余</span>
<span class="value">{{ interruptInfo?.fields?.balance_sick ?? '--' }} 天</span>
</div>
<div class="balance-item">
<span class="label">事假剩余</span>
<span class="value">{{ interruptInfo?.fields?.balance_personal ?? '--' }} 天</span>
</div>
</div>
</div>
</div>
<template #footer>
<el-button type="danger" @click="managerApprove(false)">❌ 拒绝请假</el-button>
<el-button type="success" @click="managerApprove(true)">✅ 同意请假</el-button>
</template>
</el-dialog>
</div>
</div>
</template>
<script setup>
import { ref, reactive, nextTick } from 'vue'
import { ElMessage } from 'element-plus'
const baseUrl = 'http://localhost:8081'
const form = reactive({
threadId: 'leave_001',
userId: '1001',
userName: '张三',
leaveType: '年假',
leaveDays: 2,
leaveReason: '回家处理私事'
})
const loading = ref(false)
const messages = ref([])
const showInfoConfirmDialog = ref(false)
const showApproveDialog = ref(false)
const interruptInfo = ref(null)
const chatContainer = ref(null)
// 格式化时间
const getTime = () => {
const now = new Date()
return `${now.getHours().toString().padStart(2, '0')}:${now.getMinutes().toString().padStart(2, '0')}:${now.getSeconds().toString().padStart(2, '0')}`
}
// 添加消息
const addMessage = (role, content, isHtml = false) => {
messages.value.push({
role, // 'user' 或 'assistant'
content: isHtml ? content : content.replace(/\n/g, '<br>'),
time: getTime()
})
nextTick(() => {
if (chatContainer.value) {
chatContainer.value.scrollTop = chatContainer.value.scrollHeight
}
})
}
// 清空消息
const clearMessages = () => {
messages.value = []
}
// 根据节点名称和状态数据生成友好消息
const generateFriendlyMessage = (nodeName, stateData) => {
switch (nodeName) {
case '__START__':
return `📝 收到请假申请：${stateData.user_name} 申请 ${stateData.leave_days} 天${stateData.leave_type}，原因：${stateData.leave_reason}`
case 'ai_check':
if (stateData.ai_check_result === 'ok') {
return `🔍 AI校验通过：${stateData.ai_check_msg}<br>✅ 请假信息完整，等待您确认`
} else {
return `⚠️ AI校验异常：${stateData.ai_check_msg}<br>请确认请假信息是否正确`
}
case 'info_check':
if (stateData.next_after_info === 'calc_balance') {
return `✅ 信息确认完成，继续计算假期余额...`
} else {
return `❌ 信息确认失败，流程结束`
}
case 'calc_balance':
return `💰 假期余额计算完成<br>📊 您的剩余假期：年假 ${stateData.balance_annual} 天，病假 ${stateData.balance_sick} 天，事假 ${stateData.balance_personal} 天`
case 'manager_approve':
if (stateData.next_after_manager === 'notify') {
return `👔 主管已同意请假申请，正在发送通知...`
} else {
return `❌ 主管已拒绝请假申请`
}
case 'notify':
return `📧 ${stateData.notify_result || '通知发送成功'}`
case '__END__':
return `🏁 请假流程结束，感谢使用！`
default:
return null
}
}
// 处理SSE流
const processSSEStream = async (url, onComplete) => {
const res = await fetch(url, { headers: { Accept: 'text/event-stream' } })
const reader = res.body.getReader()
const decoder = new TextDecoder()
let buffer = ''
while (true) {
const { done, value } = await reader.read()
if (done) {
if (onComplete) onComplete()
break
}
buffer += decoder.decode(value, { stream: true })
const lines = buffer.split('\n')
buffer = lines.pop() || ''
for (const line of lines) {
if (line.startsWith('data:')) {
try {
const jsonStr = line.replace('data:', '').trim()
if (!jsonStr) continue
const data = JSON.parse(jsonStr)
console.log('收到SSE数据:', data)
// 处理节点数据 - 生成友好消息
if (data.node_name && data.type && typeof data.type === 'object') {
const friendlyMsg = generateFriendlyMessage(data.node_name, data.type)
if (friendlyMsg) {
addMessage('assistant', friendlyMsg, true)
}
}
// 处理人工介入
if (data.interrupt_type) {
interruptInfo.value = data.interrupt_info
if (data.interrupt_type === 'info_confirm') {
showInfoConfirmDialog.value = true
addMessage('assistant', '⏸️ 流程已暂停，请确认以下请假信息', true)
}
else if (data.interrupt_type === 'manager_approve') {
showApproveDialog.value = true
addMessage('assistant', '⏸️ 流程已暂停，等待主管审批', true)
}
}
} catch (e) {
console.error('解析SSE数据失败:', e, line)
}
}
}
}
}
// 提交请假申请
const startLeave = async () => {
loading.value = true
messages.value = []
showInfoConfirmDialog.value = false
showApproveDialog.value = false
interruptInfo.value = null
// 先添加用户消息
addMessage('user', `提交请假申请：${form.leaveDays}天${form.leaveType}，原因：${form.leaveReason}`)
const url = `${baseUrl}/leave/submit?` + new URLSearchParams({
thread_id: form.threadId,
user_id: form.userId,
user_name: form.userName,
leave_type: form.leaveType,
leave_days: form.leaveDays,
leave_reason: form.leaveReason
})
await processSSEStream(url, () => {
loading.value = false
})
}
// 信息确认
const confirmInfo = async (ok) => {
showInfoConfirmDialog.value = false
loading.value = true
addMessage('user', ok ? '确认信息无误，继续审批' : '信息有误，取消申请')
const url = `${baseUrl}/leave/confirm-info?thread_id=${form.threadId}&info_confirm=${ok}`
await processSSEStream(url, () => {
loading.value = false
})
}
// 主管审批
const managerApprove = async (ok) => {
showApproveDialog.value = false
loading.value = true
addMessage('user', ok ? '同意请假申请' : '拒绝请假申请')
const url = `${baseUrl}/leave/approve?thread_id=${form.threadId}&manager_approve=${ok}`
await processSSEStream(url, () => {
loading.value = false
})
}
</script>
<style scoped>
.ai-chat-page {
display: flex;
height: 95vh;
gap: 20px;
padding: 20px;
background: #f0f2f5;
}
.sidebar {
width: 330px;
flex-shrink: 0;
}
.sidebar-header {
font-size: 16px;
font-weight: bold;
margin-bottom: 12px;
padding-left: 8px;
color: #303133;
}
.config-card {
border-radius: 12px;
}
.chat-container {
flex: 1;
background: #fff;
border-radius: 16px;
overflow: hidden;
display: flex;
flex-direction: column;
box-shadow: 0 2px 12px rgba(0,0,0,0.08);
}
.chat-header {
padding: 16px 20px;
border-bottom: 1px solid #e4e7ed;
font-weight: bold;
background: #fff;
display: flex;
justify-content: space-between;
align-items: center;
}
.chat-content-box {
flex: 1;
padding: 20px;
overflow-y: auto;
min-height: 0;
background: #fafafa;
}
.empty-chat {
text-align: center;
padding: 60px 20px;
color: #909399;
}
.empty-icon {
font-size: 48px;
margin-bottom: 16px;
}
.chat-message {
display: flex;
gap: 12px;
margin-bottom: 20px;
animation: fadeIn 0.3s ease;
}
@keyframes fadeIn {
from { opacity: 0; transform: translateY(10px); }
to { opacity: 1; transform: translateY(0); }
}
.chat-message.user {
flex-direction: row-reverse;
}
.chat-message.user .avatar {
background: #67c23a;
}
.chat-message.assistant .avatar {
background: #409eff;
}
.avatar {
width: 40px;
height: 40px;
background: #409eff;
color: #fff;
border-radius: 50%;
display: flex;
align-items: center;
justify-content: center;
font-size: 20px;
flex-shrink: 0;
}
.message-bubble {
max-width: 75%;
}
.chat-message.user .message-bubble {
text-align: right;
}
.message-time {
font-size: 11px;
color: #909399;
margin-bottom: 4px;
}
.chat-message.user .message-time {
text-align: right;
}
.message-text {
background: #fff;
padding: 10px 16px;
border-radius: 12px;
line-height: 1.6;
font-size: 14px;
box-shadow: 0 1px 2px rgba(0,0,0,0.05);
}
.chat-message.user .message-text {
background: #67c23a;
color: #fff;
}
.chat-message.assistant .message-text {
background: #fff;
border: 1px solid #e4e7ed;
}
.confirm-info-content,
.approve-content {
padding: 8px 0;
}
.balance-info {
margin-top: 20px;
background: #f5f7fa;
border-radius: 12px;
padding: 12px 16px;
}
.balance-title {
font-weight: bold;
margin-bottom: 12px;
font-size: 14px;
}
.balance-items {
display: flex;
gap: 16px;
}
.balance-item {
flex: 1;
text-align: center;
}
.balance-item .label {
font-size: 12px;
color: #909399;
display: block;
}
.balance-item .value {
font-size: 18px;
font-weight: bold;
color: #409eff;
}
:deep(.el-dialog) {
border-radius: 16px;
}
:deep(.el-descriptions) {
font-size: 14px;
}
</style>
六、总结
本文案例代码完整可运行，大家可直接复制到项目中，结合Spring Cloud AI依赖配置，快速搭建人工介入工作流。如果在开发过程中遇到问题，可在评论区留言，一起交流探讨！
汇聚全球AI编程工具，助力开发者即刻编程。
更多推荐
20
0- 0
已为社区贡献12条内容
所有评论(0)