跳转到内容
AI Agent 教程 AI Agent 教程

JeecgBoot 多模态聊天引擎与事件流

JeecgBoot 多模态聊天引擎与事件流

学习目标

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

  • 理解 JeecgBoot 的 SSE 事件类型体系与完整生命周期
  • 掌握多模态消息（文本 + 图片 + 文档）的构建与传输
  • 了解工具调用消息的合并展示策略
  • 掌握推理模型 thinking 过程的实时推送方案

前置知识

本章涉及流式传输的通用原理，建议先阅读：

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

项目实践

SSE 事件类型体系

JeecgBoot 定义了完整的 SSE 事件类型，覆盖聊天的完整生命周期：

class EventData {
    public static final String EVENT_INIT_REQUEST_ID = "initRequestId";  // 初始化
    public static final String EVENT_MESSAGE = "message";                 // 内容推送
    public static final String EVENT_THINKING = "thinking";               // 推理模型思考
    public static final String EVENT_THINKING_END = "thinkingEnd";        // 思考结束
    public static final String EVENT_TOOL_EXEC_BEFORE = "toolExecBefore"; // 工具调用前
    public static final String EVENT_TOOL_EXEC_DONE = "toolExecDone";     // 工具调用完成
    public static final String EVENT_MESSAGE_END = "messageEnd";          // 消息推送结束
    public static final String EVENT_FLOW_ERROR = "flowError";           // 流程异常
    public static final String EVENT_FLOW_FINISHED = "flowFinished";      // 流程完成
}

事件数据结构：

{
  "requestId": "uuid-xxx",
  "conversationId": "conv-xxx",
  "topicId": "topic-xxx",
  "event": "message",
  "data": {
    "message": "这是 AI 的流式响应内容..."
  }
}

流式响应的 TokenStream 监听链

JeecgBoot 通过链式回调处理 TokenStream 的每个阶段：

chatStream
    .onPartialResponse(token -> {
        // 流式输出：逐 token 推送
        // 如果之前处于 thinking 状态，自动发送 thinkingEnd
        send2Client(token, EVENT_MESSAGE);
    })
    .beforeToolExecution(beforeExec -> {
        // 工具调用前：推送工具名称和参数
        if (showToolProcess) {
            String tag = JeecgTagHelper.createTag(TAG_JEECG_TOOL_EXEC, toJSON(beforeExec));
            send2Client(tag, EVENT_TOOL_EXEC_BEFORE);
        }
    })
    .onToolExecuted(execResult -> {
        // 工具调用完成：推送结果
        appendMessage(messages, ToolExecutionResultMessage.from(execResult), ...);
        if (showToolProcess) {
            String tag = JeecgTagHelper.createTag(TAG_JEECG_TOOL_EXEC, toJSON(execResult));
            send2Client(tag, EVENT_TOOL_EXEC_DONE);
            send2Client(tag, EVENT_MESSAGE);  // 同时作为消息推送
        }
    })
    .onIntermediateResponse(chatResponse -> {
        // 中间响应：包含 tool_calls 的 AI 消息
        // 保存包含工具调用的 AI 消息到历史
        if (hasToolCalls(chatResponse.aiMessage())) {
            appendMessage(messages, chatResponse.aiMessage(), ...);
        }
    })
    .onPartialThinking(partialThinking -> {
        // 推理模型：逐段推送 thinking 内容
        isThinking.set(true);
        send2Client(partialThinking.text(), EVENT_THINKING);
    })
    .onCompleteResponse(response -> {
        // 完成：正常结束
        appendMessage(messages, response.aiMessage(), ...);
        saveChatConversation(chatConversation);
        closeSSE(emitter, EVENT_MESSAGE_END);
    })
    .onError(error -> {
        // 异常：翻译错误信息并推送
        String errMsg = translateError(error);
        closeSSE(emitter, EVENT_FLOW_ERROR + errMsg);
    })
    .start();

多模态消息构建

用户消息支持三种内容类型：

类型来源处理方式
文本内容用户输入框TextContent.from(content)
图片上传文件或网络 URLImageContent.from(url) 或 ImageContent.from(base64, mimeType)
文档上传 txt/pdf/docx 等Tika 解析 → 提取文本 → 注入提示词

图片处理：

// 伪代码：buildImageContents
List<ImageContent> buildImageContents(List<String> imageUrls) {
    for (String url : imageUrls) {
        if (isWebUrl(url)) {
            // 网络图片：直接使用 URL
            imageContents.add(ImageContent.from(url));
        } else {
            // 本地文件：读取 → Base64 编码
            String filePath = uploadpath + File.separator + url;
            byte[] fileContent = Files.readAllBytes(Paths.get(filePath));
            String base64 = Base64.getEncoder().encodeToString(fileContent);
            String mimeType = Files.probeContentType(Paths.get(filePath));
            imageContents.add(ImageContent.from(base64, mimeType));
        }
    }
    return imageContents;
}

文档处理：

用户上传的文档（txt/pdf/docx/pptx/xlsx）通过 Tika 解析为文本后，追加到用户消息中：

