从“让 AI 完成一个任务”，到“让 AI 稳定地完成一项业务流程”。本章围绕企业级 AI 工作流工程，深入讲解流程编排、状态机、断点恢复、异常重试、人工审批、跨系统事务补偿与全链路审计，帮助 FDE 将 AI Agent 从实验性能力转化为可交付、可运维的企业业务系统。

一、为什么企业级 AI 需要工作流工程？

在前面的章节中，我们已经逐步建立了企业 AI 系统的核心能力：

• 第48章：Enterprise AI Observability，解决 AI 系统的可观测性问题。

• 第49章：Enterprise AI Evaluation，解决 AI 输出质量如何评估的问题。

• 第50章：Enterprise AI Governance，解决企业 AI 治理与风险控制问题。

• 第51章：Enterprise AI Platform，构建统一 AI 平台与多 Agent 协同能力。

• 第52章：Enterprise AI Integration，打通 AI 与 ERP、WMS、MES、CRM 等企业系统的集成链路。

然而，当 AI 真正进入企业核心业务流程时，一个新的问题随之出现：

如果一个 AI 任务需要执行十几个步骤，中间涉及多个业务系统、人工审批和长时间等待，如何保证整个任务可靠地执行完成？

例如，企业需要构建一个 AI 采购协同助手。当系统发现某种物料库存不足时，需要依次完成：

1. 分析库存与未来生产需求。

2. 查询 ERP 中的采购订单和供应商信息。

3. 判断是否需要补货。

4. 计算建议采购数量。

5. 检查预算、采购规则和供应商限制。

6. 生成采购申请草稿。

7. 提交人工审批。

8. 审批通过后，创建正式采购申请。

9. 等待 ERP 返回业务处理结果。

10. 记录执行日志，并通知相关人员。

这不是一次简单的模型调用，而是一条跨系统、跨时间、包含风险控制的业务流程。

如果执行到第7步时，审批人需要第二天才能处理；如果第8步调用 ERP 时发生网络超时；如果系统在第9步前重启，整个流程应该如何继续？

这些问题无法仅靠 Prompt 或 Agent 的推理能力解决。

企业需要一套能够管理执行顺序、任务状态、失败恢复、人工介入、业务补偿和执行审计的工程体系，这就是 Enterprise AI Workflow Engineering。

1.1 从单次 AI 调用到业务流程执行

对比维度	单次 AI 调用	企业级 AI 工作流
执行范围	一个模型任务	多个业务步骤
执行顺序	主要由调用方控制	显式定义流程与依赖
状态管理	请求与响应	持久化的流程状态
异常处理	返回错误或重新调用	分类处理、重试、恢复与补偿
人工参与	通常需要额外开发	可配置审批与人工任务节点
跨系统协同	单次 API 调用	多系统、多步骤业务协同
长时间运行	通常不适合	支持等待、暂停、恢复
审计能力	记录请求与响应	记录每个节点、状态变化和业务决策
适用场景	摘要、分类、问答	采购、订单履约、生产协同、异常处置

这里需要明确一个概念：

Agent 负责理解目标、推理和选择行动；Workflow 负责管理行动的执行过程、状态与业务约束。

两者不是替代关系，而是协作关系。

二、什么是 Enterprise AI Workflow Engineering？

Enterprise AI Workflow Engineering，即企业级 AI 工作流工程，是指围绕企业业务目标，将 AI 推理、确定性业务逻辑、系统接口、人工审批和异常处理组织成可执行、可恢复、可追踪的流程，并通过工程化机制保证流程的可靠性与可控性。

一个完整的企业 AI 工作流通常包含以下组成部分：

1. 流程定义（Workflow Definition）

定义业务步骤、节点依赖、条件分支、并行任务与结束条件。

2. 状态管理（State Management）

保存当前步骤、业务上下文、执行结果、重试次数及待处理任务。

3. AI 决策节点（AI Decision Node）

负责意图识别、信息提取、风险分析、方案生成和受约束的任务规划。

4. 规则与权限节点（Policy Node）

执行权限校验、业务规则校验、风险检查和人工审批判断。

5. 可靠性机制（Reliability Mechanism）

支持超时、重试、幂等、断点恢复、超时补偿和人工介入。

6. 可观测与审计（Observability & Audit）

记录流程实例、节点执行、状态变更、业务决策及外部系统交互。

2.1 企业级工作流的核心目标

企业工作流不能只关注“能否执行”，还必须回答以下问题：

• 正确性： 是否遵循业务规则和执行顺序？

• 可靠性： 发生异常后能否恢复？

• 可控性： 高风险操作是否受到权限和审批约束？

• 可追踪性： 是否能解释任务为什么成功、失败或暂停？

• 可恢复性： 服务重启后是否可以从已保存的状态继续？

• 可审计性： 是否可以还原 AI 的建议、实际执行的操作及其结果？

• 可运营性： 运维人员是否能够查询、重试、取消或人工接管任务？

这些目标决定了企业级 AI 工作流不能仅依赖模型上下文或内存变量，而需要持久化的状态管理与明确的执行协议。

三、企业级 AI 工作流总体架构

一个适用于 ERP、WMS、MES、CRM 等系统的 AI 工作流架构，可以划分为六个层次。

业务入口层

Web 工作台 · 企业门户 · API · 消息事件

AI 任务规划层

Agent · 意图识别 · 计划生成 · 结果分析

工作流编排层

流程定义 · 状态机 · 分支 · 并行 · 等待 · 补偿

可靠性与治理层

持久化 · 幂等 · 重试 · 权限 · 风险控制 · 人工审批

企业能力集成层

Tool Registry · API Gateway · MCP Adapter · 业务适配器

企业业务系统层

ERP · WMS · MES · CRM · 采购 · 财务 · 人事

持久化状态存储

运行监控

审计日志

在这套架构中，AI 不是整个系统的唯一控制中心。它是工作流中的一个能力节点，而工作流引擎负责执行边界、流程状态与业务约束。

