题图来源：Microsoft Tech Community 原文页面（见文末链接）。

摘要

在需要“高自动化 + 强审计/合规”的 AI 工作流里，人类审批（human-in-the-loop）往往是绕不过去的一环：既要让多智能体协作把流程跑起来，又要在关键动作（例如停机、触发维护）前强制停下来等人确认。

Microsoft 在一篇示例文章中给出了一套脚本化方案：使用 Agent Framework 编排一个顺序（sequential）多智能体工作流，底层调用 Azure AI Foundry 的持久化（persistent）agents，并利用 **workflow checkpointing（检查点）**实现“暂停—等待审批—从检查点恢复继续执行”。

本文以第三方博主视角整理该方案的核心实现点：场景设定、资源/环境准备、持久化 agent 创建、顺序工作流编排、如何用 [PENDING] / [APPROVED] / [REJECTED] 贯穿审批状态，以及脚本结构与参考链接。

场景

这个示例工作流以“工业管线传感器数据处理”为背景，由三个专用 agent 顺序协作，并在高风险决策上引入人类审批：

  • Data Analyser Agent：收集管线泵/设备的传感器读数（pressure、temperature、flow rate），检测异常，并用结构化格式输出摘要。
  • Risk Assessor Agent V2：根据预设阈值评估风险严重性，决定下一步动作，比如安排维护或触发立即停机。
  • Maintenance Scheduler Agent：在人类审批通过后继续执行，给相关团队分配任务并确认执行步骤。

流程里最关键的机制是：一旦出现高影响决策，工作流会在检查点暂停，保存状态，并请求人类审批；待外部系统记录审批结果后，工作流从上次检查点恢复，验证审批状态，然后再执行“安排维护/停机”等关键动作。示例中外部系统用一个 JSON 文件模拟（也可以替换为真实系统的 API/数据库）。

代码概览

1) Azure 资源与本地环境

需要先准备 Azure AI Foundry 资源与一个大语言模型部署，并把 Azure AI Foundry Project 的 endpoint、以及模型部署名写入环境变量（示例格式如下）：

1
2

AZURE_AI_PROJECT_ENDPOINT="<AZURE AI FOUNDRY PROJECT ENDPOINT>"
AZURE_AI_MODEL_DEPLOYMENT_NAME="<gpt-4o>"

随后安装脚本依赖（原文以 requirements.txt 为例）。

2) 创建 Azure AI Foundry 持久化 Agents

工作流里用到的 Azure AI Foundry persistent agents 通过 Azure AI Foundry SDK 的 AIProjectClient 创建。该 SDK 在示例中会作为 Agent Framework SDK（Python）的依赖被隐式安装（同样列在 requirements.txt）。

创建完成后，这些 agents 可以在 Azure AI Foundry 门户的 Agents 页面里查看。

初次创建时，你可能会看到 tools 还没有与 agents 关联——这一步会在后续“为工作流创建 agent 实例并附加本地 tools”时完成。

另外，每个 agent 创建后都会有一个 ID：可以从 agent 的 id 属性取得，也可以在 Azure AI Foundry 门户中找到。后续在 Agent Framework 编排工作流时，需要用这些 ID 来引用对应的 Foundry agents。

原文也强调了一个实现细节：Maintenance Scheduler Agent (v2) 的指令里明确要求：当动作是 [Schedule Maintenance] 或 [Immediate Shutdown] 时，必须拿到人类审批结果 [APPROVED] 才允许执行。为了让编排逻辑能判断“是否在等审批”，该 agent 会在需要审批但尚未批准时，在消息里包含 [PENDING] 关键字；当需要时，它也可以查询审批状态。

参考脚本（原文）：A03_Create_Multiple_Foundry_Agent_Persistent.py

3) 创建顺序（Sequential）工作流

工作流侧使用 Agent Framework 的 ChatAgent 来配置所需 agents，这会调用前面在 Azure AI Foundry 中创建的持久化 agents。

然后给 ChatAgent 挂载本地 tools，用于模拟工作流里的外部动作。

原文列出的工具集合如下：

  • get_data：获取某个泵/设备的数据（temperature、pressure、flow rate），以 JSON 返回。
  • schedule_maintenance：为指定设备安排维护，返回维护请求编号。
  • send_shutdown_equipment_notification：发送设备停机通知并通知相关团队，返回通知 ID。
  • send_approval_rejection_notification：当审批被拒绝时，发送“动作被拒绝”的通知。
  • request_human_approval：通过外部系统为关键动作请求人类审批。
  • get_human_approval_status：当工作流恢复执行时，从外部 JSON 文件中查询审批状态。

用检查点实现“暂停与恢复”

脚本会初始化一个内存中的 checkpoint storage，然后构建一个顺序工作流。Agent Framework 的 Checkpoints 能在工作流执行到特定点时保存状态，并允许稍后从该点继续运行，适用于长运行/可中断流程。

脚本逻辑大致是：如果找不到 checkpoint 文件就创建新工作流；如果发现已有 checkpoint 文件就从上次检查点恢复继续。

为了判断是否“正在等待人类审批”，示例通过检测对话消息里是否出现 [PENDING] 关键字来决定是否需要保存检查点并暂停。

用外部 JSON 模拟审批系统

示例用 approval_db.json 来模拟外部审批系统（在真实生产环境中可以替换成 API 调用或数据库）。当需要审批时，脚本会生成该文件并写入状态 [PENDING]。只有当人工把状态改成 [APPROVED]，Maintenance Scheduler Agent (v2) 才会在工作流重跑/恢复时真正执行动作；如果是 [PENDING] / [APPROVED] / [REJECTED]，工作流都会从合适的步骤恢复，而不是从头再来。

参考脚本（原文）：W04_Sequential_Workflow_Human_Approval.py

参考链接

原文发布时间：2025-10-31；版本：1.0。