Flowable集成LLM实现语义决策中枢

1. 项目概述让工作流真正“思考”起来Flowable 工作流引擎在企业级业务系统中跑了十几年从审批单、采购流程到复杂的供应链协同它稳得像台老式柴油机——可靠、可预测、逻辑清晰。但问题来了当流程走到“需要判断客户投诉是否属于高危舆情”这一步时传统规则引擎要么写死一堆 if-else要么调个简单关键词匹配结果是漏判率高、误判率高、规则维护成本爆炸。这时候你盯着 Flowable 的 BPMN 图发呆心里清楚不是流程引擎不行是它缺了“理解”和“推理”的能力。而大模型 LLM 正好补上这块拼图——它不靠硬编码的规则而是靠对语义的深度理解、上下文的动态推理、以及生成式响应来处理模糊边界问题。所谓“接入 LLM 节点”本质不是给 Flowable 装个新插件而是构建一个语义决策中枢把流程中那些原本需要人工拍板、专家经验、或临时写脚本处理的“灰色地带”任务交给 LLM 去完成。比如简历筛选工作流里不是简单筛掉“学历非985”的硬条件而是让 LLM 读完整份简历JD公司文化描述输出“匹配度87%优势在项目落地能力建议进入二面并重点关注跨部门协作案例”再比如客服工单流转中LLM 实时分析对话原文历史工单产品文档直接给出“应升级为P0级技术故障转研发组并附带复现路径建议”。这不是炫技是把 LLM 从“聊天玩具”变成流程里的“智能执行单元”。我去年在一家保险科技公司落地这个方案时把理赔审核环节的自动初审率从42%拉到79%人工复核时间平均缩短63%。关键在于整个过程完全跑在 Flowable 原有架构上没动核心引擎只加了两个轻量级服务模块。下面我就把从设计思路、节点封装、参数调优到避坑实录的全过程掰开揉碎讲清楚。2. 整体架构设计与核心思路拆解2.1 为什么不能直接在 Flowable 中嵌入 LLM SDK这是新手最容易踩的第一个坑。看到 Flowable 支持 Java Delegate、Script Task就想当然地在 delegate 类里 new 一个 OpenAI 客户端然后调用 chat.completions.create()。实测下来三分钟内就会遇到三个致命问题第一Flowable 的流程引擎线程池是固定大小的默认20而 LLM API 调用是网络 I/O 密集型操作一次请求平均耗时800ms~2s线程卡死导致后续流程全部阻塞第二Java Delegate 运行在 Flowable 的 JVM 内所有 token、API Key、prompt 模板都硬编码在 class 文件里安全审计直接亮红灯第三LLM 返回的 JSON 结构千变万化Flowable 的变量序列化机制默认用 Jackson根本解析不了嵌套过深或含特殊字符的 response流程直接抛出JsonMappingException异常中断。我试过用 Spring Boot 的 WebClient 异步封装结果发现 Flowable 的异步任务调度器AsyncExecutor和 WebClient 的 Mono/Flux 线程模型根本对不上回调永远收不到。所以结论很明确LLM 必须作为独立服务存在Flowable 只负责发指令、收结果中间用标准协议桥接。这不仅是技术选型更是架构分层的铁律。2.2 三层解耦架构Flowable Adapter LLM Service我们最终采用的是严格分层的三段式架构每层职责清晰、可独立部署、可灰度升级Flowable 层决策发起者保持原样只做两件事——在 BPMN 流程图中定义一个 Service Task配置其class属性指向自定义的LlmServiceDelegate同时通过execution.setVariable(llm_input, inputMap)把结构化输入数据如用户原始文本、业务上下文、约束条件传进去。这里的关键是Flowable 不关心 LLM 怎么算只认一个约定好的输入格式和输出格式。Adapter 层协议翻译器这是整个方案的“心脏”一个独立的 Spring Boot 微服务我们叫它flowable-llm-adapter。它暴露/v1/llm/invokeREST 接口接收 Flowable 发来的 POST 请求JSON body做三件事① 校验输入合法性比如检查prompt_template_id是否存在、max_tokens是否超限② 根据model_name如 qwen2-7b 或 gpt-4o路由到对应后端③ 把 Flowable 传来的扁平化变量按目标 LLM 的 API 规范组装成标准请求OpenAI 格式 or Ollama 格式 or 自研模型 SDK 格式。Adapter 层还内置熔断器Resilience4j、重试策略最多2次、结果缓存Rediskey 为llm:${hash(input)}TTL 1h避免重复调用。LLM Service 层能力提供者完全与 Flowable 解耦。可以是云厂商 APIAzure OpenAI、本地部署模型Ollama Qwen2、或私有化大模型平台如魔搭 ModelScope 上的 finetuned 模型。这一层只管一件事收到标准请求返回标准响应必须包含choices[0].message.content字段。我们甚至用它对接过公司内部的 RAG 系统——Adapter 把用户问题知识库 ID 传过去LLM Service 先检索再生成Flowable 拿到的还是纯文本结果完全无感。这种设计带来的实际好处是当业务方说“下周要把 GPT-4 换成自家微调的 DeepSeek-V2”时运维只需改 Adapter 的配置文件Flowable 和前端流程图一动不动当 LLM Service 因为 GPU 显存不足挂了Adapter 的熔断器会自动降级返回预设的 fallback 文本如“当前智能分析繁忙请稍后重试”流程不会中断只是降级为人工处理。2.3 节点类型选择Service Task 是唯一正解BPMN 规范里能调外部服务的节点有三种Service Task、Send Task、Business Rule Task。很多人纠结选哪个其实答案非常明确必须用 Service Task。原因如下Send Task 设计初衷是发消息如 JMS、Email它的implementation属性只支持##webService或##other没有 Java Delegate 扩展点无法注入 Spring Bean也就没法调用我们的 Adapter 客户端。Business Rule Task 本质是规则引擎Drools它期望输入是事实Fact对象输出是规则结果而 LLM 的输入是 prompt输出是自由文本语义完全不匹配。强行塞进去会导致规则引擎解析失败。Service Task 的class属性天然支持自定义 Java Delegate且 Flowable 会自动注入DelegateExecution对象让我们能自由读写流程变量、控制流程走向。更重要的是Service Task 支持async属性serviceTask idllmNode flowable:asynctrue ...开启后 Flowable 会把任务扔进异步队列由独立线程池处理彻底规避主线程阻塞。我们在生产环境把 async 线程池大小设为core10, max50配合 Adapter 的熔断即使 LLM 服务抖动流程吞吐量也只下降15%远优于同步调用的雪崩效应。提示不要试图用 Script TaskJavaScript 或 Groovy调用 LLM。Groovy 的HttpBuilder在 Flowable 的沙箱环境里权限受限HTTPS 证书验证经常失败JavaScript 引擎Nashorn在 JDK15 已被移除兼容性极差。这些坑我们都踩过。3. 核心细节解析与实操要点3.1 LLM 节点的输入设计结构化 Prompt 工程LLM 不是万能的它需要精准的“指令”。在 Flowable 场景下这个指令不能是随手写的自然语言而必须是结构化、可版本化、可审计的 Prompt。我们设计了一套三层输入模型基础层Flowable 变量流程启动时业务系统通过RuntimeService.startProcessInstanceByKey()传入 Map例如MapString, Object variables new HashMap(); variables.put(customer_complaint, APP登录后一直闪退iOS 17.5iPhone 14 Pro); variables.put(product_version, v3.2.1); variables.put(user_level, VIP); variables.put(history_tickets, [{type:bug,status:resolved}, {type:feature,status:pending}]); runtimeService.startProcessInstanceByKey(complaint_flow, variables);这些变量会被LlmServiceDelegate自动收集作为 Prompt 的原始素材。模板层Prompt Template存放在数据库或配置中心我们用 Apollo每条模板有唯一 ID如complaint_risk_v2、版本号、生效时间。模板内容是 Jinja2 格式例如你是一名资深的APP质量分析师请根据以下信息判断本次投诉的风险等级高/中/低并给出理由 - 用户设备{{ device_info }} - APP版本{{ product_version }} - 用户等级{{ user_level }} - 历史工单{{ history_tickets | default(无) }} - 当前投诉{{ customer_complaint }} 输出格式必须严格为JSON包含两个字段 {risk_level: 高|中|低, reason: 不超过50字的分析}关键点① 所有变量用{{ }}包裹确保渲染安全② 强制指定输出格式JSON避免 LLM 自由发挥③ 加入角色设定“资深APP质量分析师”提升专业性。策略层Dynamic Context在LlmServiceDelegate中动态注入。比如检测到user_level VIP就额外追加一条 contextif (VIP.equals(execution.getVariable(user_level))) { context.put(vip_rule, VIP用户投诉需优先处理风险等级自动1档); }这样模板里就能用{{ vip_rule }}实现业务规则与 Prompt 的解耦。实测下来这种结构化设计让 Prompt 维护效率提升3倍运营人员只需改数据库里的模板文本开发不用发版审计时直接查模板ID和版本就能追溯每次 LLM 判断的依据。3.2 输出解析与变量映射让 Flowable “读懂”LLMLLM 返回的是一段字符串而 Flowable 流程需要的是结构化变量如risk_level用于后续网关判断。如果用正则表达式硬匹配会陷入“写一个正则修十个 bug”的深渊。我们的解决方案是强制 LLM 输出标准 JSON并用 Schema 验证。Adapter 层收到 LLM 响应后先做三步校验JSON 格式校验用 Jackson 的ObjectMapper.readTree()尝试解析捕获JsonProcessingExceptionSchema 匹配校验每个 Prompt 模板关联一个 JSON Schema存于数据库例如{ type: object, properties: { risk_level: {enum: [高, 中, 低]}, reason: {type: string, maxLength: 50} }, required: [risk_level, reason] }用json-schema-validator库验证不通过则返回错误码LLM_OUTPUT_INVALID业务逻辑校验比如risk_level为“高”时reason字段必须包含关键词“崩溃”或“闪退”。只有三重校验全通过才把 JSON 解析为 Map调用execution.setVariables(localVars)写回 Flowable。这样后续的 Exclusive Gateway 就能直接用${risk_level 高}做分支判断和普通变量毫无区别。注意绝对不要在 Flowable 的 BPMN 图里用 Expression 直接解析 LLM 返回的原始字符串。我们曾见过有人写${llm_response.contains(高风险) ? high : low}结果 LLM 一次返回“高风险已确认”另一次返回“判定为高风险级别”正则就失效了。结构化输出是底线。3.3 Token 管控与成本控制每个 token 都要精打细算LLM 调用不是免费午餐尤其在高频流程中。我们统计过一个典型客服工单分析平均消耗 1200 tokensinput 800 output 400按 GPT-4o 的价格$5/M input tokens日均 10 万次调用就是 $600 成本。必须建立 token 预估和截断机制输入预估在LlmServiceDelegate中用tiktokenPython或openai-java的TokenCount工具对拼接后的 prompt 字符串做预估。公式很简单estimated_tokens (prompt_length * 1.3) / 4英文字符按 1:1中文按 1:1.3再除以 4 得 token 数。如果预估 max_input_tokens配置项默认 2048就触发截断策略——优先保留customer_complaint和history_tickets的最新3条其余用省略号代替。输出截断Adapter 层在调用 LLM API 时强制设置max_tokens参数。但要注意LLM 可能提前结束如生成完 JSON 就停也可能超限被截断。我们的做法是在 JSON 解析前先检查 response 字符串是否以}结尾如果不是就认为被截断此时返回{risk_level:未知,reason:响应不完整请重试}并记录告警。成本监控Adapter 层每调用一次就往 Prometheus Pushgateway 推送指标llm_tokens_used{modelgpt-4o, templatecomplaint_risk_v2} 1200 llm_cost_usd{modelgpt-4o} 0.006Grafana 里就能看到各模板的 token 消耗 Top10精准定位优化点。有个真实案例把history_tickets从“全部展示”改为“仅展示未解决工单”单次调用 token 从 1800 降到 650成本直降 64%。4. 实操过程与核心环节实现4.1 开发 LlmServiceDelegateFlowable 的“LLM 驱动器”这是 Flowable 侧最核心的代码必须做到零依赖、高稳定、易调试。我们摒弃了 Spring 的RestTemplate太重用 OkHttp 手写 HTTP 客户端Component(llmServiceDelegate) public class LlmServiceDelegate implements JavaDelegate { private static final Logger log LoggerFactory.getLogger(LlmServiceDelegate.class); private final OkHttpClient httpClient new OkHttpClient.Builder() .connectTimeout(10, TimeUnit.SECONDS) .readTimeout(30, TimeUnit.SECONDS) .build(); Override public void execute(DelegateExecution execution) throws Exception { // 1. 构建输入Map取流程变量 动态上下文 MapString, Object inputMap buildInputMap(execution); // 2. 调用Adapter服务 String adapterUrl http://flowable-llm-adapter:8080/v1/llm/invoke; RequestBody body RequestBody.create( MediaType.parse(application/json), new ObjectMapper().writeValueAsString(inputMap) ); Request request new Request.Builder() .url(adapterUrl) .post(body) .build(); try (Response response httpClient.newCall(request).execute()) { if (!response.isSuccessful()) { throw new RuntimeException(Adapter call failed: response.code()); } String responseBody response.body().string(); // 3. 解析JSON响应并写回流程变量 JsonNode result new ObjectMapper().readTree(responseBody); MapString, Object outputVars new HashMap(); outputVars.put(llm_result, result); outputVars.put(risk_level, result.path(risk_level).asText()); outputVars.put(llm_reason, result.path(reason).asText()); execution.setVariables(outputVars); } catch (IOException e) { log.error(LLM invoke failed for process {}, execution.getProcessInstanceId(), e); throw new BpmnError(LLM_INVOKE_FAILED, LLM service unavailable); } } private MapString, Object buildInputMap(DelegateExecution execution) { MapString, Object inputMap new HashMap(); // 取所有流程变量 execution.getVariables().forEach((k, v) - { if (v ! null !(v instanceof byte[])) { // 过滤二进制变量 inputMap.put(k, v.toString()); } }); // 注入动态上下文 injectDynamicContext(inputMap, execution); return inputMap; } private void injectDynamicContext(MapString, Object inputMap, DelegateExecution execution) { // 示例VIP用户加权 if (VIP.equals(inputMap.get(user_level))) { inputMap.put(vip_weight, 1.5); } // 示例根据产品线注入不同知识库ID String productLine (String) inputMap.get(product_line); if (finance.equals(productLine)) { inputMap.put(kb_id, kb_finance_2024); } } }关键细节说明OkHttp 连接池复用OkHttpClient是线程安全的全局单例避免频繁创建连接异常分类处理HTTP 错误4xx/5xx抛BpmnError让 Flowable 进入错误边界事件网络异常IOException则抛运行时异常触发重试变量过滤跳过byte[]类型变量如上传的附件防止 JSON 序列化失败动态上下文注入把业务规则逻辑放在 delegate 里而不是硬编码在 prompt 模板中便于 A/B 测试。4.2 Adapter 服务开发协议转换与熔断保障Adapter 是 Spring Boot 应用核心是LlmInvokeControllerRestController RequestMapping(/v1/llm) public class LlmInvokeController { Autowired private LlmClientFactory clientFactory; // 根据model_name返回不同Client PostMapping(/invoke) CircuitBreaker(name llmInvoke, fallbackMethod fallbackInvoke) Retryable(value {IOException.class}, maxAttempts 2, backoff Backoff(delay 1000)) public ResponseEntityMapString, Object invoke(RequestBody MapString, Object inputMap) throws Exception { // 1. 校验必填字段 if (!inputMap.containsKey(prompt_template_id) || !inputMap.containsKey(model_name)) { throw new IllegalArgumentException(Missing required fields); } // 2. 获取Prompt模板从DB或Cache PromptTemplate template templateService.findById((String) inputMap.get(prompt_template_id)); if (template null) { throw new IllegalArgumentException(Template not found); } // 3. 渲染PromptJinja2 String renderedPrompt jinja2Engine.render(template.getContent(), inputMap); // 4. 构建LLM请求 LlmRequest request LlmRequest.builder() .model((String) inputMap.get(model_name)) .messages(List.of(new Message(user, renderedPrompt))) .maxTokens(Integer.parseInt(String.valueOf(inputMap.getOrDefault(max_tokens, 512)))) .build(); // 5. 调用具体LLM Client LlmResponse response clientFactory.getClient(request.getModel()).invoke(request); // 6. JSON Schema校验 JsonNode resultNode objectMapper.readTree(response.getContent()); if (!schemaValidator.validate(template.getSchema(), resultNode)) { throw new ValidationException(LLM output does not match schema); } return ResponseEntity.ok(objectMapper.convertValue(resultNode, Map.class)); } // 熔断降级方法 public ResponseEntityMapString, Object fallbackInvoke( MapString, Object inputMap, Throwable t) { log.warn(LLM invoke fallback triggered, t); MapString, Object fallback new HashMap(); fallback.put(risk_level, 未知); fallback.put(reason, 智能分析服务暂时不可用请稍后重试); return ResponseEntity.ok(fallback); } }关键组件说明LlmClientFactory工厂模式根据model_name返回不同实现OpenAiClient封装 OpenAI REST APIOllamaClient调用本地 Ollama 的/api/chat接口CustomModelClient对接公司私有模型平台的 gRPC 服务。CircuitBreaker用 Resilience4j失败率 50% 持续30秒就打开熔断器后续请求直接走fallbackInvokeRetryable网络抖动时自动重试避免单次失败影响流程Schema 校验schemaValidator是封装的json-schema-validator确保 LLM 输出始终可控。4.3 BPMN 流程图实战从草图到上线以“简历筛选工作流”为例展示如何在 Flowable Modeler 中配置 LLM 节点绘制主流程Start Event → Service TaskLLM 节点→ Exclusive Gateway → [High Match] → Human TaskHR 面试→ End[Medium Match] → Service Task发送测评链接→ End[Low Match] → End。配置 Service TaskIDllm_resume_analyzeNameLLM 简历分析ImplementationllmServiceDelegate即我们写的 delegate bean 名Async勾选 ✅关键Field Injection添加两个字段promptTemplateIdresume_screening_v3modelNameqwen2-7b配置 Exclusive GatewayCondition 1${llm_result.risk_level 高}Condition 2${llm_result.risk_level 中}Default Flow指向 Low Match 分支测试技巧在 Flowable Admin 中用“启动流程实例”功能手动输入变量 JSON{ candidate_name: 张三, resume_text: 5年Java开发主导过电商秒杀系统重构熟悉Spring Cloud、Redis集群..., job_description: 招聘高级Java工程师要求1. 3年以上分布式系统经验2. 有高并发场景实战3. 熟悉DDD... }查看流程实例日志确认llm_result变量是否正确写入在 Gateway 处设置断点验证分支是否按预期走向。实操心得第一次部署时务必在 Gateway 前加一个 Script Task内容为console.log(LLM result: execution.getVariable(llm_result));。这样能在日志里直接看到 LLM 返回的原始 JSON比查数据库快十倍。等流程跑通后再删掉。5. 常见问题与排查技巧实录5.1 典型问题速查表问题现象根本原因排查步骤解决方案流程卡在 LLM 节点日志显示AsyncExecutor线程耗尽LLM API 响应慢30sasync 线程池被占满① 查flowable-default-async-executor日志② 用jstack看线程堆栈③ 检查 Adapter 的readTimeout调大 Adapter 的readTimeout如 60s并增加 async 线程池maxPoolSizeLLM 返回 JSON但 Flowable 报Cannot construct instance of java.util.LinkedHashMapJackson 反序列化时LLM 返回的字段名含空格或特殊字符如risk level① 在LlmServiceDelegate中打印原始 response 字符串② 用在线 JSON 校验工具检查在 Prompt 模板中强制字段名用下划线risk_level并在 Adapter 层做字段名标准化同一输入LLM 有时返回高风险有时返回中风险Prompt 中未固定随机种子seedLLM 生成具有随机性① 对比两次调用的 prompt 字符串MD5② 检查 LLM API 是否传了seed参数在LlmRequest中统一设置seed42确保相同输入必得相同输出Adapter 日志报Connection refused但 LLM 服务明明在运行Docker 网络配置错误Adapter 容器无法访问 LLM 容器① 进入 Adapter 容器docker exec -it adapter sh②ping ollama或curl http://ollama:11434在 docker-compose.yml 中将两个服务放在同一 network并用 service name 互访5.2 独家避坑技巧Prompt 版本灰度发布技巧不要直接替换线上模板。我们在数据库模板表里加了status字段DRAFT/ACTIVE/OBSOLETE和weight字段0~100。新模板先设statusDRAFT, weight10意味着 10% 的流量走新模板90% 走旧模板。通过对比两组的risk_level准确率人工抽检确认效果达标后再把 weight 调到 100。这招让我们上线resume_screening_v3时0 故障。LLM 响应延迟的“感知优化”用户提交后流程页面显示“智能分析中...预计15秒”其实是前端轮询 Flowable 的taskAPI。但若 LLM 真的卡住用户等30秒会焦虑。我们的解法是在LlmServiceDelegate中调用 Adapter 前先写一个llm_status变量为pendingAdapter 收到请求后立即返回{status:accepted}前端看到llm_statuspending就显示倒计时看到llm_statusdone就刷新结果。这样用户感知的等待时间从“不确定”变成了“确定的15秒”。敏感信息脱敏前置LLM 节点处理的文本常含手机号、身份证号。我们不在 Prompt 里写“请忽略手机号”而是在buildInputMap()方法里用正则预处理String text (String) inputMap.get(customer_complaint); text text.replaceAll((?!\\d)1[3-9]\\d{9}(?!\\d), [PHONE]); text text.replaceAll(\\d{17}[0-9Xx], [ID_CARD]); inputMap.put(customer_complaint, text);这样既保护隐私又不影响 LLM 理解语义“[PHONE]”仍能提示这是联系信息。Fallback 的终极保险当熔断器打开fallbackInvoke返回“服务不可用”但业务不能停。我们在 Gateway 后加了一个兜底分支如果llm_result.risk_level 未知就触发DefaultRuleEvaluatordelegate用硬编码规则做降级判断如“VIP用户投诉含‘崩溃’字眼 → 高风险”。这保证了 LLM 服务完全宕机时流程依然能走通只是精度略低。最后分享一个小技巧在 Adapter 的/actuator/prometheus端点里我们暴露了llm_cache_hit_ratio指标。当发现某模板的缓存命中率长期低于 20%就说明这个模板的输入太个性化如含大量时间戳、随机ID不适合缓存。这时我们会把它从缓存策略里移除避免无效缓存占用内存。这个指标成了我们优化 Prompt 设计的黄金罗盘。