@Service
public class DifyWorkflowService {

    private static final Logger log = LoggerFactory.getLogger(DifyWorkflowService.class);

    private final RestTemplate restTemplate;

    @Value("${dify.base-url:https://api.dify.ai/v1}")
    private String difyBaseUrl;

    @Value("${dify.workflow-api-key}")
    private String workflowApiKey;

    public DifyWorkflowService(RestTemplate restTemplate) {
        this.restTemplate = restTemplate;
    }

    /**
     * 调用Dify工作流（同步模式）
     */
    public WorkflowResult runWorkflow(String workflowId,
                                       Map<String, Object> inputs,
                                       String userId) {
        Map<String, Object> request = new HashMap<>();
        request.put("inputs", inputs);
        request.put("response_mode", "blocking"); // 同步等待结果
        request.put("user", userId);

        HttpHeaders headers = new HttpHeaders();
        headers.set("Authorization", "Bearer " + workflowApiKey);
        headers.setContentType(MediaType.APPLICATION_JSON);

        try {
            ResponseEntity<Map> response = restTemplate.exchange(
                    difyBaseUrl + "/workflows/run",
                    HttpMethod.POST,
                    new HttpEntity<>(request, headers),
                    Map.class);

            Map body = response.getBody();
            String status = (String) body.get("status");
            Map data = (Map) body.get("data");

            if ("succeeded".equals(status)) {
                return WorkflowResult.success(
                        (Map<String, Object>) data.get("outputs"),
                        (String) body.get("workflow_run_id")
                );
            } else {
                return WorkflowResult.failure(
                        (String) data.getOrDefault("error", "工作流执行失败")
                );
            }
        } catch (Exception e) {
            log.error("Dify工作流调用失败: {}", e.getMessage(), e);
            return WorkflowResult.failure("服务调用异常: " + e.getMessage());
        }
    }

    /**
     * 流式调用Dify工作流（SSE模式）
     */
    public Flux<String> streamWorkflow(String workflowId,
                                        Map<String, Object> inputs,
                                        String userId) {
        Map<String, Object> request = new HashMap<>();
        request.put("inputs", inputs);
        request.put("response_mode", "streaming");
        request.put("user", userId);

        // 使用WebClient处理SSE流
        return WebClient.create(difyBaseUrl)
                .post()
                .uri("/workflows/run")
                .header("Authorization", "Bearer " + workflowApiKey)
                .bodyValue(request)
                .retrieve()
                .bodyToFlux(String.class)
                .filter(event -> event.startsWith("data: "))
                .map(event -> event.substring(6))
                .filter(data -> !data.equals("[DONE]"));
    }

    @Data
    public static class WorkflowResult {
        private final boolean success;
        private final Map<String, Object> outputs;
        private final String runId;
        private final String errorMessage;

        static WorkflowResult success(Map<String, Object> outputs, String runId) {
            return new WorkflowResult(true, outputs, runId, null);
        }

        static WorkflowResult failure(String error) {
            return new WorkflowResult(false, null, null, error);
        }
    }
}

3.2 轻量级自研工作流引擎

/**
 * 轻量级AI工作流引擎
 * 支持顺序、条件分支、并行节点
 */
@Service
public class AiWorkflowEngine {

    private static final Logger log = LoggerFactory.getLogger(AiWorkflowEngine.class);

    private final Map<String, NodeExecutor> executors;
    private final WorkflowDefinitionRepository workflowRepo;

    public AiWorkflowEngine(List<NodeExecutor> executorList,
                              WorkflowDefinitionRepository workflowRepo) {
        this.executors = executorList.stream()
                .collect(Collectors.toMap(NodeExecutor::getType, e -> e));
        this.workflowRepo = workflowRepo;
    }

