前面的博客有讲到dify工作流调接口查询数据以及调用Dify的聊天助手API，此博客需具备以上基础。背景：本人就职于某IT企业，当前公司正推进企业级AI应用的建设。基于微服务架构的业务复杂度较高，在技术选型中我们采用Dify工作流来实现核心需求。该方案的实现流程与此博客大体一致，可满足企业级应用的技术要求。搭建步骤已省略，核心内容在下方。Dify的工作流具备很多组件，大家可以根据需求灵活使用，此次演示为Demo，真实场景比这个要复杂哈。由于在工作流中使用http请求保存数据，需要使用postman验证该接口是否可用。在发布前，需要运行或调试工作流是否正常。遇到问题不要慌，解决问题就是，大家可以加入技术群或个人微信交流~有些人就会问，Dify不是都实现了吗，为啥还要编写java代码去对接了，前端不是能对接python接口吗？自己也不确定，特意借助DeepSeek查了下，支持国产！@RestController@RequestMapping("/workflow")public class WorkFlowController { @Autowired private WorkFlowService workFlowService;
/** * 上传文件 * * @param file * @return * @throws IOException */ @PostMapping("/upload") public WorkFlowFileVo upload(@RequestParam("file") MultipartFile file) throws IOException { return workFlowService.upload(file); }
/** * 执行工作流 * * @param workFlowRunDto * @return */ @PostMapping("/runWorkFlow") public SseEmitter runWorkFlow(@RequestBody WorkFlowRunDto workFlowRunDto) { return workFlowService.runWorkFlow(workFlowRunDto); }
/** * 获取工作流执行情况 * * @param workflowRunId * @return */ @GetMapping("/workFlowInfo") public WorkFlowExeVo workFlowRunInfo(String workflowRunId) { return workFlowService.workFlowRunInfo(workflowRunId); }
}
public interface WorkFlowService { /** * 上传文件 * * @param file * @return */ public WorkFlowFileVo upload(@RequestParam("file") MultipartFile file) throws IOException;
/** * 执行工作流 * * @param workFlowRunDto * @return */ public SseEmitter runWorkFlow(@RequestBody WorkFlowRunDto workFlowRunDto);
/** * 获取工作流执行情况 * * @param workflowRunId * @return */ public WorkFlowExeVo workFlowRunInfo(String workflowRunId);
}
/** * 上传文件 * * @param file * @return * @throws IOException */ @Override public WorkFlowFileVo upload(MultipartFile file) throws IOException { // 设置请求头 HttpHeaders headers = new HttpHeaders(); headers.setContentType(MediaType.MULTIPART_FORM_DATA); headers.set("Authorization", difyConfig.getSaveDataAuthorization()); // 创建请求实体 MultiValueMap<String, Object> body = new LinkedMultiValueMap<>(); // 处理文件内容 body.add("file", new ByteArrayResource(file.getBytes()) { @Override public String getFilename() { // 返回原始文件名 return file.getOriginalFilename(); } });
HttpEntity<MultiValueMap<String, Object>> requestEntity = new HttpEntity<>(body, headers); String uploadUrl = difyConfig.getSaveDataUrl() + "/files/upload";
// 发送请求 ResponseEntity<String> response = restTemplate.exchange(uploadUrl, HttpMethod.POST, requestEntity, String.class); log.info("上传文件的response: {}", response);
WorkFlowFileVo workFlowFileVo = JSON.parseObject(response.getBody(), WorkFlowFileVo.class); // 构建工作流请求体 WorkFlowRunDto workFlowRunDto = buildWorkFlowRunDto(workFlowFileVo.getId()); // 执行工作流 this.runWorkFlow(workFlowRunDto); }
/** * 执行工作流 * * @param workFlowRunDto * @return */ @Override public SseEmitter runWorkFlow(WorkFlowRunDto workFlowRunDto) { // 设置超时时间 300s SseEmitter emitter = new SseEmitter(300_000L);
// 使用异步线程处理流式响应 ExecutorService executor = Executors.newSingleThreadExecutor(); executor.execute(() -> { try { String runUrl = difyConfig.getSaveDataUrl() + "/workflows/run"; log.info("runUrl: {}", runUrl);
// 创建请求头 HttpHeaders headers = new HttpHeaders(); headers.set("Authorization", difyConfig.getSaveDataAuthorization()); headers.setContentType(MediaType.APPLICATION_JSON); // 设置接收流 headers.set(HttpHeaders.ACCEPT, MediaType.TEXT_EVENT_STREAM_VALUE); HttpEntity<WorkFlowRunDto> requestEntity = new HttpEntity<>(workFlowRunDto, headers);
// 使用restTemplate的execute方法处理流式响应 restTemplate.execute( runUrl, HttpMethod.POST, request -> { // 关键修复：手动设置Content-Type和请求头 request.getHeaders().setContentType(MediaType.APPLICATION_JSON); request.getHeaders().addAll(requestEntity.getHeaders());
// 写入请求体 if (requestEntity.getBody() != null) { new ObjectMapper().writeValue(request.getBody(), requestEntity.getBody()); } }, response -> { try (BufferedReader reader = new BufferedReader(new InputStreamReader(response.getBody()))) { // 标记是否已处理workflow_run_id boolean workflowRunIdProcessed = false; String line; while ((line = reader.readLine()) != null) { if (line.startsWith("event: ping")) { // 跳过心跳事件 continue; } emitter.send(line); log.info("line: {}", line); if (!workflowRunIdProcessed) { try { workflowRunIdProcessed = processLine(line); } catch (Exception e) { log.error("记录id出现异常: {}", e.getMessage()); emitter.completeWithError(e); } } } } emitter.complete(); return null; } );
} catch (Exception e) { log.error("处理过程中发生错误: {}", e.getMessage()); emitter.completeWithError(e); } finally { log.info("流式输出结束..."); } }); // 关闭线程池 executor.shutdown(); log.info("流式输出完成..."); return emitter; }
@Override public WorkFlowExeVo workFlowRunInfo(String workflowRunId) { log.info("获取到的工作流id: {}", workflowRunId);
// 设置请求头 HttpHeaders headers = new HttpHeaders(); headers.set("Authorization", difyConfig.getWorkFlowAuthorization());
String workFlowInfoUrl = difyConfig.getWorkFlowUrl() + "/workflows/run/" + workflowRunId;
HttpEntity<String> requestEntity = new HttpEntity<>(headers);
// 发送请求 ResponseEntity<String> response = restTemplate.exchange(workFlowInfoUrl, HttpMethod.GET, requestEntity, String.class); log.info("response: {}", response);
return JSON.parseObject(response.getBody(), WorkFlowExeVo.class); }
有小伙伴就纳闷了，公司不是有测试吗，怎么还要自己测啊？测试文件上传，由于Demo版代码执行文件与工作流执行在一块，如果基于业务需求，中间会进行其他操作。建议大家在开发过程中规范记录日志，以便在系统出现异常时能快速定位问题根源。日志的详细程度可根据实际需求和开发经验灵活调整。基于上述流程方案，我们已顺利完成了企业级应用需求的开发。
立即咨询 JOTO
JOTO 提供覆盖企业智能体规划与搭建、AI 平台私有化部署、RAG 知识工程、AI 安全治理、FDE 驻场共创及持续运营优化的全周期 AI 落地服务，帮助企业把验证中的 AI 能力转化为安全、可控、可持续迭代的生产力。 联系 JOTO 获取 AI 落地咨询
