基于SpringBoot和Dify实现流式响应输出

目录

• 1. 技术栈版本要求
• 2. 核心概念：流式输出
• 3. 实现步骤
• 3.1 添加依赖
• 3.2 配置 WebClient
• 3.3 实现流式输出控制器
• 3.4 实现流式输出服务层
• 3.4 实现流式输出数据访问层
• 4. 关键点说明
• 总结

在使用 Dify（假设为某种生成式 AI 模型或服务）结合 Spring Boot 和 WebClient 实现流式输出时，我们需要确保技术栈的版本兼容性，并理解流式输出的核心概念。以下是详细讲解：

1. 技术栈版本要求

Spring Boot 版本要求

最低推javascript荐版本：2.7.x 或 3.x

如果需要支持 HTTP/2 或更高级别的异步处理能力，建议使用 Spring Boot 3.x。

Spring Boot 3.x 基于 Spring Framework 6.x 和 Java 17+，提供了更好的反应式编程支持。

JDK 版本要求

最低推荐版本：Java 11

Spring Boot 2.7.x 支持 Java 8 及以上，但推荐使用 Java 11 或更高版本。

如果使用 Spring Boot 3.x，则必须使用 Java 17 或更高版本，因为 Spring Boot 3.x 已经停止支持 Java 11 以下的版本。

2. 核心概念：流式输出

流式输出（Streaming Output）是指服务器以分块的方式逐步将数据发送到客户端，而不是一次性返回完整的结果。这种方式特别适合处理大文件传输、实时数据流或生成式模型的逐词输出。

在 Spring Boot 中，可以通过以下方式实现流式输出：

• 使用 ResponseEntity<Flux<?>> 或 ResponseBodyEmitter（适用于同步场景）。
• 使用 WebClient 的反应式编程模型来处理流式请求和响应。

3. 实现步骤

3.1 添加依赖

确保在 pom.XML 中添加以下依赖项：

 <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-webflux</artifactId>
 </dependency>

spring-boot-starter-webflux 提供了反应式 Web 编程的支持。

3.2 配置 WebClient

创建一个 WebClient 实例，主要用于设置跨域资源共享（CORS, Cross-Origin Resource Sharing）。它的作用是解决前端和后端在不同域名或端口下通信时的跨域问题。

@Configuration
public class WebConfig implements WebMvcConfigurer {

    static final List<String> ORIGIN_LIST = Arrays.asList(
            // 本地
            "http://localhost:8080",
            "http://127.0.0.1:8080",
            "http://localhost:8888",
            "http://127.0.0.1:8888",
      编程      "http://localhost:8803",
            "http://127.0.0.1:8803"
    );

    @Override
    public void addCorsMappings(CorsRegistry registry) {
        // 配置全局跨域规则
        registry.addMapping("/**") // 允许所有路径的请求
                .allowedOrigins(ORIGIN_LIST.toArray(new String[0])) // 允许的源
                .allowedMethods("GET", "POST", "PUT", "DELETE", "OPTIONS") // 允许的HTTP方法
                .allowedHeaders("Content-Type", "Authorization") // 允许的请求头
                .allowCredentials(true); // 是否允许发送Cookie等凭证信息
    }
}

3.3 实现流式输出控制器

@Slf4j
@RestController
@RequestMapping("/api")
@RequiredArgsConstructor
public class DifyController {
    

    @Value("${portal.chatMessages}")
    private String chatMessages;
    
    private final DifyService difyService;
    
    @GetMapping(value = "/chatMessagesStreaming", produces = "text/event-stream")
    public Flux<StreamResponse> chatMessagesStreaming(HttpServletRequest request,
                                                  @RequestParam(value = "query", required = true) String query,
                                                  @RequestParam(value = "userName", required = true) String userName,
                                                  @RequestParam(value = "conversationId", requipythonred = false) String conversationId) throws Exception {
   
	   return difyService.streamingMessage(query, conversationId, userName).doOnNext(response -> {
	        log.info("流式结果：" + response.toString());
	        //workflow_finished节点可以获取完整答案，进行你的逻辑处keXyYP理
	        if (response.getEvent().equals("workflow_finished")) {
	            log.info("进入workflow_finished阶段");
	            String answer = response.getData().getOutputs().getAnswer();//完整答案
	        }
	         //message_end结束节点，进行你的逻辑处理
	        if (response.getEvent().equals("message_end")) {
	            log.info("进入message_end");
	        }
	
	    });
}

3.4 实现流式输出服务层

java
@Slf4j
@Service
@RequiredArgsConstructor
public class DifyService {