例如，AI 可以建议采购数量为 500 件，但正式采购申请必须经过库存核验、采购规则校验、权限校验和必要的人工审批，才能进入企业业务系统。

这意味着，模型可以提出建议，但不能绕过工作流直接决定所有业务副作用。

四、工作流、Agent、Workflow Engine：三者有什么区别？

在企业 AI 项目中，这三个概念经常被混用，但它们解决的是不同层面的问题。

能力	Agent	Workflow	Workflow Engine
核心职责	理解目标、推理、选择行动	描述业务步骤与依赖关系	实际执行和管理流程
执行方式	可以根据上下文动态决策	按定义的流程执行	调度节点、持久化状态、管理执行
是否适合动态规划	适合	可以包含动态分支	取决于引擎与流程定义
是否管理持久状态	需要额外机制或框架支持	规定需要保存哪些状态	通常提供或集成持久化机制
是否支持人工审批	可以发起审批请求	可以定义审批节点	管理等待、恢复和超时
异常恢复	依赖 Agent Runtime 和外围机制	定义异常处理路径	执行重试、恢复和补偿策略
典型应用	需求分析、风险判断、任务规划	采购、订单、生产流程	长时间运行、跨系统业务流程

4.1 推荐的职责划分

以“AI 自动处理订单交付风险”为例：

• Agent：分析订单风险，判断可能的原因，提出解决建议。

• Workflow：规定先检查库存，再核对生产计划，之后决定是否需要采购或人工介入。

• Workflow Engine：保存流程状态，执行节点，管理重试和等待，并在服务重启后恢复。

• 业务系统：提供真实库存、订单状态、生产进度等数据，并执行正式业务操作。

• 治理组件：检查权限、业务规则、风险等级和审批要求。

一个典型的设计原则是：

让 AI 负责不确定性，让工作流负责确定性，让业务系统负责业务事实。

当业务规则明确时，应优先使用确定性代码；当需要理解自然语言、归纳复杂信息或分析非结构化数据时，再调用 AI。

这样既能发挥大模型的优势，也能降低不必要的模型调用成本和业务风险。

五、工作流状态机：可靠执行的基础

企业级 AI 工作流的核心之一，是将一个业务任务建模为具有明确状态和状态转换规则的状态机。

如果仅使用一段 Python 代码顺序调用多个接口，程序一旦中断，就可能无法知道哪些步骤已经完成、哪些步骤需要重新执行。

状态机可以将流程的运行状态显式化，让每一次状态变化都有依据。

5.1 定义工作流状态

以采购申请流程为例，可以定义以下状态：

状态	含义	是否允许执行正式业务写入
CREATED	流程实例已创建	否
ANALYZING	正在分析业务需求	否
VALIDATING	正在校验库存、预算与规则	否
WAITING_APPROVAL	等待人工审批	否
APPROVED	已通过审批	仅允许进入授权执行阶段
EXECUTING	正在执行采购申请创建	允许受控执行
SUCCEEDED	业务操作成功	不再重复执行
REJECTED	审批拒绝	不允许执行
FAILED	流程执行失败	需要按策略恢复或终止
COMPENSATING	正在执行补偿操作	仅允许执行定义好的补偿
CANCELLED	流程已取消	不允许继续正常执行

这里有一个重要区别：流程状态不等于业务系统状态。

例如，工作流进入 EXECUTING，仅代表系统正在尝试执行采购申请创建，并不意味着 ERP 中的采购申请已经创建成功。

只有在获得可靠的业务结果、查询到对应业务单据或通过其他可信方式确认后，工作流才应进入 SUCCEEDED。

5.2 状态转换图

状态转换应受到约束。例如：

• WAITING_APPROVAL 不能直接跳转到 SUCCEEDED。

• REJECTED 不能继续执行正式采购操作。

• SUCCEEDED 不应因为消息重复到达而再次创建采购申请。

• CANCELLED 不能被普通重试任务重新启动。

• EXECUTING 发生网络超时时，不应直接假定业务操作失败。

在生产环境中，状态转换不仅要在代码中判断，还应通过数据库事务、版本号或条件更新机制避免并发任务同时修改同一个流程实例。

5.3 状态转换的代码实现

下面使用 Python 展示一个简化的状态转换校验示例：

from enum import Enum


class WorkflowStatus(str, Enum):
    CREATED = "CREATED"
    ANALYZING = "ANALYZING"
    VALIDATING = "VALIDATING"
    WAITING_APPROVAL = "WAITING_APPROVAL"
    APPROVED = "APPROVED"
    EXECUTING = "EXECUTING"
    SUCCEEDED = "SUCCEEDED"
    REJECTED = "REJECTED"
    FAILED = "FAILED"
    CANCELLED = "CANCELLED"


ALLOWED_TRANSITIONS = {
    WorkflowStatus.CREATED: {
        WorkflowStatus.ANALYZING,
        WorkflowStatus.CANCELLED,
    },
    WorkflowStatus.ANALYZING: {
        WorkflowStatus.VALIDATING,
        WorkflowStatus.FAILED,
        WorkflowStatus.CANCELLED,
    },
    WorkflowStatus.VALIDATING: {
        WorkflowStatus.WAITING_APPROVAL,
        WorkflowStatus.FAILED,
        WorkflowStatus.CANCELLED,
    },
    WorkflowStatus.WAITING_APPROVAL: {
        WorkflowStatus.APPROVED,
        WorkflowStatus.REJECTED,
        WorkflowStatus.CANCELLED,
    },
    WorkflowStatus.APPROVED: {
        WorkflowStatus.EXECUTING,
        WorkflowStatus.CANCELLED,
    },
    WorkflowStatus.EXECUTING: {
        WorkflowStatus.SUCCEEDED,
        WorkflowStatus.FAILED,
    },
    WorkflowStatus.FAILED: {
        WorkflowStatus.CANCELLED,
    },
    WorkflowStatus.SUCCEEDED: set(),
    WorkflowStatus.REJECTED: set(),
    WorkflowStatus.CANCELLED: set(),
}


