JeecgBoot AI 流程编排引擎

JeecgBoot AI 流程编排引擎

学习目标

通过本章学习，你将能够：

• 理解基于 LiteFlow 的可视化 AI 工作流架构
• 掌握流程节点类型体系与扩展机制
• 了解流程上下文管理与流式响应透传
• 掌握流程中途停止机制的实现方案

前置知识

本章涉及 AI 工作流编排的通用原理，建议先阅读：

• 基于规则引擎的 AI 工作流编排

下文假设你已理解上述概念，直接聚焦 JeecgBoot 的具体实现。

---

项目实践

流程与应用的关系

JeecgBoot 的 AI 应用有两种类型：

应用类型	标识	执行方式
ChatAssistant（聊天助手）	appType = "chatAssistant"	直接调用 LLM，支持 RAG/插件/MCP
ChatFlow（聊天流程）	appType = "chatFlow"	通过 airagFlowService.runFlow() 执行流程

流程（AiragFlow）是独立的实体，包含：

• flowJson：LiteFlow 流程定义（JSON 格式的节点 + 连线）
• metadata：流程元数据（输入参数定义、超时配置等）
• 应用通过 flowId 关联到流程

流程执行入口

// 伪代码：sendWithFlow
void sendWithFlow(requestId, flowId, chatConversation, topicId, messages, sendParams) {
    FlowRunParams flowRunParams = new FlowRunParams();
    flowRunParams.setRequestId(requestId);
    flowRunParams.setFlowId(flowId);
    flowRunParams.setConversationId(chatConversation.getId());
    flowRunParams.setTopicId(topicId);
    flowRunParams.setResponseMode("streaming");

    // 构建流程输入
    flowRunParams.setInputParams(Map.of(
        FLOW_INPUT_PARAM_HISTORY, chatConversation.getMessages(),  // 历史消息
        FLOW_INPUT_PARAM_QUESTION, sendParams.getContent(),         // 用户问题
        FLOW_INPUT_PARAM_IMAGES, sendParams.getImages(),            // 图片
        ...chatConversation.getFlowInputs()                         // 应用自定义入参
    ));

    // 注册事件回调（流程执行中推送 SSE）
    flowRunParams.setEventCallback(eventData -> {
        if (EVENT_FLOW_FINISHED.equals(eventData.getEvent())) {
            // 流程完成，处理最终输出
            Object outputs = eventData.getOutputs();
            AiMessage aiMessage = buildAiMessage(outputs);
            appendMessage(messages, aiMessage, chatConversation, topicId);
            saveChatConversation(chatConversation);
        }
        if (EVENT_FLOW_ERROR.equals(eventData.getEvent()) && isTimeout(data)) {
            // 超时友好提示
            sendFriendlyTimeoutMessage(emitter);
            closeSSE(emitter);
        }
    });

    // 执行流程
    airagFlowService.runFlow(flowRunParams);
}

流程输入参数

流程的输入参数通过 metadata 中的 FLOW_METADATA_INPUTS 定义：

{
  "inputs": [
    {"key": "question", "label": "用户问题", "type": "string", "required": true},
    {"key": "history", "label": "历史消息", "type": "array", "required": false},
    {"key": "customParam", "label": "自定义参数", "type": "string", "required": false}
  ]
}

在 AI 应用初始化时（initChat 接口），平台自动从流程的 metadata 提取输入参数定义，注入到应用界面的配置表单中。

流程中途停止

用户点击”停止”按钮时，流程需要优雅终止：

// 停止流程
public Result<?> stop(String requestId) {
    JeecgFlowContext flowContext = AiragLocalCache.get(FLOW_CONTEXT, requestId);
    if (flowContext != null) {
        flowContext.setStopped(true);  // AtomicBoolean
        AiragLocalCache.remove(FLOW_CONTEXT, requestId);
    }

    SseEmitter emitter = AiragLocalCache.get(SSE, requestId);
    if (emitter != null) {
        closeSSE(emitter, EVENT_MESSAGE_END);
    }
}

