dify 流程暂停与人工干预：3 种实现方案 + 避坑指南

在企业级 AI 应用开发中，完全自动化的流程往往难以应对复杂业务场景 —— 当 AI 遇到无法独立处理的审批流程、异常情况或需要人工决策的关键环节时，流程暂停与人工干预就成了必不可少的功能。本文详解如何在 Dify中实现这一关键能力，帮助你的 AI 应用在自动化与人工控制之间找到完美平衡。

为什么需要流程暂停与人工干预？5 大典型场景

在实际业务中，以下场景对人工介入的需求尤为突出：

1. 1. 内容审核与合规检查

UGC 内容生成、营销文案发布等场景需人工确认是否符合法律法规。某电商平台使用 Dify 构建的商品描述生成流程中，当 AI 生成 "限时最低价" 等敏感促销文案时，系统会自动暂停并触发人工审核，审核通过后才推送至商品详情页，使违规率下降 72%。

2.复杂决策审批流程

金融行业的贷款申请、企业内部的采购审批等多级决策场景。某银行的信贷审批工作流中，当 AI 识别到申请人征信存在异常记录时，会暂停自动审批流程，将案件推送给风控专员，处理完成后通过 API 调用恢复流程，平均处理时效提升 40%。

1. 3. 异常情况处理

系统运行中出现预料之外的错误或边界情况。物流调度系统在遇到 "地址模糊"" 库存不足 " 等异常时，自动暂停任务分配流程，通知运营人员介入处理，通过 Dify 的状态标记机制记录人工操作结果，使异常订单处理效率提升 55%。

1. 4. 个性化服务定制

需要根据用户特殊需求调整服务内容。在线教育平台的学习路径规划流程中，当 AI 检测到学生连续 3 次测试未达标时，会暂停自动推荐流程，提示辅导老师手动调整学习计划，学员满意度提升 28%。

1. 5. 数据校验与修正

AI 提取或生成的数据需要人工验证准确性。医疗系统的病历结构化流程中，AI 识别的关键诊断信息会触发人工校验节点，医生确认无误后流程才继续，数据准确率从 89% 提升至 99.2%。

技术实现：3 种方案深度解析

方案一：条件分支 + 数据库状态标记

核心原理：通过条件分支节点判断是否需要人工干预，将流程状态写入数据库，定期轮询状态变更以恢复执行。

实现步骤：

1. 1. 添加条件分支节点：在工作流中插入条件分支节点，设置判断规则（如 "content_risk_score > 0.8"）
2. 2. 配置暂停分支：在需要人工干预的分支中，添加 "数据库操作" 节点，将流程 ID 和状态（pending）写入外部数据库
3. 3. 设计轮询机制：添加 "循环节点"，设置每 30 秒查询一次数据库状态
4. 4. 恢复执行逻辑：当检测到状态变为 "approved" 时，退出循环继续执行后续节点

关键代码片段（Python 代码执行节点）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

import psycopg2
from datetime import datetime
import time

def main(workflow_id: str, risk_score: float) -> dict:
    # 连接数据库
    conn = psycopg2.connect("dbname=dify_manual host=postgres user=postgres")
    cur = conn.cursor()

    if risk_score > 0.8:
        # 写入暂停状态
        cur.execute("""
            INSERT INTO manual_tasks (workflow_id, status, created_at)
            VALUES (%s, 'pending', %s)
        """, (workflow_id, datetime.now()))
        conn.commit()

        # 轮询等待状态变更
        while True:
            cur.execute("""
                SELECT status FROM manual_tasks WHERE workflow_id = %s
            """, (workflow_id,))
            status = cur.fetchone()[0]
            if status == 'approved':
                return {"action": "resume", "message": "人工审核通过"}
            elif status == 'rejected':
                return {"action": "terminate", "message": "人工审核拒绝"}
            time.sleep(30)
    else:
        return {"action": "continue", "message": "风险评分较低，无需人工审核"}

适用场景：中小规模流程、对实时性要求不高（允许分钟级延迟）、已有数据库环境的场景。某政务审批系统采用此方案，实现了 "AI 预审→人工复核→自动归档" 的完整流程。

方案二：API 触发控制

核心原理：通过 Dify 的 HTTP 节点暴露暂停 / 恢复接口，外部系统调用 API 实现人工干预。

实现步骤：

1. 1. 创建暂停标记节点：添加 "变量赋值" 节点，将流程状态设为 "suspended"
2. 2. 配置 HTTP 请求节点：设置 POST 接口（如 /api/workflow/pause），接收人工操作指令
3. 3. 设计状态监听逻辑：使用 "代码执行" 节点创建 WebSocket 连接，监听状态变更事件
4. 4. 恢复执行分支：当接收到 "resume" 指令时，清除暂停标记继续执行

关键 API 示例：

1
2
3
4
5
6
7
8
9
10
11

# 暂停流程API
@app.post("/api/workflow/pause/{workflow_id}")
def pause_workflow(workflow_id: str):
    redis_client.set(f"workflow:{workflow_id}:status", "paused")
    return {"status": "success", "message": "流程已暂停"}
 
# 恢复流程API
@app.post("/api/workflow/resume/{workflow_id}")
def resume_workflow(workflow_id: str):
    redis_client.set(f"workflow:{workflow_id}:status", "running")
    return {"status": "success", "message": "流程已恢复"}

适用场景：需要与外部系统集成、实时性要求高（秒级响应）的场景。某智能制造平台通过此方案，实现了生产异常时的 "自动暂停→工程师处理→API 恢复" 机制。

方案三：Agent 节点 ReAct 策略

核心原理：利用 Agent 节点的自主决策能力，通过 ReAct 策略判断是否需要调用 "人工干预工具"。

实现步骤：

1. 1. 配置 Agent 节点：选择 ReAct 推理策略，添加 "人工干预" 工具
2. 2. 定义工具描述：

1
2
3
4
5
6
7
8
9
10
11
12
13
14

{
 
{
  "name": "manual_intervention",
  "description": "当遇到需要人工决策的情况时调用，参数包括task_id和reason",
  "parameters": {
    "type": "object",
    "properties": {
      "task_id": {"type": "string"},
      "reason": {"type": "string"}
    },
    "required": ["task_id", "reason"]
  }
}

1. 3. 设计提示词模板：

1
2
3
4
5
6

你是流程控制专家，当检测到以下情况时必须调用manual_intervention工具：
1. 风险评分>0.8
2. 用户明确要求人工处理
3. 连续3次自动处理失败
 
工具调用格式：

1. 4. 配置工具回调：实现工具的 HTTP 服务，将任务推送到人工处理平台，完成后通过 Agent 节点恢复流程

适用场景：复杂决策场景、需要动态判断是否干预的流程。某智能客服平台通过此方案，实现了 "AI 自主处理→疑难问题自动升级→人工解决→经验沉淀" 的闭环。

三种方案对比分析