def transition(current: WorkflowStatus,
               target: WorkflowStatus) -> WorkflowStatus:
    allowed = ALLOWED_TRANSITIONS.get(current, set())

    if target not in allowed:
        raise ValueError(
            f"非法状态转换: {current.value} -> {target.value}"
        )

    return target


status = WorkflowStatus.CREATED
status = transition(status, WorkflowStatus.ANALYZING)
status = transition(status, WorkflowStatus.VALIDATING)

print(status.value)

这段代码只负责校验转换规则，还不是完整的工作流引擎。

生产环境还需要实现：

1. 数据库持久化。

2. 并发状态更新保护。

3. 状态变更事件。

4. 节点执行记录。

5. 超时与重试调度。

6. 流程恢复与人工接管。

7. 终态保护与审计。

需要注意，FAILED 并不意味着所有失败都能重新执行。对于可能产生外部副作用的操作，必须先确认上一次执行的真实结果，再决定是否重试。

六、工作流上下文与持久化状态

工作流执行过程中，需要在多个节点之间传递数据。这些数据通常被称为 Workflow Context，即工作流上下文。

例如：

{
  "workflow_id": "wf-procurement-20261029-001",
  "business_key": "purchase-request-10086",
  "status": "WAITING_APPROVAL",
  "current_step": "human_approval",
  "input": {
    "material_code": "MAT-001",
    "required_quantity": 500
  },
  "analysis": {
    "available_stock": 120,
    "reserved_stock": 40,
    "recommended_quantity": 420
  },
  "approval": {
    "status": "PENDING",
    "approver_role": "PROCUREMENT_MANAGER"
  }
}

该示例中的建议采购数量仅为演示值，实际计算需要根据安全库存、已分配库存、在途采购、生产需求和采购规则确定。

6.1 不要把所有信息都放进工作流上下文

企业工作流经常需要处理大量数据，包括：

• 大模型输入与输出。

• ERP、WMS、MES 返回的业务数据。

• 文件、图片和报表。

• 检索到的知识库片段。

• 审批意见。

• 外部 API 请求与响应。

如果将所有数据直接保存在一个 JSON 字段中，会产生以下问题：

• 状态数据越来越大。

• 每次更新都可能带来较大的数据库写入。

• 敏感信息难以实施细粒度权限控制。

• 大模型上下文和业务状态容易混淆。

• 数据版本与保留期限难以管理。

更合理的设计是将数据分层：

数据类型	建议保存位置
当前流程状态	工作流实例表
节点输入输出摘要	节点执行记录表
关键业务参数	结构化业务数据表
大模型完整请求与响应	受控的模型调用记录或对象存储
文件与大型附件	对象存储
审批记录	审批业务表
外部系统交互记录	集成日志或事件存储
敏感数据	按业务要求加密并进行权限隔离

上下文中可以保存引用、摘要和必要的恢复数据，而不必复制所有业务系统中的原始数据。

6.2 工作流实例表设计

一个简化的 MySQL 表结构如下：

CREATE TABLE workflow_instance (
    id              BIGINT PRIMARY KEY AUTO_INCREMENT,
    workflow_id     VARCHAR(64) NOT NULL UNIQUE,
    workflow_type   VARCHAR(100) NOT NULL,
    business_key    VARCHAR(128) NOT NULL,
    status          VARCHAR(32) NOT NULL,
    current_step    VARCHAR(100),
    context_json    JSON,
    version         BIGINT NOT NULL DEFAULT 0,
    retry_count     INT NOT NULL DEFAULT 0,
    created_at      DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
    updated_at      DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
                    ON UPDATE CURRENT_TIMESTAMP,
    finished_at     DATETIME NULL,
    INDEX idx_business_key (business_key),
    INDEX idx_status_updated (status, updated_at)
);

其中：

• workflow_id 用于唯一标识流程实例。

• business_key 关联采购申请、订单或其他业务对象。

• status 表示流程当前状态。

• current_step 记录当前处理节点。

• context_json 保存必要的流程上下文。

• version 用于乐观并发控制。

• retry_count 记录流程级重试次数。

生产环境还应考虑租户标识、业务版本、取消原因、负责人、租约、超时策略和数据保留要求。对于多租户系统，唯一约束和索引通常也需要纳入租户维度。

6.3 使用乐观锁保护状态更新

假设两个 Worker 同时读取到版本号为 5 的流程实例，如果不加保护，它们可能都认为自己可以推进流程。

可以使用版本号进行条件更新：

UPDATE workflow_instance
SET status = 'VALIDATING',
    current_step = 'validate_rules',
    version = version + 1
WHERE workflow_id = 'wf-procurement-20261029-001'
  AND version = 5
  AND status = 'ANALYZING';

应用程序需要检查受影响的行数：

• 更新 1 行：本次状态转换成功。

• 更新 0 行：状态或版本已经变化，需要重新读取并判断下一步操作。

乐观锁可以保护流程状态，但不能单独保证外部业务操作只执行一次。 外部操作还需要幂等键、业务唯一约束和执行结果核对。

七、长时间运行工作流：暂停、等待与恢复

传统接口调用通常在几秒内返回结果，但企业业务流程可能持续几小时甚至几天。

例如：

• 等待采购经理审批。

• 等待供应商确认交期。

• 等待 MES 返回生产完成事件。

• 等待 WMS 完成拣货和出库。

• 等待财务系统完成付款确认。

如果采用普通的同步程序一直等待，不仅浪费计算资源，也容易受到网络连接和进程生命周期的影响。

7.1 正确的长流程执行模式

关键点在于：工作流进入等待状态时，应将必要的状态持久化，然后释放当前 Worker 的执行资源。

审批完成后，系统通过事件、队列或调度机制恢复流程，而不是要求原来的线程一直存活。

7.2 人工审批不是普通的函数调用

一个常见的错误是直接写出：

approval_result = wait_for_manager_approval()

这段伪代码容易掩盖实际的工程问题：

• 审批可能持续数小时。

• Worker 可能重启。

• 审批事件可能重复发送。

• 审批结果可能过期。

• 审批人可能失去相关权限。

• 业务对象可能在审批期间发生变化。

