跳转至

扩展系统

把 Python 文件放进 $XUN_HOME/extensions/,每个智能体在 initialize() 时都会重放它们的副作用:注册工具、挂钩子、加命令、改配置。无需注册表,也无需在调用方接线。

扩展是可信代码

扩展在导入期与初始化期以完整进程权限运行,性质等同 shell 的 rc 文件。只放你自己信任的代码。

两种形态

形态 路径 说明
包 extensions/{name}/setup_extension.py 入口文件名固定;可以相对导入同目录的兄弟模块
扁平 extensions/{name}.py 零仪式;不支持相对导入
  • 目录里没有 setup_extension.py 会告警并跳过;两种形态同名时包优先,扁平文件被屏蔽。
  • 以 . 或 __ 开头的条目被忽略;非 .py 文件(例如目录里的 README.md)静默跳过。
  • 应用顺序按名称排序,两种形态统一排序,结果确定。

入口函数

模块必须提供可调用的 setup_extension(ctx):

"""Log every tool call."""          # docstring 首行 = 扩展描述,/extensions 会显示
from xun import ExtensionContext

def setup_extension(ctx: ExtensionContext) -> None:
    ctx.agent.hooks.before_tool_call.add(lambda args: print(args.tool_calls))

ExtensionContext 暴露的公开 API 只有两项:

成员 说明
ctx.agent 正在初始化的智能体(Agent[Agent.T.Uninit])
ctx.name 扩展名(目录名或文件名)

一切效果都通过 ctx.agent 完成:toolbox.register(...)、hooks.*.add(...)、command.register(...)、config 就地改、state、system(...)、display、workspace。

版本兼容门槛

扩展可以在入口函数上声明它适配的 xun API 版本区间(两端闭区间):

"""Needs a recent xun."""
from xun import ExtensionContext, extension_attr

@extension_attr(api_min_version="1.2", api_max_version="2.0")
def setup_extension(ctx: ExtensionContext) -> None:
    ...
  • 区间随 ExtensionAttr 挂在函数上,扫描时与 xun_version() 比较:只比数字主干(1.2rc3 视作 1.2),短的一方补零。
  • 区间外的扩展状态为 skipped,不算 failed:setup_extension 不会执行,扫描时告警一次并在 /extensions 里附原因。
  • 源码运行(无包元数据)时 xun_version() 返回 None,门槛直接放行。

加载时序

flowchart TD
  A["Agent.initialize()"] --> B{"启用扩展?"}
  B -->|否| Z["跳过"]
  B -->|是| C["扫描并导入<br/>按名称排序 · 每进程一次"]
  C --> D{"版本区间内?"}
  D -->|否| S["skipped:告警"]
  D -->|是| E["执行 setup_extension"]
  E -->|成功| G["loaded"]
  E -->|异常| F["failed:告警后继续"]
  S --> H["继续初始化"]
  G --> H
  F --> H

关键语义:

  • 扫描与导入每个进程只做一次并按目录缓存(模块级副作用只跑一次,模块状态保留),setup_extension 则每个智能体各跑一次——包括每个子智能体。摘要器与命令风险评估用的两个内部辅助智能体会把 enable_extensions 置为 False,不重放扩展。
  • 扩展在模型名自动探测之前执行,因此改 config.model.name 会生效。
  • 导入失败或 setup_extension 抛异常只打印 Extension warning: 并记下状态(failed),不阻断启动,也不影响其它扩展;但同一个 setup 内后续语句会随异常中断。KeyboardInterrupt 与 CancelledError 例外——它们会向上抛出,中断 initialize()。
  • 已加载的模块不会卸载(子智能体重放依赖它们)。
  • 关闭方式:config.enable_extensions = false(配置或扩展里改),或 Agent.config.enable_extensions = False 后再 initialize()。

查看扩展状态

启动时不再打印 Loaded extensions: 汇总,/extensions 是唯一入口(导入或初始化失败时仍会打印 Extension warning:)。终端渲染为表格,Web 端由 ShowExtensionsEvent 渲染,状态共四种:

状态 含义
loaded setup_extension 至少成功执行过一次
uninitialized 已完成扫描导入,setup 还没执行
skipped 声明的版本区间不含当前 xun,原因写在 reason 字段(终端并入说明列显示)
failed 导入失败或 setup 抛异常,原因写在 reason 字段(终端并入说明列显示)

状态取值即 xun.extension.ExtensionStatus;Extension、ExtensionAttr(字段 api_min_version / api_max_version)与 ExtensionInfo 也定义在 xun.extension,包顶层只导出 ExtensionContext 与 extension_attr。

加载器 API

Agent.extension_loader 默认指向进程级的 default_loader(目录 {XUN_HOME}/extensions/),子智能体按引用共享同一个 loader。要换目录或换发现规则,就在 initialize() 之前替换它:

from pathlib import Path
from xun.extension import ExtensionLoader

agent.extension_loader = ExtensionLoader(get_extension_dir=lambda: Path("plugins"))
成员 说明
scan() 每个候选一个 Result[Extension, ExtensionIssue],按名称排序;结果按目录缓存
imported() 导入成功的扩展
infos() 全部已发现扩展的 ExtensionInfo(名称、描述、路径、状态、原因),/extensions 用的就是它
clear_scan_cache() 让下一次 scan() 重新导入,已记录的状态保留
apply(agent) 对一个未初始化智能体依次执行 setup;initialize() 内部调用

仓库内示例

仓库的 extensions/ 里现有 web_display_log.py 与 z_search.py,两个都是扁平形态;下面是去掉注释与细节后的骨架。

"""Console log for web display."""
from xun import ExtensionContext, WebDisplay

def setup_extension(ctx: ExtensionContext) -> None:
    if not isinstance(ctx.agent.display, WebDisplay):
        return
    agent_id = f"{ctx.agent.name} ({ctx.agent.identifier})"
    ctx.agent.hooks.before_display_error.add(
        lambda args: print(f"{agent_id} Error: {args.message}")
    )

先看显示层类型再挂钩子,是避免在无关智能体上产生副作用的常用写法。

"""Override web_search with an external Web Search tool."""
from xun import ExtensionContext, tool_attr

try:
    from some_sdk import Client
    _client = Client()
except Exception as exc:                       # 可选依赖或密钥缺失
    _client = None
    print(f"extension disabled: {exc}")

def setup_extension(ctx: ExtensionContext) -> None:
    if _client is None:
        return                                 # 静默失效;改成 raise 则记为 failed

    @tool_attr(name="web_search", override=True)
    def my_search(query: str, limit: int = 5) -> list[dict]:
        """Search the web."""
        return _client.search(query, count=limit)

    ctx.agent.toolbox.register(my_search)

override=True 让同名工具静默替换内置实现;可选依赖的探测放在模块级,setup 里只判断可用性——return 是静默失效,raise 会被记为 failed 并打印告警。要用包形态,把示例文件挪进 {name}/setup_extension.py 即可,其余不变。

与其它机制的关系

想做的事 更合适的机制
只给某个智能体加东西 直接调用 agent.toolbox.register() / agent.hooks.*.add()
给所有智能体加东西、跨项目共享 扩展(本文)
只改配置 配置文件;或扩展里改 ctx.agent.config
换系统提示词 agent.system(),或 setup_agent(default_system_prompt=False)