Cloudflare Agents 人机协同（Human-in-the-Loop）实战：审批闸门、工具审核与状态编排全解析

Cloudflare Agents 人机协同Human-in-the-Loop实战审批闸门、工具审核与状态编排全解析【免费下载链接】agentsBuild and deploy AI Agents on Cloudflare项目地址: https://gitcode.com/GitHub_Trending/agents1/agents人机协同Human-in-the-loop简称 HITL让 Agent 在执行关键动作前暂停下来等待人类的批准、确认或补充输入是合规、安全与可审计的 Agent 系统的基石。本文以 Cloudflare Agents 框架agents包与cloudflare/ai-chat包为背景系统讲解 HITL 的决策路径、waitForApproval工作流审批、needsApproval工具审批、onToolCall客户端工具执行、MCP Elicitation 结构化输入以及审批状态管理与超时升级等完整实战方案。读完本文你将能根据场景选对 HITL 模式并直接落地一套带审批闸门、可持久化、可恢复的 Agent 应用。为什么需要 Human-in-the-Loop纯自主的 Agent 系统在执行高风险动作时往往缺少监督窗口。引入 HITL 主要解决四类问题合规Compliance监管要求某些动作必须经过人工审批例如财务支出、对外通信安全Safety支付、批量删除、外发消息等高价值操作需要人工把关质量Quality人工复核可以捕获模型可能遗漏的错误信任Trust用户能够批准关键操作时对系统的掌控感与信任度显著提升。常见使用场景使用场景示例财务审批报销单、支付处理内容审核内容发布、邮件发送数据操作批量删除、数据导出AI 工具执行执行 LLM 工具调用前的人工确认访问控制授权、角色变更选择正确的 HITL 模式Cloudflare Agents SDK 提供多种 HITL 模式应根据业务形态选择最合适的一种使用场景模式适用场景示例长时运行工作流Workflow Approval多步骤流程、持久化审批闸门examples/workflowsAIChatAgent 工具needsApproval基于cloudflare/ai-chat的聊天式工具调用guides/human-in-the-loopOpenAI Agents SDKneedsApproval使用 OpenAI Agent SDK 的条件审批openai-sdk/human-in-the-loop客户端工具onToolCall需要浏览器 API 或用户交互的工具下文客户端工具执行一节无状态服务端Stateless Elicitation当前 MCP 工具请求结构化输入examples/mcp-elicitation-mrtr遗留服务端Legacy Elicitation既有会话型 MCP 部署examples/mcp-elicitation决策指引这是多步骤工作流的一部分吗 ├── 是 → 使用 Workflow ApprovalwaitForApproval └── 否 → 你在构建 MCP 服务器吗 ├── 是 → 使用 MCP ElicitationelicitInput └── 否 → 这是 AI 聊天交互吗 ├── 是 → 工具需要浏览器 API 吗 │ ├── 是 → 使用 onToolCall客户端执行 │ └── 否 → 使用 needsApproval服务端 审批 └── 否 → 使用 State WebSocket 做简单确认基于 Workflow 的持久化审批对于需要持久化、多步骤的业务流程使用 Cloudflare Workflows 结合waitForApproval()辅助方法。工作流会暂停执行直到人工批准或拒绝。基本模式下面的ExpenseWorkflow继承AgentWorkflow先校验报销单再通过reportProgress上报待审批状态随后调用waitForApproval挂起工作流直至审批事件到达后才继续处理import { Agent, AgentWorkflow, callable } from agents; import type { AgentWorkflowEvent, AgentWorkflowStep } from agents; // Workflow that pauses for approval export class ExpenseWorkflow extends AgentWorkflow ExpenseAgent, ExpenseParams { async run(event: AgentWorkflowEventExpenseParams, step: AgentWorkflowStep) { const expense event.payload; // Step 1: Validate the expense const validated await step.do(validate, async () { return validateExpense(expense); }); // Step 2: Wait for manager approval await this.reportProgress({ step: approval, status: pending, message: Awaiting approval for $${expense.amount} }); // This pauses the workflow until approved/rejected const approval await this.waitForApproval{ approvedBy: string }(step, { timeout: 7 days }); console.log(Approved by: ${approval.approvedBy}); // Step 3: Process the approved expense const result await step.do(process, async () { return processExpense(validated); }); await step.reportComplete(result); return result; } }源码级原理waitForApproval的实现位于 packages/agents/src/workflows.ts。它底层调用 Cloudflare Workflows 的step.waitForEvent(stepName, { type: eventType, timeout })等待类型为approval的事件。事件负载遵循 ApprovalEventPayload 结构{ approved, reason?, metadata? }当负载中approved true时将metadata作为返回值解出即上面approval.approvedBy的来源当approved false时会先通过step.reportError持久化上报错误再抛出WorkflowRejectedError该错误类型定义于 workflow-types.ts携带reason与workflowId。同时AgentWorkflow的构造函数会通过WeakSet防止重复包装run()方法并自动注入 Agent 初始化逻辑从事件参数中剥离__agentName、__agentBinding、__workflowName、__agentOrigin等内部字段再把干净的 payload 交给用户实现的run()。这意味着审批挂起期间工作流实例由 Cloudflare Workflows 保证持久化Worker 重启不会丢失等待状态。Agent 侧的审批方法Agent 提供approveWorkflow与rejectWorkflow两个方法配合callable()装饰器对外暴露成 RPC 端点供 UI 或第三方调用export class ExpenseAgent extends AgentEnv, ExpenseState { initialState: ExpenseState { pendingApprovals: [], status: idle }; // Approve a waiting workflow callable() async approve(workflowId: string, approvedBy: string): Promisevoid { await this.approveWorkflow(workflowId, { reason: Expense approved, metadata: { approvedBy, approvedAt: Date.now() } }); // Update state to reflect approval this.setState({ ...this.state, pendingApprovals: this.state.pendingApprovals.filter( (p) p.workflowId ! workflowId ) }); } // Reject a waiting workflow callable() async reject(workflowId: string, reason: string): Promisevoid { await this.rejectWorkflow(workflowId, { reason }); this.setState({ ...this.state, pendingApprovals: this.state.pendingApprovals.filter( (p) p.workflowId ! workflowId ) }); } // Track workflow progress async onWorkflowProgress( workflowName: string, workflowId: string, progress: unknown ): Promisevoid { const p progress as { step: string; status: string }; if (p.step approval p.status pending) { // Add to pending approvals list this.setState({ ...this.state, pendingApprovals: [ ...this.state.pendingApprovals, { workflowId, requestedAt: Date.now() } ] }); } } }源码级原理approveWorkflow(workflowId, data)与rejectWorkflow(workflowId, data)实现在 packages/agents/src/index.ts。两者的共同底层是sendWorkflowEvent()同文件 index.ts它先通过 SQLite 跟踪表cf_agents_workflows校验工作流是否被runWorkflow()跟踪过再向指定实例发送事件并附带最多 3 次的重试tryN(3, ...)指数退避。区别在于approveWorkflow发送{ type: approval, payload: { approved: true, reason?, metadata? } }并触发workflow:approved事件rejectWorkflow发送{ type: approval, payload: { approved: false, reason? } }并触发workflow:rejected事件。事件到达后工作流内挂起的waitForApproval恢复执行通过或抛WorkflowRejectedError拒绝。超时处理设置超时防止工作流无限期等待const approval await this.waitForApproval(step, { timeout: 7 days // or 1 hour, 30 minutes, etc. });WaitForApprovalOptions 支持三个可选字段字段默认值说明stepNamewait-for-approval传给waitForEvent的步骤名timeout无超时时长如7 days、24 hourseventTypeapproval等待的事件类型如果超时到期工作流将继续执行但没有审批数据需要显式处理该分支const approval await this.waitForApproval{ approvedBy: string }(step, { timeout: 24 hours }); if (!approval) { // Timeout expired - escalate or auto-reject await step.reportError(Approval timeout - escalating to manager); throw new Error(Approval timeout); }工作流与 Agent 的双向通信细节可进一步阅读 Workflows 集成文档。测试侧packages/agents/src/tests/workflow-integration.test.ts 通过向工作流注入{ approved: true }/{ approved: false, reason: Budget exceeded }事件验证了批准后恢复与拒绝后WorkflowRejectedError报错两条路径。使用needsApproval实现 AI 工具审批在 AI 聊天 Agent 场景中cloudflare/ai-chat与 AI SDK 的needsApproval选项可以让工具调用在真正执行前暂停等待用户批准或拒绝。服务端定义在工具定义中加入needsApproval可传布尔值总是审批或函数条件审批import { AIChatAgent } from cloudflare/ai-chat; import { createWorkersAI } from workers-ai-provider; import { streamText, tool, convertToModelMessages, stepCountIs } from ai; import { z } from zod; export class MyAgent extends AIChatAgent { async onChatMessage() { const workersai createWorkersAI({ binding: this.env.AI }); const result streamText({ model: workersai(cf/moonshotai/kimi-k2.7-code), messages: await convertToModelMessages(this.messages), tools: { // Tool with conditional approval processPayment: tool({ description: Process a payment, inputSchema: z.object({ amount: z.number(), recipient: z.string() }), // Approval required for amounts over $100 needsApproval: async ({ amount }) amount 100, execute: async ({ amount, recipient }) { return await chargeCard(amount, recipient); } }), // Tool that always requires approval deleteAccount: tool({ description: Delete a user account, inputSchema: z.object({ userId: z.string() }), needsApproval: true, execute: async ({ userId }) { return await deleteUser(userId); } }), // Tool that executes automatically (no approval) getWeather: tool({ description: Get weather for a city, inputSchema: z.object({ city: z.string() }), execute: async ({ city }) fetchWeather(city) }) }, stopWhen: stepCountIs(5) }); return result.toUIMessageStreamResponse(); } }inputSchema接受 AI SDK 的灵活 schema 格式并不局限于 Zod也可以使用 Valibot、Standard JSON Schema 兼容的 schema或用jsonSchema()包裹的原始 JSON Schema。参见 使用 Valibot 或其他 schema 库。在仓库自带的 guides/human-in-the-loop 完整示例中tools.ts 将三种工具模式集中展示getWeatherInformationneedsApproval: true的服务端审批工具、getLocalTime无execute的客户端工具、getLocalNews完全自动的服务端工具服务端入口 server.ts 则用streamTextisStepCount(5)一次管理全部工具生命周期。客户端处理审批使用useAgentChat的addToolApprovalResponse响应用户的批准/拒绝操作并渲染不同的工具状态等待审批 / 已拒绝 / 已完成import { useAgent } from agents/react; import { useAgentChat } from cloudflare/ai-chat/react; import { isToolUIPart, getToolName } from ai; function Chat() { const agent useAgent({ agent: MyAgent }); const { messages, sendMessage, addToolApprovalResponse } useAgentChat({ agent }); return ( div {messages.map((message) ( div key{message.id} {message.parts?.map((part, i) { if (part.type text) { return p key{i}{part.text}/p; } if (isToolUIPart(part)) { // Tool waiting for approval if (approval in part part.state approval-requested) { const approvalId part.approval?.id; return ( div key{part.toolCallId} classNameapproval-card p Approve strong{getToolName(part)}/strong with{ } {JSON.stringify(part.input)}? /p button onClick{() addToolApprovalResponse({ id: approvalId, approved: true }) } Approve /button button onClick{() addToolApprovalResponse({ id: approvalId, approved: false }) } Reject /button /div ); } // Tool was denied if (part.state output-denied) { return ( div key{part.toolCallId}{getToolName(part)}: Denied/div ); } // Tool completed if (part.state output-available) { return ( div key{part.toolCallId} {getToolName(part)}: {JSON.stringify(part.output)} /div ); } } return null; })} /div ))} /div ); }用addToolOutput提供自定义拒绝原因当用户拒绝工具时addToolApprovalResponse({ id, approved: false })会把工具状态置为output-denied并附带一句通用的 Tool execution denied. 消息。如果你希望把更具体的拒绝原因传给 LLM改用addToolOutput并设置state: output-errorconst { addToolOutput } useAgentChat({ agent }); // Reject with a custom error message addToolOutput({ toolCallId: part.toolCallId, state: output-error, errorText: User declined: insufficient budget for this quarter });这会以自定义错误文本向 LLM 发送tool_result让它能做出恰当的回应例如提出替代方案、追问澄清问题。addToolOutput同样适用于approval-requested与approval-responded状态的工具并非只限于input-available。两条路径的关键差异在于是否自动续跑对话addToolApprovalResponseapproved: false在autoContinueAfterToolResult开启时默认开启会自动继续对话让 LLM 看到拒绝结果并自然回应addToolOutput配合state: output-error不会自动继续给你完全的控制权——若希望 LLM 回应该错误需在之后自行调用sendMessage()。完整示例见 guides/human-in-the-loop。等待人工时如何扛住重启Durable Object 随时可能被驱逐部署、不活动超时、资源限制即使会话正停在一个审批提示或客户端工具调用上也是如此。Durable 的chatRecovery始终开启。SDK 能识别出这样的 turn 是正在等待人类而不是卡住因此不会将其封存seal当交互处于 pending 状态时no-progress 窗口、attempt cap、maxRecoveryWork与shouldKeepRecovering全部暂停生效。恢复机制会把 turn 停放park而不是判失败用户后续的审批或tool_result会通过正常续跑路径恢复对话。这意味着用户花几分钟回应一个被部署打断的提示时不会看到莫须有的 session interrupted 错误。需要注意这一保护只适用于只有客户端能解决的交互——即approval-requested部分以及客户端工具无服务端execute的input-available部分。如果一个服务端工具的execute()在执行中途被杀那是一个真正意义上的孤儿调用会走常规的 transcript 修复路径恢复。客户端工具执行onToolCall对于需要浏览器 API地理位置、摄像头、剪贴板或用户交互的工具在服务端定义工具但不提供execute函数由客户端通过onToolCall处理执行。服务端export class MyAgent extends AIChatAgent { async onChatMessage() { const workersai createWorkersAI({ binding: this.env.AI }); const result streamText({ model: workersai(cf/moonshotai/kimi-k2.7-code), messages: await convertToModelMessages(this.messages), tools: { // No execute function - client handles via onToolCall getUserLocation: tool({ description: Get the users current location from their browser, inputSchema: z.object({}) }) }, stopWhen: stepCountIs(3) }); return result.toUIMessageStreamResponse(); } }客户端const { messages, sendMessage } useAgentChat({ agent, onToolCall: async ({ toolCall, addToolOutput }) { if (toolCall.toolName getUserLocation) { const position await new Promise((resolve, reject) { navigator.geolocation.getCurrentPosition(resolve, reject); }); addToolOutput({ toolCallId: toolCall.toolCallId, output: { lat: position.coords.latitude, lng: position.coords.longitude } }); } } });服务端通过CF_AGENT_TOOL_RESULT收到工具输出。当stopWhen允许下一步时对话会自动继续让 LLM 在同一轮内针对位置数据做出回应。OpenAI Agents SDK 模式如果正在使用 OpenAI Agents SDK同样通过needsApproval函数实现条件审批import { Agent as CloudflareAgent } from agents; import { Agent as OpenAIAgent, tool, run } from openai/agents; import { z } from zod; export class WeatherAgent extends CloudflareAgentEnv { async processQuery(query: string) { const weatherTool tool({ name: get_weather, description: Get weather for a location, parameters: z.object({ location: z.string() }), // Conditional approval - only for certain locations needsApproval: async (_context, { location }) { return location San Francisco; // Require approval for SF }, execute: async ({ location }) { const conditions [sunny, cloudy, rainy]; return conditions[Math.floor(Math.random() * conditions.length)]; } }); const openaiAgent new OpenAIAgent({ name: Weather assistant, instructions: Help the user check the weather., tools: [weatherTool] }); return run(openaiAgent, query); } }完整示例见 openai-sdk/human-in-the-loop。MCP Elicitation无状态的结构化输入Stateless Elicitation基于多轮往返Multi-Round Trip RequestsMRTR。处理器返回inputRequired(...)客户端收集所需输入后在 SDK 管理的状态下重试。用户回应的过程中没有任何 Worker 保持挂起import { McpServer, acceptedContent, inputRequired } from modelcontextprotocol/server; import { createMcpHandler } from agents/mcp/server; import { z } from zod; function createServer() { const server new McpServer({ name: my-server, version: 1.0.0 }); server.registerTool( ask-name, { inputSchema: z.object({}) }, async (_args, context) { const answer acceptedContent( context.mcpReq.inputResponses, name, z.object({ name: z.string() }) ); if (!answer) { return inputRequired({ inputRequests: { name: inputRequired.elicit({ message: What is your name?, requestedSchema: { type: object, properties: { name: { type: string } }, required: [name] } }) } }); } return { content: [{ type: text, text: Hello ${answer.name} }] }; } ); return server; } export default { fetch(request, env, ctx) { return createMcpHandler(createServer, { legacy: reject })( request, env, ctx ); } } satisfies ExportedHandler;MCP 客户端会渲染请求中携带的 JSON Schema 表单从应用视角看原始操作保持 pending而 SDK 在后台完成多轮往返。参见 Stateless Elicitation 示例。Legacy Elicitation则是既有 Legacy 部署中的模式通过有状态传输推送elicitation/create请求。当仍在使用McpAgent或createLegacyMcpHandlerWorkerTransport时参考明确标记为遗留的 examples/mcp-elicitation 示例。审批的状态模式把待审批项记录在 Agent state 中便于 UI 渲染与持久化type PendingApproval { id: string; workflowId?: string; type: expense | publish | delete; description: string; amount?: number; requestedBy: string; requestedAt: number; expiresAt?: number; }; type ApprovalRecord { id: string; approvalId: string; decision: approved | rejected; decidedBy: string; decidedAt: number; reason?: string; }; type ApprovalState { pending: PendingApproval[]; history: ApprovalRecord[]; };多人审批模式对于需要多个审批人的敏感操作通过计数 追加记录的方案实现达到阈值才放行type MultiApproval { id: string; requiredApprovals: number; // e.g., 2 currentApprovals: Array{ userId: string; approvedAt: number; }; rejections: Array{ userId: string; rejectedAt: number; reason: string; }; }; callable() async approveMulti(approvalId: string, userId: string): Promiseboolean { const approval this.state.pending.find(p p.id approvalId); if (!approval) throw new Error(Approval not found); // Add this users approval approval.currentApprovals.push({ userId, approvedAt: Date.now() }); // Check if we have enough approvals if (approval.currentApprovals.length approval.requiredApprovals) { // Execute the approved action await this.executeApprovedAction(approval); return true; } this.setState({ ...this.state }); return false; // Still waiting for more approvals }该模式可与前文onWorkflowProgress回调联动工作流上报pending进度时写入pendingApprovals审批通过后再从列表中移除形成一条完整的提交 → 展示 → 审批 → 清理状态闭环。超时与升级设置审批超时waitForApproval接受人类可读的时长字符串const approval await this.waitForApproval(step, { timeout: 24 hours });用定时任务做升级结合schedule()可以安排催办与升级提醒。下面的submitForApproval在提交后 4 小时触发sendReminder催办24 小时触发escalateApproval升级callable() async submitForApproval(request: ApprovalRequest): Promisestring { const approvalId crypto.randomUUID(); // Add to pending this.setState({ ...this.state, pending: [...this.state.pending, { id: approvalId, ...request }] }); // Schedule reminder after 4 hours await this.schedule( Date.now() 4 * 60 * 60 * 1000, sendReminder, { approvalId } ); // Schedule escalation after 24 hours await this.schedule( Date.now() 24 * 60 * 60 * 1000, escalateApproval, { approvalId } ); return approvalId; }完整示例汇总模式位置说明Workflow approvalexamples/workflows带审批闸门的多步骤任务处理AIChatAgent toolsguides/human-in-the-loop用needsApprovalonToolCall的聊天工具审批OpenAI Agents SDKopenai-sdk/human-in-the-loop带弹窗的条件工具审批Stateless Elicitationexamples/mcp-elicitation-mrtr无状态多轮输入Legacy Elicitationexamples/mcp-elicitation有状态推送式输入请求延伸阅读Workflows 文档waitForApproval()、approveWorkflow()、rejectWorkflow()的完整 API 说明MCP Servers 文档inputRequired()与遗留elicitInput()Callable Methods 文档callable()装饰器如何把审批方法暴露为 RPC 端点Client Tools 与自动续跑文档客户端工具执行与CF_AGENT_TOOL_RESULT续跑机制的细节。【免费下载链接】agentsBuild and deploy AI Agents on Cloudflare项目地址: https://gitcode.com/GitHub_Trending/agents1/agents创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考