更合理的做法是创建一个独立的审批任务，并将审批任务 ID 与工作流实例关联。

审批服务接收到结果后，应校验：

1. 审批任务是否属于该流程实例。

2. 当前用户是否具有审批权限。

3. 审批任务是否仍然处于待审批状态。

4. 审批结果是否符合当前业务状态。

5. 是否已经处理过相同的审批事件。

6. 审批期间关键业务数据是否发生变化。

通过校验后，审批服务可以发送一个带唯一事件 ID 的结果事件，再由工作流引擎推进状态。

7.3 审批期间的数据一致性

假设 AI 在上午建议采购 500 件物料，采购经理下午才完成审批。

在此期间，WMS 中的库存可能已经增加，其他采购单也可能覆盖了同一需求。

因此，审批通过并不一定意味着系统应该原样执行上午生成的采购数量。

对于重要业务操作，应在执行前重新核验关键数据：

• 当前可用库存。

• 已分配库存。

• 在途采购数量。

• 当前需求数量。

• 预算和采购限制。

• 审批授权范围。

如果关键数据发生变化，可以重新计算、请求二次确认或将任务转入人工处理。

八、断点恢复与节点执行记录

断点恢复是企业级工作流的重要能力。

所谓断点恢复，是指流程在某个节点发生异常或系统重启后，能够根据持久化状态继续执行，而不必从头开始重复整个业务流程。

但必须明确：

保存了工作流状态，不代表外部业务副作用就可以安全地自动重放。

例如，系统在创建 ERP 采购申请后发生崩溃，数据库中却尚未记录成功。如果恢复机制直接重新调用创建接口，就可能产生重复采购申请。

因此，恢复机制必须结合节点执行记录、幂等键和外部业务状态核验。

8.1 节点执行记录表

CREATE TABLE workflow_step_execution (
    id                  BIGINT PRIMARY KEY AUTO_INCREMENT,
    workflow_id         VARCHAR(64) NOT NULL,
    step_key            VARCHAR(100) NOT NULL,
    execution_no        INT NOT NULL,
    status              VARCHAR(32) NOT NULL,
    idempotency_key     VARCHAR(128),
    input_summary       JSON,
    output_summary      JSON,
    error_code          VARCHAR(100),
    error_message       VARCHAR(1000),
    started_at          DATETIME NULL,
    finished_at         DATETIME NULL,
    created_at          DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
    UNIQUE KEY uk_step_execution (
        workflow_id, step_key, execution_no
    ),
    UNIQUE KEY uk_idempotency_key (idempotency_key),
    INDEX idx_workflow_step (workflow_id, step_key, status)
);

这里的 idempotency_key 允许为空时，MySQL 可以在唯一索引中存储多个 NULL。但如果所有业务执行节点都必须具备幂等键，应由应用层强制校验，不能仅依赖数据库索引。

8.2 节点执行状态

一个节点可以具有如下执行状态：

• PENDING：等待执行。

• RUNNING：正在执行。

• SUCCEEDED：已经确认成功。

• RETRY_WAIT：等待重试。

• FAILED：执行失败。

• UNKNOWN：无法确认外部操作是否成功。

• SKIPPED：根据流程条件跳过。

其中 UNKNOWN 非常重要。

当调用 ERP 创建采购申请时，如果请求发出后连接断开，客户端可能不知道 ERP 是否已经完成处理。

这时不能简单地把节点设置为 FAILED 并立即重新执行。应先通过幂等键查询、业务单据查询或集成事件核对结果。

只有确认上一次操作未产生副作用，或者确认重试具备幂等保障后，才应继续执行。

8.3 一个可靠的节点执行伪代码

def execute_business_step(
    workflow_id: str,
    step_key: str,
    idempotency_key: str,
    business_adapter,
):
    # 1. 检查本地执行记录
    record = load_step_execution(
        workflow_id=workflow_id,
        step_key=step_key,
    )

    if record and record.status == "SUCCEEDED":
        return record.output_summary

    # 2. 对于可能产生外部副作用的操作，
    #    先检查外部系统是否已经处理该幂等键
    existing = business_adapter.find_by_idempotency_key(
        idempotency_key
    )

    if existing:
        save_step_success(
            workflow_id,
            step_key,
            existing,
        )
        return existing

    # 3. 记录执行意图
    mark_step_running(
        workflow_id,
        step_key,
        idempotency_key,
    )

    try:
        # 4. 使用同一个幂等键调用业务系统
        result = business_adapter.execute(
            idempotency_key=idempotency_key
        )

        # 5. 核验结果后记录成功
        verify_business_result(result)

        save_step_success(
            workflow_id,
            step_key,
            result,
        )

        return result

    except TimeoutError:
        # 超时并不等于业务执行失败
        mark_step_unknown(
            workflow_id,
            step_key,
            "OUTCOME_UNKNOWN",
        )
        raise

这段代码是工程模式示例，load_step_execution、mark_step_running 等函数需要自行实现事务、并发控制和持久化。

如果 ERP 不支持幂等键或根据请求标识查询结果，就需要通过业务唯一编号、外部单据查询接口、事件回执等机制实现核对；在无法可靠核验时，应将任务转交人工处理，而不是无限重试。

九、重试、超时与熔断：构建可靠的执行机制

企业工作流通常会调用多个外部系统，异常是不可避免的。

但重试不能简单理解为“失败后再调用一次”。

不同类型的异常，需要不同的处理策略。

9.1 异常分类

异常类型	示例	推荐处理
参数错误	缺少物料编码	修正参数或终止
权限错误	无权创建采购申请	拒绝执行并记录审计
业务规则错误	预算不足	进入业务异常分支
限流错误	API 返回 429	根据服务端提示延迟重试
临时网络错误	连接建立失败	有界重试
服务端暂时不可用	HTTP 503	退避重试，必要时熔断
请求超时且结果未知	创建请求已发送但无响应	查询外部状态或使用幂等机制
模型输出不符合格式	JSON 校验失败	有界修复或重新生成
人工审批超时	审批任务长期未处理	提醒、升级或按规则终止

