Agent Zero 扩展机制深度解析helpers/extension.py 的扩展点发现、调度与缓存架构【免费下载链接】agent-zeroAgent Zero AI framework项目地址: https://gitcode.com/GitHub_Trending/ag/agent-zero本篇技术指南围绕 Agent Zero 框架的扩展运行时核心模块 helpers/extension.py 及其维护文档 helpers/extension.py.dox.md 展开系统讲解 Agent Zero 中 Python 扩展与 WebUI 扩展的发现、调度、缓存与热更新机制。读完本文你将掌握Extension基类的编写契约、call_extensions_async/call_extensions_sync两类调度入口、extensible装饰器如何为既有函数隐式注入start/end扩展点以及扩展清单如何被注入 WebUI 页面并能在自己的 Agent Zero 实例中动手编写与调试扩展。模块定位Agent Zero 扩展运行时的唯一入口在 Agent Zero 的架构中helpers/目录存放的是被核心代码与插件复用的框架级 API见 helpers/AGENTS.md。extension.py是其中专门负责扩展Extension机制的模块其职责可以概括为三句话发现在各级扩展目录中扫描符合某个「扩展点extension point」的扩展类调度在核心流程的特定时机以同步或异步方式执行这些扩展暴露 WebUI 扩展把浏览器端HTML/JS扩展资产按扩展点分组注入到渲染后的 WebUI 页面。官方文档 docs/developer/extensions.md 指出扩展是「改变 Agent Zero 行为的高级方式」如果读者是新手应优先从插件plugin入手——插件更容易创建、测试、禁用和移除。扩展则适合在以下场景使用在某个特定的生命周期节点添加行为以可复用的方式塑造提示词prompt与核心工具紧密集成在任务开始前准备框架自有的状态。而简单的 UI 改动、一次性脚本或应该易于移除的功能都不适合用扩展实现。extension.py.dox.md明确声明了模块的所有权边界extension.py拥有运行时实现.dox.md文件则承载关于职责、契约、副作用与验证的持久化说明并要求「每当公共函数、类、持久化行为、路径/安全假设、副作用或跨模块契约发生变化时同步更新本文档」。核心常量与配置扩展目录与缓存分区模块顶部定义了一组决定扩展行为边界的常量理解它们是后续所有机制的基石常量值含义DEFAULT_EXTENSIONS_FOLDERpython/extensions内置扩展在仓库中的根目录相对仓库根USER_EXTENSIONS_FOLDERusr/extensions用户扩展根目录_EXTENSIONS_CACHE_AREAextension_folder_classes(extensions)按目录缓存「文件夹→扩展类」的分区名_CLASSES_CACHE_AREAextension_classes(extensions)按「agent扩展点」缓存「扩展点→类列表」的分区名_WEBUI_MANIFEST_CACHE_AREAwebui_extension_manifest(extensions)(plugins)WebUI 扩展清单缓存分区_WEBUI_MANIFEST_SUFFIXES{html: (.html, .htm, .xhtml), js: (.js, .mjs)}WebUI 资产类型与其文件后缀的映射_UNSET_Unset()哨兵实例用于标记「结果尚未设置」的内部哨兵_EXTENSIONS_LOG_COUNTSdict[str, int]扩展调用计数供调试日志使用值得注意源码中还保留了cache.toggle_area(_EXTENSIONS_CACHE_AREA, False)与cache.toggle_area(_CLASSES_CACHE_AREA, False)的注释示例说明这两个分区可以按需被关闭见 helpers/cache.py 的toggle_area实现。当前仓库中内置的 Python 扩展点目录位于 extensions/python实际存在的扩展点包括agent_init、banners、before_main_llm_call、error_format、hist_add_before、hist_add_tool_result、job_loop、message_loop_end、message_loop_prompts_after、message_loop_prompts_before、message_loop_start、monologue_end、monologue_start、process_chain_end、reasoning_stream、reasoning_stream_chunk、reasoning_stream_end、response_stream、response_stream_chunk、response_stream_end、startup_migration、system_prompt、tool_execute_after、tool_execute_before、user_message_ui、util_model_call_before、webui_ws_connect、webui_ws_disconnect、webui_ws_event等另有_functions/目录专门承载extensible装饰器生成的隐式扩展点下文详述。Extension 基类编写一个扩展的最小契约所有 Python 扩展都必须继承Extension抽象基类其定义如下class Extension: def __init__(self, agent: Agent|None, **kwargs): self.agent: Agent|None agent self.kwargs kwargs abstractmethod def execute(self, **kwargs) - None | Awaitable[None]: pass契约要点构造器接收当前agent可能为None以及任意关键字参数并把它们分别保存在self.agent与self.kwargs上execute是唯一的抽象方法返回类型可以是None或可等待对象Awaitable[None]即扩展既可以写成普通同步函数也可以写成async def调度方会根据返回值是否为Awaitable决定是否需要await。仓库内置了一个可直接参考的最小示例 agents/_example/extensions/agent_init/_10_example_extension.py它在 agent 初始化时把 agent 改名为 SuperAgent numberfrom helpers.extension import Extension # this is an example extension that renames the current agent when initialized # see /extensions folder for all available extension points class ExampleExtension(Extension): async def execute(self, **kwargs): # rename the agent to SuperAgent0 self.agent.agent_name SuperAgent str(self.agent.number)注意文件命名_10_example_extension.py中的_10前缀不是装饰而是排序与覆盖控制的手段——_get_extension_classes最终会对类按文件名排序详见下文「扩展点调度」一节文件名前缀可用于控制同一扩展点内多个扩展的执行顺序。扩展点调度异步与同步双入口extension.py对外暴露两个对称的调度函数分别用于异步与同步上下文async def call_extensions_async( extension_point: str, agent: Agent|None None, **kwargs ): _log_extension_call(extension_point) # fetch classes for this extension point and agent classes _get_extension_classes(extension_point, agentagent, **kwargs) # execute unique extensions for cls in classes: result cls(agentagent).execute(**kwargs) if isinstance(result, Awaitable): await result def call_extensions_sync(extension_point: str, agent: Agent|None None, **kwargs): _log_extension_call(extension_point) # fetch classes for this extension point and agent classes _get_extension_classes(extension_point, agentagent, **kwargs) # execute unique extensions for cls in classes: result cls(agentagent).execute(**kwargs) if isinstance(result, Awaitable): raise ValueError( fExtension {cls.__name__} returned awaitable in sync mode )两者共享同一套逻辑先调用_log_extension_call记录调用再通过_get_extension_classes拿到该扩展点且针对该 agent应执行的类列表然后逐类实例化并调用execute(**kwargs)。唯一的区别是异步入口遇到Awaitable返回值会await它同步入口若遇到可等待返回值会直接抛出ValueError——这保证了在同步调用链上不会出现「协程被静默丢弃」的隐患。_log_extension_call是一个内置的调试计数器。它读取环境变量EXTENSIONS_LOG取整数值作为「每 N 次调用打印一次」的间隔每调用一次就把该扩展点计数加一并维护_total总数当总数达到 N 的整数倍时把当前所有计数打印出来。开发者可通过设置EXTENSIONS_LOG100之类的环境变量观察扩展调用频率无需改动任何代码。核心流程中的调用示例在 agent.py 中这些调度入口被大量使用例如# 同步扩展点agent 初始化完成后调用 extension.call_extensions_sync(agent_init, self) # 异步扩展点流程链结束时调用 await extension.call_extensions_async(process_chain_end, agentself.get_agent(), data{})此外api/banners.py 在生成横幅时调用await call_extensions_async(banners, agentNone, bannersbanners, frontend_contextfrontend_context)说明扩展点同样支持agentNone的全局场景。extensible 装饰器为既有函数注入隐式扩展点除了显式调用扩展点extension.py还提供了一种「无侵入」的扩展方式用extensible装饰任意函数该函数在执行前后会自动各产生一个扩展点。装饰器文档对此有完整说明从被包装函数自动推导两个扩展点目录路径_functions/模块路径/限定名路径/start_functions/模块路径/限定名路径/end模块路径段来自func.__module__按.切分限定名路径段来自完整的嵌套func.__qualname__按.切分并剔除locals段。示例模块helpers.something、限定名Outer.Inner.__init__会生成_functions/helpers/something/Outer/Inner/__init__/start_functions/helpers/something/Outer/Inner/__init__/end被包装函数被调用时装饰器构造一个可变的data载荷传给两个扩展点键初始值扩展可执行的操作data[args]位置参数替换/修改后影响原函数调用data[kwargs]关键字参数替换/修改后影响原函数调用data[result]内部哨兵_UNSET若设置则短路原函数不再执行data[exception]None若设置为BaseException实例则最终强制抛出执行流程为start扩展先执行可以修改入参或直接设置data[result]/data[exception]实现短路若data[result]仍是_UNSET装饰器用可能被修改过的data[args]/data[kwargs]调用原函数end扩展最后执行可以改写data[result]或替换/清除data[exception]最终若data[exception]中有异常则抛出否则返回data[result]。核心实现中_prepare_inputs负责从args/kwargs中探测Agent实例通过_get_agent先查kwargs.get(agent)再遍历args且要求该实例有非空__dict__并把扩展点路径、agent 与data打包返回若函数缺少__module__或__qualname__则跳过扩展逻辑、直接调用原函数。同步与异步的判别发生在装饰器装配阶段if inspect.iscoroutinefunction(func): return wraps(func)(_run_async) return wraps(func)(_run_sync)即被装饰函数若是async def则使用_run_async包装内部通过call_extensions_async调度且对原函数返回值做await处理否则使用_run_sync内部通过call_extensions_sync调度。wraps保证了元数据如__name__、__doc__不被破坏。从源码结构看agent.py中有 30 余处方法使用了extension.extensible见 agent.py 中诸如extension.extensible的注解api/plugins.py 中也有 3 处覆盖了消息处理、工具执行、流程链等关键路径。仓库测试 tests/test_extensions_stress.py 专门对extensible做了 10000 次迭代的性能剖析cProfile验证class PerfAgent(Agent): extensible def perf_hook(self, value: int): return value 1 pytest.mark.parametrize(iterations, [10000]) def test_extensible_method_performance_trace(iterations: int): ... for i in range(iterations): result agent.perf_hook(i) ... assert result iterations该测试断言连续调用 10000 次带扩展点的同步方法后结果正确并打印累计耗时统计说明extensible被设计为可高频使用而不会破坏调用语义的机制。扩展发现与去重_get_extension_classes 的覆盖语义_get_extension_classes是扩展调度背后的核心查找函数def _get_extension_classes( extension_point: str, agent: Agent|None None, **kwargs ) - list[Type[Extension]]: from helpers import subagents cache_key cache.determine_cache_key(agent, extension_point) cached cache.get(_CLASSES_CACHE_AREA, cache_key) if cached is not None: return cached # search for extension folders in all agents paths paths subagents.get_paths(agent, extensions/python, extension_point) all_exts [cls for path in paths for cls in _get_extensions(path)] # merge: first ocurrence of file name is the override unique {} for cls in all_exts: file _get_file_from_module(cls.__module__) if file not in unique: unique[file] cls classes sorted( unique.values(), keylambda cls: _get_file_from_module(cls.__module__) ) cache.add(_CLASSES_CACHE_AREA, cache_key, classes) return classes关键语义有三点路径来源subagents.get_paths(agent, extensions/python, extension_point)会收集该 agent 全部生效路径下的同名扩展点目录内置extensions/python、用户级usr/extensions、agent 级与项目级扩展目录等这与 helpers/extension.py 中register_extensions_watchdogs监控的根目录一一对应。文件名覆盖override合并时以「文件名的最后一个模块段」为键做去重首次出现即胜出——即路径列表中靠前的目录里的同名扩展文件会覆盖后面的。这一机制让用户可以用同名文件覆盖内置扩展行为。确定性排序最终按文件名_get_file_from_module返回module_name.split(.)[-1]排序因此_10_xxx.py会排在_20_xxx.py之前执行顺序稳定可控。_get_extensions则负责单目录内的类加载def _get_extensions(folder: str): folder files.get_abs_path(folder) cached cache.get(_EXTENSIONS_CACHE_AREA, folder) if cached is not None: return cached if not files.exists(folder): return [] classes modules.load_classes_from_folder(folder, *, Extension) cache.add(_EXTENSIONS_CACHE_AREA, folder, classes) return classes底层类加载由 helpers/modules.py 的load_classes_from_folder完成按字母序扫描目录内所有匹配*的.py文件用importlib.util.spec_from_file_location逐个导入再通过inspect.getmembers反向遍历类成员筛选出「是Extension的真子类」的类one_per_fileTrue时每个文件只取第一个从而把「每个文件一个扩展」固化为约定。缓存体系扩展点与 Agent 的绑定关系扩展类的查找结果被缓存在 helpers/cache.py 中其键由cache.determine_cache_key(agent, *additional)决定def determine_cache_key(agent, *additional): if agent: profile agent.config.profile or none project agent.context.get_data(project) or none return (profile, project, *additional) return (none, none, *additional)也就是说扩展类缓存的键 (agent 配置 profile, 当前项目, 扩展点)。这带来一个重要推论不同的 agent profile 与项目会得到彼此独立的扩展集合扩展的生效范围天然与 profile/项目绑定。cache.add/cache.get会检查分区是否被toggle_area关闭源码中保留了关闭_EXTENSIONS_CACHE_AREA与_CLASSES_CACHE_AREA的注释示例而cache.clear(area)支持*?[通配符模糊清理多个分区。WebUI 扩展资产发现与清单注入WebUI浏览器端扩展是另一类一等公民由get_webui_extensions与get_webui_extension_manifest两个函数支撑。按需拉取get_webui_extensionsget_webui_extensions(agent, extension_point, filters)用于按扩展点且可选文件过滤器拉取 WebUI 资产路径def get_webui_extensions( agent: Agent | None, extension_point: str, filters: list[str] | None None ): from helpers import subagents entries: list[str] [] effective_filters filters or [*] # search for extension folders in all agents paths folders subagents.get_paths( agent, extensions/webui, extension_point, ) extensions [] for folder in folders: for filter in effective_filters: pattern files.get_abs_path(folder, filter) extensions.extend(files.find_existing_paths_by_pattern(pattern)) for extension in extensions: rel_path files.deabsolute_path(extension) entries.append(rel_path) return entries逻辑要点默认过滤器是[*]全部文件对每个生效目录 × 每个过滤器组合出绝对路径模式用files.find_existing_paths_by_pattern找到实际存在的文件最后统一转为相对路径返回。该函数被 api/load_webui_extensions.py 暴露为 HTTP 接口前端可动态请求某个扩展点的资源。全量清单get_webui_extension_manifestget_webui_extension_manifest一次性返回「所有 WebUI 扩展 URL按资产类型和扩展点分组」的清单其结构为dict[str, dict[str, list[str]]]第一层键为html与js对应_WEBUI_MANIFEST_SUFFIXES第二层键为扩展点资产所在目录值为 URL 列表cache_key cache.determine_cache_key(agent) cached cache.get(_WEBUI_MANIFEST_CACHE_AREA, cache_key) if cached is not None: return cached manifest: dict[str, dict[str, list[str]]] { asset_type: {} for asset_type in _WEBUI_MANIFEST_SUFFIXES } roots subagents.get_paths(agent, extensions/webui) for root in roots: relative_files sorted(files.list_files_in_dir_recursively(root)) for asset_type, suffixes in _WEBUI_MANIFEST_SUFFIXES.items(): for suffix in suffixes: for relative_file in relative_files: if not relative_file.lower().endswith(suffix): continue extension_point os.path.dirname(relative_file).replace( os.sep, / ) if not extension_point or extension_point .: continue absolute_path files.get_abs_path(root, relative_file) relative_path files.deabsolute_path(absolute_path).replace( os.sep, / ) manifest[asset_type].setdefault(extension_point, []).append( / relative_path.lstrip(/) ) cache.add(_WEBUI_MANIFEST_CACHE_AREA, cache_key, manifest) return manifest实现细节对每个 WebUI 扩展根目录做递归文件列举并排序从而保持根目录与过滤器顺序的确定性资产按后缀归入html.html/.htm/.xhtml或js.js/.mjs资产所在目录os.path.dirname规范化后的相对路径即扩展点根目录下散落的文件extension_point .会被跳过最终 URL 以/开头并缓存在_WEBUI_MANIFEST_CACHE_AREA分区键为determine_cache_key(agent)即 profile项目。该清单被 helpers/ui_server.py 在渲染 WebUI 首页时注入先json.dumps序列化再对、、做 HTML 转义\u0026、\u003c、\u003e最后通过files.replace_placeholders_text替换 index 模板中的占位符从而把扩展清单安全地嵌进页面。对应的测试 tests/test_webui_extension_surfaces.py 验证了清单的语义def test_webui_extension_manifest_groups_plugin_assets_by_type_and_surface() - None: surface manifest-probe with _temporary_probe_plugin(surface) as (plugin_id, probe_file_name): manifest get_webui_extension_manifest(agentNone) expected_suffix ( f/{plugin_id}/extensions/webui/{surface}/{probe_file_name} ) assert any( path.endswith(expected_suffix) for path in manifest[html].get(surface, []) ) assert surface not in manifest[js]该测试临时创建一个探测插件确认其 HTML 资产被正确归入manifest[html][surface]且不会出现在js分组中——印证了「插件内extensions/webui/扩展点/目录下的资产会并入全局 WebUI 扩展清单」的机制。热更新扩展文件系统看门狗扩展代码在运行期发生变化时框架通过register_extensions_watchdogs()注册三类文件系统看门狗来失效缓存从而实现无需重启即可让新扩展生效def register_extensions_watchdogs(): from helpers import watchdog, projects def extensions_changed(items: list[watchdog.WatchItem]): cache.clear(_EXTENSIONS_CACHE_AREA) cache.clear(_CLASSES_CACHE_AREA) PrintStyle.debug(Extensions watchdog triggered:, items) # extensions and usr/extensions watchdog.add_watchdog( idextensions_base, roots[ files.get_abs_path(files.EXTENSIONS_DIR), files.get_abs_path(files.USER_DIR, files.EXTENSIONS_DIR), ], handlerextensions_changed, ) # usr/projects/**/extensions watchdog.add_watchdog( idextensions_projects, roots[projects.PROJECTS_PARENT_DIR], patterns[f*/{projects.PROJECT_META_DIR}/**/{files.EXTENSIONS_DIR}/**/*], handlerextensions_changed, ) # agents and usr/agents watchdog.add_watchdog( idextensions_agents, roots[ files.get_abs_path(files.AGENTS_DIR), files.get_abs_path(files.USER_DIR, files.AGENTS_DIR), ], patterns[f*/{files.EXTENSIONS_DIR}/**/*], handlerextensions_changed, )三个看门狗覆盖了扩展可能存放的全部位置看门狗 ID监控根目录监控模式目的extensions_base内置extensions与usr/extensions整个目录内置/用户级扩展变化extensions_projects项目父目录projects.PROJECTS_PARENT_DIR*/项目元目录/**/extensions/**/*项目级扩展变化extensions_agentsagents与usr/agents*/extensions/**/*agent 级扩展变化任一事件触发后处理器extensions_changed会同时清除_EXTENSIONS_CACHE_AREA与_CLASSES_CACHE_AREA两个缓存分区并通过PrintStyle.debug打印触发详情——这解释了为什么新增/修改扩展文件后无需重启框架即可生效。验证与测试矩阵extension.py.dox.md的 Verification 章节强调对 helper 行为改动要运行针对性测试对涉及鉴权、文件系统、WebSocket、隧道、上传或密钥处理的 helper 要做安全回归。仓库中与扩展机制直接相关的测试包括tests/test_extensions_stress.pyextensible高频调用正确性与性能剖析tests/test_webui_extension_surfaces.pyWebUI 扩展接口get_webui_extensions/get_webui_extension_manifest的资产发现与分组语义文档中列出的关联回归测试tests/test_a0_connector_prompt_gating.py、tests/test_api_chat_lifetime.py、tests/test_browser_agent_regressions.py、tests/test_error_retry_plugin.py、tests/test_history_compression_wait.py、tests/test_model_config_api_keys.py、tests/test_oauth_codex.py——这些测试覆盖了依赖扩展机制的周边功能改动扩展运行时后应一并回归。从零编写一个可运行的扩展综合以上机制编写一个 Python 扩展的最小流程如下选择扩展点在extensions/python/扩展点/下查看现有扩展点如before_main_llm_call、tool_execute_before、hist_add_tool_result等或在agent.py中搜索call_extensions_async/call_extensions_sync确认该扩展点实际被调用的时机与传入的**kwargs放置文件在合适的扩展目录内置、usr/extensions、agent 级或项目级创建extensions/python/扩展点/文件名.py文件命名可用_NN_前缀控制执行顺序或与既有文件名重名以实现覆盖编写类继承Extension实现execute(self, **kwargs)同步或async def均可在方法内通过self.agent访问当前 Agent按需等待生效修改文件后register_extensions_watchdogs注册的看门狗会自动清除类缓存无需重启如需调试调用频率可设置环境变量EXTENSIONS_LOGNWebUI 扩展将.html/.js资产放入extensions/webui/扩展点/目录其 URL 会自动进入get_webui_extension_manifest生成的清单并注入页面或通过api/load_webui_extensions.py按过滤器拉取。相关文档docs/developer/extensions.md扩展机制的官方入门说明与适用场景判断docs/guides/create-plugin.md更轻量的插件开发路径扩展机制的推荐替代方案docs/guides/agent-profiles.mdagent profile 与扩展缓存键中profile维度的关系docs/guides/projects.md项目级扩展目录与determine_cache_key中project维度的关系【免费下载链接】agent-zeroAgent Zero AI framework项目地址: https://gitcode.com/GitHub_Trending/ag/agent-zero创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 
长河君 · 2026/9/14 7:06:56 · 阅读 1.2k Agent Zero 扩展机制深度解析：helpers/extension.py 的扩展点发现、调度与缓存架构
Agent Zero 扩展机制深度解析helpers/extension.py 的扩展点发现、调度与缓存架构【免费下载链接】agent-zeroAgent Zero AI framework项目地址: https://gitcode.com/GitHub_Trending/ag/agent-zero本篇技术指南围绕 Agent Zero 框架的扩展运行时核心模块 helpers/extension.py 及其维护文档 helpers/extension.py.dox.md 展开系统讲解 Agent Zero 中 Python 扩展与 WebUI 扩展的发现、调度、缓存与热更新机制。读完本文你将掌握Extension基类的编写契约、call_extensions_async/call_extensions_sync两类调度入口、extensible装饰器如何为既有函数隐式注入start/end扩展点以及扩展清单如何被注入 WebUI 页面并能在自己的 Agent Zero 实例中动手编写与调试扩展。模块定位Agent Zero 扩展运行时的唯一入口在 Agent Zero 的架构中helpers/目录存放的是被核心代码与插件复用的框架级 API见 helpers/AGENTS.md。extension.py是其中专门负责扩展Extension机制的模块其职责可以概括为三句话发现在各级扩展目录中扫描符合某个「扩展点extension point」的扩展类调度在核心流程的特定时机以同步或异步方式执行这些扩展暴露 WebUI 扩展把浏览器端HTML/JS扩展资产按扩展点分组注入到渲染后的 WebUI 页面。官方文档 docs/developer/extensions.md 指出扩展是「改变 Agent Zero 行为的高级方式」如果读者是新手应优先从插件plugin入手——插件更容易创建、测试、禁用和移除。扩展则适合在以下场景使用在某个特定的生命周期节点添加行为以可复用的方式塑造提示词prompt与核心工具紧密集成在任务开始前准备框架自有的状态。而简单的 UI 改动、一次性脚本或应该易于移除的功能都不适合用扩展实现。extension.py.dox.md明确声明了模块的所有权边界extension.py拥有运行时实现.dox.md文件则承载关于职责、契约、副作用与验证的持久化说明并要求「每当公共函数、类、持久化行为、路径/安全假设、副作用或跨模块契约发生变化时同步更新本文档」。核心常量与配置扩展目录与缓存分区模块顶部定义了一组决定扩展行为边界的常量理解它们是后续所有机制的基石常量值含义DEFAULT_EXTENSIONS_FOLDERpython/extensions内置扩展在仓库中的根目录相对仓库根USER_EXTENSIONS_FOLDERusr/extensions用户扩展根目录_EXTENSIONS_CACHE_AREAextension_folder_classes(extensions)按目录缓存「文件夹→扩展类」的分区名_CLASSES_CACHE_AREAextension_classes(extensions)按「agent扩展点」缓存「扩展点→类列表」的分区名_WEBUI_MANIFEST_CACHE_AREAwebui_extension_manifest(extensions)(plugins)WebUI 扩展清单缓存分区_WEBUI_MANIFEST_SUFFIXES{html: (.html, .htm, .xhtml), js: (.js, .mjs)}WebUI 资产类型与其文件后缀的映射_UNSET_Unset()哨兵实例用于标记「结果尚未设置」的内部哨兵_EXTENSIONS_LOG_COUNTSdict[str, int]扩展调用计数供调试日志使用值得注意源码中还保留了cache.toggle_area(_EXTENSIONS_CACHE_AREA, False)与cache.toggle_area(_CLASSES_CACHE_AREA, False)的注释示例说明这两个分区可以按需被关闭见 helpers/cache.py 的toggle_area实现。当前仓库中内置的 Python 扩展点目录位于 extensions/python实际存在的扩展点包括agent_init、banners、before_main_llm_call、error_format、hist_add_before、hist_add_tool_result、job_loop、message_loop_end、message_loop_prompts_after、message_loop_prompts_before、message_loop_start、monologue_end、monologue_start、process_chain_end、reasoning_stream、reasoning_stream_chunk、reasoning_stream_end、response_stream、response_stream_chunk、response_stream_end、startup_migration、system_prompt、tool_execute_after、tool_execute_before、user_message_ui、util_model_call_before、webui_ws_connect、webui_ws_disconnect、webui_ws_event等另有_functions/目录专门承载extensible装饰器生成的隐式扩展点下文详述。Extension 基类编写一个扩展的最小契约所有 Python 扩展都必须继承Extension抽象基类其定义如下class Extension: def __init__(self, agent: Agent|None, **kwargs): self.agent: Agent|None agent self.kwargs kwargs abstractmethod def execute(self, **kwargs) - None | Awaitable[None]: pass契约要点构造器接收当前agent可能为None以及任意关键字参数并把它们分别保存在self.agent与self.kwargs上execute是唯一的抽象方法返回类型可以是None或可等待对象Awaitable[None]即扩展既可以写成普通同步函数也可以写成async def调度方会根据返回值是否为Awaitable决定是否需要await。仓库内置了一个可直接参考的最小示例 agents/_example/extensions/agent_init/_10_example_extension.py它在 agent 初始化时把 agent 改名为 SuperAgent numberfrom helpers.extension import Extension # this is an example extension that renames the current agent when initialized # see /extensions folder for all available extension points class ExampleExtension(Extension): async def execute(self, **kwargs): # rename the agent to SuperAgent0 self.agent.agent_name SuperAgent str(self.agent.number)注意文件命名_10_example_extension.py中的_10前缀不是装饰而是排序与覆盖控制的手段——_get_extension_classes最终会对类按文件名排序详见下文「扩展点调度」一节文件名前缀可用于控制同一扩展点内多个扩展的执行顺序。扩展点调度异步与同步双入口extension.py对外暴露两个对称的调度函数分别用于异步与同步上下文async def call_extensions_async( extension_point: str, agent: Agent|None None, **kwargs ): _log_extension_call(extension_point) # fetch classes for this extension point and agent classes _get_extension_classes(extension_point, agentagent, **kwargs) # execute unique extensions for cls in classes: result cls(agentagent).execute(**kwargs) if isinstance(result, Awaitable): await result def call_extensions_sync(extension_point: str, agent: Agent|None None, **kwargs): _log_extension_call(extension_point) # fetch classes for this extension point and agent classes _get_extension_classes(extension_point, agentagent, **kwargs) # execute unique extensions for cls in classes: result cls(agentagent).execute(**kwargs) if isinstance(result, Awaitable): raise ValueError( fExtension {cls.__name__} returned awaitable in sync mode )两者共享同一套逻辑先调用_log_extension_call记录调用再通过_get_extension_classes拿到该扩展点且针对该 agent应执行的类列表然后逐类实例化并调用execute(**kwargs)。唯一的区别是异步入口遇到Awaitable返回值会await它同步入口若遇到可等待返回值会直接抛出ValueError——这保证了在同步调用链上不会出现「协程被静默丢弃」的隐患。_log_extension_call是一个内置的调试计数器。它读取环境变量EXTENSIONS_LOG取整数值作为「每 N 次调用打印一次」的间隔每调用一次就把该扩展点计数加一并维护_total总数当总数达到 N 的整数倍时把当前所有计数打印出来。开发者可通过设置EXTENSIONS_LOG100之类的环境变量观察扩展调用频率无需改动任何代码。核心流程中的调用示例在 agent.py 中这些调度入口被大量使用例如# 同步扩展点agent 初始化完成后调用 extension.call_extensions_sync(agent_init, self) # 异步扩展点流程链结束时调用 await extension.call_extensions_async(process_chain_end, agentself.get_agent(), data{})此外api/banners.py 在生成横幅时调用await call_extensions_async(banners, agentNone, bannersbanners, frontend_contextfrontend_context)说明扩展点同样支持agentNone的全局场景。extensible 装饰器为既有函数注入隐式扩展点除了显式调用扩展点extension.py还提供了一种「无侵入」的扩展方式用extensible装饰任意函数该函数在执行前后会自动各产生一个扩展点。装饰器文档对此有完整说明从被包装函数自动推导两个扩展点目录路径_functions/模块路径/限定名路径/start_functions/模块路径/限定名路径/end模块路径段来自func.__module__按.切分限定名路径段来自完整的嵌套func.__qualname__按.切分并剔除locals段。示例模块helpers.something、限定名Outer.Inner.__init__会生成_functions/helpers/something/Outer/Inner/__init__/start_functions/helpers/something/Outer/Inner/__init__/end被包装函数被调用时装饰器构造一个可变的data载荷传给两个扩展点键初始值扩展可执行的操作data[args]位置参数替换/修改后影响原函数调用data[kwargs]关键字参数替换/修改后影响原函数调用data[result]内部哨兵_UNSET若设置则短路原函数不再执行data[exception]None若设置为BaseException实例则最终强制抛出执行流程为start扩展先执行可以修改入参或直接设置data[result]/data[exception]实现短路若data[result]仍是_UNSET装饰器用可能被修改过的data[args]/data[kwargs]调用原函数end扩展最后执行可以改写data[result]或替换/清除data[exception]最终若data[exception]中有异常则抛出否则返回data[result]。核心实现中_prepare_inputs负责从args/kwargs中探测Agent实例通过_get_agent先查kwargs.get(agent)再遍历args且要求该实例有非空__dict__并把扩展点路径、agent 与data打包返回若函数缺少__module__或__qualname__则跳过扩展逻辑、直接调用原函数。同步与异步的判别发生在装饰器装配阶段if inspect.iscoroutinefunction(func): return wraps(func)(_run_async) return wraps(func)(_run_sync)即被装饰函数若是async def则使用_run_async包装内部通过call_extensions_async调度且对原函数返回值做await处理否则使用_run_sync内部通过call_extensions_sync调度。wraps保证了元数据如__name__、__doc__不被破坏。从源码结构看agent.py中有 30 余处方法使用了extension.extensible见 agent.py 中诸如extension.extensible的注解api/plugins.py 中也有 3 处覆盖了消息处理、工具执行、流程链等关键路径。仓库测试 tests/test_extensions_stress.py 专门对extensible做了 10000 次迭代的性能剖析cProfile验证class PerfAgent(Agent): extensible def perf_hook(self, value: int): return value 1 pytest.mark.parametrize(iterations, [10000]) def test_extensible_method_performance_trace(iterations: int): ... for i in range(iterations): result agent.perf_hook(i) ... assert result iterations该测试断言连续调用 10000 次带扩展点的同步方法后结果正确并打印累计耗时统计说明extensible被设计为可高频使用而不会破坏调用语义的机制。扩展发现与去重_get_extension_classes 的覆盖语义_get_extension_classes是扩展调度背后的核心查找函数def _get_extension_classes( extension_point: str, agent: Agent|None None, **kwargs ) - list[Type[Extension]]: from helpers import subagents cache_key cache.determine_cache_key(agent, extension_point) cached cache.get(_CLASSES_CACHE_AREA, cache_key) if cached is not None: return cached # search for extension folders in all agents paths paths subagents.get_paths(agent, extensions/python, extension_point) all_exts [cls for path in paths for cls in _get_extensions(path)] # merge: first ocurrence of file name is the override unique {} for cls in all_exts: file _get_file_from_module(cls.__module__) if file not in unique: unique[file] cls classes sorted( unique.values(), keylambda cls: _get_file_from_module(cls.__module__) ) cache.add(_CLASSES_CACHE_AREA, cache_key, classes) return classes关键语义有三点路径来源subagents.get_paths(agent, extensions/python, extension_point)会收集该 agent 全部生效路径下的同名扩展点目录内置extensions/python、用户级usr/extensions、agent 级与项目级扩展目录等这与 helpers/extension.py 中register_extensions_watchdogs监控的根目录一一对应。文件名覆盖override合并时以「文件名的最后一个模块段」为键做去重首次出现即胜出——即路径列表中靠前的目录里的同名扩展文件会覆盖后面的。这一机制让用户可以用同名文件覆盖内置扩展行为。确定性排序最终按文件名_get_file_from_module返回module_name.split(.)[-1]排序因此_10_xxx.py会排在_20_xxx.py之前执行顺序稳定可控。_get_extensions则负责单目录内的类加载def _get_extensions(folder: str): folder files.get_abs_path(folder) cached cache.get(_EXTENSIONS_CACHE_AREA, folder) if cached is not None: return cached if not files.exists(folder): return [] classes modules.load_classes_from_folder(folder, *, Extension) cache.add(_EXTENSIONS_CACHE_AREA, folder, classes) return classes底层类加载由 helpers/modules.py 的load_classes_from_folder完成按字母序扫描目录内所有匹配*的.py文件用importlib.util.spec_from_file_location逐个导入再通过inspect.getmembers反向遍历类成员筛选出「是Extension的真子类」的类one_per_fileTrue时每个文件只取第一个从而把「每个文件一个扩展」固化为约定。缓存体系扩展点与 Agent 的绑定关系扩展类的查找结果被缓存在 helpers/cache.py 中其键由cache.determine_cache_key(agent, *additional)决定def determine_cache_key(agent, *additional): if agent: profile agent.config.profile or none project agent.context.get_data(project) or none return (profile, project, *additional) return (none, none, *additional)也就是说扩展类缓存的键 (agent 配置 profile, 当前项目, 扩展点)。这带来一个重要推论不同的 agent profile 与项目会得到彼此独立的扩展集合扩展的生效范围天然与 profile/项目绑定。cache.add/cache.get会检查分区是否被toggle_area关闭源码中保留了关闭_EXTENSIONS_CACHE_AREA与_CLASSES_CACHE_AREA的注释示例而cache.clear(area)支持*?[通配符模糊清理多个分区。WebUI 扩展资产发现与清单注入WebUI浏览器端扩展是另一类一等公民由get_webui_extensions与get_webui_extension_manifest两个函数支撑。按需拉取get_webui_extensionsget_webui_extensions(agent, extension_point, filters)用于按扩展点且可选文件过滤器拉取 WebUI 资产路径def get_webui_extensions( agent: Agent | None, extension_point: str, filters: list[str] | None None ): from helpers import subagents entries: list[str] [] effective_filters filters or [*] # search for extension folders in all agents paths folders subagents.get_paths( agent, extensions/webui, extension_point, ) extensions [] for folder in folders: for filter in effective_filters: pattern files.get_abs_path(folder, filter) extensions.extend(files.find_existing_paths_by_pattern(pattern)) for extension in extensions: rel_path files.deabsolute_path(extension) entries.append(rel_path) return entries逻辑要点默认过滤器是[*]全部文件对每个生效目录 × 每个过滤器组合出绝对路径模式用files.find_existing_paths_by_pattern找到实际存在的文件最后统一转为相对路径返回。该函数被 api/load_webui_extensions.py 暴露为 HTTP 接口前端可动态请求某个扩展点的资源。全量清单get_webui_extension_manifestget_webui_extension_manifest一次性返回「所有 WebUI 扩展 URL按资产类型和扩展点分组」的清单其结构为dict[str, dict[str, list[str]]]第一层键为html与js对应_WEBUI_MANIFEST_SUFFIXES第二层键为扩展点资产所在目录值为 URL 列表cache_key cache.determine_cache_key(agent) cached cache.get(_WEBUI_MANIFEST_CACHE_AREA, cache_key) if cached is not None: return cached manifest: dict[str, dict[str, list[str]]] { asset_type: {} for asset_type in _WEBUI_MANIFEST_SUFFIXES } roots subagents.get_paths(agent, extensions/webui) for root in roots: relative_files sorted(files.list_files_in_dir_recursively(root)) for asset_type, suffixes in _WEBUI_MANIFEST_SUFFIXES.items(): for suffix in suffixes: for relative_file in relative_files: if not relative_file.lower().endswith(suffix): continue extension_point os.path.dirname(relative_file).replace( os.sep, / ) if not extension_point or extension_point .: continue absolute_path files.get_abs_path(root, relative_file) relative_path files.deabsolute_path(absolute_path).replace( os.sep, / ) manifest[asset_type].setdefault(extension_point, []).append( / relative_path.lstrip(/) ) cache.add(_WEBUI_MANIFEST_CACHE_AREA, cache_key, manifest) return manifest实现细节对每个 WebUI 扩展根目录做递归文件列举并排序从而保持根目录与过滤器顺序的确定性资产按后缀归入html.html/.htm/.xhtml或js.js/.mjs资产所在目录os.path.dirname规范化后的相对路径即扩展点根目录下散落的文件extension_point .会被跳过最终 URL 以/开头并缓存在_WEBUI_MANIFEST_CACHE_AREA分区键为determine_cache_key(agent)即 profile项目。该清单被 helpers/ui_server.py 在渲染 WebUI 首页时注入先json.dumps序列化再对、、做 HTML 转义\u0026、\u003c、\u003e最后通过files.replace_placeholders_text替换 index 模板中的占位符从而把扩展清单安全地嵌进页面。对应的测试 tests/test_webui_extension_surfaces.py 验证了清单的语义def test_webui_extension_manifest_groups_plugin_assets_by_type_and_surface() - None: surface manifest-probe with _temporary_probe_plugin(surface) as (plugin_id, probe_file_name): manifest get_webui_extension_manifest(agentNone) expected_suffix ( f/{plugin_id}/extensions/webui/{surface}/{probe_file_name} ) assert any( path.endswith(expected_suffix) for path in manifest[html].get(surface, []) ) assert surface not in manifest[js]该测试临时创建一个探测插件确认其 HTML 资产被正确归入manifest[html][surface]且不会出现在js分组中——印证了「插件内extensions/webui/扩展点/目录下的资产会并入全局 WebUI 扩展清单」的机制。热更新扩展文件系统看门狗扩展代码在运行期发生变化时框架通过register_extensions_watchdogs()注册三类文件系统看门狗来失效缓存从而实现无需重启即可让新扩展生效def register_extensions_watchdogs(): from helpers import watchdog, projects def extensions_changed(items: list[watchdog.WatchItem]): cache.clear(_EXTENSIONS_CACHE_AREA) cache.clear(_CLASSES_CACHE_AREA) PrintStyle.debug(Extensions watchdog triggered:, items) # extensions and usr/extensions watchdog.add_watchdog( idextensions_base, roots[ files.get_abs_path(files.EXTENSIONS_DIR), files.get_abs_path(files.USER_DIR, files.EXTENSIONS_DIR), ], handlerextensions_changed, ) # usr/projects/**/extensions watchdog.add_watchdog( idextensions_projects, roots[projects.PROJECTS_PARENT_DIR], patterns[f*/{projects.PROJECT_META_DIR}/**/{files.EXTENSIONS_DIR}/**/*], handlerextensions_changed, ) # agents and usr/agents watchdog.add_watchdog( idextensions_agents, roots[ files.get_abs_path(files.AGENTS_DIR), files.get_abs_path(files.USER_DIR, files.AGENTS_DIR), ], patterns[f*/{files.EXTENSIONS_DIR}/**/*], handlerextensions_changed, )三个看门狗覆盖了扩展可能存放的全部位置看门狗 ID监控根目录监控模式目的extensions_base内置extensions与usr/extensions整个目录内置/用户级扩展变化extensions_projects项目父目录projects.PROJECTS_PARENT_DIR*/项目元目录/**/extensions/**/*项目级扩展变化extensions_agentsagents与usr/agents*/extensions/**/*agent 级扩展变化任一事件触发后处理器extensions_changed会同时清除_EXTENSIONS_CACHE_AREA与_CLASSES_CACHE_AREA两个缓存分区并通过PrintStyle.debug打印触发详情——这解释了为什么新增/修改扩展文件后无需重启框架即可生效。验证与测试矩阵extension.py.dox.md的 Verification 章节强调对 helper 行为改动要运行针对性测试对涉及鉴权、文件系统、WebSocket、隧道、上传或密钥处理的 helper 要做安全回归。仓库中与扩展机制直接相关的测试包括tests/test_extensions_stress.pyextensible高频调用正确性与性能剖析tests/test_webui_extension_surfaces.pyWebUI 扩展接口get_webui_extensions/get_webui_extension_manifest的资产发现与分组语义文档中列出的关联回归测试tests/test_a0_connector_prompt_gating.py、tests/test_api_chat_lifetime.py、tests/test_browser_agent_regressions.py、tests/test_error_retry_plugin.py、tests/test_history_compression_wait.py、tests/test_model_config_api_keys.py、tests/test_oauth_codex.py——这些测试覆盖了依赖扩展机制的周边功能改动扩展运行时后应一并回归。从零编写一个可运行的扩展综合以上机制编写一个 Python 扩展的最小流程如下选择扩展点在extensions/python/扩展点/下查看现有扩展点如before_main_llm_call、tool_execute_before、hist_add_tool_result等或在agent.py中搜索call_extensions_async/call_extensions_sync确认该扩展点实际被调用的时机与传入的**kwargs放置文件在合适的扩展目录内置、usr/extensions、agent 级或项目级创建extensions/python/扩展点/文件名.py文件命名可用_NN_前缀控制执行顺序或与既有文件名重名以实现覆盖编写类继承Extension实现execute(self, **kwargs)同步或async def均可在方法内通过self.agent访问当前 Agent按需等待生效修改文件后register_extensions_watchdogs注册的看门狗会自动清除类缓存无需重启如需调试调用频率可设置环境变量EXTENSIONS_LOGNWebUI 扩展将.html/.js资产放入extensions/webui/扩展点/目录其 URL 会自动进入get_webui_extension_manifest生成的清单并注入页面或通过api/load_webui_extensions.py按过滤器拉取。相关文档docs/developer/extensions.md扩展机制的官方入门说明与适用场景判断docs/guides/create-plugin.md更轻量的插件开发路径扩展机制的推荐替代方案docs/guides/agent-profiles.mdagent profile 与扩展缓存键中profile维度的关系docs/guides/projects.md项目级扩展目录与determine_cache_key中project维度的关系【免费下载链接】agent-zeroAgent Zero AI framework项目地址: https://gitcode.com/GitHub_Trending/ag/agent-zero创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考