ADK Python Workflow 人机协作实战request_input_rerun 单节点审批模式深度解析【免费下载链接】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_rerun讲解如何在 ADK Workflows 中用RequestInput事件配合node(rerun_on_resumeTrue)装饰器把向人类要输入和处理人类输入合并进同一个节点。读完后你将掌握两种 Human-in-the-Loop人机协作模式的区别、可直接运行的完整示例代码、事件日志级别的执行流程验证以及该机制在框架源码中的底层实现原理。两种模式对比request_input 与 request_input_rerunADK 提供了两种在 Workflow 中暂停并等待人工输入的模式二者模拟的都是同一个客服场景AI 起草一封回复客户投诉的邮件人工审核后决定放行、驳回或退回修改。核心差异在于恢复执行时人类输入是如何被消费的。维度标准模式request_input重执行模式request_input_rerun暂停方式某节点 yieldRequestInput事件后暂停同样 yieldRequestInput事件恢复后输入去向自动作为下一个节点的入参node_input通过执行Context的resume_inputs被同一个节点读取节点数量需要两个节点一个请求输入一个处理输入单个节点自包含请求 处理边定义复杂度链路多一跳更简洁从源码结构看标准模式的实现位于 request_input 示例request_human_review节点负责 yieldRequestInput工作流恢复后由框架把用户输入作为参数喂给后续节点handle_human_review(node_input: str)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)而重执行模式把这两个职责合并进一个节点该节点被node(rerun_on_resumeTrue)装饰恢复时框架重新运行这个节点本身节点内部通过ctx.resume_inputs判断自己是首次执行应暂停还是被恢复应处理输入。这样请求与处理逻辑保持在同一处状态更内聚边定义也更短。完整示例客服邮件审批工作流完整代码见 agent.py整体结构如下保留 Apache 2.0 版权头省略from google.adk import Agent from google.adk import Context from google.adk import Event from google.adk import Workflow from google.adk.events import RequestInput from google.adk.workflow import node 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, ) node(rerun_on_resumeTrue) def human_review(draft: str, ctx: Context): resume_input ctx.resume_inputs.get(human_review) if not resume_input: yield RequestInput( interrupt_idhuman_review, message( Please review the following draft email and provide approve, f reject, or feedback to revise.\n\n---\n{draft}\n--- ), ) return if resume_input reject: yield Event(routerejected) elif resume_input approve: yield Event(routeapproved) else: yield Event(state{feedback: resume_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_rerun, edges[ (START, process_input, draft_email, human_review), ( human_review, { revise: draft_email, approved: send_email, rejected: reject_email, }, ), ], )工作流图继承自示例 README各节点职责process_input把用户输入的投诉文本写入共享 statecomplaint并预置空feedback。draft_email一个 LLMAgent节点。instruction 中的{complaint}从 state 插值{feedback?}带?后缀表示可选——仅当 state 中存在feedback时才拼入提示词output_keydraft把模型输出写回 state 的draft键供下游节点以参数形式引用。human_review核心的请求 处理双职责节点详见下文实现要点。reject_email/send_email两条终态分支分别输出草稿被驳回与草稿已发送。官方 README 给出的典型测试输入包括The delivery was a week late、I received the wrong item、My account was charged twice。实现要点四个关键步骤以下四步完整继承自示例 README 的 How To 章节是编写任何 rerun 型审批节点的标准流程。1. 用node(rerun_on_resumeTrue)装饰节点并在签名中声明Contextfrom google.adk.workflow import node from google.adk import Context node(rerun_on_resumeTrue) def human_review(draft: str, ctx: Context): # ...rerun_on_resumeTrue告诉框架当该节点因RequestInput暂停、后续又带着用户输入恢复时不要推进到下一个节点而是重新执行本节点。函数签名必须包含 workflow 的Context才能访问resume_inputs。2. 通过ctx.resume_inputs判断是否处于恢复执行resume_input ctx.resume_inputs.get(human_review)resume_inputs是一个以interrupt_id为键的字典。示例中查找用的键human_review与 yieldRequestInput时声明的interrupt_id一致恰好也与节点同名。3. 首次执行时 yieldRequestInput并立即 return 暂停if not resume_input: yield RequestInput( interrupt_idhuman_review, messagePlease review the draft..., ) return # Important: Stop execution of this node for now注意return的作用节点首次运行到这里后停止工作流进入等待状态事件流中会挂起一个待响应的中断。4. 恢复执行时消费输入并产出路由事件if resume_input reject: yield Event(routerejected) elif resume_input approve: yield Event(routeapproved) else: yield Event(state{feedback: resume_input}, routerevise)任何非approve/reject的输入都被当作修改意见写入 state 的feedback供draft_email的{feedback?}插值并沿revise路由回退给 LLM 节点重写形成草稿—审阅—修改循环。边定义单节点带来的简化由于human_review一个节点承担了请求与处理全部逻辑边定义比标准模式更短Workflow( namerequest_input_rerun, edges[ (START, process_input, draft_email, human_review), (human_review, {revise: draft_email, approved: send_email}), ], )对比标准模式的边定义(START, process_input, draft_email, request_human_review, handle_human_review)——多出一个handle_human_review节点。注意两条边定义中{revise: ..., approved: ...}这个字典表示条件路由目标节点取决于节点 yield 的Event(route...)取值示例还额外映射了rejected: reject_email见 agent.py。事件流验证一次完整的修改后批准执行记录test 会话记录 完整回放了一次phone broke投诉的处理过程是验证该模式行为的最佳证据。11 个事件的演进如下e-1用户输入phone broke。e-2process_input节点产出 stateDelta{complaint: phone broke, feedback: }nodeInfo.path为request_input_rerun1/process_input1。e-3draft_email首次运行写出第一版草稿stateDelta.draft路径后缀draft_email1。e-4human_review首次执行对外表现为一次名为adk_request_input的函数调用参数含interruptIdfc-1、message附完整草稿正文、payload、response_schema事件携带longRunningToolIds: [fc-1]表示工作流在此挂起等待。e-5用户以functionResponse返回shorter——既非 approve 也非 reject。e-6human_review被重执行产出route: revise及stateDelta: {feedback: shorter}。e-7draft_email2同一节点的第二次执行路径中的2即执行序号基于反馈重写出一版更短的邮件。e-8human_review2再次 yieldRequestInputfc-2等待二次审阅——证明同一节点可在循环中反复中断/恢复。e-9用户返回approve。e-10节点第三次执行产出route: approved。e-11send_email输出Draft approved and sent successfully.。最终 state 中保留了complaint、最新版draft和feedback: shorter。这条记录证实了三件事RequestInput在事件层被序列化为adk_request_input函数调用调用与响应的配对按interruptId匹配rerun_on_resume节点确实被原样重执行2序号循环中断复用同一interrupt_id是被支持的——RequestInput 模型注释明确说明跨循环迭代复用同一interrupt_id是受支持的框架按计数匹配函数调用与响应但为了事件日志清晰仍建议每次迭代使用唯一 ID。源码级原理resume_inputs 如何被注入与清理结合框架源码src/google/adk可以还原该机制的调用链事件定义RequestInput是 Pydantic 模型见 request_input.py。三个关键字段interrupt_id默认自动生成 UUID显式指定可让恢复端确定性定位、payload可选的恢复载荷、message展示给用户的提示文案。Context 侧agents/context.py 中Context.resume_inputs返回dict[str, Any]文档注释说明其键为 interrupt id构造函数中初始化为空字典self._resume_inputs resume_inputs or {}因此查不到值 首次执行这一判空逻辑是安全的。工作流调度侧_workflow.py 负责在恢复时把输入路由回原节点。从源码结构看节点执行结果中的resume_inputs会被存回节点状态node_state.resume_inputs result.resume_inputs or {}约 L664恢复路径上再从节点状态取出并构造带resume_inputs的子 Context 重新执行该节点约 L678-L691重执行完成后会node_state.resume_inputs.clear()约 L810-L811保证输入只被消费一次、不会在下一次正常执行时残留命中。此外若提供了resume_inputs却找不到可恢复的执行记录框架会发出告警resume_inputs provided but no recovered executions约 L254-L256这是排查输入没被接住问题的第一信号。装饰器侧workflow/_node.py 的node装饰器文档以node(namemy_node, rerun_on_resumeTrue)为例说明该参数会覆盖内部节点的rerun_on_resume属性约 L113使节点具备恢复时原地重跑的语义。这套机制的语义是中断状态只记录哪个节点在等输入恢复时把用户输入按interrupt_id塞回该节点的Context节点自己决定下一步路由——这正是能把请求与处理合并进单节点的底层原因。选型建议与适用前提当请求输入与解释输入逻辑简单、只需一个后继节点做分发时标准request_input模式两节点更直白可参考 request_input 示例。当输入处理逻辑复杂如示例中的三路分发 写 state 回环修订或你希望审阅上下文草稿内容、判断逻辑集中在一个函数内便于维护与测试时request_input_rerun的单节点模式更内聚。适用前提节点函数签名需声明ContextRequestInput应显式指定interrupt_id以便resume_inputs精确命中revise回环在语义上允许无限次循环生产场景中可考虑在节点内对迭代次数加限。所有代码基于当前仓库的 ADK Python 实现验证示例可放入独立的 app 目录后用adk run/adk web交互式调试该方式适用于已按 ADK CLI 约定组织好agent.py的示例目录。延伸阅读示例源码contributing/samples/workflows/request_input_rerun/agent.py事件回放数据contributing/samples/workflows/request_input_rerun/tests/phone_broke.json对照示例contributing/samples/workflows/request_input/agent.py核心实现src/google/adk/events/request_input.py、src/google/adk/workflow/_node.py、src/google/adk/workflow/_workflow.py、src/google/adk/agents/context.py【免费下载链接】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），仅供参考 ADK Python Workflow 人机协作实战：request_input_rerun 单节点审批模式深度解析
ADK Python Workflow 人机协作实战request_input_rerun 单节点审批模式深度解析【免费下载链接】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_rerun讲解如何在 ADK Workflows 中用RequestInput事件配合node(rerun_on_resumeTrue)装饰器把向人类要输入和处理人类输入合并进同一个节点。读完后你将掌握两种 Human-in-the-Loop人机协作模式的区别、可直接运行的完整示例代码、事件日志级别的执行流程验证以及该机制在框架源码中的底层实现原理。两种模式对比request_input 与 request_input_rerunADK 提供了两种在 Workflow 中暂停并等待人工输入的模式二者模拟的都是同一个客服场景AI 起草一封回复客户投诉的邮件人工审核后决定放行、驳回或退回修改。核心差异在于恢复执行时人类输入是如何被消费的。维度标准模式request_input重执行模式request_input_rerun暂停方式某节点 yieldRequestInput事件后暂停同样 yieldRequestInput事件恢复后输入去向自动作为下一个节点的入参node_input通过执行Context的resume_inputs被同一个节点读取节点数量需要两个节点一个请求输入一个处理输入单个节点自包含请求 处理边定义复杂度链路多一跳更简洁从源码结构看标准模式的实现位于 request_input 示例request_human_review节点负责 yieldRequestInput工作流恢复后由框架把用户输入作为参数喂给后续节点handle_human_review(node_input: str)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)而重执行模式把这两个职责合并进一个节点该节点被node(rerun_on_resumeTrue)装饰恢复时框架重新运行这个节点本身节点内部通过ctx.resume_inputs判断自己是首次执行应暂停还是被恢复应处理输入。这样请求与处理逻辑保持在同一处状态更内聚边定义也更短。完整示例客服邮件审批工作流完整代码见 agent.py整体结构如下保留 Apache 2.0 版权头省略from google.adk import Agent from google.adk import Context from google.adk import Event from google.adk import Workflow from google.adk.events import RequestInput from google.adk.workflow import node 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, ) node(rerun_on_resumeTrue) def human_review(draft: str, ctx: Context): resume_input ctx.resume_inputs.get(human_review) if not resume_input: yield RequestInput( interrupt_idhuman_review, message( Please review the following draft email and provide approve, f reject, or feedback to revise.\n\n---\n{draft}\n--- ), ) return if resume_input reject: yield Event(routerejected) elif resume_input approve: yield Event(routeapproved) else: yield Event(state{feedback: resume_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_rerun, edges[ (START, process_input, draft_email, human_review), ( human_review, { revise: draft_email, approved: send_email, rejected: reject_email, }, ), ], )工作流图继承自示例 README各节点职责process_input把用户输入的投诉文本写入共享 statecomplaint并预置空feedback。draft_email一个 LLMAgent节点。instruction 中的{complaint}从 state 插值{feedback?}带?后缀表示可选——仅当 state 中存在feedback时才拼入提示词output_keydraft把模型输出写回 state 的draft键供下游节点以参数形式引用。human_review核心的请求 处理双职责节点详见下文实现要点。reject_email/send_email两条终态分支分别输出草稿被驳回与草稿已发送。官方 README 给出的典型测试输入包括The delivery was a week late、I received the wrong item、My account was charged twice。实现要点四个关键步骤以下四步完整继承自示例 README 的 How To 章节是编写任何 rerun 型审批节点的标准流程。1. 用node(rerun_on_resumeTrue)装饰节点并在签名中声明Contextfrom google.adk.workflow import node from google.adk import Context node(rerun_on_resumeTrue) def human_review(draft: str, ctx: Context): # ...rerun_on_resumeTrue告诉框架当该节点因RequestInput暂停、后续又带着用户输入恢复时不要推进到下一个节点而是重新执行本节点。函数签名必须包含 workflow 的Context才能访问resume_inputs。2. 通过ctx.resume_inputs判断是否处于恢复执行resume_input ctx.resume_inputs.get(human_review)resume_inputs是一个以interrupt_id为键的字典。示例中查找用的键human_review与 yieldRequestInput时声明的interrupt_id一致恰好也与节点同名。3. 首次执行时 yieldRequestInput并立即 return 暂停if not resume_input: yield RequestInput( interrupt_idhuman_review, messagePlease review the draft..., ) return # Important: Stop execution of this node for now注意return的作用节点首次运行到这里后停止工作流进入等待状态事件流中会挂起一个待响应的中断。4. 恢复执行时消费输入并产出路由事件if resume_input reject: yield Event(routerejected) elif resume_input approve: yield Event(routeapproved) else: yield Event(state{feedback: resume_input}, routerevise)任何非approve/reject的输入都被当作修改意见写入 state 的feedback供draft_email的{feedback?}插值并沿revise路由回退给 LLM 节点重写形成草稿—审阅—修改循环。边定义单节点带来的简化由于human_review一个节点承担了请求与处理全部逻辑边定义比标准模式更短Workflow( namerequest_input_rerun, edges[ (START, process_input, draft_email, human_review), (human_review, {revise: draft_email, approved: send_email}), ], )对比标准模式的边定义(START, process_input, draft_email, request_human_review, handle_human_review)——多出一个handle_human_review节点。注意两条边定义中{revise: ..., approved: ...}这个字典表示条件路由目标节点取决于节点 yield 的Event(route...)取值示例还额外映射了rejected: reject_email见 agent.py。事件流验证一次完整的修改后批准执行记录test 会话记录 完整回放了一次phone broke投诉的处理过程是验证该模式行为的最佳证据。11 个事件的演进如下e-1用户输入phone broke。e-2process_input节点产出 stateDelta{complaint: phone broke, feedback: }nodeInfo.path为request_input_rerun1/process_input1。e-3draft_email首次运行写出第一版草稿stateDelta.draft路径后缀draft_email1。e-4human_review首次执行对外表现为一次名为adk_request_input的函数调用参数含interruptIdfc-1、message附完整草稿正文、payload、response_schema事件携带longRunningToolIds: [fc-1]表示工作流在此挂起等待。e-5用户以functionResponse返回shorter——既非 approve 也非 reject。e-6human_review被重执行产出route: revise及stateDelta: {feedback: shorter}。e-7draft_email2同一节点的第二次执行路径中的2即执行序号基于反馈重写出一版更短的邮件。e-8human_review2再次 yieldRequestInputfc-2等待二次审阅——证明同一节点可在循环中反复中断/恢复。e-9用户返回approve。e-10节点第三次执行产出route: approved。e-11send_email输出Draft approved and sent successfully.。最终 state 中保留了complaint、最新版draft和feedback: shorter。这条记录证实了三件事RequestInput在事件层被序列化为adk_request_input函数调用调用与响应的配对按interruptId匹配rerun_on_resume节点确实被原样重执行2序号循环中断复用同一interrupt_id是被支持的——RequestInput 模型注释明确说明跨循环迭代复用同一interrupt_id是受支持的框架按计数匹配函数调用与响应但为了事件日志清晰仍建议每次迭代使用唯一 ID。源码级原理resume_inputs 如何被注入与清理结合框架源码src/google/adk可以还原该机制的调用链事件定义RequestInput是 Pydantic 模型见 request_input.py。三个关键字段interrupt_id默认自动生成 UUID显式指定可让恢复端确定性定位、payload可选的恢复载荷、message展示给用户的提示文案。Context 侧agents/context.py 中Context.resume_inputs返回dict[str, Any]文档注释说明其键为 interrupt id构造函数中初始化为空字典self._resume_inputs resume_inputs or {}因此查不到值 首次执行这一判空逻辑是安全的。工作流调度侧_workflow.py 负责在恢复时把输入路由回原节点。从源码结构看节点执行结果中的resume_inputs会被存回节点状态node_state.resume_inputs result.resume_inputs or {}约 L664恢复路径上再从节点状态取出并构造带resume_inputs的子 Context 重新执行该节点约 L678-L691重执行完成后会node_state.resume_inputs.clear()约 L810-L811保证输入只被消费一次、不会在下一次正常执行时残留命中。此外若提供了resume_inputs却找不到可恢复的执行记录框架会发出告警resume_inputs provided but no recovered executions约 L254-L256这是排查输入没被接住问题的第一信号。装饰器侧workflow/_node.py 的node装饰器文档以node(namemy_node, rerun_on_resumeTrue)为例说明该参数会覆盖内部节点的rerun_on_resume属性约 L113使节点具备恢复时原地重跑的语义。这套机制的语义是中断状态只记录哪个节点在等输入恢复时把用户输入按interrupt_id塞回该节点的Context节点自己决定下一步路由——这正是能把请求与处理合并进单节点的底层原因。选型建议与适用前提当请求输入与解释输入逻辑简单、只需一个后继节点做分发时标准request_input模式两节点更直白可参考 request_input 示例。当输入处理逻辑复杂如示例中的三路分发 写 state 回环修订或你希望审阅上下文草稿内容、判断逻辑集中在一个函数内便于维护与测试时request_input_rerun的单节点模式更内聚。适用前提节点函数签名需声明ContextRequestInput应显式指定interrupt_id以便resume_inputs精确命中revise回环在语义上允许无限次循环生产场景中可考虑在节点内对迭代次数加限。所有代码基于当前仓库的 ADK Python 实现验证示例可放入独立的 app 目录后用adk run/adk web交互式调试该方式适用于已按 ADK CLI 约定组织好agent.py的示例目录。延伸阅读示例源码contributing/samples/workflows/request_input_rerun/agent.py事件回放数据contributing/samples/workflows/request_input_rerun/tests/phone_broke.json对照示例contributing/samples/workflows/request_input/agent.py核心实现src/google/adk/events/request_input.py、src/google/adk/workflow/_node.py、src/google/adk/workflow/_workflow.py、src/google/adk/agents/context.py【免费下载链接】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），仅供参考