智能体工作流进入企业生产环境后，难点很快会从“模型能否完成任务”转向“任务由谁负责、处理到哪一步、异常如何恢复、何时需要人工介入”。Amazon Quick Automate 的原生案件管理把每项业务工作包装成可跟踪的 case，并将创建、处理、人工审核和解决串成完整生命周期，使自动化能够从单次演示扩展到持续运行的企业流程。
Case 是业务状态的持久载体
普通自动化常把一次输入直接交给模型或工具，执行结束后只留下日志。企业流程则需要一个稳定的业务对象，用来记录上下文、责任人、当前状态、处理历史和最终结果。Case 正是这个对象。
一条典型生命周期可以表示为：
NEW -> IN_PROGRESS -> WAITING_FOR_HUMAN -> IN_PROGRESS -> RESOLVED
|
+-> REJECTED / NEEDS_MORE_INFO
任意处理阶段 -> FAILED -> RETRYING -> IN_PROGRESS
状态不只是展示字段，还应驱动后续动作：
NEW表示案件已经创建，可以进入待处理队列。IN_PROGRESS表示处理器或智能体已经接手，避免重复消费。WAITING_FOR_HUMAN表示自动化暂停，等待审批、补充材料或业务判断。FAILED应同时记录错误类型、重试次数和可恢复性。RESOLVED表示结果已经落库，并满足案件关闭条件。
摘要介绍的能力覆盖单个或批量案件管理、状态自动跟踪、异常处理和 Human-in-the-loop（HITL）。这意味着团队可以把案件状态当作工作流控制面，而不是依赖散落在提示词、日志和通知中的隐式状态。
Creator-Processor 模式如何实现动态扩展
Creator-Processor 模式把“识别待办事项”和“完成业务处理”拆成两个角色。
Creator 接收邮件、表单、API 请求或批量文件，为每个独立业务事项创建 case。Processor 从待处理案件中领取工作，调用智能体、规则引擎或外部系统，然后持续更新案件状态。两者解耦后，创建速度不再受单个处理器吞吐量限制，也可以根据案件积压量动态增加 Processor。
业务输入
|
v
Case Creator ---> 案件队列/案件存储 ---> Processor 1
---> Processor 2
---> Processor N
|
v
HITL / 外部系统 / 解决
落地时需要特别处理三类并发问题：
- 幂等创建：同一业务事件重放时，不能生成多个重复案件。
- 原子领取：多个 Processor 不应同时处理同一个 case。
- 幂等更新：外部 API 超时后重试，不能重复付款、重复发送通知或覆盖较新的状态。
可以使用业务事件 ID 作为创建幂等键，并在领取案件时执行带版本号的条件更新。对于有副作用的外部操作，还应保存操作 ID 和执行结果，避免把“请求超时”误判成“操作未发生”。
可以这样实践：搭建一个最小案件处理器
下面是一个可直接运行的本地 Python 示例，用来演示 Creator-Processor、自动状态更新、异常重试和 HITL 分支。它不是 Amazon Quick Automate 的真实 API 定义，而是便于设计工作流和字段模型的最小原型；接入产品时，应将其中的内存存储和处理函数替换为实际的案件、工作流及连接器能力。
将代码保存为 case_workflow.py，使用 Python 3.10 或更高版本运行：
from dataclasses import dataclass, field
from enum import Enum
from typing import Any
from uuid import uuid4
class Status(str, Enum):
NEW = "NEW"
IN_PROGRESS = "IN_PROGRESS"
WAITING_FOR_HUMAN = "WAITING_FOR_HUMAN"
RETRYING = "RETRYING"
FAILED = "FAILED"
RESOLVED = "RESOLVED"
@dataclass
class Case:
id: str
source_id: str
payload: dict[str, Any]
status: Status = Status.NEW
attempts: int = 0
result: dict[str, Any] | None = None
history: list[str] = field(default_factory=list)
def transition(self, status: Status, note: str) -> None:
self.status = status
self.history.append(f"{status.value}: {note}")
cases: dict[str, Case] = {}
source_index: dict[str, str] = {}
def create_case(source_id: str, payload: dict[str, Any]) -> Case:
# source_id acts as the idempotency key.
if source_id in source_index:
return cases[source_index[source_id]]
case = Case(id=str(uuid4()), source_id=source_id, payload=payload)
case.history.append("NEW: case created")
cases[case.id] = case
source_index[source_id] = case.id
return case
def process_case(case: Case) -> None:
case.attempts += 1
case.transition(Status.IN_PROGRESS, f"attempt {case.attempts}")
amount = float(case.payload["amount"])
if amount >= 10_000 and not case.payload.get("approved"):
case.transition(Status.WAITING_FOR_HUMAN, "high-value approval required")
return
if case.payload.get("simulate_error") and case.attempts < 2:
raise RuntimeError("temporary downstream failure")
case.result = {"decision": "accepted", "amount": amount}
case.transition(Status.RESOLVED, "processing completed")
def run_processor(case: Case, max_attempts: int = 3) -> None:
while case.status not in {Status.RESOLVED, Status.WAITING_FOR_HUMAN}:
try:
process_case(case)
except RuntimeError as exc:
if case.attempts >= max_attempts:
case.transition(Status.FAILED, str(exc))
break
case.transition(Status.RETRYING, str(exc))
if __name__ == "__main__":
normal = create_case("event-1001", {"amount": 850, "simulate_error": True})
review = create_case("event-1002", {"amount": 25_000})
for item in cases.values():
run_processor(item)
print(item.id, item.status.value, item.history)
# Simulate a human approval and resume the suspended workflow.
review.payload["approved"] = True
run_processor(review)
print("after approval:", review.id, review.status.value, review.history)
运行命令：
python3 case_workflow.py
接入 Quick Automate 时，可以把这个原型映射为三段工作流：创建案件、处理案件、人工审核后恢复案件。字段至少应包括 source_id、status、attempts、assigned_to、payload、result、error_code 和 updated_at。
HITL 不是兜底按钮，而是一条明确分支
人工介入如果只靠“出错后发邮件”，案件很容易停在无人负责的状态。更可靠的做法是为 HITL 定义进入条件、负责人、截止时间、可执行动作和恢复点。
例如，在供应商发票处理流程中，智能体可以提取发票字段并与采购订单核对。金额超过阈值、字段置信度不足或银行账户发生变化时，案件转为 WAITING_FOR_HUMAN。审核人只能选择批准、拒绝或要求补充信息；选择结果写回 case，工作流再从固定节点恢复。
建议同时记录以下审计信息：
- 触发人工审核的规则或模型判断。
- 提交给审核人的原始材料与模型建议。
- 审核人的身份、操作、备注和时间。
- 审核前后的字段差异。
- 恢复工作流所使用的案件版本。
涉及付款、账户权限、合规决定等高风险操作时，不应让模型自行降低审核门槛。模型输出也不应直接覆盖原始证据，应作为独立建议保存。
上线前检查吞吐量，也检查可恢复性
扩展智能体工作流不能只增加并发 Processor。上线前应验证以下问题：
- 是否为每个输入定义了稳定的幂等键？
- Processor 崩溃后，案件能否被重新领取？
- 重试是否区分临时错误和永久错误？
- HITL 是否有负责人、超时策略和升级路径？
- 状态变化是否保留审计记录，而不是只保存最终状态？
- 批量案件是否支持限流，避免压垮下游系统？
- 关闭后的案件是否允许重开，重开后从哪个节点继续？
原生案件管理的核心价值，不只是给智能体任务加一个编号，而是把自动化变成可分配、可暂停、可恢复、可审计的业务过程。Creator-Processor 模式解决吞吐量问题，明确的状态机和 HITL 解决控制问题；两者结合，才适合承载长期运行的企业流程。