尧图网站建设 
Conductor 中的 Human Task配置人工介入任务并实现 Human-in-the-Loop 审批流【免费下载链接】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/conductorConductor 系统任务中的HUMAN类型任务用于在工作流中插入一个人工闸门工作流运行到该任务时会被暂停任务保持IN_PROGRESS状态直到外部触发Task Update API、事件处理器或队列更新 API将其标记为COMPLETED或FAILED。本文基于 Human Task 配置文档 展开完整讲解 Human Task 的 JSON 配置、两种完成方式、五种监控/回调模式以及一个可直接复用的完整审批流示例并结合开源仓库源码说明其底层执行机制帮助你在工作流中可靠地落地人工审批与外部信号等待场景。核心概念Human Task 是什么Human Tasktype: HUMAN的作用是暂停工作流并等待外部信号。它充当一个 gate任务创建后立即进入IN_PROGRESS状态此后一直挂起直到被外部触发器标记为COMPLETED或FAILED。典型使用场景包括工作流需要暂停并等待人工干预例如手动审批manual approval等待来自外部来源的事件例如 Kafka、SQS或 Conductor 内部的队列机制。从源码结构看Human Task 的极简实现印证了它只做等待的语义。任务实现类 Human 继承自WorkflowSystemTask其start方法仅将任务状态置为IN_PROGRESScancel方法仅置为CANCELED——没有任何轮询或回调逻辑真正的状态推进完全依赖外部更新Override public void start(WorkflowModel workflow, TaskModel task, WorkflowExecutor workflowExecutor) { task.setStatus(IN_PROGRESS); }JSON 配置Human Task 不需要任何专用参数即可配置。最小配置如下{ name: human, taskReferenceName: human_ref, inputParameters: {}, type: HUMAN }需要注意的是inputParameters可以为空但并非不能携带数据——任务映射器 HumanTaskMapper 在执行时会通过ParametersUtils.getTaskInputV2对inputParameters做变量替换并写入TaskModel同时记录startTime并将状态直接置为IN_PROGRESSTaskModel humanTask taskMapperContext.createTaskModel(); humanTask.setTaskType(TASK_TYPE_HUMAN); humanTask.setInputData(humanTaskInput); humanTask.setStartTime(System.currentTimeMillis()); humanTask.setStatus(TaskModel.Status.IN_PROGRESS);这意味着你在inputParameters里写入的${...}表达式会被正常求值可用于把上下文信息如申请单号、发起人带给等待中的人工任务供审批人参考。完成 Human Task 的两种官方方式方式一Task Update API通过 Task Update APIPOST api/tasks完成 Human Task需要提供taskId、任务状态和期望的任务输出。使用 CLI 的等价命令conductor task update-execution --workflow-id {workflowId} --task-ref-name waiting_around_ref --status COMPLETED --output {data_key:somedatatoWait1,data_key2:somedatatoWAit2}在服务端 REST 层该能力对应 TaskResource 中的多个端点POST /api/tasks提交TaskResult、POST /api/tasks/update-v2以及按 ref name 更新的POST /api/tasks/{workflowId}/{taskRefName}/{status}可选workerid查询参数请求体即任务输出。此外还有同步变体{workflowId}/{taskRefName}/{status}/sync会直接返回更新后的 workflow适合脚本化审批流程。方式二事件处理器 / Update Queue APIs若启用了 SQS 等事件队列集成Human Task 还可以通过 Update Queue API 完成对应实现位于 QueueAdminResourcePOST api/queue/update/{workflowId}/{taskRefName}/{status}POST api/queue/update/{workflowId}/task/{taskId}/{status}POST 消息体中携带的任何参数都会作为该任务的输出重复。例如发送如下 COMPLETED 消息curl -X POST {{ server_host }}{{ api_prefix }}/queue/update/{workflowId}/waiting_around_ref/COMPLETED \ -H Content-Type: application/json \ -d {data_key:somedatatoWait1,data_key2:somedatatoWAit2}则 Human Task 的输出为{ data_key:somedatatoWait1, data_key2:somedatatoWAit2 }从源码看该端点将请求委托给DefaultEventQueueProcessor.updateByTaskRefName / updateByTaskId位于 DefaultEventQueueProcessor即通过在事件队列中发布消息来驱动任务状态变更——这正是它能与 SQS/Kafka 等外部事件源解耦对接的原因。另外也可以配置一个使用complete_taskaction 的事件处理器来完成 Human Task这样外部系统只需发布事件无需直接调用 HTTP API。监控 Human Task回调与通知模式当工作流到达 Human Task 时通常需要收到通知或回调以触发下一步动作发邮件、通知 Slack 频道、触发外部系统等。文档推荐了五种模式模式 1轮询 Workflow Status API最简单的方式是轮询工作流执行状态查找处于IN_PROGRESS状态的 Human 任务# Get workflow execution status curl {{ server_host }}/api/workflow/{workflowId} \ -H accept: application/json解析响应找到taskType: HUMAN且status: IN_PROGRESS的任务。优点实现简单无需额外配置缺点需要轮询非实时模式 2基于 Conductor 内部事件的事件处理器Conductor 在任务状态变化时可以发布内部事件可以配置事件处理器监听{ name: human_task_notification_handler, event: conductor:TASK_STATUS_CHANGE, condition: $.taskType HUMAN $.status IN_PROGRESS, actions: [ { action: start_workflow, start_workflow: { name: notification_workflow, input: { workflowId: ${workflowId}, taskRefName: ${taskRefName}, taskStatus: ${status} } } } ] }每当 Human Task 进入IN_PROGRESS状态时即触发一个通知工作流。该处理器可配合 事件处理器配置文档 中的完整字段说明使用。模式 3通过 EVENT Task 做 Webhook 集成在 Human Task 之前加一个EVENT任务用于发送 webhook 通知{ name: notify_human_task, taskReferenceName: notify_ref, type: EVENT, sink: kafka:human-task-notifications, inputParameters: { workflowId: ${workflow.input.workflowId}, taskRefName: human_ref, eventType: HUMAN_TASK_PENDING } }, { name: human_approval, taskReferenceName: human_ref, type: HUMAN }随后配置事件处理器或外部消费者处理这些通知。sink的前缀如kafka:指向对应的消息中间件需要相应的事件队列模块已启用。模式 4完成时携带回调输出的 Task Update完成 Human Task 时把回调信息放进输出中curl -X POST {{ server_host }}/api/tasks \ -H Content-Type: application/json \ -d { taskId: ${taskId}, status: COMPLETED, output: { approvedBy: userexample.com, approvedAt: 2026-04-22T10:30:00Z, comments: Approved for production deployment } }输出会供下游任务使用可用于审计追踪或进一步通知。模式 5外部系统集成用于实时通知的外部集成途径包括Slack/Teams用事件处理器触发一个通知工作流向其 webhook 发帖邮件通过 SMTP 或邮件服务 API 发送通知短信/推送集成 Twilio、Pushover 或类似服务自定义 Webhook当有人工任务待处理时 POST 到你的内部系统最佳实践使用关联 ID所有通知中都包含workflowId和taskRefName便于追踪设置超时考虑为未审批的人工任务添加超时/升级逻辑审计追踪记录所有人工任务完成的 timestamp 与用户信息幂等性确保通知处理器幂等以处理重复事件完整示例带通知的审批工作流下面是一个端到端的审批工作流定义Slack 发起审批请求 → 等待人工审批 → 发送审批结果。{ name: approval_workflow, version: 1, tasks: [ { name: send_approval_request, taskReferenceName: send_request_ref, type: HTTP, inputParameters: { http_request: { method: POST, url: https://hooks.slack.com/services/xxx, body: { text: Approval needed for workflow ${workflow.input.requestId} } } } }, { name: wait_for_approval, taskReferenceName: approval_ref, type: HUMAN }, { name: send_approval_result, taskReferenceName: send_result_ref, type: HTTP, inputParameters: { http_request: { method: POST, url: https://hooks.slack.com/services/xxx, body: { text: Approval ${approval_ref.output.status} by ${approval_ref.output.approvedBy} } } } } ] }该工作流的执行过程发送 Slack 通知告知需要审批工作流停在wait_for_approvalHUMAN任务处可通过前文任一方式Task Update API、Update Queue API 或complete_task事件处理器提交审批结果例如携带approvedBy、approvedAt的输出审批完成后第三个 HTTP 任务引用${approval_ref.output.status}与${approval_ref.output.approvedBy}发送包含审批结果与审批人的后续通知——这体现了 Human Task 输出可被下游任务变量引用的完整数据链路。小结Human Task 用零参数的极简配置实现了 Conductor 中最关键的可恢复执行能力之一工作流可以暂停但不丢失等待任意时长的人工决策或外部事件。结合 Human 与 HumanTaskMapper 的源码可以看出挂起逻辑完全无状态状态推进全部经由 TaskResource 的更新端点或 QueueAdminResource 的队列更新端点完成因此可以灵活对接审批系统、IM 通知、消息队列等多种外部信号源。对于需要 Human-in-the-Loop 的 Agent 工作流例如让 LLM 产出结果后必须经人工确认再执行副作用操作Human Task 是最直接的实现手段。【免费下载链接】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），仅供参考 Conductor 中的 Human Task：配置人工介入任务并实现 Human-in-the-Loop 审批流
Conductor 中的 Human Task配置人工介入任务并实现 Human-in-the-Loop 审批流【免费下载链接】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/conductorConductor 系统任务中的HUMAN类型任务用于在工作流中插入一个人工闸门工作流运行到该任务时会被暂停任务保持IN_PROGRESS状态直到外部触发Task Update API、事件处理器或队列更新 API将其标记为COMPLETED或FAILED。本文基于 Human Task 配置文档 展开完整讲解 Human Task 的 JSON 配置、两种完成方式、五种监控/回调模式以及一个可直接复用的完整审批流示例并结合开源仓库源码说明其底层执行机制帮助你在工作流中可靠地落地人工审批与外部信号等待场景。核心概念Human Task 是什么Human Tasktype: HUMAN的作用是暂停工作流并等待外部信号。它充当一个 gate任务创建后立即进入IN_PROGRESS状态此后一直挂起直到被外部触发器标记为COMPLETED或FAILED。典型使用场景包括工作流需要暂停并等待人工干预例如手动审批manual approval等待来自外部来源的事件例如 Kafka、SQS或 Conductor 内部的队列机制。从源码结构看Human Task 的极简实现印证了它只做等待的语义。任务实现类 Human 继承自WorkflowSystemTask其start方法仅将任务状态置为IN_PROGRESScancel方法仅置为CANCELED——没有任何轮询或回调逻辑真正的状态推进完全依赖外部更新Override public void start(WorkflowModel workflow, TaskModel task, WorkflowExecutor workflowExecutor) { task.setStatus(IN_PROGRESS); }JSON 配置Human Task 不需要任何专用参数即可配置。最小配置如下{ name: human, taskReferenceName: human_ref, inputParameters: {}, type: HUMAN }需要注意的是inputParameters可以为空但并非不能携带数据——任务映射器 HumanTaskMapper 在执行时会通过ParametersUtils.getTaskInputV2对inputParameters做变量替换并写入TaskModel同时记录startTime并将状态直接置为IN_PROGRESSTaskModel humanTask taskMapperContext.createTaskModel(); humanTask.setTaskType(TASK_TYPE_HUMAN); humanTask.setInputData(humanTaskInput); humanTask.setStartTime(System.currentTimeMillis()); humanTask.setStatus(TaskModel.Status.IN_PROGRESS);这意味着你在inputParameters里写入的${...}表达式会被正常求值可用于把上下文信息如申请单号、发起人带给等待中的人工任务供审批人参考。完成 Human Task 的两种官方方式方式一Task Update API通过 Task Update APIPOST api/tasks完成 Human Task需要提供taskId、任务状态和期望的任务输出。使用 CLI 的等价命令conductor task update-execution --workflow-id {workflowId} --task-ref-name waiting_around_ref --status COMPLETED --output {data_key:somedatatoWait1,data_key2:somedatatoWAit2}在服务端 REST 层该能力对应 TaskResource 中的多个端点POST /api/tasks提交TaskResult、POST /api/tasks/update-v2以及按 ref name 更新的POST /api/tasks/{workflowId}/{taskRefName}/{status}可选workerid查询参数请求体即任务输出。此外还有同步变体{workflowId}/{taskRefName}/{status}/sync会直接返回更新后的 workflow适合脚本化审批流程。方式二事件处理器 / Update Queue APIs若启用了 SQS 等事件队列集成Human Task 还可以通过 Update Queue API 完成对应实现位于 QueueAdminResourcePOST api/queue/update/{workflowId}/{taskRefName}/{status}POST api/queue/update/{workflowId}/task/{taskId}/{status}POST 消息体中携带的任何参数都会作为该任务的输出重复。例如发送如下 COMPLETED 消息curl -X POST {{ server_host }}{{ api_prefix }}/queue/update/{workflowId}/waiting_around_ref/COMPLETED \ -H Content-Type: application/json \ -d {data_key:somedatatoWait1,data_key2:somedatatoWAit2}则 Human Task 的输出为{ data_key:somedatatoWait1, data_key2:somedatatoWAit2 }从源码看该端点将请求委托给DefaultEventQueueProcessor.updateByTaskRefName / updateByTaskId位于 DefaultEventQueueProcessor即通过在事件队列中发布消息来驱动任务状态变更——这正是它能与 SQS/Kafka 等外部事件源解耦对接的原因。另外也可以配置一个使用complete_taskaction 的事件处理器来完成 Human Task这样外部系统只需发布事件无需直接调用 HTTP API。监控 Human Task回调与通知模式当工作流到达 Human Task 时通常需要收到通知或回调以触发下一步动作发邮件、通知 Slack 频道、触发外部系统等。文档推荐了五种模式模式 1轮询 Workflow Status API最简单的方式是轮询工作流执行状态查找处于IN_PROGRESS状态的 Human 任务# Get workflow execution status curl {{ server_host }}/api/workflow/{workflowId} \ -H accept: application/json解析响应找到taskType: HUMAN且status: IN_PROGRESS的任务。优点实现简单无需额外配置缺点需要轮询非实时模式 2基于 Conductor 内部事件的事件处理器Conductor 在任务状态变化时可以发布内部事件可以配置事件处理器监听{ name: human_task_notification_handler, event: conductor:TASK_STATUS_CHANGE, condition: $.taskType HUMAN $.status IN_PROGRESS, actions: [ { action: start_workflow, start_workflow: { name: notification_workflow, input: { workflowId: ${workflowId}, taskRefName: ${taskRefName}, taskStatus: ${status} } } } ] }每当 Human Task 进入IN_PROGRESS状态时即触发一个通知工作流。该处理器可配合 事件处理器配置文档 中的完整字段说明使用。模式 3通过 EVENT Task 做 Webhook 集成在 Human Task 之前加一个EVENT任务用于发送 webhook 通知{ name: notify_human_task, taskReferenceName: notify_ref, type: EVENT, sink: kafka:human-task-notifications, inputParameters: { workflowId: ${workflow.input.workflowId}, taskRefName: human_ref, eventType: HUMAN_TASK_PENDING } }, { name: human_approval, taskReferenceName: human_ref, type: HUMAN }随后配置事件处理器或外部消费者处理这些通知。sink的前缀如kafka:指向对应的消息中间件需要相应的事件队列模块已启用。模式 4完成时携带回调输出的 Task Update完成 Human Task 时把回调信息放进输出中curl -X POST {{ server_host }}/api/tasks \ -H Content-Type: application/json \ -d { taskId: ${taskId}, status: COMPLETED, output: { approvedBy: userexample.com, approvedAt: 2026-04-22T10:30:00Z, comments: Approved for production deployment } }输出会供下游任务使用可用于审计追踪或进一步通知。模式 5外部系统集成用于实时通知的外部集成途径包括Slack/Teams用事件处理器触发一个通知工作流向其 webhook 发帖邮件通过 SMTP 或邮件服务 API 发送通知短信/推送集成 Twilio、Pushover 或类似服务自定义 Webhook当有人工任务待处理时 POST 到你的内部系统最佳实践使用关联 ID所有通知中都包含workflowId和taskRefName便于追踪设置超时考虑为未审批的人工任务添加超时/升级逻辑审计追踪记录所有人工任务完成的 timestamp 与用户信息幂等性确保通知处理器幂等以处理重复事件完整示例带通知的审批工作流下面是一个端到端的审批工作流定义Slack 发起审批请求 → 等待人工审批 → 发送审批结果。{ name: approval_workflow, version: 1, tasks: [ { name: send_approval_request, taskReferenceName: send_request_ref, type: HTTP, inputParameters: { http_request: { method: POST, url: https://hooks.slack.com/services/xxx, body: { text: Approval needed for workflow ${workflow.input.requestId} } } } }, { name: wait_for_approval, taskReferenceName: approval_ref, type: HUMAN }, { name: send_approval_result, taskReferenceName: send_result_ref, type: HTTP, inputParameters: { http_request: { method: POST, url: https://hooks.slack.com/services/xxx, body: { text: Approval ${approval_ref.output.status} by ${approval_ref.output.approvedBy} } } } } ] }该工作流的执行过程发送 Slack 通知告知需要审批工作流停在wait_for_approvalHUMAN任务处可通过前文任一方式Task Update API、Update Queue API 或complete_task事件处理器提交审批结果例如携带approvedBy、approvedAt的输出审批完成后第三个 HTTP 任务引用${approval_ref.output.status}与${approval_ref.output.approvedBy}发送包含审批结果与审批人的后续通知——这体现了 Human Task 输出可被下游任务变量引用的完整数据链路。小结Human Task 用零参数的极简配置实现了 Conductor 中最关键的可恢复执行能力之一工作流可以暂停但不丢失等待任意时长的人工决策或外部事件。结合 Human 与 HumanTaskMapper 的源码可以看出挂起逻辑完全无状态状态推进全部经由 TaskResource 的更新端点或 QueueAdminResource 的队列更新端点完成因此可以灵活对接审批系统、IM 通知、消息队列等多种外部信号源。对于需要 Human-in-the-Loop 的 Agent 工作流例如让 LLM 产出结果后必须经人工确认再执行副作用操作Human Task 是最直接的实现手段。【免费下载链接】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），仅供参考