Dify 스트리밍 API 연계 어댑터 구현

작성 상황은 Dify 로 WorkFlow 를 작성하고 해당 Dify 에 입력부분과 출력 부분에 Adapter 를 추가하여

구현이 필요한 상황이다.

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

    @Autowired
    private WebClient difyStudioAdapter;

    @PostMapping(
        value = "/chat/stream",
        produces = MediaType.TEXT_EVENT_STREAM_VALUE
    )
    public ResponseEntity<Flux<ServerSentEvent<String>>> chatStream(
            @RequestBody ChatRequestDto request) {

        String apiKey = "서버 설정에서 가져온 Dify API Key";
        Map<String, Object> body = new HashMap<>();
        
        body.put("inputs", Map.of());
        body.put("query", query);
        body.put("user", userId);
        body.put("response_mode", "streaming");

	String requestBody = objectMapper. writeValueAsString(body); 
        
	Flux<ServerSentEvent<String>> stream = difyStudioAdapter.post() 
            .uri ( uri: "/chat-messages") 
            .header( headerName: "Authorization" .headerValues: "Bearer" + "api-key")
            .contentType(MediaType.APPLICATION_JSON) 
            .accept(MediaType.TEXT_EVENT_STREAM)
            .bodyValue(requestBody)
            .retrieve()
            .bodyToFlux(new ParameterizedTypeReference<ServerSentEvent<String>>() {})
            // .doOnNext(data -> log.info("data: {}", data)) 
            .filter(sse -> sse.data() != null) 
            .map (sse -> { return objectMapper.readTree(sse.data());})
            .doOnNext(data -> log.info("data: {}", data))
            .filter( json -> { 
                String event = json.path("event").asString(); 
                return "activity".equals(event) ||"message".equals(event) 
                    ||"message_end".equals(event) 
                    || "workflow_finished".equals(event); 
                }) 
            .doOnNext(data -> log.info("data: {}", data))
            .filter(Objects::nonNull) 
            .flatMap(data -> { 
                String event = data.path("event").asString(); 
                switch (event) { 
                    case "activity": 
                        return Flux.just(data.path("data"));
                    default:
                        return Flux.just(data);
                }        
            }) 
            .map( json -> ServerSentEvent.<String>builder().event("data").data(json.tostring()).build());        
        
        return ResponseEntity.ok()
            .contentType (MediaType.TEXT_EVENT_STREAM)
            .header(HttpHeaders.CACHE_CONTROL, "no-cache")
            .body (stream);
    }
}

 

WebClientConfig

@Configuration
public class WebClientConfig {

	@Bean
    public WebClient difyAdapter (WebClient.Builder builder) {
    	return builder
        	.baseUrl ("http://url/v1")
            .build();
    }
}