节点层响应：正在执行的节点（特别是 AI 节点）需要检测 flowContext.isStopped() 标志，检测到后主动终止 LLM 输出，不等待完整响应。

流式响应透传

流程引擎在执行过程中，通过 eventCallback 将中间结果实时推送：

事件	来源节点	数据内容
AI 节点输出	LLM TokenStream	流式 token、thinking 过程、工具调用
知识库节点输出	RAG 检索	检索到的文档片段
脚本节点输出	Python/JS 执行	脚本标准输出
HTTP 节点输出	REST API 调用	API 响应体

关键设计：流程引擎不解析各节点的具体输出格式，而是将事件原样透传给回调函数。这意味着每个节点负责将自己产生的数据包装为标准 EventData 格式。

JeecgTagHelper：自定义标签注入

流程中的工具执行结果通过自定义标签注入到流式输出中：

// 伪代码：工具执行结果注入
ToolExecutionVo vo = ToolExecutionVo.build(toolExecution);
String execTag = JeecgTagHelper.createTag(
    JeecgTagHelper.TAG_JEECG_TOOL_EXEC,  // 标签类型
    JSON.toJSONString(vo)                 // 标签内容（JSON）
);
// execTag 示例：<jeecg-tool-exec>{"name":"search","args":{...},"result":"..."}</jeecg-tool-exec>

send2Client.accept(execTag, EVENT_TOOL_EXEC_DONE);

前端解析自定义标签，将工具调用过程以可折叠的方式展示，而不干扰正常回答内容的渲染。

超时处理

流程超时时的友好提示：

if (message.contains(FLOW_ERROR_MSG_LLM_TIMEOUT)) {
    message = "当前用户较多，排队中，请稍后再试！";
    sendFriendlyMessage(emitter, message);
    closeSSE(emitter, EVENT_MESSAGE_END);  // 主动关闭，防止流程切面返回异常消息
}

设计意图：超时时主动关闭 SSE 而非等待异常传播，确保用户先看到”排队中”的友好提示，而不是”调用大模型接口失败”的技术错误。

---

问题与规避

陷阱	表现	对策
流程执行异常导致 SSE 未关闭	前端一直等待响应	onError 回调中确保调用 closeSSE
流程中途停止不生效	点击停止后 LLM 仍继续输出	确保每个节点在执行前检查 flowContext.isStopped()
流程输入参数不匹配	流程定义有自定义入参但应用未传递	initChat 接口自动从流程 metadata 提取入参定义
超时后异常消息覆盖友好提示	LLM 超时错误返回技术异常信息	超时场景优先发送友好提示，再关闭 SSE
流式响应超时断开	SSE 连接长时间无数据被网关断开	SseEmitter(-0L) 设置无超时，配合应用层心跳

---

设计取舍

流程 vs 直接 LLM 调用

两种路径的适用场景：

场景	推荐方案	理由
简单问答	ChatAssistant	无需编排，直接调用 LLM
需要 RAG + 工具调用	ChatAssistant	插件和知识库直接绑定到应用
需要多步骤 AI 处理	ChatFlow	多个 AI 节点、分支判断、结果聚合
需要业务代码介入	ChatFlow	Java 节点、脚本节点可执行自定义逻辑

核心区别：ChatAssistant 是”单轮 LLM 调用 + 增强”，ChatFlow 是”多步骤流程 + LLM 作为其中一步”。

基于缓存的流程上下文 vs 数据库持久化

JeecgBoot 选择：AiragLocalCache（内存缓存）存储流程上下文。

方案	优势	代价
内存缓存	读写快，无需序列化	服务重启丢失，多实例不共享
Redis	多实例共享，持久化	额外序列化开销
数据库	最可靠	查询延迟高

流程上下文是临时状态（请求级生命周期），使用内存缓存是合理选择。如果需要跨实例共享，可升级为 Redis。

---

参考来源

• LiteFlow 官方文档
• JeecgBoot 源码 v3.9.2，AiragChatServiceImpl.java、JeecgFlowContext.java