尧图网站建设 
Genkit Python 工具中断机制详解Interrupt、restart_tool 与 respond_to_interrupt 实现 Human-in-the-Loop【免费下载链接】genkitOpen-source framework for building agentic apps in JavaScript, Go, Dart, and Python, built and used in production by Google项目地址: https://gitcode.com/GitHub_Trending/ge/genkit本文以 Genkit Python SDK 的 tool-interrupts 示例为主线讲解「工具执行中途暂停、向人类请示、再继续执行」的完整机制如何通过Interrupt异常暂停工具、用restart_tool重跑工具、用respond_to_interrupt直接向对话注入结果而跳过工具执行并深入到 py/packages/genkit/src/genkit/_ai/_tools.py 中的实现原理帮助你在银行转账、下单等高风险 Agent 应用中落地人机协作human-in-the-loop检查点。场景与运行前提官方示例位于 py/samples/tool-interrupts/其 README 用三句话概括了核心能力A tool can stop and ask a human before it finishes. You see the reason, thenrestart_toolre-runs it orrespond_to_interruptinjects a result without running the tool.即工具可以在完成前停下来向人请示你看到原因后选择让restart_tool重新运行该工具或让respond_to_interrupt不运行工具、直接把一个结果注入对话。运行前提见 pyproject.tomlPython ≥ 3.10使用uv管理依赖依赖genkit、genkit-google-genai、pydantic2.10.5需要GEMINI_API_KEY环境变量示例使用gemini-flash-latest模型。运行方式继承自原 READMEexport GEMINI_API_KEYyour-api-key uv sync uv run src/main.py完整示例代码解析示例入口 src/main.py 演示了一个「财务桌面treasury desk」场景模型被要求调用转账工具工具首次执行时暂停等待人工审批。定义可中断的工具from genkit_google_genai import GoogleAI from pydantic import BaseModel, Field from genkit import Genkit, Interrupt, ToolRunContext, respond_to_interrupt, restart_tool ai Genkit(plugins[GoogleAI()], modelGoogleAI.gemini_model(gemini-flash-latest)) class TransferRequest(BaseModel): to_account: str Field(descriptionrecipient name or masked account) amount_usd: str Field(descriptionamount as a string, e.g. 250.00) memo: str ai.tool() async def request_transfer(body: TransferRequest, ctx: ToolRunContext) - dict: # 首次调用会暂停恢复后 is_resumed() 为 True工具真正执行。 if not ctx.is_resumed(): raise Interrupt({summary: fWire ${body.amount_usd} to {body.to_account} — {body.memo}}) return {status: confirmed, resumed: ctx.resumed_metadata}关键点工具签名工具函数是async最多接收 2 个参数——输入体由 Pydantic 模型TransferRequest描述SDK 会据此生成发给模型的 input schema和ToolRunContext执行上下文。这一「0–2 参数」约束在 py/packages/genkit/src/genkit/_ai/_tools.py 的参数分发逻辑match len(input_spec.args)中得到印证。暂停即抛异常工具通过raise Interrupt(metadata)暂停。Interrupt是一个可带 metadata 的异常类继承自GenkitInterrupt见 py/packages/genkit/src/genkit/_ai/_tools.pymetadata 会挂在被暂停的工具请求tool request上随响应返回若处于 OpenTelemetry span 中metadata 还会写入genkit:metadata:interrupt属性用于可观测性。恢复的判定ToolRunContext.is_resumed()依据resumed_metadata is not None判断本次执行是否为中断后的恢复执行py/packages/genkit/src/genkit/_ai/_tools.py。示例中工具「首次抛 Interrupt、恢复后才真正执行」等价于一次审批门禁人批准前 wire 汇款这个副作用根本不会发生。第一次交互restart_tool 让工具重跑response await ai.generate( promptPlease wire $250.00 to Jane Doe (account ending in 4521) for April rent., systemYou are a treasury desk. Call request_transfer to send money., tools[request_transfer], ) print(response.text) if not response.interrupts: return interrupt response.interrupts[0] print(fpaused: {interrupt.metadata}) # restart_tool人说了 yes 之后让工具重新运行。 approved await ai.generate( messagesresponse.messages, resume_restartrestart_tool(interruptinterrupt, resumed_metadata{approved: True}), tools[request_transfer], ) print(approved.text)流程是ai.generate(...)返回的响应上response.interrupts属性暴露出所有被中断的ToolRequestPart实现见 py/packages/genkit/src/genkit/_core/_model.py 与 #L690-L694每个元素携带metadata——也就是工具抛出Interrupt时附上的{summary: ...}。拿到中断句柄后restart_tool(interrupt..., resumed_metadata{approved: True})构造一个用于resume_restart的ToolRequestPart。源码见 py/packages/genkit/src/genkit/_ai/_tools.py它的行为细节值得注意默认沿用原tool_request.input若传入replace_input则替换输入并把旧输入存入 metadata 的replacedInput键resumed_metadata会被写入新 part 的metadata[resumed]不传时写True随后经ContextVar传播进工具成为ToolRunContext.resumed_metadata因此恢复执行时ctx.is_resumed()返回True、ctx.resumed_metadata即{approved: True}工具返回{status: confirmed, resumed: {approved: True}}。恢复时必须把原响应的messages与同一组tools一起回传给ai.generate以保持对话历史与工具注册一致。第二次交互respond_to_interrupt 直接注入结果declined await ai.generate( promptPlease wire $80.00 to Sam Lee (account ending in 9910) for lunch., systemYou are a treasury desk. Call request_transfer to send money., tools[request_transfer], ) if not declined.interrupts: return print(fpaused: {declined.interrupts[0].metadata}) # respond_to_interrupt 直接注入结果——拒绝汇款工具不再执行。 done await ai.generate( messagesdeclined.messages, resume_respondrespond_to_interrupt({status: declined}, interruptdeclined.interrupts[0]), tools[request_transfer], ) print(done.text)respond_to_interrupt(response, *, interrupt, metadataNone)返回一个ToolResponsePart实现见 py/packages/genkit/src/genkit/_ai/_tools.py它复用了被中断请求的ref和name把第一个位置参数作为output写入工具响应并在 part 的metadata上标记interruptResponse。把这个 part 传给generate(..., resume_respond...)后工具函数不会被再次调用{status: declined}直接作为工具结果进入模型可见的对话历史模型据此生成「已拒绝」的最终回复这正好演示了两种恢复语义的分野resume_restart 批准重跑工具resume_respond 否决/代答跳过工具执行。generate的两个恢复参数resume_respond/resume_restart是ai.generate的一等参数接受单个或列表见 py/packages/genkit/src/genkit/_ai/_prompt.pyresume_respond: ToolResponsePart | list[ToolResponsePart] | None resume_restart: ToolRequestPart | list[ToolRequestPart] | None列表形式意味着一次generate可以批量恢复多个并行中断例如模型在同一轮同时请求了多笔转账逐一审批。底层实现要点结合 py/packages/genkit/src/genkit/_ai/_tools.py 可以补充几个示例注释里没有的机制细节恢复元数据的传播链run_tool_request#L503-L531从 part metadata 读取resumed/replacedInput写入两个ContextVar_tool_resumed_metadata、_tool_original_input工具包装器在按参数个数分发调用时用这两个值构造ToolRunContext(resumed_metadata..., original_input...)#L685-L714。所以ctx.resumed_metadata与ctx.original_input分别告诉你「人批了什么」和「重跑前原输入是什么」。重启期间不允许再次中断run_tool_after_restart#L557-L593捕获恢复执行中再次抛出的Interrupt包装为FAILED_PRECONDITION错误消息形如 Tool interrupted again during restart: ...见restart_interrupt_error#L534-L554。从源码注释看这是有意约束防止恢复流程无限循环等待审批。纯中断工具define_interrupt除了「有时执行、有时中断」的普通工具外SDK 还提供define_interrupt#L797-L850注册「每次调用都抛Interrupt」的工具支持静态 metadata 或(input) - dict回调适合显式的人机检查点而本文示例属于「有时运行逻辑、有时中断」用ai.tool() 手动raise Interrupt更合适该函数的 docstring 明确给出了这一选择建议。可观测性Interrupt的 metadata 会写入当前 span 的genkit:metadata:interrupt属性恢复执行时resumed元数据会写入genkit:metadata:resumed#L685-L694便于在 trace 中复盘审批决策。选型建议restart_tool 还是 respond_to_interrupt维度restart_toolresume_restartrespond_to_interruptresume_respond语义人类批准让工具真正执行人类否决或由人代答直接注入结果工具是否执行会重新执行不会执行返回类型ToolRequestPart新的工具请求ToolResponsePart工具响应额外能力replace_input可在批准时改写输入旧输入存replacedInputmetadata可标注中断响应通道内容典型用法转账批准后发起 wire转账拒绝直接告诉模型 declined限制恢复执行中再次抛Interrupt会报FAILED_PRECONDITION注入内容成为对话历史的一部分模型据此继续小结tool-interrupts 示例py/samples/tool-interrupts/README.md用最少的代码覆盖了 Genkit Python 中断机制的完整闭环Interrupt携带 metadata 暂停工具并把原因暴露在response.interrupts上restart_tool配合resumed_metadata实现「批准后重跑」并可用is_resumed()在工具内区分首次执行与恢复执行respond_to_interrupt实现「不重跑、直接注入结果」。这套机制对转账、支付、删除数据等副作用工具尤为关键——在模型自主调用工具与真实业务动作之间插入了一道可审计的人工确认。进一步阅读可参考 py/samples/tool-interrupts/src/main.py 与 py/packages/genkit/src/genkit/_ai/_tools.py 中Interrupt、restart_tool、respond_to_interrupt、run_tool_after_restart的完整实现。【免费下载链接】genkitOpen-source framework for building agentic apps in JavaScript, Go, Dart, and Python, built and used in production by Google项目地址: https://gitcode.com/GitHub_Trending/ge/genkit创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Genkit Python 工具中断机制详解：Interrupt、restart_tool 与 respond_to_interrupt 实现 Human-in-the-Loop
Genkit Python 工具中断机制详解Interrupt、restart_tool 与 respond_to_interrupt 实现 Human-in-the-Loop【免费下载链接】genkitOpen-source framework for building agentic apps in JavaScript, Go, Dart, and Python, built and used in production by Google项目地址: https://gitcode.com/GitHub_Trending/ge/genkit本文以 Genkit Python SDK 的 tool-interrupts 示例为主线讲解「工具执行中途暂停、向人类请示、再继续执行」的完整机制如何通过Interrupt异常暂停工具、用restart_tool重跑工具、用respond_to_interrupt直接向对话注入结果而跳过工具执行并深入到 py/packages/genkit/src/genkit/_ai/_tools.py 中的实现原理帮助你在银行转账、下单等高风险 Agent 应用中落地人机协作human-in-the-loop检查点。场景与运行前提官方示例位于 py/samples/tool-interrupts/其 README 用三句话概括了核心能力A tool can stop and ask a human before it finishes. You see the reason, thenrestart_toolre-runs it orrespond_to_interruptinjects a result without running the tool.即工具可以在完成前停下来向人请示你看到原因后选择让restart_tool重新运行该工具或让respond_to_interrupt不运行工具、直接把一个结果注入对话。运行前提见 pyproject.tomlPython ≥ 3.10使用uv管理依赖依赖genkit、genkit-google-genai、pydantic2.10.5需要GEMINI_API_KEY环境变量示例使用gemini-flash-latest模型。运行方式继承自原 READMEexport GEMINI_API_KEYyour-api-key uv sync uv run src/main.py完整示例代码解析示例入口 src/main.py 演示了一个「财务桌面treasury desk」场景模型被要求调用转账工具工具首次执行时暂停等待人工审批。定义可中断的工具from genkit_google_genai import GoogleAI from pydantic import BaseModel, Field from genkit import Genkit, Interrupt, ToolRunContext, respond_to_interrupt, restart_tool ai Genkit(plugins[GoogleAI()], modelGoogleAI.gemini_model(gemini-flash-latest)) class TransferRequest(BaseModel): to_account: str Field(descriptionrecipient name or masked account) amount_usd: str Field(descriptionamount as a string, e.g. 250.00) memo: str ai.tool() async def request_transfer(body: TransferRequest, ctx: ToolRunContext) - dict: # 首次调用会暂停恢复后 is_resumed() 为 True工具真正执行。 if not ctx.is_resumed(): raise Interrupt({summary: fWire ${body.amount_usd} to {body.to_account} — {body.memo}}) return {status: confirmed, resumed: ctx.resumed_metadata}关键点工具签名工具函数是async最多接收 2 个参数——输入体由 Pydantic 模型TransferRequest描述SDK 会据此生成发给模型的 input schema和ToolRunContext执行上下文。这一「0–2 参数」约束在 py/packages/genkit/src/genkit/_ai/_tools.py 的参数分发逻辑match len(input_spec.args)中得到印证。暂停即抛异常工具通过raise Interrupt(metadata)暂停。Interrupt是一个可带 metadata 的异常类继承自GenkitInterrupt见 py/packages/genkit/src/genkit/_ai/_tools.pymetadata 会挂在被暂停的工具请求tool request上随响应返回若处于 OpenTelemetry span 中metadata 还会写入genkit:metadata:interrupt属性用于可观测性。恢复的判定ToolRunContext.is_resumed()依据resumed_metadata is not None判断本次执行是否为中断后的恢复执行py/packages/genkit/src/genkit/_ai/_tools.py。示例中工具「首次抛 Interrupt、恢复后才真正执行」等价于一次审批门禁人批准前 wire 汇款这个副作用根本不会发生。第一次交互restart_tool 让工具重跑response await ai.generate( promptPlease wire $250.00 to Jane Doe (account ending in 4521) for April rent., systemYou are a treasury desk. Call request_transfer to send money., tools[request_transfer], ) print(response.text) if not response.interrupts: return interrupt response.interrupts[0] print(fpaused: {interrupt.metadata}) # restart_tool人说了 yes 之后让工具重新运行。 approved await ai.generate( messagesresponse.messages, resume_restartrestart_tool(interruptinterrupt, resumed_metadata{approved: True}), tools[request_transfer], ) print(approved.text)流程是ai.generate(...)返回的响应上response.interrupts属性暴露出所有被中断的ToolRequestPart实现见 py/packages/genkit/src/genkit/_core/_model.py 与 #L690-L694每个元素携带metadata——也就是工具抛出Interrupt时附上的{summary: ...}。拿到中断句柄后restart_tool(interrupt..., resumed_metadata{approved: True})构造一个用于resume_restart的ToolRequestPart。源码见 py/packages/genkit/src/genkit/_ai/_tools.py它的行为细节值得注意默认沿用原tool_request.input若传入replace_input则替换输入并把旧输入存入 metadata 的replacedInput键resumed_metadata会被写入新 part 的metadata[resumed]不传时写True随后经ContextVar传播进工具成为ToolRunContext.resumed_metadata因此恢复执行时ctx.is_resumed()返回True、ctx.resumed_metadata即{approved: True}工具返回{status: confirmed, resumed: {approved: True}}。恢复时必须把原响应的messages与同一组tools一起回传给ai.generate以保持对话历史与工具注册一致。第二次交互respond_to_interrupt 直接注入结果declined await ai.generate( promptPlease wire $80.00 to Sam Lee (account ending in 9910) for lunch., systemYou are a treasury desk. Call request_transfer to send money., tools[request_transfer], ) if not declined.interrupts: return print(fpaused: {declined.interrupts[0].metadata}) # respond_to_interrupt 直接注入结果——拒绝汇款工具不再执行。 done await ai.generate( messagesdeclined.messages, resume_respondrespond_to_interrupt({status: declined}, interruptdeclined.interrupts[0]), tools[request_transfer], ) print(done.text)respond_to_interrupt(response, *, interrupt, metadataNone)返回一个ToolResponsePart实现见 py/packages/genkit/src/genkit/_ai/_tools.py它复用了被中断请求的ref和name把第一个位置参数作为output写入工具响应并在 part 的metadata上标记interruptResponse。把这个 part 传给generate(..., resume_respond...)后工具函数不会被再次调用{status: declined}直接作为工具结果进入模型可见的对话历史模型据此生成「已拒绝」的最终回复这正好演示了两种恢复语义的分野resume_restart 批准重跑工具resume_respond 否决/代答跳过工具执行。generate的两个恢复参数resume_respond/resume_restart是ai.generate的一等参数接受单个或列表见 py/packages/genkit/src/genkit/_ai/_prompt.pyresume_respond: ToolResponsePart | list[ToolResponsePart] | None resume_restart: ToolRequestPart | list[ToolRequestPart] | None列表形式意味着一次generate可以批量恢复多个并行中断例如模型在同一轮同时请求了多笔转账逐一审批。底层实现要点结合 py/packages/genkit/src/genkit/_ai/_tools.py 可以补充几个示例注释里没有的机制细节恢复元数据的传播链run_tool_request#L503-L531从 part metadata 读取resumed/replacedInput写入两个ContextVar_tool_resumed_metadata、_tool_original_input工具包装器在按参数个数分发调用时用这两个值构造ToolRunContext(resumed_metadata..., original_input...)#L685-L714。所以ctx.resumed_metadata与ctx.original_input分别告诉你「人批了什么」和「重跑前原输入是什么」。重启期间不允许再次中断run_tool_after_restart#L557-L593捕获恢复执行中再次抛出的Interrupt包装为FAILED_PRECONDITION错误消息形如 Tool interrupted again during restart: ...见restart_interrupt_error#L534-L554。从源码注释看这是有意约束防止恢复流程无限循环等待审批。纯中断工具define_interrupt除了「有时执行、有时中断」的普通工具外SDK 还提供define_interrupt#L797-L850注册「每次调用都抛Interrupt」的工具支持静态 metadata 或(input) - dict回调适合显式的人机检查点而本文示例属于「有时运行逻辑、有时中断」用ai.tool() 手动raise Interrupt更合适该函数的 docstring 明确给出了这一选择建议。可观测性Interrupt的 metadata 会写入当前 span 的genkit:metadata:interrupt属性恢复执行时resumed元数据会写入genkit:metadata:resumed#L685-L694便于在 trace 中复盘审批决策。选型建议restart_tool 还是 respond_to_interrupt维度restart_toolresume_restartrespond_to_interruptresume_respond语义人类批准让工具真正执行人类否决或由人代答直接注入结果工具是否执行会重新执行不会执行返回类型ToolRequestPart新的工具请求ToolResponsePart工具响应额外能力replace_input可在批准时改写输入旧输入存replacedInputmetadata可标注中断响应通道内容典型用法转账批准后发起 wire转账拒绝直接告诉模型 declined限制恢复执行中再次抛Interrupt会报FAILED_PRECONDITION注入内容成为对话历史的一部分模型据此继续小结tool-interrupts 示例py/samples/tool-interrupts/README.md用最少的代码覆盖了 Genkit Python 中断机制的完整闭环Interrupt携带 metadata 暂停工具并把原因暴露在response.interrupts上restart_tool配合resumed_metadata实现「批准后重跑」并可用is_resumed()在工具内区分首次执行与恢复执行respond_to_interrupt实现「不重跑、直接注入结果」。这套机制对转账、支付、删除数据等副作用工具尤为关键——在模型自主调用工具与真实业务动作之间插入了一道可审计的人工确认。进一步阅读可参考 py/samples/tool-interrupts/src/main.py 与 py/packages/genkit/src/genkit/_ai/_tools.py 中Interrupt、restart_tool、respond_to_interrupt、run_tool_after_restart的完整实现。【免费下载链接】genkitOpen-source framework for building agentic apps in JavaScript, Go, Dart, and Python, built and used in production by Google项目地址: https://gitcode.com/GitHub_Trending/ge/genkit创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考