跳转至

Agent 初始化与执行

两条路径

from xun import setup_agent

agent = setup_agent(
    name="agent",
    tools=[my_tool],             # 自己写的工具函数
    default_tools=True,            # 注册全部内置工具 + 子智能体工具
    default_system_prompt=True,
    default_commands=True,
    workdir=".",
)
answer = agent.instruct("...").execute()
参数 类型 默认 作用
name str "agent" 智能体名
tools list[Callable] [] 追加注册的工具函数
default_tools bool False 执行 toolbox.with_defaults().with_subagent_provider();policy 命令组要 default_tools 与 default_commands 同时为真才注册
default_system_prompt bool True 注入内置系统提示词
default_commands bool True 注册内置会话命令
display DisplayAbstract \| None None 显示层,缺省为控制台 Display()
workdir Path \| str \| None None 工作目录,缺省为 Path.cwd()

内部顺序为「先配置、后初始化」:initialize() 最后调用,因此 after_initialize 钩子看到的是完整配置好的智能体。返回值类型是 Agent[Agent.T.Init]。

Agent 是 dataclass,所有字段都有默认值,可按需覆盖:

from xun import Agent, ToolBox

agent = Agent(name="writer", toolbox=ToolBox().with_defaults())
agent = agent.initialize()
字段 类型 默认 说明
name str agent-<uuid8> 显示名;子智能体为 <父名>-child-<id8>
identifier str 随机 UUID 显示层用于区分智能体的键
display DisplayAbstract NullDisplay() 静默显示;此时 get_choice 抛 NotImplementedError(auto_confirm 为真时直接返回默认项,不会走到显示层)
conversation Conversation 新建 消息历史
toolbox ToolBox 空 需自行 with_defaults() 或 register()
command CommandRegistry 空 会话命令
workspace Workspace workdir=Path.cwd() 工作目录 + 惰性临时目录
cancel_event ChainedEvent 新建 可挂到父智能体令牌之下
config AgentConfig load_config().clone() 全局配置的深拷贝;AgentConfig.default() 不读 config.json(.env 也不由它加载),仍用 os.environ 替换 ${XUN_*} 占位符,缺失即报错(仅 XUN_OPENAI_MODEL、XUN_AUTO_CONFIRM 有内置回落值)
api_call_semaphore Semaphore 3 并发模型调用上限,父子共享同一信号量
state dict[str, Any] {} 自由挂载点(当前用于子智能体深度与内置工具的策略对象)
hooks Hooks 新建 钩子注册表
compactor CompactorAbstract AutoCompactor() 初始化时安装到 before_execution_step
extension_loader ExtensionLoader 进程级 default_loader 按引用共享,子智能体复用同一次扫描

生命周期

stateDiagram-v2
  [*] --> Uninit: Agent(...)
  Uninit --> Init: initialize() 或 with
  Init --> Init: 轮次
  Init --> Final: finalize()
  Uninit --> Final: finalize()
  Final --> [*]

图中的「执行一轮」指 system / instruct / execute 的反复调用;with 语句块在进入时执行 initialize(),退出时执行 finalize()。

initialize() 依次完成:应用扩展 → 模型名自动探测(config.model.name 为空时)→ 自动确认告警 → 绑定显示层 → 准备工作区 → 安装压缩器 → 触发 after_initialize。它对已初始化的智能体是幂等的(原样返回),但对已 finalize() 的智能体抛 RuntimeError。

with Agent(toolbox=ToolBox().with_defaults()) as agent:   # 进入时 initialize
    agent.instruct("...").execute()
                                                          # 退出时 finalize

finalize() 触发 before_finalize 并解绑显示层。若智能体被垃圾回收却尚未 finalize,weakref.finalize 注册的兜底逻辑会补一次收尾。

注入消息

方法 签名 说明
system system(content: str) -> self 覆盖系统消息(会话第 0 条);可链式调用
instruct instruct(instruction, images=None, _emit_event=True) -> self 追加用户消息;images 接受路径、URL 或 PIL 图像;_emit_event=False 表示不向界面广播(子智能体内部使用)