    /**
     * 执行工作流
     */
    public WorkflowExecutionResult execute(String workflowId,
                                            Map<String, Object> initialInput) {
        WorkflowDefinition definition = workflowRepo.findById(workflowId)
                .orElseThrow(() -> new IllegalArgumentException("工作流不存在: " + workflowId));

        // 初始化执行上下文
        WorkflowContext context = new WorkflowContext(workflowId, initialInput);
        context.setStartTime(System.currentTimeMillis());

        log.info("[工作流] 开始执行: {}", workflowId);

        try {
            // 找到起始节点
            NodeDefinition startNode = definition.findStartNode();
            executeNode(startNode, definition, context);

            context.setStatus(WorkflowStatus.COMPLETED);
            log.info("[工作流] 执行完成: {}，耗时{}ms",
                    workflowId, System.currentTimeMillis() - context.getStartTime());

        } catch (Exception e) {
            context.setStatus(WorkflowStatus.FAILED);
            context.setError(e.getMessage());
            log.error("[工作流] 执行失败: {}", workflowId, e);
        }

        return new WorkflowExecutionResult(context);
    }

    private void executeNode(NodeDefinition node,
                              WorkflowDefinition definition,
                              WorkflowContext context) {
        log.debug("[工作流节点] 执行: {} ({})", node.getId(), node.getType());

        // 获取节点执行器
        NodeExecutor executor = executors.get(node.getType());
        if (executor == null) {
            throw new IllegalStateException("未知节点类型: " + node.getType());
        }

        // 执行节点
        NodeResult result = executor.execute(node, context);
        context.setNodeResult(node.getId(), result);

        // 根据结果选择下一个节点
        String nextNodeId = resolveNextNode(node, result, context);
        if (nextNodeId != null) {
            NodeDefinition nextNode = definition.findNode(nextNodeId);
            executeNode(nextNode, definition, context);
        }
    }

    private String resolveNextNode(NodeDefinition node,
                                    NodeResult result,
                                    WorkflowContext context) {
        // 条件分支
        if (node.getEdges() != null) {
            for (EdgeDefinition edge : node.getEdges()) {
                if (edge.getCondition() == null ||
                        evaluateCondition(edge.getCondition(), result, context)) {
                    return edge.getTargetNodeId();
                }
            }
        }
        return null; // 终止节点
    }

    private boolean evaluateCondition(String condition,
                                       NodeResult result,
                                       WorkflowContext context) {
        // 简单的条件表达式求值（实际项目可以用SpEL）
        if (condition.startsWith("output.contains:")) {
            String value = condition.substring(16);
            Object output = result.getOutput();
            return output != null && output.toString().contains(value);
        }
        if (condition.startsWith("context.equals:")) {
            String[] parts = condition.substring(15).split("=", 2);
            Object ctxValue = context.get(parts[0]);
            return ctxValue != null && ctxValue.toString().equals(parts[1]);
        }
        return true; // 无条件边
    }
}

3.3 工作流节点执行器（LLM节点示例）

@Component
public class LlmNodeExecutor implements NodeExecutor {

    private final ChatClient chatClient;

    public LlmNodeExecutor(ChatClient.Builder builder) {
        this.chatClient = builder.build();
    }

    @Override
    public String getType() {
        return "LLM";
    }

    @Override
    public NodeResult execute(NodeDefinition node, WorkflowContext context) {
        // 从节点配置获取参数
        String promptTemplate = node.getConfig().get("prompt").toString();
        String model = (String) node.getConfig().getOrDefault("model", "gpt-4o");
        double temperature = ((Number) node.getConfig()
                .getOrDefault("temperature", 0.7)).doubleValue();

        // 从上下文渲染Prompt中的变量
        String prompt = renderTemplate(promptTemplate, context);

        // 调用LLM
        long start = System.currentTimeMillis();
        String output = chatClient.prompt()
                .options(ChatOptions.builder()
                        .model(model)
                        .temperature(temperature)
                        .build())
                .user(prompt)
                .call()
                .content();

        return NodeResult.success(output,
                Map.of("model", model,
                       "latency_ms", System.currentTimeMillis() - start,
                       "prompt_length", prompt.length()));
    }

    private String renderTemplate(String template, WorkflowContext context) {
        String result = template;
        // 替换 {{variable}} 形式的变量
        Matcher m = Pattern.compile("\\{\\{([^}]+)\\}\\}").matcher(template);
        while (m.find()) {
            String varName = m.group(1).trim();
            Object value = context.get(varName);
            if (value != null) {
                result = result.replace(m.group(), value.toString());
            }
        }
        return result;
    }
}

