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

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

@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等凭证信息
    }
}

@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", required = false) String conversationId) throws Exception {
   
	   return difyService.streamingMessage(query, conversationId, userName).doOnNext(response -> {
	        log.info("流式结果：" + response.toString());
	        //workflow_finished节点可以获取完整答案，进行你的逻辑处理
	        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");
	        }
	
	    });
}

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;
    }

@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;
}