<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) {
        // 配置全局跨域規(guī)則
        registry.addMapping("/**") // 允許所有路徑的請求
                .allowedOrigins(ORIGIN_LIST.toArray(new String[0])) // 允許的源
                .allowedMethods("GET", "POST", "PUT", "DELETE", "OPTIONS") // 允許的HTTP方法
                .allowedHeaders("Content-Type", "Authorization") // 允許的請求頭
                .allowCredentials(true); // 是否允許發(fā)送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("流式結(jié)果：" + response.toString());
	        //workflow_finished節(jié)點(diǎn)可以獲取完整答案，進(jìn)行你的邏輯處理
	        if (response.getEvent().equals("workflow_finished")) {
	            log.info("進(jìn)入workflow_finished階段");
	            String answer = response.getData().getOutputs().getAnswer();//完整答案
	        }
	         //message_end結(jié)束節(jié)點(diǎn)，進(jìn)行你的邏輯處理
	        if (response.getEvent().equals("message_end")) {
	            log.info("進(jìn)入message_end");
	        }
	
	    });
}

java
@Slf4j
@Service
@RequiredArgsConstructor
public class DifyService {

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

    @Value("${dify.key}")
    private String apiKey;
	/**
     * 流式調(diào)用dify.
     *
     * @param query 查詢文本
     * @param conversationId id
     * @param userName  用戶名
     * @return Flux 響應(yīng)流
     */
    public Flux<StreamResponse> streamingMessage(String query, String conversationId, String userName) {
        //1.設(shè)置請求體
        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);
        }
        //如果存在自定義入?yún)⒖梢约拥饺缦翸ap中
        //Map<String, Object> commoninputs = new HashMap<>();
        //commoninputs.put("search_type", searchType);
        //body.setInputs(commoninputs);

        //2.使用webclient發(fā)送post請求
        return webClient.post()
                .uri(url)
                .headers(httpHeaders -> {
                    httpHeaders.setContentType(MediaType.APPLICATION_JSON);
                    httpHeaders.setBearerAuth(apiKey);
                })
                .bodyValue(JSON.toJSONString(body))
                .retrieve()
                .bodyToFlux(StreamResponse.class);//實(shí)體轉(zhuǎn)換
                .filter(this::shouldInclude) // 過濾掉不需要的數(shù)據(jù)【根據(jù)需求增加】
                //.map(this::convertToCustomResponseAsync) // 異步轉(zhuǎn)換【如果返回格式自定義則通過異步轉(zhuǎn)換實(shí)現(xiàn)】
                .onErrorResume(throwable -> {
                	log.info("異常輸出："+throwable.getMessage())
                })
                //.concatWith(Mono.just(createCustomFinalMessage())); // 添加自定義的最終消息【根據(jù)需求增加】
    }

    private boolean shouldInclude(StreamResponse streamResponse) {
        // 示例：只要message節(jié)點(diǎn)的數(shù)據(jù)和message_end節(jié)點(diǎn)的數(shù)據(jù)
        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;

    /**
     * 任務(wù)ID.
     */
    private String task_id;

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

    /**
     * LLM 返回文本塊內(nèi)容.
     */
    private String answer;

    /**
     * 創(chuàng)建時間戳.
     */
    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;
}