9.2 指数退避

一种常见的重试策略是指数退避：

其中：

• ：第 n 次重试前的等待时间。

• ：初始等待时间。

• ：最大等待时间。

• ：随机抖动，用于避免大量任务同时重试。

例如，初始等待 1 秒，最大等待 30 秒，重试间隔可以逐渐增加。

在生产环境中，还应设置：

• 最大重试次数。

• 整个流程的截止时间。

• 节点级超时。

• 每个依赖系统的并发限制。

• 全局限流。

• 重试任务的监控告警。

对于明确的业务拒绝、权限错误和不可恢复的参数错误，不应进行无意义的自动重试。

9.3 重试必须考虑幂等性

假设一次采购申请创建请求因为网络超时没有返回结果。

如果立即重试，可能发生以下两种情况：

• ERP 没有创建成功，重试后正常创建。

• ERP 已经创建成功，但响应丢失，重试又创建一张申请单。

因此，写操作的可靠执行通常需要以下组合：

1. 客户端生成稳定的业务幂等键。

2. 服务端对幂等键进行去重。

3. 业务系统设置合理的唯一性约束。

4. 重试时复用同一个幂等键。

5. 对结果未知的操作进行状态核验。

6. 记录请求、响应及业务单据关联关系。

幂等键应绑定一次明确的业务意图，而不是每次重试都重新生成。

例如：

workflow_id: wf-10086
step_key: create_purchase_request
idempotency_key: wf-10086:create-purchase-request:v1

如果用户修改了采购数量并重新发起新的业务意图，应建立新的版本或新的业务操作标识，而不能将不同的操作错误地合并为同一个幂等请求。

十、跨系统事务：Saga 与补偿机制

企业 AI 工作流往往需要操作多个系统，而这些系统通常各自拥有独立的数据库和事务边界。

例如，一条订单履约流程可能涉及：

1. ERP 确认销售订单。

2. WMS 预留库存。

3. MES 安排生产。

4. WMS 执行出库。

5. ERP 更新交付状态。

如果第4步失败，前面的操作已经完成，系统应该如何处理？

不能假设多个系统之间天然存在一个可以覆盖所有操作的数据库事务。

一种常见的设计模式是 Saga。

10.1 Saga 的基本思想

Saga 将一个长流程拆分为多个本地事务，每个事务有自己的业务操作；当后续步骤无法继续时，根据业务定义执行补偿操作。

需要注意，补偿并不一定是简单的反向操作。

例如，库存预留可能可以释放，但已经完成的物料生产、实际发货或已发生的财务结算，通常不能通过删除记录来“撤销”。

在企业系统中，补偿往往意味着执行新的业务操作，例如：

• 释放库存预留。

• 取消尚未开始的生产任务。

• 创建退货单。

• 发起退款。

• 创建冲销凭证。

• 生成异常处理工单。

补偿操作本身也可能失败，因此必须被记录、重试和审计。

10.2 编排式 Saga 与事件式 Saga

常见的两种实现方式是：

编排式 Saga（Orchestration）

由一个中心协调器管理每一步操作、状态和补偿。

优点是流程关系清晰，便于审计、运维和处理人工审批。

适合流程复杂、业务规则集中、需要统一治理的企业场景。

事件式 Saga（Choreography）

各业务服务根据事件进行响应，并发布新的业务事件推动流程继续。

优点是服务之间耦合较低，但当业务步骤越来越多时，流程追踪、循环依赖和异常补偿会更难管理。

对于包含 AI 决策、人工审批和多系统写操作的复杂业务，通常可以优先考虑编排式 Saga；对于相对独立、事件驱动的业务服务，则可以采用事件式协作。

十一、Transactional Outbox：解决数据库与消息发送的一致性

假设工作流引擎完成了一次数据库状态更新：

工作流状态：VALIDATING → WAITING_APPROVAL

随后需要发送一条消息，通知审批服务创建审批任务。

如果数据库更新成功，但消息发送失败，就可能出现流程已经等待审批，却没有审批任务的情况。

反过来，如果先发消息，再更新数据库，消息消费者也可能收到一个尚未持久化的状态变化。

Transactional Outbox 是解决这一类问题的常用模式。

11.1 基本实现

在同一个本地数据库事务中完成两项操作：

1. 更新工作流状态。

2. 写入一条待发布的 Outbox 事件。

示例：

START TRANSACTION;

UPDATE workflow_instance
SET status = 'WAITING_APPROVAL',
    current_step = 'human_approval',
    version = version + 1
WHERE workflow_id = 'wf-10086'
  AND status = 'VALIDATING';

INSERT INTO outbox_event (
    event_id,
    aggregate_id,
    event_type,
    payload,
    status,
    created_at
)
VALUES (
    'evt-wf-10086-approval',
    'wf-10086',
    'ApprovalRequested',
    JSON_OBJECT(
        'workflow_id', 'wf-10086',
        'approval_type', 'PROCUREMENT'
    ),
    'PENDING',
    CURRENT_TIMESTAMP
);

COMMIT;

这只是示意 SQL。实际实现需要检查状态更新的受影响行数，并确保在状态更新未成功时不插入错误的事件。

提交事务后，后台发布器读取待发布事件并发送到消息总线。

即使发布器崩溃，也可以在恢复后继续发送。

但需要理解：Outbox 通常保证的是本地业务状态与待发布事件的原子记录，而不是消息只会发送一次。发布器可能在消息成功发送、尚未来得及标记完成时崩溃，从而再次发送同一事件。

因此，消息消费者仍然需要通过事件 ID、业务唯一键或消费记录实现幂等处理。

十二、AI 决策节点：如何将不确定性限制在可控范围内？

企业工作流中，AI 可以承担很多复杂任务：

• 从自然语言中提取业务参数。

• 分析订单延迟的可能原因。

• 判断异常属于哪一类。

• 根据规则和业务数据提出解决建议。

• 将复杂信息整理成结构化结果。

• 为人工审批提供解释和证据。

但 AI 节点的输出不应直接成为未经校验的业务指令。

