跳转至

显示层与 Web 服务

智能体不直接 print,而是向 DisplayAbstract 广播事件;所有确认请求也经同一层发出。这样同一份执行逻辑可以落到终端、浏览器或完全静默。

调用侧 API(智能体内部使用)

方法 触发的钩子 发出的事件
display_event(payload) 无 任意事件(不限状态)
info(message) before_display_info InfoEvent
warning(message) before_display_warning WarningEvent
error(message) before_display_error ErrorEvent
get_choice(prompt, choices, ...) 无 ConfirmEvent(source 为 "user" 或 "auto")
get_confirm(prompt, ..., default=True) 无 同上(选项为 Yes/No)

除 display_event 外都要求智能体处于 Init 状态;三个显示钩子可就地改写 args.message,常用于过滤或加前缀(见 扩展系统 的示例)。

事件模型

事件为 pydantic 模型,外层统一信封 DisplayEvent:timestamp、name(类名)、agent(AgentInfo:name / identifier / workdir)、payload。

事件 关键字段
UserMessageEvent content、images(kind 为 url 或 base64)
UserCommandEvent name、arguments
ModelWorkingEvent model_call_id、remaining_iterations
ModelMessageEvent model_call_id、content、reasoning、total_tokens
ToolCallEvent tool_call_id、tool_name、args
ToolResultEvent tool_call_id、result
ShowHistoryEvent / ShowHelpEvent / ShowToolsEvent 历史、命令表、工具表
InfoEvent / WarningEvent / ErrorEvent message
ConfirmEvent choice、choices、source、prompt、message
AgentBindEvent / AgentUnbindEvent 无(绑定/解绑信号)

自定义显示层

只需实现两个抽象方法:

from xun import Agent, DisplayAbstract, ToolBox

class RawDisplay(DisplayAbstract):
    def on_event(self, event):
        print(f"{event.name}: {event.payload}")

    def get_choice(self, request) -> str:
        print(f"[{request.prompt}] {request.choices}")
        return input(">>> ")

with Agent(display=RawDisplay(), toolbox=ToolBox().with_defaults()) as agent:
    agent.instruct("...").execute()

get_choice 收到 DisplayAbstract.ChoiceRequest:agent_info、prompt、choices、message、title、subtitle、default、allow_extra;返回选中的字符串。

内置实现:

实现 行为
NullDisplay() 丢弃事件;get_choice 抛 NotImplementedError。裸 Agent() 的默认值
Display() 控制台输出(rich),setup_agent 与 xun 的默认值
WebDisplay(expose_files=False, max_events=2000) Web 后端,保留事件历史供前端回放

Web 服务编程接口

xuns 只是下面这段的封装:

from xun import WebDisplay, WebDisplayService, setup_agent

display = WebDisplay(expose_files=True)
setup_agent(display=display, default_tools=True, workdir=".test")

service = WebDisplayService(host="localhost", port=18960, token="", base_path="")
service.mount("/", display)
service.start(blocking=False)     # 打印每个会话的访问 URL
...
service.stop()
  • token="" 时自动生成随机令牌;--token 的三种携带方式与路由约定见 xuns · Web 服务。
  • port=0 表示随机端口,真实端口写在 service.port 上。
  • mount() 可挂多个显示实例,路径互不重叠;/srv 段被保留。
  • 站点根路径自动重定向到 /chat/;FastAPI 自带的 /docs 被禁用,/docs/ 留给打包进来的文档站点。

会话工厂

传入 session_manager 后,界面可以自行创建/删除会话。工厂是一个同步上下文管理器,产出 (mount_path, display),finally 负责清理:

from contextlib import contextmanager
from uuid import uuid4
from xun import WebDisplay, WebDisplayService, setup_agent

@contextmanager
def session_manager():
    display = WebDisplay(expose_files=True)
    agent = setup_agent(display=display, default_tools=True, workdir=".test")
    try:
        yield f"/{uuid4()}", display
    finally:
        agent.finalize()

service = WebDisplayService(session_manager=session_manager)

web_session(manage_sessions=True) 提供的就是这套默认工厂(为每个会话分配独立临时工作目录)。

前后端契约

端点 说明
WS /ws 主通道:message、command、choice、cancel 四类消息
GET /api/events 断线重连时补拉事件
GET /api/prompts、POST /api/prompts/{id}/resolve 待回答的确认请求
GET /api/agents、GET /api/running、GET /api/config 智能体列表、运行状态、是否开放文件接口
GET /api/commands/{agent_id}、GET /api/capabilities/{agent_id} 命令表、模型与能力
GET/POST /api/files/*、/api/serve/* 文件浏览与临时静态托管(仅 expose_files=True 时注册)

WebSocket 消息体:message(agent_id、content、最多 8 张图片)、command(name、arguments)、choice(prompt_id、value)、cancel(agent_id)。

文档站点

WebDisplayService 会把 src/xun/assets/docs 作为静态站点挂在 {base}/docs/,不需要令牌;这份站点正是本目录用 MkDocs 生成后由 make doc-dist 拷贝进去的产物。