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_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 |
conversation |
Conversation |
新建 | 消息历史 |
toolbox |
ToolBox |
空 | 需自行 with_defaults() 或 register() |
command |
CommandRegistry |
空 | 会话命令 |
workspace |
Workspace |
workdir=Path.cwd() |
工作目录 + 惰性临时目录 |
cancel_event |
ChainedEvent |
新建 | 可挂到父智能体令牌之下 |
config |
AgentConfig |
load_config().clone() |
全局配置的深拷贝;想完全不读磁盘与环境变量用 AgentConfig.default() |
api_call_semaphore |
Semaphore |
3 | 并发模型调用上限,父子共享同一信号量 |
state |
dict[str, Any] |
{} |
自由挂载点(子智能体深度、浏览器会话标记等) |
hooks |
Hooks |
新建 | 钩子注册表 |
compactor |
CompactorAbstract |
AutoCompactor() |
初始化时安装到 before_execution_step |
生命周期¶
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。它是幂等的:对已初始化的智能体调用会原样返回。
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,状态不变。
执行¶
@overload
def execute(self, schema: type[T], max_iterations: int = 256, context: Any = None) -> T
@overload
def execute(self, schema: None = None, max_iterations: int = 256, 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默认256,可用内部环境变量_XUN_DEFAULT_MAX_ITER覆盖;用尽视为错误,返回Err(ErrorInfo("Maximum tool call iterations exceeded.", ...))。context会原样传给声明了ToolCallContext参数的工具,见 工具系统。- 传入
schema(pydantic 模型)即结构化输出:同时使用response_format(strict=False)与提示词内嵌 schema 双保险,解析时先用json_repair修复再model_validate。结构化输出要求会话里已存在用户消息(先instruct())。
KeyboardInterrupt 与 CancelledError 不会被吞掉,会向上抛出。
循环内部¶
- 触发
before_execution(参数可就地修改:schema、max_iterations、context_value)。 - 每一轮:
check_cancel()→before_execution_step(自动压缩在此)→ 流式调用模型(受api_call_semaphore限制,stream_options={"include_usage": True})。 - 文本/推理事件通过
model_text_delta、model_reasoning_delta流出(钩子可改写内容)。 - 若有工具调用:
before_tool_call(列表可编辑)→ 参数用json_repair修复并回写到 assistant 消息 → 逐个执行 →after_tool_call→ 写入会话。 - 工具异常不会中断运行,转成
Err结果交给模型继续处理;模型调用异常最多重试 3 次,每次询问Retry?(自动确认下默认同意)。 - 某一轮不再调用工具即结束,该轮文本即结果。
只有当 toolbox.list_tools_json() 非空时才带 tools 与 tool_choice="auto";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)。作用域退出时令牌会被清除,因此被取消过的智能体还能继续复用。
派生子智能体:Agent.inherit()¶
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, # 取消级联
)
config 取父配置的克隆,api_call_semaphore 与父共享(并发预算合并);state、hooks、compactor 不继承(按默认值新建,扩展会在子智能体 initialize() 时重放)。详见 子智能体与取消。