扩展系统¶
把 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) |