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