我要提问
ARTICLE DETAIL

资讯详情

前沿编程新知与开发实战干货的深度解读。

adk-python 工具级自愈重试机制:ReflectAndRetryToolPlugin 源码解析与实战指南

adk-python 工具级自愈重试机制:ReflectAndRetryToolPlugin 源码解析与实战指南 adk-python 工具级自愈重试机制ReflectAndRetryToolPlugin 源码解析与实战指南【免费下载链接】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 中Agent 的工具调用经常因为参数错误、远程 API 超时或模型幻觉出不存在的函数名而失败。ReflectAndRetryToolPlugin提供了一套自愈self-healing、并发安全concurrent-safe的工具级故障恢复机制它拦截工具抛出的异常与错误结果把结构化的反思引导reflection guidance回传给模型并允许工具调用自动重试到可配置的上限。读完本文你将掌握该插件的注册方式、全部配置项、底层计数与锁机制以及如何通过覆写钩子处理软错误soft error与模型幻觉调用等实战场景。为什么需要工具级重试工具失败的原因多种多样函数抛出异常、远程 API 超时、模型用畸形参数调用工具等。如果不做处理一次失败的调用要么直接击穿整个 invocation要么让模型陷入重复执行同一条错误调用的死循环。ReflectAndRetryToolPlugin在工具运行期间捕获这类失败注入一条包含错误详情与导致失败参数的反思消息给模型一次纠正调用并再次尝试的机会。它继承自BasePlugin由PluginManager驱动通过after_tool_callback与on_tool_error_callback两个钩子观察工具执行过程并从ToolContext读取当前 invocation 信息。任何在其上注册了该插件的App都能获得工具自纠错能力是 ReflectAndRetryModelPlugin 在工具层的对应实现后者处理模型层失败如畸形函数调用、安全拦截等。插件核心特性自愈重试把失败的工具调用转成反思消息并自动重试。并发安全计数使用锁保护的计数器并行工具执行之间不会互相污染状态。按工具独立计数按工具名分别跟踪连续失败次数一个工具的失败连击不会消耗另一个工具的重试预算。可配置作用域失败计数默认按 invocation 隔离也可通过TrackingScope枚举切换为进程级全局共享。可扩展错误检测覆写extract_error_from_result即可把返回值里藏着错误对象的软错误也纳入重试逻辑。快速开始创建插件实例并把它和 Agent 一起注册到App上即可生效源码见 reflect_retry_tool_plugin.pyfrom google.adk.agents import LlmAgent from google.adk.apps import App from google.adk.plugins import ReflectAndRetryToolPlugin def get_stock_price(symbol: str) - float: Looks up the current price for a stock ticker symbol. prices {SYMBOL: 100.0} return prices[symbol] # Raises KeyError for an unknown symbol. agent LlmAgent( nameresilient_agent, descriptionAssistant equipped with tool error reflection., instructionYou are a helpful assistant., tools[get_stock_price], ) # Retry a failing tool call up to 3 times before giving up. retry_plugin ReflectAndRetryToolPlugin(max_retries3) app App( nametool_retry_demo, root_agentagent, plugins[retry_plugin], )当get_stock_price抛出异常例如传入未知股票代码时插件不会把缺失的结果返回给模型而是返回一条描述错误的反思消息让 Agent 重新分析参数并再试一次。第三次反思重试之后若下一次仍然连续失败插件会重新抛出原始异常。值得注意的是ReflectAndRetryToolPlugin在 plugins/init.py 中通过__getattr__惰性导入lazy importfrom google.adk.plugins import ReflectAndRetryToolPlugin只有在真正访问该符号时才加载对应模块避免不必要的启动开销。工作原理钩子、计数器与反射消息插件通过BasePlugin暴露的两个钩子观察工具执行两个钩子的签名定义在 base_plugin.py异常处理on_tool_error_callback工具抛出异常时该钩子把异常转发给中央的、由锁保护的_handle_tool_error例程见 reflect_retry_tool_plugin.py。结果检查after_tool_callback工具正常返回后插件先跳过自己的反射响应通过REFLECT_AND_RETRY_RESPONSE_TYPE标记识别避免重试自己的输出然后调用extract_error_from_result检测软错误如果是干净的成功则重置该工具对应的失败计数器见 reflect_retry_tool_plugin.py。计数与重试捕获到失败后通过ScopedFailureTracker递增该工具名下的计数器。当current_retries max_retries时返回一个ToolFailureResponse——一条结构化反思消息包含工具名、错误详情、所用参数与当前尝试次数并明确告诉模型不要重复完全相同的调用。重试耗尽当计数超过max_retries后根据throw_exception_if_retry_exceeded的值要么重新抛出原始异常要么返回一条放弃这个工具的ToolFailureResponse。计数逻辑依赖_reflect_retry_utils模块中的三个核心构件见 _reflect_retry_utils.pyScopedFailureTracker内部维护dict[str, PerItemFailuresCounter]作用域键 → 工具名 → 连续失败次数所有increment与reset操作都在asyncio.Lock下原子执行。TrackingScopeINVOCATION按 invocation 隔离或GLOBAL全局共享两种计数生命周期。resolve_scope_key把作用域解析为实际键INVOCATION要求必须提供invocation_id缺失时抛出ValueErrorGLOBAL则固定返回GLOBAL_SCOPE_KEY__global_reflect_and_retry_scope__。作用域键由_get_scope_key方法基于tool_context.invocation_id推导见 reflect_retry_tool_plugin.py该方法可以在子类中覆写以实现按用户、按会话等自定义跟踪逻辑。插件的__init__中会同时保留_tracker、_scoped_failure_counters与_lock的引用reflect_retry_tool_plugin.py这也是并发安全的实现基础。从源码注释可以确认一个重要的并发边界asyncio.Lock只序列化同一事件循环上的协程不提供线程间的互斥因此一个ScopedFailureTracker实例不应跨线程共享。并发测试与行为验证可参考 test_reflect_retry_tool_plugin.py其中覆盖了默认/自定义初始化参数、成功调用后计数复位、异常重试、软错误提取与重试耗尽等完整行为路径。配置项详解ReflectAndRetryToolPlugin引入的配置项如下继承自BasePlugin的选项不在列选项类型默认值说明namestrreflect_retry_tool_plugin插件实例标识符。max_retriesint3放弃前允许的最大连续失败次数。必须非负0表示禁用重试。throw_exception_if_retry_exceededboolTrue为True时超限后重新抛出最终异常为False时返回反思引导。tracking_scopeTrackingScopeTrackingScope.INVOCATION失败计数器生命周期按 invocation 隔离或全局共享。关键语义说明max_retries的判定条件是current_retries max_retries因此3允许第 13 次尝试第 4 次连续失败触发耗尽逻辑。负数会在构造时直接抛出ValueErrormax_retries must be a non-negative integer.0则在第一次失败时就放弃、不做任何重试见 reflect_retry_tool_plugin.py。throw_exception_if_retry_exceeded决定重试耗尽后的失败模式重新抛出让外层监督者捕获或返回一条指示模型放弃该工具的最终ToolFailureResponse。注意非Exception的错误例如extract_error_from_result返回的普通 dict在抛出前会被_ensure_exception包装成Exception见 reflect_retry_tool_plugin.py。tracking_scope默认INVOCATIONGLOBAL会在所有 invocation 之间共享同一个计数器。反射消息的结构当重试未耗尽时_create_tool_reflection_response生成的ToolFailureResponsePydantic 模型字段见 reflect_retry_tool_plugin.py包含error_type、error_details、retry_count和reflection_guidance四个字段。其中reflection_guidance是一段完整的 Markdown 引导包含失败的工具名与错误详情错误类型: 错误信息格式触发失败的参数 JSONjson.dumps(tool_args, indent2, defaultstr)序列化保证任意类型的参数都能展示当前是第几次重试如This is retry attempt 1 of 3五条明确的反思指令检查参数是否非法/格式错误/缺失、检查前置状态与资源是否就绪、评估是否选对了工具、尝试拆解任务、以及错误函数名提示要求模型只使用可用工具。而重试耗尽时的_get_tool_retry_exceed_msg则直接给出最终指令Do not attempt to use the{tool.name}tool again for this task要求模型换一种策略或如实告知用户任务无法完成见 reflect_retry_tool_plugin.py。进阶应用检测软错误soft error有些工具从不抛异常而是返回{status: error}之类的错误对象。要让这类失败也触发反思与重试可以子类化插件并覆写extract_error_from_result返回需要重试的错误或返回None表示结果正常from google.adk.plugins import ReflectAndRetryToolPlugin class CustomRetryPlugin(ReflectAndRetryToolPlugin): async def extract_error_from_result( self, *, tool, tool_args, tool_context, result ): if isinstance(result, dict) and result.get(status) error: return result return None retry_plugin CustomRetryPlugin(max_retries5)当覆写方法返回非None值时插件会像处理抛出的异常一样处理它递增计数、生成反射消息此时error_type会使用ToolError兜底、或触发耗尽逻辑。优雅降级而不是抛出异常与其重新抛出异常让整个调用失败不如让插件返回ToolFailureResponse指示模型停止使用该工具并尝试其他方案retry_plugin ReflectAndRetryToolPlugin( max_retries2, throw_exception_if_retry_exceededFalse, )这种模式适合希望 Agent 自行绕路而非直接崩溃的场景——例如工具临时不可用时模型可以改用备用工具或直接向用户说明情况。自定义作用域默认按 invocation 跟踪失败次数切换为GLOBAL即可在所有 invocation 之间共享同一个计数器from google.adk.plugins import ReflectAndRetryToolPlugin from google.adk.plugins.reflect_retry_tool_plugin import TrackingScope retry_plugin ReflectAndRetryToolPlugin( max_retries5, tracking_scopeTrackingScope.GLOBAL, )更进一步源码注释明确指出_get_scope_key可以被覆写用于实现按用户、按会话等自定义跟踪维度例如每个用户独立的失败预算。实战样例仓库中的两个可运行示例仓库 plugin_reflect_tool_retry 目录提供了两个可直接运行的示例1. Basic Usage基本用法示例 basic/agent.py 中的guess_number_tool被注入了两种失败接近目标数字时返回{status: error, error_message: Number is almost valid.}软错误猜测过大/过小时抛出ValueError硬异常。配合CustomRetryPlugin(max_retries20, throw_exception_if_retry_exceededFalse)与LoggingPlugin两种错误都能驱动模型持续反思并逼近正确答案 3。运行方式$ adk web contributing/samples/plugins/plugin_reflect_tool_retry选择 basic 后输入提示词Please guess a number! Tell me what number you guess and how is it.即可看到模型从 50 开始逐步根据反馈调整猜测的完整重试过程。2. Hallucinating tool calls幻觉工具名示例 hallucinating_func_name/agent.py 通过after_model_callback在模型正常返回roll_die调用时把函数名篡改为不存在的roll_die_wrong_name模拟模型幻觉出错误工具名的场景。由于插件捕获了工具不存在的错误模型会在反思引导下重新使用正确的工具名完成掷骰子任务$ adk web contributing/samples/plugins/plugin_reflect_tool_retry选择 hallucinating_func_name 并输入Roll a 6 sided die即可复现。这也印证了反射消息中第 5 条指令Wrong Function Name: Does the error indicates the tool is not found? Please check again and only use available tools.的实战价值。局限性与适用边界仅覆盖工具层失败该插件只恢复工具调用失败模型层失败如畸形函数调用、安全拦截请使用ReflectAndRetryModelPlugin。软错误需要覆写以返回值形式报告失败不抛异常的工具默认被视为成功必须覆写extract_error_from_result才会触发重试。依赖模型遵循引导反射消息是以工具响应形式投递给模型的。如果模型无视引导并重复同样的调用就会消耗重试预算。只统计连续失败一次成功调用会重置该工具的计数器因此只有不间断的失败连击才会达到重试上限。线程边界基于asyncio.Lock的计数只在同一事件循环内保证原子性实例不应跨线程共享详见源码注释。小结ReflectAndRetryToolPlugin把工具失败 → 结构化反思 → 自动重试这条链路封装成开箱即用的插件max_retries控制重试预算throw_exception_if_retry_exceeded决定耗尽后的行为TrackingScope控制计数作用域extract_error_from_result让软错误也能进入重试管道。配合仓库中的两个示例你可以在几分钟内为自己的 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),仅供参考
返回列表