12.1 推荐的 AI 决策流程

12.2 使用结构化输出

例如，AI 分析订单风险后，可以返回如下 JSON：

{
  "risk_level": "HIGH",
  "risk_reasons": [
    "可用库存不足",
    "生产计划晚于要求交期"
  ],
  "recommended_actions": [
    {
      "action": "CHECK_ALTERNATIVE_STOCK",
      "requires_approval": false
    },
    {
      "action": "CREATE_PROCUREMENT_DRAFT",
      "requires_approval": true
    }
  ],
  "confidence": 0.91
}

这份结果仍然只是 AI 的分析结果。

系统必须进一步校验：

• risk_level 是否属于允许的枚举值。

• action 是否在已注册的工具范围内。

• AI 是否有权获取相关订单与库存数据。

• 推荐动作是否符合业务规则。

• 采购数量和供应商是否来自可信业务数据。

• 是否需要人工审批。

• 当前用户是否有权执行相应操作。

此外，confidence 是模型提供的信号，不是经过校准的真实成功概率，也不能单独作为高风险业务自动执行的授权依据。

12.3 AI 输出与业务事实分离

假设 AI 返回：

{
  "available_stock": 500,
  "recommended_quantity": 0
}

但 WMS 的实时可用库存只有 120 件。

系统不能因为 AI 输出了一个格式正确的 JSON，就直接信任其中的库存数值。

更合理的做法是：

1. 从 WMS 获取可信的库存数据。

2. 从 ERP 获取采购和订单数据。

3. 让 AI 分析这些数据并提出建议。

4. 通过确定性业务规则重新校验建议。

5. 在执行前再次检查关键业务条件。

结构化输出解决的是格式问题，不等于解决了事实真实性、业务正确性和执行安全问题。

十三、构建一个完整的企业 AI 采购工作流

下面将前面的机制组合起来，设计一个可落地的业务场景。

13.1 业务背景

某制造企业希望建立 AI 采购协同助手。当系统发现物料供应风险时，自动分析缺口、评估风险并生成采购申请草稿。

系统包含：

• WMS：管理库存和库存预留。

• ERP：管理采购申请、供应商、采购订单和预算。

• MES：提供生产计划、工单和物料需求。

• AI 服务：分析异常并生成处理建议。

• 工作流引擎：管理流程状态、重试、审批与恢复。

• 审批服务：管理采购经理的审批任务。

13.2 工作流步骤

步骤	执行组件	主要操作	结果
1	事件监听器	接收低库存或物料短缺事件	创建流程实例
2	WMS Adapter	查询可用库存与预留量	库存快照
3	MES Adapter	查询生产需求和计划	需求数据
4	ERP Adapter	查询在途采购和预算	采购与预算数据
5	AI Decision Node	分析风险并提出建议	结构化风险报告
6	Business Rule Node	校验数量、预算和采购规则	通过或拒绝
7	Approval Service	创建审批任务	等待人工审批
8	Workflow Engine	接收审批事件并恢复	进入执行阶段
9	ERP Adapter	创建采购申请	返回业务单据标识
10	Verification Node	查询并核验采购申请	确认成功或进入异常处理
11	Audit Service	记录决策、审批和执行结果	完整审计记录

13.3 采购数量的确定性计算

采购建议不能仅由大模型生成。

例如，在一个简化的计算模型中：

其中：

• D：规划周期内的预计需求量。

• S：目标安全库存。

• A：当前可用库存。

• P：规划周期内预计可用的在途采购数量。

假设：

• 预计需求量：500 件。

• 目标安全库存：100 件。

• 当前可用库存：120 件。

• 预计可用的在途采购数量：80 件。

则：

因此，简化模型下的建议补货数量为 400 件。

但在真实业务中，还需要考虑采购批量、最小订购量、供应商交期、质量冻结库存、已分配库存、采购周期以及单位换算等因素。正式采购数量必须由企业规则进行最终校验。

13.4 工作流定义示例

下面使用接近实际工程的 Python 伪代码描述流程：

async def procurement_workflow(event, services):
    # 1. 创建或获取唯一业务流程实例
    workflow = await services.workflow_store.get_or_create(
        business_key=event["material_risk_id"],
        workflow_type="PROCUREMENT_ASSISTANT",
    )

    if workflow.status == "SUCCEEDED":
        return workflow.result

    # 2. 获取可信业务数据
    stock = await services.wms.get_available_stock(
        event["material_code"]
    )

    demand = await services.mes.get_material_demand(
        event["material_code"]
    )

    purchasing = await services.erp.get_open_purchase_data(
        event["material_code"]
    )

    # 3. 使用确定性规则计算基础需求
    quantity = calculate_replenishment(
        demand=demand,
        stock=stock,
        purchasing=purchasing,
    )

    # 4. 调用 AI 分析风险与提出建议
    analysis = await services.ai.analyze_procurement_risk(
        stock=stock,
        demand=demand,
        purchasing=purchasing,
        suggested_quantity=quantity,
    )

    # 5. 校验 AI 输出与业务规则
    validated = validate_analysis(
        analysis=analysis,
        stock=stock,
        demand=demand,
        purchasing=purchasing,
    )

    if not validated:
        await services.workflow_store.mark_failed(
            workflow.workflow_id,
            reason="ANALYSIS_VALIDATION_FAILED",
        )
        return {"status": "FAILED"}

    # 6. 创建审批任务并等待结果
    approval = await services.approvals.create_or_get(
        workflow_id=workflow.workflow_id,
        business_type="PROCUREMENT",
        payload={
            "material_code": event["material_code"],
            "recommended_quantity": quantity,
            "risk_reasons": analysis["risk_reasons"],
        },
    )

    await services.workflow_store.wait_for_approval(
        workflow.workflow_id,
        approval["approval_id"],
    )

    # 注意：真实的长时间工作流应持久化等待状态，
    # 由审批事件恢复流程，而不是长期占用当前 Worker。

    return {
        "status": "WAITING_APPROVAL",
        "workflow_id": workflow.workflow_id,
        "approval_id": approval["approval_id"],
    }

