Spring WebFlux 实现 AI 流式对话：后端架构深度解析与实战

在传统的 AI 对话交互中，用户发送请求后，后端需要等待生成式 AI 模型（例如基于 Transformer 的模型）完成整个回复，才能将结果一次性返回给前端。这种方式在生成内容较长时，会导致用户长时间的等待，影响用户体验。尤其是在需要实时反馈的场景下，例如在线客服、智能助手等，这种延迟是无法接受的。本文将深入探讨如何利用 Spring WebFlux 调用生成式 AI 提供的 stream 流式接口，实现返回实时对话，大幅提升用户体验。

Spring WebFlux 与响应式编程

Spring WebFlux 是 Spring 框架提供的响应式 Web 编程模型。它基于 Reactor 库，利用非阻塞 I/O 和事件驱动机制，可以处理高并发请求，提高服务器的吞吐量和响应速度。与传统的 Spring MVC 基于 Servlet 的阻塞式模型不同，WebFlux 能够更好地处理需要长时间运行的任务，例如调用 AI 模型生成文本。

WebFlux 的核心概念包括：

• Mono: 代表 0 或 1 个元素的异步序列。
• Flux: 代表 0 到 N 个元素的异步序列。

通过使用 Mono 和 Flux，我们可以将 AI 模型返回的 stream 流式数据转换为响应式的数据流，并通过 Server-Sent Events (SSE) 或 WebSocket 等方式实时推送给前端。

实现实时对话的关键技术

1. 生成式 AI 的 Stream 流式接口

首先，需要确保所使用的生成式 AI 模型提供了 stream 流式接口。这些接口通常会以文本块的形式逐步返回生成的内容，而不是一次性返回全部结果。例如，某些云服务商提供的 AI 平台会提供类似的接口，允许开发者以流的形式接收响应。具体的接口调用方式取决于 AI 模型的 API 文档。

2. Spring WebFlux Controller 的实现

在 Spring WebFlux 中，需要创建一个 Controller 来处理客户端的请求，并调用 AI 模型的 stream 流式接口。以下是一个示例代码：

@RestController
@RequestMapping("/ai")
public class AIController {

    @Autowired
    private AIStreamService aiStreamService; // 假设有一个 AIStreamService 来调用 AI 模型

    @GetMapping(value = "/chat", produces = MediaType.TEXT_EVENT_STREAM_VALUE) // 使用 TEXT_EVENT_STREAM 返回 SSE
    public Flux<String> chat(@RequestParam String message) {
        return aiStreamService.generateStreamResponse(message); // 调用 AIStreamService 获取流式响应
    }
}

在这个示例中，我们使用了 @GetMapping 注解来映射 /ai/chat 路径，并使用 MediaType.TEXT_EVENT_STREAM_VALUE 指定返回的 Content-Type 为 text/event-stream，这是 Server-Sent Events 的标准 Content-Type。Flux<String> 表示返回的是一个包含多个字符串元素的异步序列，每个字符串代表 AI 模型生成的一个文本块。

3. AIStreamService 的实现

AIStreamService 负责调用 AI 模型的 stream 流式接口，并将返回的响应转换为 Flux 对象。以下是一个示例代码：

@Service
public class AIStreamService {

    private final WebClient webClient; // 使用 WebClient 发起 HTTP 请求
    private final String aiApiUrl = "https://api.example.com/ai/generate"; // 替换为实际的 AI API 地址

    public AIStreamService(WebClient.Builder webClientBuilder) {
        this.webClient = webClientBuilder.baseUrl(aiApiUrl).build();
    }

    public Flux<String> generateStreamResponse(String message) {
        return webClient.post()
                .uri("?message=" + message)
                .accept(MediaType.TEXT_EVENT_STREAM) // 接收 TEXT_EVENT_STREAM 类型的数据
                .retrieve()
                .bodyToFlux(String.class) // 将响应体转换为 Flux<String>
                .onErrorReturn("Error occurred."); // 错误处理
    }
}

在这个示例中，我们使用了 Spring WebFlux 提供的 WebClient 来发起 HTTP 请求。WebClient 是一个非阻塞的 HTTP 客户端，可以很好地与 Reactor 库集成。我们指定了 AI 模型的 API 地址，并设置了请求的 Content-Type 和 Accept 类型。bodyToFlux(String.class) 方法将响应体转换为 Flux<String> 对象，以便后续处理。

4. 前端的实现

前端需要使用支持 Server-Sent Events 的技术来接收后端推送的实时数据。以下是一个使用 JavaScript 的示例代码：

const eventSource = new EventSource('/ai/chat?message=你好'); // 创建 EventSource 对象

eventSource.onmessage = function(event) { // 监听 message 事件
  const message = event.data; // 获取数据
  console.log(message); // 显示数据
  // 将数据添加到页面上，例如：
  document.getElementById('chat-content').innerHTML += message;
};

eventSource.onerror = function(error) { // 监听 error 事件
  console.error('EventSource failed:', error);
};

这段代码创建了一个 EventSource 对象，并指定了后端的 API 地址。onmessage 事件监听器负责处理接收到的数据，并将数据添加到页面上。onerror 事件监听器负责处理错误情况。

实战避坑经验总结

1. 字符编码问题：确保后端返回的数据和前端接收的数据使用相同的字符编码，避免出现乱码问题。通常建议使用 UTF-8 编码。
2. 流的关闭：在 AI 模型生成完成后，需要显式地关闭 stream 流，释放资源。可以使用 Flux.doFinally 或 Mono.doFinally 方法来执行清理操作。
3. 错误处理：需要对 AI 模型返回的错误信息进行处理，并将其展示给用户。可以使用 Flux.onErrorReturn 或 Mono.onErrorResume 方法来处理错误。
4. 并发控制：如果需要处理大量的并发请求，可以使用 Spring WebFlux 提供的 RateLimiter 或其他限流组件来控制请求的速率，避免服务器过载。可以考虑使用 Redis 来实现分布式限流。
5. Nginx 反向代理：在生产环境中，通常会使用 Nginx 作为反向代理服务器。Nginx 需要配置支持 Server-Sent Events 或 WebSocket 的代理，才能正确地转发数据。确保 Nginx 的 proxy_buffering 设置为 off，以避免缓存 SSE 数据。此外，注意调整 proxy_read_timeout 和 proxy_send_timeout，避免长时间连接被 Nginx 关闭。

通过以上步骤，我们可以使用 Spring WebFlux 调用生成式 AI 提供的 stream 流式接口，实现返回实时对话，从而提升用户体验。在实际应用中，还需要根据具体的场景进行调整和优化。使用宝塔面板可以方便地进行 Nginx 的配置和管理，例如修改配置文件、重启服务等。同时，需要关注服务器的并发连接数，合理分配资源，确保系统的稳定性和性能。