Facade cn

本页系统阐述 Jianmu 中两个核心外观类 —— Agent（单 Agent 入口）与 AgentTeam（多 Agent 入口）—— 的设计理念、组装逻辑和生命周期管理。二者构成面向用户的顶层 API，将行为树组装、状态管理、运行时上下文、检查点持久化以及交互挂起/恢复等内部复杂性统一封装为简洁接口。

在阅读本页之前，建议先了解底层的 ReactiveRunner：事件驱动的异步 tick 调度与挂起恢复机制 和 类型化状态管理：Pydantic Schema、Reducer 合并与 Ephemeral 字段，外观类的底层委托正指向这些组件。

外观模式的设计动机

Jianmu 的运行内核由多个低层组件协作构成：ReactiveRunner 驱动行为树 tick、StateManager 管理线程安全的状态读写、RunContext 携带不可序列化的运行时依赖（模型客户端、审批管理器、沙箱句柄等）、检查点后端负责持久化与恢复。如果让用户在每次创建和运行 Agent 时都手动组装这些组件，不仅冗长易错，还破坏了关注点分离原则。

外观模式在 Jianmu 中扮演的角色如下：

graph TB
    subgraph "用户层"
        AG[Agent 外观]
        AT[AgentTeam 外观]
    end

    subgraph "组装层（外观内部）"
        RR[ReactiveRunner]
        SM[StateManager]
        RC[RunContext]
        CH[CheckpointerProtocol]
        DF[_AgentRuntimeDefaults]
    end

    subgraph "基础设施层"
        BT[行为树 root]
        EB[事件总线]
        CFG[jianmu.yaml 配置]
    end

    AG -->|"委托"| RR
    AG -->|"持有"| SM
    AG -->|"持有"| RC
    AG -->|"可选的"| CH
    AG -->|"解析默认值"| DF
    AG -->|"接收"| BT
    DF -->|"读取"| CFG

    AT -->|"委托"| AR[AgentRuntime]
    AT -->|"持有"| TD[AgentTeamDefaults]
    AT -->|"可选的"| CH2[CheckpointerProtocol]
    AT -->|"通过"| EB
    AT -->|"接收 roles + providers"| AR
    TD -->|"传播到 spawn 的 Host"| AR

两张外观在设计上遵循相同的核心原则：接受显式覆盖，兜底配置推导。用户在构造 Agent 时可以显式传入 state_manager、checkpointer、thread_id 等参数；未传入的则通过 _resolve_agent_runtime_defaults() 和 _resolve_default_checkpointer() 从 jianmu.yaml 的 runtime 段自动推导。

Agent：单 Agent 的构造与运行时组装

构造参数与默认值推导

Agent.__init__ 的必选参数只有 root（行为树根节点），但必须提供 state_schema 或 state_manager 之一，否则抛出 ValueError。这一约束在构造期即完成校验，避免了运行时才发现状态层缺失的隐蔽错误。

# 紧凑用法：仅提供 root + state_schema，其余由配置推导
agent = Agent(root=my_tree, state_schema=MyState)

# 完整用法：显式控制所有组件
agent = Agent(
    root=my_tree,
    state_manager=prebuilt_state_manager,
    context=RunContext(model_client=mc, constraints=ct),
    setup_timeout=30.0,
    checkpointer=FileCheckpointer(),
    thread_id="session-42",
    checkpoint_interval=5,
    max_fps=30.0,
)

构造过程中的默认值解析链条如下表：

参数	优先级逻辑	推导函数	配置来源
state_manager	显式传入 → 由 state_schema 自动创建并 initialize()	构造函数内联	无
context	显式传入 → 空 RunContext()	构造函数内联	无
checkpointer	显式传入 → _resolve_default_checkpointer()	agent.py#L48-L63	runtime.checkpoint.enabled + backend
thread_id	显式传入 → "default_thread"	_resolve_agent_runtime_defaults()	无（硬编码默认值）
checkpoint_interval	显式传入 → runtime.checkpoint.interval	_resolve_agent_runtime_defaults()	jianmu.yaml → runtime.checkpoint.interval
max_fps	显式传入 → runtime.max_fps	_resolve_agent_runtime_defaults()	jianmu.yaml → runtime.max_fps

