子智能体与取消¶
子智能体是「上下文隔离的一次性智能体」:默认空白会话、可命名、可级联取消、深度受限。
内置工具¶
ToolBox().with_subagent_provider()(setup_agent(default_tools=True) 已包含)注册两个工具:
| 工具 | 参数 | 返回 |
|---|---|---|
agent_run |
task: str、name: str \| None = None |
Result[str, ErrorInfo] |
agent_run_parallel |
tasks: list[str]、names: list[str] \| None = None |
Result[list[Result[str, ErrorInfo]], ErrorInfo]:正常时是 Ok(list),列表顺序与输入一致;只有工具自身出错才是外层 Err |
agent_run_parallel 用线程池(默认 4 worker)并发;tasks/names 也可以整体写成字符串,由通用参数校验层解析(JSON 数组或逗号分隔,见 工具系统);给了 names 却与 tasks 长度不符时,返回的列表里只有一条 Err。只有 Ctrl+C 会打断兄弟任务(子线程收不到 SIGINT)。
自定义 getter¶
给 with_subagent_provider() 传入 getter 即可完全决定子智能体的构成:
import math
from xun import Agent, AgentGetterParam, ToolBox, ToolCallContext as Context
def sqrt(ctx: Context, a: float) -> float:
"""Square root."""
return math.sqrt(a)
def agent_getter(param: AgentGetterParam) -> Agent[Agent.T.Uninit]:
# 必须返回**未初始化**的智能体,并落实 param 里的要求(name 为 None 表示不挑名字)
agent = Agent(toolbox=ToolBox().register(sqrt))
agent.name = param.name or "subagent"
return agent
agent = Agent(toolbox=ToolBox().with_subagent_provider(agent_getter)).initialize()
agent.instruct("Compute sqrt(114514) with a sub-agent.").execute()
AgentGetterParam 只有两个字段:
| 字段 | 类型 | 说明 |
|---|---|---|
tool_context |
ToolCallContext |
由此取到父智能体 param.tool_context.agent |
name |
str \| None |
请求的名字;AgentGetterProtocol = Callable[[AgentGetterParam], Agent[Agent.T.Uninit]] |
默认 getter 的行为等价于下面这段伪代码(ctx = param.tool_context,get_subagent_prompt() 在 xun.prompt 里,未从顶层导出):
agent = Agent.inherit(ctx.agent).system(get_subagent_prompt())
if param.name:
agent.name = param.name
agent.state["__subagent_depth"] = ctx.agent.state.get("__subagent_depth", 0) # 先继承父深度
if agent.state["__subagent_depth"] >= ToolBox.SUBAGENT_MAX_DEPTH: # 3
agent.toolbox.disable_subagent()
agent.state["__subagent_depth"] += 1 # 再为下一层 +1
继承语义¶
Agent.inherit() 参数 |
默认 | 效果 |
|---|---|---|
share_workspace |
True |
父子共用 workdir 与同一个惰性临时目录 |
share_display |
True |
子智能体的事件也出现在同一界面;False 用 NullDisplay |
copy_toolbox |
True |
工具表克隆,子智能体可再改 |
copy_command |
True |
会话命令可用 |
copy_conversation |
False |
空白上下文——隔离的核心 |
inherit_cancel_event |
True |
取消令牌挂到父令牌之下 |
另外:config 克隆父配置;api_call_semaphore 与父共享(并发的子智能体共同遵守同一个模型调用上限,默认 3);hooks、state、compactor 不继承,但扩展会在 initialize() 时重放,因此扩展注册的钩子仍会出现在子智能体上。
copy_toolbox 与 copy_command 都是浅拷贝:容器是新的,里面的 Function / Command 对象仍与父共享。
深度限制:SUBAGENT_MAX_DEPTH = 3。检查发生在创建时、用的是父智能体的深度,因此最多能生成 3 代子智能体,第 4 代(其父深度已达 3)拿到的智能体里 agent_run / agent_run_parallel 已被 disable_subagent() 关掉。
取消级联¶
flowchart TD
P["父智能体取消"] --> C1["子智能体一"]
P --> C2["子智能体二"]
C2 --> C3["孙智能体"]
C3 --> X["check_cancel 抛出 CancelledError"]
ChainedEvent.is_set()沿parent链递归判断,父取消自动影响所有后代。- 主线程收到 SIGINT 时,
cancellable_execution()会主动调用agent.cancel(),从而通知工作线程里的子智能体(子线程收不到 SIGINT)。 agent_run收到CancelledError时:若是父智能体驱动取消,原样上抛;否则返回Err(ErrorInfo("Execution cancelled by user.")),让父智能体把失败当作工具结果继续处理。- 长时间运行的工具需要在循环里自行调用
ctx.agent.check_cancel():
import time
from xun import ToolCallContext as Context
def long_running(ctx: Context, steps: int = 10) -> str:
"""Do something slow but cancellable."""
for i in range(steps):
time.sleep(1)
ctx.agent.check_cancel()
return "done"
取消是协作式的
没有检查点就无法中断。bash 工具在读取输出的循环里检查并终止整个进程组;自写工具若要中断路径/网络操作,需要自己加 check_cancel()。