	@Value("${dify.url}")
    private String url;

    @Value("${dify.key}")
    private String apiKey;
	/**
     * 流式调用dify.
     *
     * @param query 查询文本
     * @param conversationId id
     * @param userName  用户名
     * @return Flux 响应流
     */
    public Flux<StreamResponse> streamingMessage(String query, String conversationId, String userName) {
        //1.设置请求体
        DifyRequestBody body = new DifyRequestBody();
        body.setInputs(new HashMap<>());
        body.setQuery(query);
        body.setResponseMode("streaming");
        body.setConversationId("");
        body.setUser(userName);
        if (StringUtils.isNotEmpty(conversationId)) {
            body.setConversationId(conversationId);
        }
        //如果存在自定义入参可以加到如下Map中
        //Map<String, Object> commoninputs = new HashMap<>();
        //commoninputs.put("search_type", searchType);
        //body.setInputs(commoninputs);

        //2.使用webclient发送post请求
        return webClient.post()
                .uri编程客栈(url)
                .headers(httpHeaders -> {
                    httpHeaders.setContentType(MediaType.APPLICATION_jsON);
                    httpHeaders.setBearerAuth(apiKey);
                })
                .bodyValue(JSON.toJSONString(body))
                .retrieve()
                .bodyToFlux(StreamResponse.class);//实体转换
                .filter(this::shouldInclude) // 过滤掉不需要的数据【根据需求增加】
                //.map(this::convertToCustomResponseAsync) // 异步转换【如果返回格式自定义则通过异步转换实现】
                .onErrorResume(throwable -> {
                	log.info("异常输出："+throwable.getMessage())
                })
                //.concatWith(Mono.just(createCustomFinalMessage())); // 添加自定义的最终消息【根据需求增加】
    }

    private boolean shouldInclude(StreamResponse streamResponse) {
        // 示例：只要message节点的数据和message_end节点的数据
        if (streamResponse.getEvent().equals("message")
                || streamResponse.getEvent().equals("message_end")) {
            return true;
        }
        return false;
    }

3.4 实现流式输出数据访问层

和dify返回流式输出格式一致

@Data
public class StreamResponse implements Serializable {

    /**
     * 不同模式下的事件类型.
     */
    private String event;

    /**
     * agent_thought id.
     */
    private String id;

    /**
     * 任务ID.
     */
    private String task_id;

    /**
     * 消息唯一ID.
     */
    private String message_id;

    /**
     * LLM 返回文本块内容.
     */
    private String answer;

    /**
     * 创建时间戳.
     */
    private Long created_at;

    /**
     * 会话 ID.
     */
    private String conversation_id;
    
    private StreamResponseData data;
}

@Data
public class StreamResponseData implements Serializable {
    private String id;
    private String workflow_id;
    private String status;
    private Long created_at;
    private Long finished_at;
    private OutputsData outputs;
}

@Data
public class OutputsData implements Serializable {
    private String answer;
}

4. 关键点说明

1.MediaType.TEXT_EVENT_STREAM_VALUE

表示使用 Server-Sent Events (SSE) 协议进行流式传输。

客户端可以通过浏览器或支持 SSE 的工具（如 Postman）接收流式数据。

2.Flux

Flux 是 Reactor 库中的核心类型，表示一个可以包含零个或多个元素的异步序列。

在这里，Flux 表示从 Dify 接收到的逐词或逐句生成的文本流。

3.WebClient 的反应式特性

WebClient 是 Spring 提供的反应式 HTTP 客户端，能够高效处理流式数据。

它不会阻塞线程，而是通过事件驱动的方式逐步处理数据

总结

通过上述步骤，我们可以使用 Spring Boot 和 WebClient 实现流式输出功能。关键在于利用反应式编程模型（Reactor 的 Flux 和 WebClient），以及正确配置流式传输协议（如 SSE）。根据需求选择合适的 Spring Boot 和 JDK 版本，可以确保项目的性能和稳定性。

到此这篇关于基于SpringBoot和Dify实现流式响应输出的文章就介绍到这了,更多相关SpringBoot Dify流式响应输出内容请搜索编程客栈(www.devze.com)以前的文章或继续浏览下面的相关文章希望大家以后多多支持编程客栈(www.devze.com)！