_resolve_default_checkpointer() 仅在 runtime.checkpoint.enabled = true 时创建 FileCheckpointer，且 backend 必须为 "file"（其他后端抛出 ValueError）。如果检查点被禁用，则返回 None，后续 run() 调用可在运行时再覆盖。

ReactiveRunner 的自动组装

Agent 外观在构造期间即完成 ReactiveRunner 的完整组装：

runner = ReactiveRunner(
    root,
    state_manager,
    ctx=ctx,
    setup_timeout=setup_timeout,
    max_pending_wakeups=None,   # 使用配置默认值
    hot_loop_warn_factor=None,  # 使用配置默认值
)

ReactiveRunner 的构造函数内部会遍历 root.iterate() 并为每个 JianmuNodeMixin 节点调用 inject_runtime_deps(node, payload)，将 RunContext、StateManager 和 _on_wake_signal 回调注入到每个节点。这一注入过程由外观隐式完成，用户无需感知 InjectPayload 的存在。

暴露的属性

Agent 将内部关键组件暴露为只读属性，方便用户检查运行时状态而不破坏封装：

属性	类型	说明
runner	ReactiveRunner	底层响应式运行器，可访问 root、tree、state_manager 等
state	Any	快捷方式：等价于 state_manager.get()，返回当前 Pydantic 状态快照
state_manager	StateManager	线程安全状态管理器实例
context	RunContext	运行时依赖容器
root	Behaviour	行为树根节点

Agent 的运行时生命周期方法

Agent 外观提供四类执行入口，覆盖从简单端到端运行到细粒度挂起恢复的完整生命周期。

run()：端到端执行

run() 是最常用的入口，驱动行为树直到根节点达到 SUCCESS 或 FAILURE。它在每次调用时均可覆盖检查点和 FPS 参数：

status = await agent.run(
    input_data={"task": "分析这个仓库"},
    reset_tree=True,        # 中断之前的树状态
    reset_data=False,       # 保留状态管理器中非 ephemeral 字段
    max_ticks=100,
    timeout_s=30.0,
    checkpointer=custom_cp, # 覆盖构造期的默认检查点
    thread_id="run-42",
)

run() 的内部委托路径为：Agent.run() → ReactiveRunner.run() → _event_loop()。_event_loop() 在每个 checkpoint_interval 次 tick 后执行一次检查点保存，在每次 tick 后检查 max_ticks 和 timeout_s 是否耗尽。

step()：交互式单步执行

step() 为需要逐 tick 控制的场景（如 TUI Chat、可视化调试器）设计。每次调用执行一次树 tick，并在前后处理 Ephemeral("step") 字段的自动重置：

actions = await agent.step(obs={"user_input": "继续"}, yield_to_async=True)

step() 返回当前 step 范围的 action 字段字典，来自 state_manager.get_step_fields()。它与 run() 互斥：当 run() 正在执行时调用 step() 会抛出 RuntimeError，反之亦然。

run_until_suspend()：可挂起的执行

run_until_suspend() 是交互场景的核心入口——它驱动树执行，直到达到终态（success/failure）或遇到可发布的挂起点（如审批等待、用户输入请求）。返回 RunResult 对象：

result = await agent.run_until_suspend(
    input_data={"task": "..."},
    suspension_mode=SuspensionMode.YIELD,
    restore=RestorePolicy.NEVER,
)
if result.outcome == "suspended":
    print(f"挂起: {result.suspension.reason}, request_id={result.suspension.request_id}")

