ADK Workflows 人机协同实战：用 RequestInput 中断事件构建「批准 / 驳回 / 修订」循环工作流

ADK Workflows 人机协同实战用 RequestInput 中断事件构建「批准 / 驳回 / 修订」循环工作流【免费下载链接】adk-pythonAn open-source, code-first Python toolkit for building, evaluating, and deploying sophisticated AI agents with flexibility and control.项目地址: https://gitcode.com/GitHub_Trending/ad/adk-python本文基于 ADK Python 仓库中的官方示例文档 request_input 示例 展开讲解如何在 ADK Workflows 中利用RequestInput事件实现 Human-in-the-LoopHITL工作流以一个「客服邮件草稿人工审核」场景为例演示如何让工作流执行到某节点时挂起、向人类请求输入并根据人类返回的approve/reject/ 自定义反馈三条路由分别完成、终止或回环给 AI 修订。读完本文你将掌握RequestInput事件的字段含义、节点 yield 的规范化机制、底层adk_request_input伪函数调用伪 FunctionCall协议以及如何在会话 JSON 中验证整个中断—恢复流程。场景与整体结构示例实现位于 contributing/samples/workflows/request_input/agent.py描述的是一个客服场景LLM 代理节点draft_email针对客户投诉起草一封回复邮件工作流随后挂起执行提示人类用户审核request_human_review节点根据人类的输入approve、reject或自定义修改反馈工作流要么完成发送邮件、要么中止驳回要么携带反馈回到 AI 重新起草。原始文档指出这种模式对于「AI 的动作在执行前需要人类校验」的任务至关重要crucial。README 给出的完整流程图为可以注意到两个结构特点线性主干 单一分叉点START到handle_human_review是顺序执行的链唯一的条件路由发生在handle_human_review之后的三条边revise/approved/rejected回环边cyclerevise路由指回draft_email使工作流形成环。这也是RequestInput场景中最有价值的形态——人类反馈可以驱动任意多轮的 AI 修订。示例推荐的三条典型用户输入来自 README 的 Sample Inputs 一节为My phone battery drains too fastI never received my orderThe software crashes when I open the settings仓库中附带了两份可直接用于回归验证的会话测试文件tests/phone_broke.json首轮反馈「shorter」触发修订、二轮approve触发发送和 tests/phone_broke_reject.json直接reject触发驳回。完整示例代码解析下面是 agent.py 的完整实现省略了 Apache 2.0 许可头配合逐段说明from google.adk import Agent from google.adk import Event from google.adk import Workflow from google.adk.events import RequestInput def process_input(node_input: str): Takes the initial customer complaint as input and sets it in the state. yield Event(state{complaint: node_input, feedback: }) draft_email Agent( namedraft_email, instruction Please write a polite, helpful response email to the following customer complaint: {complaint} If there is any feedback from the manager to revise the draft, please incorporate it: {feedback?} , output_keydraft, ) def request_human_review(draft: str): yield RequestInput( message( Please review the following draft email and provide approve, f reject, or feedback to revise.\n\n---\n{draft}\n--- ), ) def handle_human_review(node_input: str): if node_input reject: yield Event(routerejected) elif node_input approve: yield Event(routeapproved) else: yield Event(state{feedback: node_input}, routerevise) def reject_email(): yield Event(messageDraft rejected.) def send_email(draft: str): yield Event(messageDraft approved and sent successfully.) root_agent Workflow( namerequest_input, edges[ ( START, process_input, draft_email, request_human_review, handle_human_review, ), ( handle_human_review, { revise: draft_email, approved: send_email, rejected: reject_email, }, ), ], )各节点的职责拆解如下1. process_input把用户输入写入 stateprocess_input是工作流的第一个函数节点它接收START传入的用户输入即最初的客户投诉文本并通过Event(state{...})把complaint写入会话 state同时把feedback初始化为空字符串。初始化feedback为空串很关键——它保证了第一轮起草时 instruction 模板中的{feedback?}不会因键缺失而报错。2. draft_emailLLM 代理节点与状态变量注入draft_email是一个AgentLLM 代理节点其要点有三个instruction 模板引用 state{complaint}会取 state 中的complaint键{feedback?}中的?表示可选变量——键不存在时按空处理而不是抛错。修订回环时handle_human_review写入的新feedback会被下一轮模板自动拾取这是整个「反馈驱动修订」机制的数据通路output_keydraft把 LLM 的输出写入 state 的draft键使后续节点能通过参数名引用它节点即 Agent在 Workflow 中普通Agent可以直接作为节点参与边edge的编排不需要额外包装。3. request_human_reviewyield RequestInput 挂起工作流这是 HITL 的核心动作。README 第一步指出的就是从一个节点 yield 一个RequestInput事件来挂起工作流并向用户请求输入。该节点接收draft: str参数由上游draft_email的output_key提供将草稿原文嵌入提示消息中然后from google.adk.events import RequestInput def request_human_review(draft: str): yield RequestInput( messagePlease review the draft..., )注意这里 yield 的是RequestInput对象本身而不是Event。节点执行到yield后工作流即挂起等待外部CLI、Web UI 或测试文件中的用户事件回填输入。4. handle_human_review用 node_input 决定路由README 第二步指出下一个节点会把人类的输入作为参数node_input接收你可以用它决定下一条路由。handle_human_review是request_human_review之后唯一的后继节点人类响应会作为它的node_input传入def handle_human_review(node_input: str): if node_input reject: yield Event(routerejected) elif node_input approve: yield Event(routeapproved) else: yield Event(state{feedback: node_input}, routerevise)三个分支分别对应 README 流程图中的三条边node_input approve→routeapproved→ 走send_emailnode_input reject→routerejected→ 走reject_email其他任意文本 → 视为修改意见先写入state[feedback]供下一轮draft_email的 instruction 使用再routerevise指回draft_email。这里体现了 README 第三步的设计在边中处理不同的路由包括为修订回环looping back建模。路由字典与回环边一起定义在Workflow.edges中Workflow( namerequest_input, edges[ (START, process_input, draft_email, request_human_review, handle_human_review), (handle_human_review, {revise: draft_email, approved: send_email}), ], )示例完整代码中路由字典还额外包含rejected: reject_email一条。Event的route动作字段会告诉调度器「本节点执行完毕后沿哪条边继续」这是 ADK Workflows 条件分叉的标准机制。RequestInput 事件字段详解RequestInput的完整定义在 src/google/adk/events/request_input.py是一个 Pydantic 模型共四个字段字段类型 / 默认值说明interrupt_idstr默认由platform_uuid.new_uuid自动生成中断标识通常对应一个函数调用 ID用于把「中断请求」与「用户响应」一一匹配。源码文档注明在循环迭代如驳回/重试循环中复用同一个interrupt_id是被支持的——框架按计数匹配函数调用与响应但出于事件日志可读性考虑仍建议每次迭代使用唯一 IDmessageOptional[str]默认None展示给用户的提示信息示例中就是「请审核草稿并给出 approve / reject / 修改意见」payloadOptional[Any]默认None自定义载荷供恢复resume阶段携带额外上下文response_schemaOptional[SchemaType]默认None即Any期望的响应结构可接受 Python 类型如 PydanticBaseModel类、泛型别名如list[str]或原始 JSON Schema dict。本示例不设置该字段因为人类的响应就是自由文本approve / reject / 意见理解这四个字段后可以推断该示例选择了最轻量的用法只设置message让响应保持自由文本如果你的审核场景需要结构化结果例如强制返回{verdict: approve|reject, note: str}可以为request_human_review节点补上response_schema。底层机制RequestInput 如何变成「中断」从源码结构看yield RequestInput到「工作流挂起」之间经过了明确的规范化与协议转换分三层节点层yield 的规范化节点执行入口在 src/google/adk/workflow/_base_node.py 的BaseNode.run中。它把_run_impl中 yield 的任意对象统一规范化为Event规则如下源码 docstring 原文归纳None→ 跳过Event→ 直接透传RequestInput→转换为中断 Eventinterrupt Event其他值 → 包装为Event(outputvalue)。即request_human_review节点里那行yield RequestInput(...)在框架内部自动完成了到中断事件的转换开发者不需要手工构造Event。对于 LLM 代理节点_llm_agent_wrapper包装的Agentsrc/google/adk/workflow/_function_node.py 同样把RequestInput列入直通类型pass-through types保证函数节点与代理节点的中断行为一致。协议层adk_request_input 伪函数调用转换逻辑在 src/google/adk/workflow/utils/_workflow_hitl_utils.py 的create_request_input_event中它把RequestInput的字段interrupt_id、payload、message以及经 JSON Schema 序列化后的response_schema装进一个FunctionCall函数名固定为常量REQUEST_INPUT_FUNCTION_CALL_NAME adk_request_input并把interrupt_id写入事件的long_running_tool_ids。也就是说在事件日志里一次人工输入请求就表现为一条 role 为 model、携带adk_request_input函数调用的事件且被标记为长时运行工具。这也解释了为什么 HITL 在 ADK 里与长时运行工具long-running tools共享同一套「调用—暂停—响应—恢复」的会话协议同一常量在 LLM 主流程src/google/adk/flows/llm_flows/functions.py、CLIsrc/google/adk/cli/cli.py 中的_REQUEST_INPUT adk_request_input以及测试运行器src/google/adk/cli/agent_test_runner.py中反复出现说明 Web UI、CLI 和自动化测试都是通过识别这个名字来完成中断渲染与响应回填的。用户侧的响应则由create_request_input_response同文件 L95-L115构造为FunctionResponse其id必须回填中断时的interrupt_id。运行层node_input 的来源节点被再次调度时执行循环在 src/google/adk/workflow/_node_runner.py 中调用node.run(ctxctx, node_inputnode_input)。对于handle_human_review这样的节点node_input就是外部回填的用户响应值approve/reject/ 自定义文本它随后作为位置参数绑定到函数节点签名上的形参。从函数节点的参数解析机制看request_human_review(draft: str)的draft参数则来自上游draft_email的output_key所写入的 state——两类输入state 数据 vs 人类响应在节点入口处被统一归一为node_input语义。用会话 JSON 验证完整的中断—恢复流程tests/phone_broke.json 是一份完整的会话快照覆盖了「首轮修订 二轮批准」的两次中断逐事件拆解如下nodeInfo.path中的n是同一节点的运行序号回环后序号递增事件 ID作者关键内容说明e-1user文本phone broke客户投诉作为初始输入进入STARTe-2request_inputstateDelta: {complaint: phone broke, feedback: }路径process_input1投诉写入 statee-3draft_email生成客服草稿stateDelta.draft写入路径draft_email1第一轮起草outputFor指向本节点e-4request_inputfunctionCallnameadk_request_inputidfc-1args 含message内嵌草稿全文事件携带longRunningToolIds: [fc-1]工作流在此挂起即RequestInput的协议形态e-5userfunctionResponseidfc-1response.result shorter人类第一轮响应给出修改意见e-6request_inputactions: {route: revise, stateDelta: {feedback: shorter}}路径handle_human_review1分叉节点把意见写入 state 并路由回环e-7draft_email生成更简短的第二版草稿路径draft_email2回环后的第二轮起草instruction 中的{feedback?}已取到shortere-8request_inputadk_request_input第二次调用idfc-2路径request_human_review2第二次挂起e-9userfunctionResponseidfc-2response.result approve人类第二轮响应批准e-10request_inputactions: {route: approved}路径handle_human_review2路由到send_emaile-11request_input文本Draft approved and sent successfully.路径send_email1工作流正常结束几个值得注意的实现细节两次中断使用不同的 interruptIdfc-1 / fc-2与request_input.py中「每次迭代建议使用唯一 ID」的注释一致1/2的节点路径序号让事件日志中可以清楚区分回环的前后两轮feedback的流转完全走 statee-6 的stateDelta写入feedback: shorter最终 state 快照文件末尾state段保留了complaint、draft、feedback三个键draft为最后一版草稿挂起事件与 LLM 输出事件在协议上是同构的e-3/e-7 是常规 model 输出e-4/e-8 是携带长时运行工具标记的中断调用二者都在同一invocationIdi-1内恢复时不需要切换会话上下文。配套的 tests/phone_broke_reject.json 则覆盖reject分支两份文件与 agent.py 一起构成了该示例的端到端回归验证这些测试文件由仓库的示例测试框架统一执行参见 tests/unittests/test_samples.py。可复用的 HITL 工作流设计模式从该示例及其文档 README.md可以提炼出三条通用设计要点「请求」与「裁决」分离成两个节点request_human_review只负责 yieldRequestInput并嵌入需要审核的上下文这里是草稿全文handle_human_review只负责解析响应、写 state、发路由。这样中断提示的文案与路由逻辑各自独立演进也符合 Workflow 节点单一职责的结构。反馈通过 state 回灌 LLM 节点修订回环不靠节点间直接传参而是「反馈写 state → 下一轮 instruction 模板读取 state」。只要 LLM 节点的 instruction 里预留了{feedback?}这样的占位回环即可携带任意轮次的累积修改意见如果需要区分不同轮次的意见可以把feedback设计为追加式例如覆盖前拼接历史意见。三条路由是完备的最小集approved继续执行下游动作、rejected显式终止并给出终止消息如reject_email的Draft rejected.、其余输入一律按修改意见回环。这个「二元裁决 兜底回环」的划分避免了把人类任意输入误判为终态。小结本示例展示了 ADK Workflows 中 Human-in-the-Loop 的完整闭环节点 yield RequestInput 事件挂起工作流 → 框架将其转换为adk_request_input长时运行工具调用见 _workflow_hitl_utils.py→ 用户以FunctionResponse回填approve/reject/ 修改意见 → 分叉节点依据node_input发出approved/rejected/revise路由其中revise经 state 回灌反馈形成修订回环。相关资源一览示例文档contributing/samples/workflows/request_input/README.md示例实现contributing/samples/workflows/request_input/agent.py会话测试数据tests/phone_broke.json、tests/phone_broke_reject.json事件定义src/google/adk/events/request_input.py节点规范化src/google/adk/workflow/_base_node.pyHITL 工具函数src/google/adk/workflow/utils/_workflow_hitl_utils.py同一套机制也适用于工作流之外的 LLM 主流程adk_request_input在 flows/llm_flows 中同样生效因此在 LLM Agent 场景里请求人工确认时可以复用本文描述的「中断—响应—恢复」协议心智模型。【免费下载链接】adk-pythonAn open-source, code-first Python toolkit for building, evaluating, and deploying sophisticated AI agents with flexibility and control.项目地址: https://gitcode.com/GitHub_Trending/ad/adk-python创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考