3.4 YAML格式工作流定义示例

# workflow-content-review.yaml
# 内容审核工作流：分析 -> 分类 -> 决策
id: content-review-v1
name: 内容审核工作流
version: 1.0

nodes:
  - id: start
    type: INPUT
    edges:
      - targetNodeId: analyze

  - id: analyze
    type: LLM
    config:
      prompt: |
        请分析以下内容，识别可能的问题：
        
        内容：{{input.content}}
        
        请从以下维度分析：违规内容、语言质量、事实准确性
        输出JSON格式：{"issues": [...], "overall_risk": "high/medium/low"}
      model: gpt-4o-mini
      temperature: 0.1
    edges:
      - condition: "output.contains:high"
        targetNodeId: escalate
      - condition: "output.contains:medium"
        targetNodeId: human_review
      - targetNodeId: approve

  - id: escalate
    type: NOTIFICATION
    config:
      channel: dingding
      message: "发现高风险内容，需要立即处理：{{analyze.output}}"
    edges:
      - targetNodeId: reject

  - id: human_review
    type: HUMAN_TASK
    config:
      assignee_role: CONTENT_REVIEWER
      timeout_hours: 24
    edges:
      - condition: "context.equals:decision=approved"
        targetNodeId: approve
      - targetNodeId: reject

  - id: approve
    type: BUSINESS_ACTION
    config:
      action: APPROVE_CONTENT
      data: "{{input.content_id}}"

  - id: reject
    type: BUSINESS_ACTION
    config:
      action: REJECT_CONTENT
      data: "{{input.content_id}}"

四、效果评估与优化

三种方案的综合对比：

评估维度	Dify	FlowiseAI	自研工作流
上手时间	1-2小时	1-2天	2-4周
非技术人员可操作	是	部分	否
定制化程度	低	中	高
与Java系统集成	REST API	REST API	原生集成
高并发性能	受限于Dify服务	受限于Flowise服务	完全可控
私有化部署	支持（开源版）	支持	天然私有
可观测性	内置基础功能	基础功能	完全自定义
工程维护成本	低	中	高
适合规模	中小型	中型	大型/核心系统

选型建议：

快速验证想法、非技术人员参与：选Dify
需要定制节点且愿意用低代码：选FlowiseAI
核心业务、高SLA要求、深度定制：自研
大部分场景的性价比选择：先用Dify，遇到瓶颈再考虑其他

五、踩坑实录

坑1：Dify的API限流在高并发下成为瓶颈

我们一个AI审核流程，高峰期每分钟要处理200个请求，Dify的Rate Limit默认是60 requests/min（云服务版）。私有化部署版可以调，但调大之后发现单机Dify服务的CPU成了瓶颈。Dify毕竟是Python服务，并发处理能力不如Java系统，高吞吐场景要做好容量规划。

坑2：FlowiseAI的自定义节点调试体验很差

FlowiseAI的自定义节点是JavaScript写的，出了问题只能在服务器日志里找错误，没有断点调试，也没有本地测试框架。一个逻辑错误经常要改代码、部署、测试反复好几次。建议把复杂逻辑封装成Java REST API，FlowiseAI只调用API，JavaScript节点只做参数转换，这样Java侧可以正常做单元测试。

坑3：自研工作流的状态管理远比想象中复杂

工作流执行到中途如果服务重启了，如何恢复？节点执行超时了如何补偿？并行分支有一个失败了整个工作流怎么处理？这些都是我一开始没想到的问题，等到生产环境遇到才一个一个填坑。建议在开始自研之前，认真评估这些边缘情况，看看引入一个成熟的工作流引擎（如Activiti、Flowable）是否更合适。

六、总结

AI工作流引擎的选型，本质上是在"灵活性"和"成本"之间取舍。Dify的价值在于让非技术人员能独立管理AI工作流，减少对开发资源的依赖；FlowiseAI在灵活性和易用性之间找到了中间地带；自研方案给你最大的控制权，但维护成本是真实存在的。

建议的演进路径：新产品从Dify快速起步，遇到瓶颈时迁移到FlowiseAI，只有真正核心且差异化的业务才值得自研。不要一开始就自研，浪费的工程时间远大于收益。