RunResult 的 outcome 字段取值为 "success"、"failure" 或 "suspended"。suspended 时 suspension 字段为 SuspensionRecord 实例，包含 request_id（用于后续恢复）、reason（APPROVAL_PENDING / AWAITING_USER_INPUT / EXTERNAL_RESULT_PENDING）和 category（Host 视角的标准化分类）。

resume_interaction()：挂起恢复

resume_interaction() 是 ReactiveRunner.resume_and_continue() 的外观封装，将挂起的交互点恢复到运行态。关键区别在于它将底层的 ResumeError 异常转换为 ResumeInteractionResult 结构化返回，避免调用方需要捕获异常：

result = await agent.resume_interaction(
    thread_id="session-42",
    request_id="sr_abc123...",
    payload={"approved": True, "comment": "允许执行"},
)
if result.resume.status == ResumeStatus.SUCCESS:
    run_result = result.run  # RunResult | None

该方法内部保留 ResumeError 语义：当 runner.resume_and_continue() 抛出的异常携带 result 属性时，外观将其转换为 ResumeInteractionResult(resume=exc.result, run=None)，否则向上传播。

resume() 与 reset()

resume() 是 run(reset_tree=False, reset_data=False) 的便捷别名——它继续当前内存态的运行器而不重置树或状态。reset() 中断树执行并可选择重新初始化状态管理器：

await agent.resume(max_ticks=50)     # 等价于 run(reset_tree=False, reset_data=False, max_ticks=50)
agent.reset(reset_data=True)         # 中断树 + 重置状态

生命周期方法总览

下面将 Agent 的完整生命周期入口与其底层委托关系一并呈现：

flowchart LR
    subgraph "Agent 外观方法"
        A_run["run()"]
        A_step["step()"]
        A_sus["run_until_suspend()"]
        A_res["resume_interaction()"]
        A_rst["reset()"]
    end

    subgraph "ReactiveRunner 方法"
        R_run["run() → _event_loop()"]
        R_step["step()"]
        R_sus["run_until_suspend()"]
        R_rc["resume_and_continue()"]
        R_rst["reset()"]
    end

    A_run --> R_run
    A_step --> R_step
    A_sus --> R_sus
    A_res --> R_rc
    A_rst --> R_rst

AgentTeam：多 Agent 的运行时组装

AgentTeam 将 AgentRuntime 的复杂构造参数（角色注册、事件总线、消息存储、调度策略、重启策略等）封装为简洁的面向用户 API。它与 Agent 外观遵循相同的设计理念：接受显式覆盖，兜底配置推导。

构造参数与组件组装

AgentTeam.__init__ 的所有参数均为可选——即使不传任何参数，也能构建一个可工作的 AgentTeam 实例：

team = AgentTeam()  # 最简构造：所有默认值来自配置

team = AgentTeam(
    roles={"assistant": SwarmRole(...), "planner": FunctionalRole(...)},
    default_model_client=ModelClient.resolve(env_override=True),
    tool_providers=[MCPToolProvider(config_path="mcp.json")],
    checkpointer=FileCheckpointer(),
    runtime_event_bus=RuntimeEventBus(),
    context_builder=my_context_builder,
    session_id="session-42",
    auto_start=True,  # spawn 后立即自动运行
)

和节点级 API 一样，AgentTeam 期望拿到的是已经解析好的 ModelClient 外观，而不是裸 provider 对象。公开主路径应传入 ModelClient.resolve(...) 或其他已经构造完成的 ModelClient 实例。

构造过程完成以下关键组装：

1. 检查点解析：_resolve_default_checkpointer() 与单 Agent 外观完全一致，从 runtime.checkpoint 配置推导。
2. 团队默认值捕获：创建不可变的 AgentTeamDefaults 数据类，封装 default_model_client、context_builder、tool_providers。
3. AgentRuntime 实例化：将角色映射、模型客户端、工具提供者、检查点后端、事件总线和上下文构建器全部注入 AgentRuntime 构造函数。

