企业官网建设流程全解析

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-Loop（HITL）工作流：以一个「客服邮件草稿人工审核」场景为例，演示如何让工作流执行到某节点时挂起、向人类请求输入，并根据人类返回的approve/reject/ 自定义反馈三条路由分别完成、终止或回环给 AI 修订。读完本文，你将掌握RequestInput事件的字段含义、节点 yield 的规范化机制、底层adk_request_input伪函数调用（伪 FunctionCall）协议，以及如何在会话 JSON 中验证整个中断—恢复流程。

场景与整体结构

示例实现位于 contributing/samples/workflows/request_input/agent.py，描述的是一个客服场景：

1. LLM 代理节点draft_email针对客户投诉起草一封回复邮件；
2. 工作流随后挂起执行，提示人类用户审核（request_human_review节点）；
3. 根据人类的输入（approve、reject或自定义修改反馈），工作流要么完成（发送邮件）、要么中止（驳回），要么携带反馈回到 AI 重新起草。

原始文档指出，这种模式对于「AI 的动作在执行前需要人类校验」的任务至关重要（crucial）。README 给出的完整流程图为：

可以注意到两个结构特点：

• 线性主干 + 单一分叉点：START到handle_human_review是顺序执行的链，唯一的条件路由发生在handle_human_review之后的三条边（revise/approved/rejected）；
• 回环边（cycle）：revise路由指回draft_email，使工作流形成环。这也是RequestInput场景中最有价值的形态——人类反馈可以驱动任意多轮的 AI 修订。

示例推荐的三条典型用户输入（来自 README 的 Sample Inputs 一节）为：

• My phone battery drains too fast
• I never received my order
• The 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( name="draft_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_key="draft", ) 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(route="rejected") elif node_input == "approve": yield Event(route="approved") else: yield Event(state={"feedback": node_input}, route="revise") def reject_email(): yield Event(message="Draft rejected.") def send_email(draft: str): yield Event(message="Draft approved and sent successfully.") root_agent = Workflow( name="request_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：把用户输入写入 state

process_input是工作流的第一个函数节点，它接收START传入的用户输入（即最初的客户投诉文本），并通过Event(state={...})把complaint写入会话 state，同时把feedback初始化为空字符串。初始化feedback为空串很关键——它保证了第一轮起草时 instruction 模板中的{feedback?}不会因键缺失而报错。

2. draft_email：LLM 代理节点与状态变量注入

draft_email是一个Agent（LLM 代理）节点，其要点有三个：

• instruction 模板引用 state："{complaint}"会取 state 中的complaint键；"{feedback?}"中的?表示可选变量——键不存在时按空处理，而不是抛错。修订回环时handle_human_review写入的新feedback会被下一轮模板自动拾取，这是整个「反馈驱动修订」机制的数据通路；
• output_key="draft"：把 LLM 的输出写入 state 的draft键，使后续节点能通过参数名引用它；
• 节点即 Agent：在 Workflow 中，普通Agent可以直接作为节点参与边（edge）的编排，不需要额外包装。

3. request_human_review：yield 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( message="Please 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(route="rejected") elif node_input == "approve": yield Event(route="approved") else: yield Event(state={"feedback": node_input}, route="revise")

三个分支分别对应 README 流程图中的三条边：

• node_input == "approve"→route="approved"→ 走send_email；
• node_input == "reject"→route="rejected"→ 走reject_email；
• 其他任意文本 → 视为修改意见，先写入state["feedback"]（供下一轮draft_email的 instruction 使用），再route="revise"指回draft_email。

这里体现了 README 第三步的设计：在边中处理不同的路由，包括为修订回环（looping back）建模。路由字典与回环边一起定义在Workflow.edges中：

Workflow( name="request_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 模型，共四个字段：

理解这四个字段后，可以推断该示例选择了最轻量的用法：只设置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→转换为中断 Event（interrupt Event）；
• 其他值 → 包装为Event(output=value)。

即request_human_review节点里那行yield RequestInput(...)在框架内部自动完成了到中断事件的转换，开发者不需要手工构造Event。对于 LLM 代理节点（_llm_agent_wrapper包装的Agent），src/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）、CLI（src/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(ctx=ctx, node_input=node_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是同一节点的运行序号，回环后序号递增）：

几个值得注意的实现细节：

1. 两次中断使用不同的 interruptId（fc-1 / fc-2），与request_input.py中「每次迭代建议使用唯一 ID」的注释一致；@1/@2的节点路径序号让事件日志中可以清楚区分回环的前后两轮；
2. feedback的流转完全走 state：e-6 的stateDelta写入feedback: "shorter"，最终 state 快照（文件末尾"state"段）保留了complaint、draft、feedback三个键，draft为最后一版草稿；
3. 挂起事件与 LLM 输出事件在协议上是同构的：e-3/e-7 是常规 model 输出，e-4/e-8 是携带长时运行工具标记的中断调用，二者都在同一invocationId（i-1）内，恢复时不需要切换会话上下文。

配套的 tests/phone_broke_reject.json 则覆盖reject分支，两份文件与 agent.py 一起构成了该示例的端到端回归验证（这些测试文件由仓库的示例测试框架统一执行，参见 tests/unittests/test_samples.py）。

可复用的 HITL 工作流设计模式

从该示例（及其文档 README.md）可以提炼出三条通用设计要点：

1. 「请求」与「裁决」分离成两个节点：request_human_review只负责 yieldRequestInput并嵌入需要审核的上下文（这里是草稿全文）；handle_human_review只负责解析响应、写 state、发路由。这样中断提示的文案与路由逻辑各自独立演进，也符合 Workflow 节点单一职责的结构。
2. 反馈通过 state 回灌 LLM 节点：修订回环不靠节点间直接传参，而是「反馈写 state → 下一轮 instruction 模板读取 state」。只要 LLM 节点的 instruction 里预留了{feedback?}这样的占位，回环即可携带任意轮次的累积修改意见；如果需要区分不同轮次的意见，可以把feedback设计为追加式（例如覆盖前拼接历史意见）。
3. 三条路由是完备的最小集：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.py
• HITL 工具函数：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