这段代码主要展示业务步骤和组件职责，不是完整的生产级工作流实现。

真实部署中，审批通过后的业务执行通常由单独的恢复入口继续处理，而不是让一个函数一直等待。还需要为每个步骤实现状态持久化、幂等、并发控制、异常分类和审计记录。

13.5 审批通过后执行正式业务操作

审批通过后，应再次验证：

1. 审批任务属于当前工作流。

2. 审批结果有效且尚未被消费。

3. 当前业务数据满足执行条件。

4. 采购数量没有超出审批授权范围。

5. 当前执行用户或服务身份具备相应权限。

6. 本次业务操作尚未成功创建。

7. ERP 返回结果能够被核验。

执行成功后，系统记录采购申请编号，并将流程推进到 SUCCEEDED。

如果 ERP 返回超时，则应进入 UNKNOWN 或等效的待核验状态，先查询是否已生成采购申请，再决定下一步操作。

十四、工作流引擎选型：自研、Temporal、Camunda 还是 LangGraph？

企业 AI 项目不一定都需要引入复杂的工作流引擎。选择技术时，应根据流程的持续时间、业务复杂度、恢复要求、现有技术栈和运维能力综合判断。

Temporal

适合需要可靠执行、持久化恢复、定时器、重试和长时间运行任务的应用工作流。适用于复杂后端业务流程，但需要团队理解其工作流确定性、Activity 和部署运维机制。

重点：可靠执行、持久化恢复、长时间运行。

Camunda / BPMN 工作流引擎

适合业务流程建模、人工任务、流程可视化和企业级流程治理。对于采购审批、订单履约、合规检查等需要业务人员参与的流程尤其值得评估。

重点：BPMN 建模、人工任务、业务流程管理。

LangGraph

适合构建有状态的 LLM 应用与 Agent 图流程，例如多轮推理、条件路由、人工介入和 AI 任务状态管理。它解决的是 AI 应用编排问题，不应未经评估就替代所有企业业务流程能力。

重点：Agent 图编排、AI 状态、人工介入。

Apache Airflow

更适合以定时调度、数据管道和任务依赖为中心的工作流，例如定期汇总 ERP 数据、生成运营报表和执行数据质量检查。对于实时交互式审批流程，需要评估其调度模型是否匹配需求。

重点：定时调度、数据任务、任务依赖。

14.1 选型对照

需求	优先评估方向
长时间运行、失败后可靠恢复	Temporal 等持久化工作流方案
BPMN、审批、人机协同	Camunda 等业务流程引擎
LLM 多轮推理、Agent 状态图	LangGraph 等 AI 编排框架
定时批处理、数据依赖链	Apache Airflow
简单、短生命周期的固定流程	自研轻量状态机也可能足够

以上不是严格的一对一分类。实际选型应以具体版本的能力、持久化语义、许可证、部署模式、团队经验和业务约束为准。

14.2 推荐的企业组合架构

在复杂企业项目中，可以将 AI 编排与业务流程编排分开：

• AI 编排层：负责模型调用、Agent 推理、工具选择和结构化输出。

• 业务工作流层：负责业务状态、审批、超时、恢复和补偿。

• 集成层：负责 API、MCP、消息、幂等和业务适配。

• 治理层：负责身份、权限、规则、风险控制与审计。

例如，LangGraph 可以承担一段复杂的 AI 分析流程，而 Temporal 或 BPMN 引擎负责整条采购业务流程的生命周期。

但不应为了技术完整性而堆叠多个引擎。只有当职责边界清楚、状态同步机制可靠且运维成本可接受时，多引擎组合才有价值。

十五、企业级工作流的可观测性与审计

工作流运行后，系统不仅要回答“成功了吗”，还需要回答：

• 为什么创建这个任务？

• AI 使用了哪些业务数据？

• 哪个节点耗时最长？

• 为什么进入人工审批？

• 谁批准了这项业务操作？

• 是否发生过重试？

• 是否存在重复请求？

• 外部系统实际返回了什么结果？

• 为什么流程最终失败或被取消？

因此，工作流需要建立统一的追踪和审计机制。

15.1 推荐的关联标识

建议至少记录以下标识：

字段	作用
workflow_id	关联整个流程实例
business_key	关联业务对象
step_key	标识工作流节点
execution_id	标识某次节点执行
trace_id	关联分布式调用链
event_id	标识事件并辅助去重
idempotency_key	标识一次业务操作意图
approval_id	关联人工审批任务

这些标识应该在工作流引擎、AI Runtime、API Gateway、业务适配器和外部系统之间尽可能保持关联。

15.2 推荐的工作流指标

还应关注以下业务指标：

• 自动处理完成率。

• 人工审批等待时间。

• 需要人工接管的任务比例。

• 外部业务操作重复率。

• 未核实的未知执行结果数量。

• 补偿成功率。

• 流程积压数量。

• 单个流程的模型调用成本。

• AI 建议被人工修改的比例。

• 流程执行后产生的真实业务价值。

不要只关注模型响应时间。对于长时间运行的工作流，人工审批可能占据绝大多数耗时，单独统计 AI 推理时间并不能反映业务流程效率。

十六、FDE 实战：从 PoC 到生产级工作流的实施路线

FDE 在工作流项目中，不仅需要设计技术方案，还要把业务过程、异常处理和验收标准落实到可执行的工程任务中。

阶段一：业务流程发现

目标：弄清楚流程真实的业务边界。

主要任务：

1. 访谈业务人员，识别流程发起条件。

2. 梳理正常路径和异常路径。

3. 识别所有外部系统及其数据来源。

4. 明确哪些步骤允许自动执行。

5. 明确哪些步骤必须人工审批。

6. 确认流程完成的业务判定标准。

交付物：

• 业务流程图。

• 业务规则清单。

• 异常场景清单。

• 人工审批矩阵。

• 业务系统接口清单。

阶段二：工作流建模

目标：将业务流程转化为可执行的状态与节点。

主要任务：

1. 定义流程实例与业务唯一键。