AgentRuntime 内部组装细节

AgentTeam 外观隐藏了 AgentRuntime 内部的复杂组装逻辑：

内部组件	组装方式	代码位置
事件总线	默认 InMemoryEventBus()，可通过 runtime_event_bus 参数注入 RuntimeEventBus 实现跨模块桥接	core.py#L147
消息存储	默认 InMemoryMessageStore()	core.py#L148
状态工厂	lambda profile: default_state_factory(DefaultAgentState, profile)	core.py#L149
运行器工厂	lambda root, state, ctx: ReactiveRunner(root, state, ctx=ctx)	core.py#L150
工具提供者链	[SwarmToolProvider(self)] → [MCPToolProvider] → 用户传入的 tool_providers	core.py#L183-L187
Prompt 运行时	通过 build_prompt_runtime(skills_dir, context_builder) 构建	core.py#L167-L170
调度参数	mailbox_drain_max_iterations=100、restart_policy="on_failure"、max_restarts=3 等	core.py#L89-L95

暴露的属性

AgentTeam 暴露两个关键属性：

属性	类型	说明
runtime	AgentRuntime	底层多 Agent 运行时，可访问 spawn()、send_message()、step_all()、list_agents() 等完整 API
defaults	AgentTeamDefaults	构造期捕获的团队默认值（default_model_client、context_builder、tool_providers）

用户可以通过 team.runtime 访问底层运行时的完整能力（如 step_agent()、run_until_idle()、subscribe_events() 等），而 team.defaults 则提供一个不可变的快照，便于了解团队级别的配置传播路径。

AgentTeam 的生命周期管理

异步上下文管理器

AgentTeam 支持 async with 语法，自动调用 start() 初始化工具提供者并注册资源，退出时调用 close() 停止所有托管 Agent 并关闭提供者：

async with AgentTeam(roles=roles, default_model_client=mc) as team:
    agent_id = team.spawn("assistant", task="分析日志")
    await team.runtime.run_until_idle()
# 退出时自动清理

start() 委托给 AgentRuntime.initialize_async()，它会遍历所有 ToolProvider 并调用其 initialize() 方法。close() 委托给 AgentRuntime.close_async()，先 kill_async 所有根宿主，再按逆序调用每个提供者的 close()。

spawn()：创建 Agent 宿主

spawn() 是团队协作的核心入口。它委托给 AgentRuntime.spawn()，后者完成以下内部组装：

1. 角色解析：通过 _resolve_role() 将角色名映射为 AgentRole 对象或直接透传。
2. Profile 构建：创建 AgentProfile(id=uuid, role=..., parent_id=..., task=..., budget=...)。
3. 行为树构建：调用 role_obj.build_tree(profile) 生成 Behaviour 根节点。
4. 状态初始化：通过 state_factory 创建 StateManager，注入 agent_id、role、task。
5. RunContext 组装：创建携带 model_client、runtime（自引用）、constraints、prompt_runtime、runtime_event_bus 的上下文。
6. Runner 组装：通过 runner_factory(root, state, ctx) 创建 ReactiveRunner。
7. AgentHost 包装：将所有组件封装为 AgentHost 数据类并注册到 _hosts 字典。

agent_id = team.spawn(
    role="assistant",       # 已注册的角色名称或 AgentRole 实例
    task="总结这篇论文",
    parent_id=None,         # 父 Agent ID（用于 attached 模式）
    mode="detached",        # "detached" | "attached"
    constraints=my_ct,      # 可选的执行约束
)

send()：消息路由

send() 将消息发送委托给 AgentRuntime.send_message()，支持点对点通信（to_agent_id）、组播（group_id）和主题订阅（topic）：

msg_count = await team.send(
    sender_id="agent-a",
    content="请 review 我的输出",
    to_agent_id="agent-b",
    content_type="text",
    metadata={"priority": "high"},
)