两者对 Agent[T.Alive] 可用,返回 self,状态不变。

执行

# 两个重载(实际返回类型见下)
def execute(self, schema: type[T], max_iterations: int = 512, context: Any = None) -> T: ...
def execute(self, schema: None = None, max_iterations: int = 512, context: Any = None) -> str: ...

由于 execute 带有 @except_safe,实际返回类型是 Result[str, ErrorInfo] 或 Result[T, ErrorInfo]:

Result API 说明
is_ok() / is_err() 判定
unwrap() / unwrap_err() 取值(对 Err 调用 unwrap() 会抛异常)
.value 直接取原始值
value_json() / value_str() JSON 对象 / 字符串形式(工具结果落盘用)
  • max_iterations 默认 512,可用内部环境变量 _XUN_DEFAULT_MAX_ITER 覆盖;用尽即失败:先广播一条 ErrorEvent,再抛 RuntimeError,最终返回 Err(ErrorInfo("Maximum tool call iterations exceeded.", ...))。
  • context 被包进 ToolCallContext 后注入给声明了该参数的工具,工具里用 ctx.value 取回原值,见 工具系统。
  • 传入 schema(pydantic 模型)即结构化输出:同时使用 response_format(strict=False)与提示词内嵌 schema 双保险,解析时先用 json_repair 修复再 model_validate。结构化输出要求会话的最后一条消息是用户消息:schema 说明被追加到末条用户消息上,所以每轮 execute(schema=...) 前都得先 instruct(),否则 ValueError(同样被转成 Err)。

KeyboardInterrupt 与 CancelledError 不会被吞掉,会向上抛出。

循环内部

  1. 触发 before_execution(参数可就地修改:schema、max_iterations、context_value;若钩子设置了 takeover_result,整个循环被跳过并直接返回它,见 钩子与会话命令)。
  2. 每一轮:check_cancel() → before_execution_step(自动压缩在此)→ 流式调用模型(受 api_call_semaphore 限制,单次请求超时 900 秒——首 token 往往要等很久,stream_options={"include_usage": True})。
  3. 文本/推理事件通过 model_text_delta、model_reasoning_delta 流出(钩子可改写内容)。
  4. 若有工具调用:before_tool_call(列表可编辑)→ 参数用 json_repair 修复并回写到 assistant 消息 → 逐个执行 → after_tool_call → 写入会话 → after_execution_step(defer_tool_image 注入的图片在这一步落地)。
  5. 工具异常不会中断运行,转成 Err 结果交给模型继续处理;模型调用异常(只回了 usage 的空响应也算)最多重试 3 次,每次询问 Retry?(自动确认下默认同意);provider 上报不出 token usage 时本次执行以错误结束。
  6. 某一轮不再调用工具即结束,该轮文本即结果。

只有当 toolbox.list_tools_json(config.model.capabilities) 非空时才带 tools 与 tool_choice="auto"(按模型能力过滤,required_capabilities 不满足的工具不下发);temperature、reasoning_effort 仅在非 None 时下发。

取消

API 说明
agent.cancel() 置位取消令牌,返回是否为「运行中取消」;空闲时返回 False 且不改状态
agent.check_cancel() 长工具内部自检,命中即抛 CancelledError
agent.is_running 是否处于运行中
with agent.cancellable_execution(): 手动建立运行作用域(run_start / run_end 由此触发,可嵌套)

取消是协作式的:循环在每轮开始、每个流式增量、每个工具调用前检查;父智能体取消会级联到所有子智能体(ChainedEvent.parent)。作用域退出时令牌被清除,因此被取消过的智能体还能继续复用。级联细节见 子智能体与取消。

派生子智能体

child = Agent.inherit(
    parent,
    share_workspace=True,      # 共用 workdir 与同一个惰性临时目录
    share_display=True,        # False 则子智能体使用 NullDisplay
    copy_toolbox=True,         # 工具表浅拷贝
    copy_command=True,
    copy_conversation=False,   # 子智能体默认空白上下文
    inherit_cancel_event=True, # 取消级联
)

逐参数说明、浅拷贝语义与取消级联见 子智能体与取消。