ai-agents-for-beginners 如何在 Agent 工作流中为高风险工具添加 human approval 暂停审批【免费下载链接】ai-agents-for-beginners18 Lessons to Get Started Building AI Agents项目地址: https://gitcode.com/GitHub_Trending/ai/ai-agents-for-beginners在 ai-agents-for-beginners 这个课程仓库里第 14 课 Microsoft Agent FrameworkMAF部分给出了完整的 Human-in-the-Loop 实现Agent 工作流运行到需要人工把关的分支时暂停等待人输入yes/no之后才继续路由。本文以仓库中的酒店预订工作流为例演示如何把一个高风险后续动作无房时推荐替代城市、或取消预订改成必须经过人工审批才能执行并覆盖从依赖安装、环境配置到验证输出的完整路径。仓库另在第 6 课提供了工具级的前置审批门pre-action gate 风险分级 审计日志方案作为可选分支介绍。准备条件主路径使用 MAF 工作流14-human-loop.ipynb前置条件如下安装依赖。仓库根目录 requirements.txt 已锁定关键版本其中 MAF 核心包被固定为agent-framework-core1.10.0注释说明 1.11.0 移除了课程 notebook 使用的ChatMessage、HostedWebSearchTool等 API同时需要agent-framework-foundry~1.10.0和agent-framework-openai~1.10.0pip install -r requirements.txt配置环境变量。MAF notebook 从.env加载两个变量并用 Azure CLI 做免密钥认证需先执行az loginexport AZURE_AI_PROJECT_ENDPOINT你的 Foundry 项目端点 export AZURE_AI_MODEL_DEPLOYMENT_NAME模型部署名 az login创建聊天客户端这是 notebook 中配置 LLM 的方式Microsoft Foundry provider AzureCliCredentialfrom agent_framework.foundry import FoundryChatClient from azure.identity import AzureCliCredential from dotenv import load_dotenv import os load_dotenv() chat_client FoundryChatClient( project_endpointos.environ[AZURE_AI_PROJECT_ENDPOINT], modelos.environ[AZURE_AI_MODEL_DEPLOYMENT_NAME], credentialAzureCliCredential(), )第一步定义发给人的审批请求 payload审批请求必须是类型化的这样才能在暂停事件里携带上下文问什么、针对哪个城市。notebook 用一个 dataclass 定义请求负载HumanFeedbackRequest包含prompt要问人的问题和destination不可用的城市from dataclasses import dataclass dataclass class HumanFeedbackRequest: Request sent to the human asking if user wants alternatives. This is what gets sent to the RequestInfoExecutor. prompt: str # The question to ask the user destination: str # The unavailable destination for context同时用 Pydantic 模型约束确认 Agent 的 JSON 输出保证暂停前拿到结构化问题from pydantic import BaseModel class ConfirmationQuestion(BaseModel): confirmation_agent 的 response_formatAgent 输出为 JSON。 question: str # The question to ask the user destination: str # The unavailable destination for context第二步创建执行暂停与审批路由的 Executor核心是自定义ExecutorDecisionManager它承担两个职责调用ctx.request_info()暂停工作流收到人回复后根据内容路由。两个关键装饰器是handler暴露工作流步骤和response_handler处理人的回复from typing import Any from agent_framework import ( AgentExecutorRequest, AgentExecutorResponse, Message, Executor, WorkflowContext, handler, response_handler, ) class DecisionManager(Executor): 根据人工反馈协调工作流路由。 def __init__(self, id: str | None None): super().__init__(idid or decision_manager) handler async def on_confirmation( self, response: AgentExecutorResponse, ctx: WorkflowContext, ) - None: 解析确认问题并暂停工作流等待人工输入。 confirmation ConfirmationQuestion.model_validate_json(response.agent_run_response.text) # 暂停工作流人回复字符串会投递给 on_human_feedback。 await ctx.request_info( request_dataHumanFeedbackRequest( promptconfirmation.question, destinationconfirmation.destination, ), response_typestr, ) response_handler async def on_human_feedback( self, original_request: HumanFeedbackRequest, feedback: str, ctx: WorkflowContext[AgentExecutorRequest, str], ) - None: 根据人的 yes/no 回复路由工作流。 user_reply (feedback or ).strip().lower() destination original_request.destination or unknown if user_reply yes: # 审批通过 → 路由到 alternative_agent user_msg Message( roleuser, contents[fThe user wants to see alternative destinations near {destination}. Please suggest one.], ) await ctx.send_message(AgentExecutorRequest(messages[user_msg], should_respondTrue)) elif user_reply no: # 审批拒绝 → 路由到 cancellation_agent user_msg Message( roleuser, contents[The user has declined to see alternatives. Please acknowledge their decision.], ) await ctx.send_message(AgentExecutorRequest(messages[user_msg], should_respondTrue)) else: # 文档给出的默认策略意外输入按拒绝处理 user_msg Message( roleuser, contents[The user has declined to see alternatives. Please acknowledge their decision.], ) await ctx.send_message(AgentExecutorRequest(messages[user_msg], should_respondTrue))注意文档明确强调暂停组件本身不收集输入它只负责暂停工作流由你的应用代码监听request_info事件、收集人的答案再按request_id把回复发回工作流。另外yes/no之外的输入会被当作拒绝decline处理这是文档给出的兜底策略。第三步组装工作流只让高风险分支经过审批审批暂停只挂在需要人工决策的分支上低风险路径有房直接预订不暂停。完整建图使用WorkflowBuilder 条件边from agent_framework import AgentExecutor, WorkflowBuilder, tool # 工具模拟酒店可用性检查示例中有房的固定城市列表 tool(descriptionCheck hotel room availability for a destination city) def hotel_booking(destination: str) - str: cities_with_rooms [stockholm, seattle, tokyo, london, amsterdam] has_rooms destination.lower() in cities_with_rooms return json.dumps({has_availability: has_rooms, destination: destination}) # 五个 AgentExecutorinstructions 从原 notebook 复制 # availability_agent 带 hotel_booking 工具response_formatBookingCheckResult # confirmation_agent 无房时生成确认问题response_formatConfirmationQuestion # alternative_agent 审批通过时推荐替代城市response_formatAlternativeResult # booking_agent 有房时鼓励预订response_formatBookingConfirmation # cancellation_agent 审批拒绝时生成取消消息response_formatCancellationMessage # 各自形如 # availability_agent AgentExecutor( # chat_client.as_agent(instructions..., tools[hotel_booking], # default_options{response_format: BookingCheckResult}), # idavailability_agent, # ) decision_manager DecisionManager(iddecision_manager) # display_result 为 executor 定义的输出 executorctx.yield_output 产出最终结果 workflow ( WorkflowBuilder( start_executoravailability_agent, output_executors[display_result], ) # 无房分支走人工审批decision_manager 内部通过 ctx.request_info 暂停 .add_edge(availability_agent, confirmation_agent, conditionno_availability_condition) .add_edge(confirmation_agent, decision_manager) .add_edge(decision_manager, alternative_agent, conditionuser_wants_alternatives_condition) .add_edge(decision_manager, cancellation_agent, conditionuser_declines_alternatives_condition) .add_edge(alternative_agent, display_result) .add_edge(cancellation_agent, display_result) # 有房分支直接预订不经过人工审批 .add_edge(availability_agent, booking_agent, conditionhas_availability_condition) .add_edge(booking_agent, display_result) .build() )条件函数has_availability_condition、no_availability_condition解析上游AgentExecutorResponse中的BookingCheckResultJSON 并返回布尔值user_wants_alternatives_condition、user_declines_alternatives_condition则检查decision_manager发出的AgentExecutorRequest消息文本判断走哪条审批结果分支。完整的 agents 与条件函数代码见 14-human-loop.ipynb。第四步运行工作流暂停时收集审批再恢复执行模式是首轮流式运行 → 监听暂停事件 → 带responses恢复运行的循环。notebook 中用脚本化答案SCRIPTED_ANSWER yes代替input()让 notebook 无人值守运行在真实应用里把它换成input()或 UI 回调即可import json request_paris AgentExecutorRequest( messages[Message(roleuser, contents[I want to book a hotel in Paris])], should_respondTrue, ) SCRIPTED_ANSWER yes # 真实应用替换为 input() 或 UI 回调 workflow_output: str | None None # 首轮流式运行直到工作流暂停或产出结果 stream workflow.run(request_paris, streamTrue) while True: requests: list[tuple[str, HumanFeedbackRequest]] [] async for event in stream: if event.type request_info and isinstance(event.data, HumanFeedbackRequest): print(f\n⏸️ WORKFLOW PAUSED - Human input requested!) print(f Request ID: {event.request_id}) print(f Destination: {event.data.destination}) print(f Question: {event.data.prompt}) requests.append((event.request_id, event.data)) elif event.type output: workflow_output str(event.data) print(\n✅ Workflow completed with output!) if not requests: break # 为每个待审批请求提供人的答案{request_id: 答案} responses: dict[str, str] {} for req_id, req in requests: responses[req_id] SCRIPTED_ANSWER # 恢复运行带上 responses 继续 stream workflow.run(streamTrue, responsesresponses)恢复的关键在于event.request_id与回复的一一对应responses字典的键是暂停事件里的request_id不要手工管理其他状态。一个值得注意的细节notebook 的总结章节把恢复 API 描述为send_responses_streaming(pending_responses)而实际可运行的代码单元格使用的是workflow.run(streamTrue, responsesresponses)两者指向同一语义——把{request_id: 答案}发回工作流恢复执行。验证结果按上面的代码跑通后文档展示的输出示例结果Request ID每次运行不同是暂停时打印⏸️ WORKFLOW PAUSED - Human input requested!附带Request ID如文档示例中的032c8fce-b9d1-400e-ba8d-afd2248e2926、Destination: Paris以及QUESTION FOR YOU:下确认 Agent 生成的问题示例为 Unfortunately, there are no rooms available in Paris. Would you like to explore nearby alternative destinations?。恢复后打印Sending human responses: {request_id: yes}工作流继续最终经alternative_agent产出包含alternative_destination和reason字段的 JSON 输出审批通过路径若回答no则经cancellation_agent产出status: cancelled的取消确认。对照路径请求 Stockholm 这类有房城市时workflow.run(request_stockholm)直接完成不发出任何request_info事件输出为BookingConfirmationdestination/action/message字段全程无人工审批——这验证了审批只挂在该挂的分支上。可选分支工具级前置审批门与风险分级第 6 课如果你要审批的不是工作流分支而是单个高风险工具动作本身06-human-in-the-loop.ipynb 给出了另一条路径在动作产生副作用之前设门pre-action gate按风险分级决定是否需要人。该 notebook 需要 Azure OpenAI 环境AZURE_OPENAI_ENDPOINT、AZURE_OPENAI_DEPLOYMENT同样先az login走 Responses API 的/openai/v1/端点。核心机制风险分级classify_risk()用关键词启发式把动作分为low/medium/high三级。高风险关键词包括send、email、post、publish、charge、pay、transfer、delete、drop、cancel、refund无法识别的动作默认归入medium批量人工复核而不是low。low和medium自动放行只有high阻塞等待人。审批门gate_action()在非 demo 模式下询问input(approve / deny / escalate?)EOF、空输入或非法输入一律按deny兜底。DEMO_MODE True时整个流程无需交互高风险动作在attempt0被脚本化拒绝、attempt1自动批准用于演示拒绝→重试→批准的循环机制。审计日志每个门决策以 JSONL 追加写入含decision、reason、action、risk_tier、时间戳文件名带 UTC 时间戳避免覆盖历史日志。修订循环run_with_revision(goal, max_revisions)在被deny时把拒绝理由反馈给 LLM 重新提案直到批准或达到max_revisions此时结果为max_revisions_reached。注意文档明确说明DEMO_MODE 下的自动批准只是演示循环机制的脚本化行为真正的根据修订内容重新分类需要DEMO_MODE False加人工操作员。该 notebook 开头也指出认证与访问控制、工具调用中间件MAF 侧等属于其他课程内容。限制与适用边界MAF 路径的暂停发生在decision_manager的ctx.request_info()调用处人回复的类型由response_type指定示例中为str审批只约束经过该 executor 的分支有房直订路径完全绕过审批。意外输入既非yes也非no按文档给出的策略一律视为拒绝并走取消分支。版本约束MAF 依赖锁定在agent-framework-core1.10.0一系升级 1.11.0 会因 API 变更移除ChatMessage、HostedWebSearchToolMessage构造器变化导致课程 notebook 不能直接运行。更完整的组件对照RequestInfoExecutor、RequestInfoMessage、RequestResponse的角色分工以及多审批点、超时处理asyncio.wait_for、Web/Slack UI 集成等扩展写法可直接在 14-human-loop.ipynb 末尾的 Key Takeaways 一节查看无审批的对照工作流见 hotel_booking_workflow_sample.py。【免费下载链接】ai-agents-for-beginners18 Lessons to Get Started Building AI Agents项目地址: https://gitcode.com/GitHub_Trending/ai/ai-agents-for-beginners创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