2. 定义节点及其输入输出。

3. 定义状态机和合法状态转换。

4. 定义并行任务、分支和等待节点。

5. 定义节点超时和重试策略。

6. 定义流程取消和补偿规则。

交付物：

• 工作流模型。

• 状态转换表。

• 节点契约。

• 异常处理设计。

• 补偿流程设计。

阶段三：集成与持久化

目标：确保工作流能够安全调用业务系统，并在异常后恢复。

主要任务：

1. 实现 ERP、WMS、MES 等业务适配器。

2. 建立流程实例和节点执行记录。

3. 实现幂等键和重复请求保护。

4. 实现消息 Outbox 或等效可靠事件机制。

5. 实现外部状态核验。

6. 建立工作流与业务系统之间的关联标识。

交付物：

• 业务适配器。

• 状态持久化方案。

• 幂等与并发控制方案。

• 消息投递机制。

• 集成测试用例。

阶段四：接入 AI 与治理

目标：让 AI 以受控方式参与业务流程。

主要任务：

1. 明确 AI 节点的职责边界。

2. 设计结构化输出 Schema。

3. 建立业务数据来源与事实校验机制。

4. 配置权限和风险规则。

5. 接入人工审批。

6. 记录 AI 输入输出摘要与决策依据。

交付物：

• AI 节点定义。

• Prompt 与 Schema 版本。

• AI 评估集。

• 权限策略。

• 风险控制策略。

• 审批与审计机制。

阶段五：可靠性测试与生产验收

目标：验证工作流在真实故障场景下仍然可控。

主要任务：

1. 模拟 Worker 重启。

2. 模拟 API 超时和限流。

3. 模拟重复消息。

4. 模拟审批超时和重复审批事件。

5. 模拟外部系统已执行但响应丢失。

6. 模拟补偿操作失败。

7. 验证流程恢复和审计完整性。

8. 验证权限边界和人工接管机制。

交付物：

• 故障注入测试报告。

• 性能与容量测试报告。

• 安全测试报告。

• 生产部署与回滚方案。

• 运行手册和告警规则。

• 业务验收报告。

十七、生产环境上线前检查清单

下面的清单可以直接用于项目评审和上线验收。

生产上线自检

23 项

1. 流程建模

每个流程都有唯一标识和明确的业务完成条件

所有状态转换都有合法性校验

分支、并行、等待和取消路径已定义

2. 可靠性

关键节点状态已经持久化

重试次数、退避和截止时间已配置

未知执行结果具有核验机制

关键写操作具备幂等保护

3. 跨系统一致性

跨系统操作的失败路径已经定义

补偿操作经过业务确认

消息发送与数据库更新具备可靠协调机制

重复消息不会导致重复业务操作

4. AI 与治理

AI 输出具有结构化 Schema

业务事实来自可信系统

权限和风险校验独立于模型输出

高风险操作具有审批或其他授权机制

5. 可观测性

流程、节点、事件和业务单据可以关联

失败、超时和积压有监控告警

审计日志记录状态变化和关键操作

支持人工查询、接管和恢复

6. 上线准备

已完成故障注入与恢复测试

已定义容量、限流和降级策略

已准备回滚和应急处理手册

业务负责人已确认验收指标

十八、企业 AI 工作流最常见的八个错误

错误一：把整个业务流程写成一个 Agent。

让一个 Agent 自由决定所有步骤，会导致流程边界不清晰、失败恢复困难，也很难进行业务审计。

建议将流程骨架交给工作流引擎，将需要推理的环节交给 Agent。

错误二：只保存对话历史，不保存流程状态。

对话记录不能替代流程状态。系统需要知道当前节点、已完成操作、待审批任务和下一步允许执行的动作。

错误三：把接口超时当成业务失败。

超时只说明客户端没有在指定时间内获得结果。外部系统可能已经执行成功，必须进行核验。

错误四：认为重试就能解决所有问题。

重试可能引发重复写入。必须根据异常类型、幂等能力和业务状态决定是否重试。

错误五：将 AI 输出直接传给业务系统。

AI 输出需要经过 Schema 校验、业务规则、权限和风险控制，不能仅凭模型生成的内容执行高风险操作。

错误六：把补偿理解为删除或回滚数据库记录。

企业业务操作可能已经产生真实的库存、生产、物流和财务影响。补偿应采用合法的业务操作，并保留审计轨迹。

错误七：忽略人工审批的生命周期。

审批不仅是一个布尔值，还涉及审批人权限、重复事件、超时、业务数据变化和审批结果有效性。

错误八：只监控 AI 响应时间，不监控业务流程结果。

流程可能因为审批等待、系统积压或外部接口异常而失败。必须同时监控技术指标与业务指标。

十九、本章总结

企业级 AI 工作流工程解决的是 AI 从“能够给出答案”走向“能够可靠地完成业务”的关键问题。

本章围绕状态机、工作流上下文、持久化、长时间运行、人工审批、断点恢复、异常重试、幂等机制、Saga 补偿、Transactional Outbox 和 AI 决策节点，建立了一套面向企业真实业务的工程方法。

需要记住以下五条原则：

1. 流程显式化： 将业务步骤、状态和异常路径建模，而不是依赖隐式的模型推理。

2. 状态持久化： 让工作流在服务重启、任务等待和节点失败后仍然能够恢复。

3. 副作用可控： 通过幂等、核验、业务规则和授权机制控制外部系统写操作。

4. AI 受约束： 让模型参与分析与决策，但不允许模型绕过业务规则和治理机制。

5. 结果可审计： 记录流程状态、关键输入输出、审批、重试和最终业务结果。

对于 FDE 而言，真正的工作不只是把 Agent 接入 ERP 或 WMS，而是把 AI 能力融入业务流程，保证系统在正常执行、异常中断和人工介入等各种情况下，都能表现出可预期、可解释、可恢复的行为。

下一章将继续深入企业 AI 工程化的另一个重要主题：

第54章：Enterprise AI Deployment Engineering——构建可灰度发布、可回滚、可持续交付的企业级 AI 部署体系。