跳转至

子智能体与取消

子智能体是「上下文隔离的一次性智能体」:默认空白会话、可命名、可级联取消、深度受限。

内置工具

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()。