底层消息经过 routing_impl.enqueue_hot_envelope() 入队到 InMemoryMessageStore，目标 Agent 在其下一轮 tick 或 mailbox drain 循环中消费。

save_snapshot() 与 restore_snapshot()

两个方法提供团队级别的持久化快照：

snapshot = await team.save_snapshot(thread_id="checkpoint-42", step=10)
restored = await team.restore_snapshot(thread_id="checkpoint-42")

save_snapshot() 委托给 snapshot_impl.save_snapshot()，序列化所有 AgentHost 状态和消息存储。restore_snapshot() 先 kill_root_hosts 清除现有宿主，再从快照重建宿主、消息存储和路由索引。

两张外观的对比与协作

架构层级对比

graph TB
    subgraph "单 Agent 路径"
        AG[Agent 外观] --> RR[ReactiveRunner]
        RR --> BT[行为树 root]
        RR --> SM1[StateManager]
        RR --> RC1[RunContext]
    end

    subgraph "多 Agent 路径"
        AT[AgentTeam 外观] --> AR[AgentRuntime]
        AR --> H1[AgentHost 1]
        AR --> H2[AgentHost 2]
        AR --> MS[MessageStore]
        AR --> EB[EventBus]
        H1 --> RR1[ReactiveRunner]
        H1 --> SM2[StateManager]
        H1 --> RC2[RunContext]
        H2 --> RR2[ReactiveRunner]
        H2 --> SM3[StateManager]
        H2 --> RC3[RunContext]
    end

维度	Agent	AgentTeam
核心委托	ReactiveRunner	AgentRuntime
行为树数量	1 个 root	N 个 role.build_tree() 产物
状态管理	1 个 StateManager	每个 Host 独立的 StateManager
依赖注入	RunContext 直接注入	RunContext 由 spawn 过程组装
跨 Agent 通信	不适用	send_message() + mailbox
事件总线	可选的 RuntimeEventBus	内置 InMemoryEventBus + 可选的 RuntimeEventBus 桥接
检查点	直接传给 ReactiveRunner	传递给 AgentRuntime，由 snapshot 实现管理
异步上下文管理器	不支持	支持 async with
默认值捕获	_AgentRuntimeDefaults（内部）	AgentTeamDefaults（公开属性）

协作场景：Team 中使用 Agent 的挂起/恢复模式

当一个 Swarm 中的 Agent 需要挂起等待审批时，运行时通过 AgentRuntime.resolve_suspension() 桥接到该 Agent 的 ReactiveRunner.resume_and_continue()。这一路径使用了与 Agent.resume_interaction() 相同的底层机制，但通过 AgentHost.runner 引用定位正确的运行器实例。

正确使用指南与反模式

推荐模式： - 将 Agent 用于单一任务的端到端执行，通过 run() 获取最终状态。 - 将 AgentTeam 用于多角色协作编排，通过 spawn() + send() + run_until_idle() 构建工作流。 - 在交互式应用中，使用 Agent.step() 和 Agent.run_until_suspend() / resume_interaction() 实现细粒度控制。 - 利用 async with AgentTeam(...) 确保工具提供者的正确初始化和清理。

反模式： - 不要在 run() 执行期间从外部线程直接修改 state_manager（应通过 wake_up 回调或 ResumeData 机制）。 - 不要绕过外观直接操作 _runner 的内部方法——外观提供了参数验证和默认值解析保护。 - 不要在 AgentTeam 未调用 start() 或未使用 async with 时调用 spawn()（此时工具提供者未初始化）。

---

后续阅读建议：了解外观内部的运行机制后，建议深入 ReactiveRunner：事件驱动的异步 tick 调度与挂起恢复机制 了解 tick 循环细节，或转入 Swarm 运行时：AgentRuntime 的邮箱路由、事件总线与生命周期管理 了解 AgentRuntime 的完整能力集。