String buildContentWithFiles(String content, List<String> files) {
    StringBuilder sb = new StringBuilder(content);
    for (String file : files) {
        String fileText = TikaDocumentParser.parse(file);
        sb.append("\n\n--- 文件内容 ---\n").append(fileText);
    }
    return sb.toString();
}

文档文件的内容上限为 20,000 字符，最多 3 个文件。

工具调用消息合并展示

在聊天历史中，一次完整的工具调用涉及四条消息：

  1. AI 消息（包含 tool_calls）
  2. 工具请求（tool name + args）
  3. 工具结果（tool result）
  4. AI 最终回复

如果逐条展示，用户会看到大量技术细节。JeecgBoot 通过 mergeToolMessages 合并：

// 伪代码：合并工具消息
List<MessageHistory> mergeToolMessages(histories, showToolProcess) {
    for (MessageHistory message : histories) {
        if (message.role == USER) {
            // 用户消息原样保留
            mergedMessages.add(message);
            continue;
        }


        if (message.role == AI) {
            // 合并连续 AI 消息和工具调用
            if (currentAiMsg == null) {
                currentAiMsg = createEmptyAiMessage();
            }
            merge(currentAiMsg, message);


            if (message.hasToolExecutionRequests()) {
                // 缓存工具请求，匹配后续结果
                for (req : message.toolExecutionRequests) {
                    requestCache.put(req.id, req);
                }
            }
        }


        if (message.role == TOOL && showToolProcess) {
            // 将工具结果包装为自定义标签，合并到当前 AI 消息
            ToolExecutionVo vo = ToolExecutionVo.build(
                requestCache.get(message.content),  // content 存储 toolId
                message.toolExecutionResult
            );
            String execTag = JeecgTagHelper.createTag(TAG_JEECG_TOOL_EXEC, toJSON(vo));
            merge(currentAiMsg, execTag);
        }
    }
    return mergedMessages;
}

合并效果：用户看到一条完整的 AI 回复，工具调用过程以可折叠的标签形式嵌入其中，而非独立的消息条。

断线重连机制

SSE 连接可能因网络波动断开。JeecgBoot 支持断线重连时继续接收消息：

// 伪代码：断线重连
SseEmitter receiveByRequestId(String requestId) {
    // 从缓存获取已推送的历史消息
    List<EventData> historyMsgs = AiragLocalCache.get(SSE_HISTORY_MSG, requestId);


    SseEmitter emitter = createSSE(requestId);


    // 后台线程继续推送新消息
    executor.submit(() -> {
        int lastIndex = 0;
        while (true) {
            if (lastIndex < historyMsgs.size()) {
                // 推送新消息
                emitter.send(historyMsgs.get(lastIndex++));
            } else {
                // 没有新消息，等待或超时
                if (主线程已关闭 SSE) break;
                if (等待超过 120 秒) break;
                Thread.sleep(500);
            }
        }
        emitter.send(EVENT_MESSAGE_END);
        emitter.complete();
    });


    return emitter;
}

推理模型 Thinking 推送

针对 DeepSeek 等推理模型，onPartialThinking 回调捕获 reasoning_content 并单独推送：

.onPartialThinking(partialThinking -> {
    isThinking.set(true);
    EventData thinkingEvent = new EventData(requestId, null, EVENT_THINKING, ...);
    thinkingEvent.setData(EventMessageData.builder().message(partialThinking.text()).build());
    sendMessage2Client(emitter, thinkingEvent);
})

当前端检测到 event = "thinking" 时，以特殊的思考样式（如灰色文字、斜体）展示。thinkingEnd 事件标记思考结束，后续内容恢复正常样式。

问题与规避

陷阱表现对策
工具调用过程展示过多前端展示大量 JSON 参数和结果showToolProcess 配置控制，默认 true，用户可在应用设置中关闭
图片上传路径遍历恶意构造的文件路径读取系统文件SsrfFileTypeFilter.checkPathTraversal() 校验 + 规范化路径确保在 upload 目录内
断线重连线程泄漏每次重连创建新线程，旧线程未退出120 秒超时自动退出，主线程关闭 SSE 时检测并退出后台线程
Thinking 推送与正常推送混淆thinking token 与正常 token 混在同一条消息中使用 isThinking 原子标志区分状态，收到 thinking token 时自动发送 thinkingEnd

设计取舍

工具过程展示开关

JeecgBoot 选择：通过 app.metadata.showToolProcess 控制，默认展示。

方案优势代价
始终展示用户始终看到完整过程，透明度高工具调用多时界面杂乱
始终隐藏界面简洁用户不知道 AI 调用了什么
可配置（当前方案）用户/应用可自主选择需要额外的配置管理

对于调试阶段的 AI 应用，展示工具过程很重要；对于生产阶段的应用，用户通常只需要最终结果。

SSE vs WebSocket

JeecgBoot 选择：SSE（Server-Sent Events）。

方案优势代价
SSE单向推送（服务器→客户端）足够，HTTP 兼容，断线自动重连不支持客户端→服务器消息
WebSocket双向通信额外连接管理开销，对 AI 聊天场景过度设计

AI 聊天场景中，客户端主要发送请求，服务器持续推送响应，SSE 的单向模型已完全满足需求。

参考来源