Agent 初始化¶
推荐路径是直接用 Agent 类:它的每个部件都是一个可替换的对象。
只想快速跑起来时用 setup_agent,它把内置工具、系统提示、会话命令与显示层一次性配好并返回已初始化的智能体。
最短路径:setup_agent¶
from xun import setup_agent, interactive_session
def weather(city: str) -> str:
"""Query the weather for a given city."""
return f"Sunny in {city}"
agent = setup_agent(tools=[weather]) # 已 initialize()
interactive_session(agent, task="What is the weather in Paris?")
def setup_agent(
name: str = "agent",
tools: list[Callable] = [],
default_tools: bool = False,
default_system_prompt: bool = True,
default_commands: bool = True,
display: DisplayAbstract | None = None,
workdir: Path | str | None = None,
) -> Agent[Agent.T.Init]: ...
| 参数 | 默认 | 说明 |
|---|---|---|
name |
"agent" |
智能体名字,出现在显示层与事件中 |
tools |
[] |
追加的自定义工具函数 |
default_tools |
False |
是否装载 33 个内置工具;同时会开启子智能体派生工具与工具类会话命令 |
default_system_prompt |
True |
写入内置系统提示 |
default_commands |
True |
注册内置会话命令(/help 等) |
display |
None |
None 时使用控制台显示 Display();程序化调用想静默请传 NullDisplay() |
workdir |
None |
工作目录,None 取当前目录 |
初始化在最后
setup_agent 在所有配置完成后才调用 initialize(),
因此 after_initialize 钩子看到的是一个配置完整的智能体。
完整路径:Agent 类¶
Agent 是一个 dataclass,每个字段都是一个可以被替换的核心对象:
| 字段 | 默认 | 派生时是否继承 | 作用 |
|---|---|---|---|
name |
agent-<uuid8> |
新名字 | 显示与事件中的标识 |
identifier |
uuid | 新 uuid | 进程内唯一 ID,显示层按它区分智能体 |
display |
NullDisplay() |
共享引用 | 显示层,见显示扩展 |
conversation |
Conversation() |
默认不拷贝 | 消息序列与系统提示段落 |
toolbox |
ToolBox() |
克隆 | 工具箱,见工具设置 |
command |
CommandRegistry() |
克隆 | 会话命令 |
workspace |
Workspace() |
共享同一对象 | 工作目录 + 惰性临时目录 |
cancel_event |
ChainedEvent() |
挂到父事件下 | 取消链,见链式取消 |
config |
load_config().clone() |
克隆 | 该智能体的独立配置副本 |
api_call_semaphore |
3 | 共享引用 | 全进程模型并发上限 |
extension_loader |
全局加载器 | 共享引用 | 扩展在每次 initialize() 重放 |
state |
AgentState() |
不继承 | 类型化状态容器 |
hooks |
Hooks() |
不继承 | 生命周期钩子 |
compactor |
AutoCompactor() |
不继承 | 会话压缩策略 |
from xun import Agent, ToolBox
def weather(city: str) -> str:
return "The weather in " + city + " is sunny."
agent = Agent(toolbox=ToolBox().register(weather)).initialize()
answer = agent.instruct("What is the weather in Paris?").execute()
print(answer.unwrap())
链式写法(README 中的形式):
greeting = (
Agent() # Agent[Uninit]
.initialize() # Agent[Init]
.instruct("Hi~") # Agent[Init]
.execute() # Result[str, ErrorInfo]
.unwrap() # str
)
initialize() 做了什么¶
flowchart TD
A["Agent(...)"] --> B["initialize()"]
B --> B1["扩展 setup 重放"]
B1 --> B2["模型名为空时自动探测"]
B2 --> B3["display.bind(agent)"]
B3 --> B4["workspace.prepare()"]
B4 --> B5["compactor.install(hooks)"]
B5 --> B6["after_initialize 钩子"]
B6 --> B7["同步 AGENTS.md"]
B7 --> C["Agent[Init]:可 execute"]
C --> D["finalize()"]
D --> D1["before_finalize 钩子"]
D1 --> D2["display.unbind + AgentUnbindEvent"]
D2 --> E["Agent[Final]:不可再用"]
execute() 前必须初始化,运行时也会检查并抛出带提示的 RuntimeError。
状态迁移的细节见 生命周期与类型状态。
执行与返回值¶
result = agent.instruct("写一个 100 字的故事").execute() # Result[str, ErrorInfo]
story = agent.instruct(...).execute(schema=StoryModel) # Result[StoryModel, ErrorInfo]
agent.execute_command("help") # Result[None, ErrorInfo]
execute / execute_command 被 except_safe 包装,异常一律转成 Result.Err(ErrorInfo),
但 CancelledError 与 KeyboardInterrupt 会原样抛出。Result 的接口:
| 方法 | 行为 |
|---|---|
is_ok() / is_err() |
判断分支 |
unwrap() / unwrap_err() |
取值;分支不符时抛普通 Exception |
value |
原始值(可能是 T 或 E) |
value_str() / value_json() |
字符串 / JSON 形式 |
Result.from_payload(data, ok_factory, err_factory) |
从 JSON 还原 |
max_iterations 默认 512,超限抛 RuntimeError 并表现为 Result.Err;
context= 传入的对象会作为 ToolCallContext.value 交给工具,见工具设置。
派生子智能体¶
Agent.inherit(parent, ...) 返回一个 未初始化 的子智能体,可选择共享哪些部件:
child = Agent.inherit(
parent,
share_workspace=True, # 同一 Workspace 对象(同 workdir 与惰性 tempdir)
share_display=True, # 同一显示层
copy_toolbox=True, # 工具箱浅拷贝
copy_command=True, # 命令注册表浅拷贝
copy_conversation=False, # 会话默认独立
inherit_cancel_event=True, # 取消事件挂到父链上
)
config 始终克隆,state / hooks / compactor 始终新建。
子智能体在扩展与压缩策略上都重新走一遍默认,配合内置派生策略使用见 子智能体派生。
手动挂部件¶
没有 add_tool() 这样的方法——所有配置都是往对应对象上挂:
agent.toolbox.register(my_tool) # 工具(返回 ToolBox,可链式)
agent.command.register(my_command) # 会话命令
agent.hooks.before_tool_call.add(handler) # 钩子
agent.conversation.set_system_persistent_section("rules", "…") # 系统提示的持久段落
agent.state["counter"] = JsonEntry(0) # 状态条目
agent.system("你是一个……") # 覆盖系统提示正文(不覆盖持久段落)