Conductor HITL 审批工作流实战让模型起草、让人类裁决、让送达恰好一次【免费下载链接】conductorConductor is an event driven agentic workflow engine providing durable and highly resilient execution engine for applications and AI Agents项目地址: https://gitcode.com/GitHub_Trending/co/conductor导读本文以 Conductor 开源仓库中的 HITLHuman-in-the-Loop审批配方为骨架讲解如何构建一个模型起草对外动作 → 工作流持久化暂停等待人类审批 → 仅在显式批准后以幂等方式发送的完整工作流。读完本文你将掌握HUMAN任务、JSON_JQ_TRANSFORM严格归一化、SWITCH条件路由与幂等送达的完整搭配并理解为什么缺失的审批必须默认视为拒绝。核心场景需要人类裁决的 AI 动作许多 AI 驱动的业务动作发送客户退款通知、执行对外写操作、发布营销文案不能由模型独自拍板。本配方构建的工作流hitl_approved_action正是为此设计运行结果预期模型起草一份面向客户的回复工作流持久化暂停等待人类裁决只有显式的批准才会触发送达——而拒绝与超时被记录为两种截然不同的结果绝不会被静默当作同意处理。这套流程之所以可靠依赖三类 Conductor 原语HUMAN系统任务Human.java提供持久化暂停与异步完成语义JSON_JQ_TRANSFORM系统任务JsonJqTransform.java在路由前把人类输入强制归一化成严格结构HTTP任务配合幂等键实现恰好一次送达。第一条安全原则缺失的批准 ≠ 批准本配方所对抗的典型失败模式是一个直接读取${human_decision.output.approved}并据此路由的工作流。如果审批人完成任务时没有携带该字段、或该字段以字符串false形式到达、或任务超时truthiness 检查可能放行动作通过。因此默认必须是拒绝。为此引入normalize_decision任务在任何路由发生之前把人类的原始 payload 强制收敛为严格结构{approved: ((.decision.approved // false) true), approver: (.decision.approver // unknown), note: (.decision.note // )}该表达式的语义非常明确缺失字段通过// false变为false非布尔值如字符串false、数字、对象经 true比较后一律为false只有字面量true才算批准approver缺失时回退为unknownnote缺失时回退为空字符串保证下游输出结构始终完整。SWITCH任务只基于这个归一化后的值路由绝不直接使用人类的原始输出。源码佐证JSON_JQ_TRANSFORM 的归一化行为JSON_JQ_TRANSFORM在仓库中的实现位于 JsonJqTransform.java它从任务输入读取queryExpression参数缺失则任务直接FAILED使用net.thisptr.jackson.jq.JsonQuery编译并执行 jq 表达式编译结果通过 CaffeineLoadingCache缓存过期 1 小时、上限 1000 条避免每个实例重复编译执行结果写入输出result首个结果与resultList全部结果结果类型会被精确还原对象 →Map、数组 →List、布尔 →Boolean、整数 →Long、浮点 →Double、其他 → 文本。这意味着本配方中normalize_decision.output.result.approved是一个真正的布尔类型由(… true)保证可供下游SWITCH可靠取值。三种结局全部可持久、全部可区分工作流的三种结局全部落盘且互相可区分结局delivery.status审批人批准sent并记录使用的幂等键审批人拒绝withheld_by_reviewer附带其备注超时无人裁决工作流超时approval.status保持pending这种显式区分是审计与对账的基础拒绝和过期都不是同意且不能互相混同。源码佐证HUMAN 任务的持久化暂停语义HUMAN系统任务在 Human.java 中实现start()时将任务置为IN_PROGRESScancel()时置为CANCELED。配合定义中的asyncComplete: true任务不会由 worker 轮询自动完成而是等待外部UI 或 API显式提交结果——这正是持久化暂停的机制。其映射器 HumanTaskMapper.java 在任务创建时通过ParametersUtils.getTaskInputV2解析输入参数并将任务初始状态置为IN_PROGRESS印证了创建即挂起、等待人类的语义。前置条件运行本配方需要两样东西OpenAI 集成草拟任务draft_customer_action使用LLM_CHAT_COMPLETE类型对应 LLMWorkers.java 中的WorkerTask(LLM_CHAT_COMPLETE)与 ChatCompletion.java调用openaiprovider 与gpt-4o-mini模型。可送达的端点send_approved_action会向传入的deliveryUrl发起 POST并在Idempotency-Key请求头中携带actionKey。请指向你自己的服务该服务必须真正遵守这个幂等头https://httpbin.org/post可用于试运行它会原样回显收到的内容。关于超时设计文档特别强调HUMAN任务携带20 小时超时而整个工作流的超时是86,400 秒24 小时——这才能支撑真实的审批队列。如果给审批只留 1 小时任何身处其他时区、需要睡眠的人类审批人都会让流程每晚过期。可运行的工作流定义将以下定义保存为hitl-approval.json即仓库中的 hitl-approval.json 完整内容{ name: hitl_approved_action, description: A model drafts a customer-facing action, a human decides, and only an explicit approval reaches the idempotent send. Rejection and expiry are distinct, recorded outcomes \u2014 neither is treated as consent., version: 1, schemaVersion: 2, timeoutSeconds: 86400, timeoutPolicy: TIME_OUT_WF, inputParameters: [ customerId, conversation, actionKey, deliveryUrl ], variables: { approval: { status: not_requested, approver: , decidedAt: }, delivery: { status: not_attempted } }, tasks: [ { name: draft_customer_action, taskReferenceName: draft, type: LLM_CHAT_COMPLETE, inputParameters: { llmProvider: openai, model: gpt-4o-mini, messages: [ { role: system, message: Draft a customer-facing resolution for review. Return JSON: {\summary\: string, \proposedMessage\: string, \riskFlags\: [string]}. Never promise a refund amount, credit, or deadline that is not stated in the conversation. Put anything you are unsure about in riskFlags. }, { role: user, message: Customer: ${workflow.input.customerId}\nConversation: ${workflow.input.conversation} } ], temperature: 0.2, maxTokens: 800, jsonOutput: true } }, { name: request_approval, taskReferenceName: request_approval, type: SET_VARIABLE, inputParameters: { approval: { status: pending, approver: , decidedAt: } } }, { name: await_human_decision, taskReferenceName: human_decision, type: HUMAN, asyncComplete: true }, { name: normalize_decision, taskReferenceName: normalize_decision, type: JSON_JQ_TRANSFORM, inputParameters: { decision: ${human_decision.output}, queryExpression: {approved: ((.decision.approved // false) true), approver: (.decision.approver // \unknown\), note: (.decision.note // \\)} } }, { name: record_decision, taskReferenceName: record_decision, type: SET_VARIABLE, inputParameters: { approval: { status: decided, approved: ${normalize_decision.output.result.approved}, approver: ${normalize_decision.output.result.approver}, note: ${normalize_decision.output.result.note} } } }, { name: route_on_decision, taskReferenceName: route_decision, type: SWITCH, evaluatorType: value-param, expression: approved, inputParameters: { approved: ${normalize_decision.output.result.approved} }, decisionCases: { true: [ { name: send_approved_action, taskReferenceName: send_action, type: HTTP, inputParameters: { http_request: { uri: ${workflow.input.deliveryUrl}, method: POST, headers: { Idempotency-Key: ${workflow.input.actionKey} }, body: { customerId: ${workflow.input.customerId}, message: ${draft.output.result.proposedMessage}, approvedBy: ${normalize_decision.output.result.approver} }, connectionTimeOut: 5000, readTimeOut: 15000 } } }, { name: record_delivery, taskReferenceName: record_delivery, type: SET_VARIABLE, inputParameters: { delivery: { status: sent, idempotencyKey: ${workflow.input.actionKey} } } } ], false: [ { name: record_rejection, taskReferenceName: record_rejection, type: SET_VARIABLE, inputParameters: { delivery: { status: withheld_by_reviewer, note: ${normalize_decision.output.result.note} } } } ] }, defaultCase: [] } ], outputParameters: { draft: ${draft.output.result}, approval: ${workflow.variables.approval}, delivery: ${workflow.variables.delivery} } }逐步拆解这个定义整个定义只有 6 个任务职责单一、链路清晰draft_customer_actionLLM_CHAT_COMPLETE系统提示强制模型输出固定 JSON 结构{summary, proposedMessage, riskFlags}并明确禁止承诺对话中未出现的退款金额、信用额度或截止日期拿不准的全部放入riskFlags。temperature: 0.2压低随机性jsonOutput: true强制 JSON 输出。注意模型输出经过jsonOutput后proposedMessage通过${draft.output.result.proposedMessage}引用。request_approvalSET_VARIABLE把工作流变量approval.status从not_requested置为pending为超时无人裁决的可区分结局埋下状态位。await_human_decisionHUMANasyncComplete: true表示任务不会自动完成工作流在此持久化挂起等待 UI 或 API 显式提交。normalize_decisionJSON_JQ_TRANSFORM核心安全关卡把人类提交的任意 payload 强制归一化为{approved, approver, note}严格结构详见上文表达式语义。record_decisionSET_VARIABLE将归一化结果落盘到approval变量包含status: decided、approved、approver、note。route_on_decisionSWITCHevaluatorType: value-param、expression: approved只根据归一化后的布尔值分支true分支依次执行send_actionHTTP POST 携带幂等键与record_delivery记录sent与所用幂等键false分支仅执行record_rejection记录withheld_by_reviewer与备注defaultCase为空数组兜底。工作流级timeoutSeconds: 86400timeoutPolicy: TIME_OUT_WF定义了 24 小时总预算超时则整个工作流超时approval.status停留在pending这是第三种结局的机制来源。注册并运行conductor workflow create hitl-approval.json conductor workflow start -w hitl_approved_action -i {customerId:C-123,conversation:Customer reports being charged twice for a returned order. Order 8891, two charges of $49.00 on 12 July.,actionKey:refund-note-C-123-0001,deliveryUrl:https://httpbin.org/post}打开 Conductor UI 的Executions页面本地默认http://localhost:8080/executions选中新执行即可查看任务图以及每个任务的输入输出。运行会停在human_decision任务上。先审查草稿及其riskFlags再完成任务。在 UI 中可直接从执行视图完成任务在 OSS Conductor 上等效的 API 调用是替换 WORKFLOW_IDcurl -X POST http://localhost:8080/api/tasks/WORKFLOW_ID/human_decision/COMPLETED \ -H Content-Type: application/json \ -d {approved:true,approver:support-oncall,note:Verified duplicate charge in the ledger.}值得尝试的三种完成方式以下三种提交都必须拒绝发送缺一不可{approved:false,approver:support-oncall,note:Amount not verified.} # 显式拒绝 {} # 审批人什么都没提交 {approved:false,approver:bot} # 字符串而非布尔第一种是显式拒绝approved为字面量false第二种字段完全缺失// false兜底第三种false是字符串 true比较后同样为false。每一种都会让工作流成功完成delivery.status: withheld_by_reviewer且执行过程中完全没有send_action任务。成功完成但没有发送任何东西在这里是正确的结局而不是故障——这正是本配方与把异常当成错误式实现的本质区别。生产环境注意事项文档为落地到真实业务给出了六条硬性准则批准什么就发送什么。批准后不要再重新运行模型否则人类批准的东西与最终发送的东西不一致。幂等键由调用方生成。若在工作流内部生成幂等键重试就会变成第二条消息。本配方中actionKey是入参由调用方在每次业务动作时生成如refund-note-C-123-0001。不是字面量true就是否。缺失字段和字符串false都会拒绝送达。记录批准人以及批准时间。受监管的业务还应额外记录策略版本policy version与审批人所见内容的摘要digest。约束草稿而非只约束审批。一个每小时清掉二十份草稿的审批人不可能发现模型编造的退款金额——这就是草拟任务系统提示中绝不承诺对话中未出现的金额/信用/期限不确定的放入 riskFlags的原因。提示词前先脱敏。去除支付细节附件改为按引用传递。与其他 HITL 配方的边界本仓库的 cookbook 中还有一篇姊妹配方 human-approved-action.md工作流定义见 human-approved-action.json它解决的是部署型 Agent 的工具审批场景——Agent 在调用受保护工具前暂停人类批准后同一 Agent 执行按executionId恢复继续运行而本文的hitl-approval解决的是工作流级动作送达场景——模型起草、人类裁决、幂等送达。两者共同构成 Conductor 上人工参与的两条主干路径前者面向 Agent 运行中的工具边界后者面向面向客户的最终动作。小结本文给出的hitl_approved_action工作流是AI 起草 人类裁决 幂等送达的可运行参考实现HUMAN任务提供持久化暂停Human.java、HumanTaskMapper.javaJSON_JQ_TRANSFORM提供严格的拒绝优先归一化JsonJqTransform.javaSWITCH与SET_VARIABLE把三种结局显式落盘HTTP 幂等键保证送达恰好一次。把缺失的批准永远视为拒绝是这一整套设计最值得迁移到任何 AI 工作流中的安全准则。【免费下载链接】conductorConductor is an event driven agentic workflow engine providing durable and highly resilient execution engine for applications and AI Agents项目地址: https://gitcode.com/GitHub_Trending/co/conductor创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考