Roles and team cn

在 Jianmu 的 Swarm 多 Agent 协作体系中，三个抽象层协同工作：AgentRole 定义了"一个 Agent 应该如何被构建"的协议契约，SwarmRole 提供了开箱即用的 LLM 驱动角色实现，而 AgentTeam 则对底层的 AgentRuntime 进行外观封装，将角色注册、运行时组装、生命周期管理和快照持久化收敛为统一的用户入口。三者之间的层次关系如下：

graph TD
    subgraph "用户层"
        AT[AgentTeam 外观]
    end
    subgraph "角色层"
        AR[AgentRole 协议]
        SR[SwarmRole]
        FR[FunctionalRole / @role]
        CR[Custom Role 类]
    end
    subgraph "运行时层"
        ART[AgentRuntime 核心引擎]
        AH[AgentHost 每个 Agent 的包装]
        RR[ReactiveRunner 行为树执行器]
    end
    AT -->|"封装并代理"| ART
    AR -->|"实现为"| SR
    AR -->|"实现为"| FR
    AR -->|"实现为"| CR
    ART -->|"创建并管理"| AH
    AH -->|"持有"| RR
    AH -->|"持有"| AR
    SR -->|"build_tree() →"| SN[SwarmNode]
    CR -->|"build_tree() →"| BT[自定义行为树]
    FR -->|"build_tree() →"| FN[_FunctionalNode]

角色对象的职责是 工厂化生产行为树：给定一个 AgentProfile（包含 id、role、parent_id、task、budget、metadata），输出合法的 py_trees.behaviour.Behaviour 根节点。运行时接管后将此行为树挂载到 ReactiveRunner 中驱动执行。

AgentRole：角色工厂协议¶

AgentRole 是 Jianmu Swarm 体系中最核心的抽象——它是一个 Protocol，不强制继承关系，只要对象拥有 name: str 属性并实现 build_tree(profile: AgentProfile) -> Any 方法即为合法角色。

class AgentRole(Protocol):
    name: str
    def build_tree(self, profile: AgentProfile) -> Any: ...

build_tree 接收的 AgentProfile 是 dataclass 定义的身份快照，携带运行时分配给该 Agent 的完整信息：

字段	类型	说明
`id`	`str`	运行时生成的 UUID 标识符
`role`	`str`	注册到运行时的角色名称
`parent_id`	`str \\| None`	父 Agent ID（attached 模式）
`task`	`str \\| None`	当前分配的任务描述
`budget`	`BudgetSpec \\| None`	Token 和工具调用预算上限
`metadata`	`dict[str, Any]`	应用级附加元数据

build_tree 在 AgentRuntime.spawn() 内部被调用，且调用期间 profile 会被临时写入运行时的 _pending_profiles 字典，使得角色在构造行为树时可以通过 runtime.get_profile(agent_id) 查询自身或其他 Agent 的身份信息。方法返回的结果必须通过 isinstance(root, Behaviour) 校验，否则抛出 ValueError。

SwarmRole：开箱即用的 LLM Agent 角色¶

SwarmRole 是 AgentRole 协议的最完整实现，专为 liquid-topology（液态拓扑）场景设计——Agent 可以在运行时动态创建子 Agent、发送直接/群组/主题消息来重新组织协作拓扑。

构造参数与职责矩阵¶

SwarmRole(
    name,                    # 角色名称，用于运行时注册
    runtime,                 # AgentRuntimeProtocol 引用
    system_prompt,           # 基础系统提示词
    model_client,            # 共享的模型客户端（或 model_client_factory）
    model_client_factory,    # 按 AgentProfile 动态创建模型客户端
    tools,                   # 附加工具列表
    runtime_tool_names,      # 白名单过滤运行时工具
    max_iterations=15,       # ReAct 循环最大迭代次数
    constraints,             # 执行约束（预算/沙箱等）
    skills_dir,              # 技能目录
    enabled_skills,          # 默认启用的技能列表
    skill_files,             # 额外的 SKILL.md 路径
    skill_prompt_mode,       # "summary" | "full"
)

模型客户端解析遵循三级优先级链：model_client_factory(profile) → model_client → runtime.default_model_client。当未提供任何一方时抛出 ValueError。

build_tree：从角色到 SwarmNode¶

SwarmRole.build_tree(profile) 的核心产出物是一个 SwarmNode（继承自 SkillNode）。构造过程如下：

flowchart LR
    P[AgentProfile] --> RC[resolve model client]
    P --> RT[resolve runtime tools]
    RC --> CFG[SkillNodeConfig + ReActConfig]
    RT --> CFG
    CFG --> SN[SwarmNode]
    SN -->|"内嵌"| SK[skill/react agent-loop]

SwarmNode 本质是一个带技能执行能力的 ReAct 循环节点。其内建的 _skill_react_system_prompt() 在基础提示词之后追加了对话完成策略：

"A turn is only complete after you call send_message or send_direct_message when a visible reply is needed."

这确保 Swarm 中的 LLM Agent 在产出最终回复前一定会调用消息发送工具，而非仅停留在内部推理阶段。

工具解析通过 _resolve_runtime_tools(profile) 完成：优先使用 runtime.get_all_tools(profile.id)（如果可用），回退到 SwarmToolProvider。若设置了 runtime_tool_names 白名单，则过滤保留名称匹配的工具。

SwarmToolProvider：运行时工具集¶

SwarmToolProvider 是注册到运行时的第一个（也是唯一内置的）工具提供者，向每个 Agent 暴露 14 个运行时管理工具。工具列表采用"最后写入按名称覆盖"策略，内置提供者排在首位，后续注册的 MCPToolProvider 和用户自定义提供者可以覆盖同名工具。

工具名称	类别	功能
`create` / `create_agent`	生命周期	创建子 Agent，支持 `reuse_key` 幂等重试
`send_message`	消息	发送直接/群组/主题消息
`send_direct_message`	消息	发送直接消息到指定 Agent ID
`send_topic_message`	消息	广播主题消息
`list_agents`	查询	列出所有活跃 Agent
`self`	查询	获取当前 Agent 自身 Profile
`pause_agent`	生命周期	暂停指定 Agent
`resume_agent`	生命周期	恢复暂停的 Agent
`kill_agent`	生命周期	终止指定 Agent 及其附属子 Agent
`join_group`	路由	加入广播群组
`leave_group`	路由	离开广播群组
`subscribe_topic`	路由	订阅主题流
`unsubscribe_topic`	路由	取消主题订阅

create 和 send_* 工具内置了 side_effects 去重机制：通过 (kind, key) 元组在 AgentHost.side_effects 字典中记录已执行的操作，当 LLM 因重试等原因发出重复调用时直接返回缓存结果，避免创建多余 Agent 或重复发送消息。

FunctionalRole 与 @role 装饰器：函数式角色定义¶

对于不需要 LLM 推理的确定性 Agent（如工作流编排器、数据管道节点），Jianmu 提供了函数式角色路径：

from jianmu.swarm import role

@role(name="validator", description="验证输入数据的完整性")
def validate_input(profile, state_manager, ctx):
    data = state_manager.get("input_data")
    if not data:
        return False  # → Status.FAILURE
    state_manager.update({"validated": True})
    return True  # → Status.SUCCESS

@role 装饰器将同步函数包装为 FunctionRole 实例，其 build_tree() 返回一个 _FunctionalNode。返回值到行为树状态的映射规则如下：

返回值类型	映射结果
`Status` 枚举	直接使用；非 RUNNING 状态标记 `_done=True`
`bool`	`True` → `Status.SUCCESS`，`False` → `Status.FAILURE`
`dict`	写入 `state_manager.update()` → `Status.SUCCESS`
`None`	`Status.SUCCESS`
其他	`Status.SUCCESS`

注意：@role 当前仅支持同步函数，传入异步函数会触发 TypeError。_FunctionalNode 内部仅调用一次 handler（_done 锁防止重复执行），适用于"一次性判定"场景而非持续运行的 Agent。

自定义 Role 类：手动组合行为树¶

当需要精确控制行为树结构时，直接实现 AgentRole 协议并手动构造节点序列。以下来自 basic_collaboration_demo.py 的 CoordinatorRole 展示了完整的协作流水线：

class CoordinatorRole:
    name = "coordinator"
    def __init__(self, runtime):
        self.runtime = runtime

    def build_tree(self, profile):
        return Sequence(name="CoordinatorFlow", memory=True, children=[
            SpawnAgent("SpawnWorker", role="worker",
                       output_key="worker_id", runtime=self.runtime,
                       reuse_if_present=True),
            PrepareTaskNode(),
            SendMessage("SendTaskToWorker",
                        to_agent_key="worker_id", content_key="task",
                        runtime=self.runtime),
            WaitMessage("WaitWorkerReply"),
            CaptureReplyNode(),
        ])

此模式中使用的三个 Swarm 原语节点（详见 [Swarm 节点](swarm_nodes_cn.md)）构成了所有手动角色组合的基石：

SpawnAgent：调用 runtime.spawn() 并将新 Agent ID 写入状态。reuse_if_present=True 时跳过重复创建。
SendMessage：从状态读取路由目标和内容，调用 runtime.send_message()。支持 to_agent_key / group_key / topic_key 三种路由。
WaitMessage：阻塞直到 ctx.incoming_messages 非空。

AgentTeam：运行时装配与生命周期外观¶

AgentTeam 是面向用户的一站式入口，将 AgentRuntime 的构造参数收敛为统一接口，同时提供异步上下文管理器支持：

async with AgentTeam(
    roles={"assistant": swarm_role},
    default_model_client=model_client,
    tool_providers=[custom_provider],
    checkpointer=FileCheckpointer(),
    runtime_event_bus=RuntimeEventBus(),
    context_builder=my_context_builder,
    auto_start=True,
) as team:
    agent_id = team.spawn("assistant", task="分析数据")
    # ... 使用 team.runtime 进行更细粒度控制

AgentTeam 与 AgentRuntime 的职责边界¶

关注点	AgentTeam（外观）	AgentRuntime（引擎）
角色注册	通过构造函数 `roles` 传入	持有 `_roles: dict`，支持 `register_role()`
工具提供者	通过构造函数传入	管理 `_tool_providers` 列表和初始化/关闭生命周期
默认模型客户端	存储到 `AgentTeamDefaults`	注入到每个 AgentHost 的 RunContext
上下文构建器	共享到 `AgentTeamDefaults` 和 PromptRuntime	用于邮箱历史合并和 Prompt 上下文装配
Checkpoint	解析配置中的默认 FileCheckpointer	执行实际的 save_snapshot / restore_snapshot
运行时事件总线	透传给 `AgentRuntime`	桥接 Swarm 生命周期事件与单 Agent 流
生命周期	`start()` / `close()` + async context manager	`initialize_async()` / `close_async()`

AgentTeam.spawn() 直接委托给 self._runtime.spawn()，AgentTeam.send() 委托给 self._runtime.send_message()，而 save_snapshot() / restore_snapshot() 在委托前会优先使用 AgentTeam 构造时记录的 _checkpoint_thread_id。

AgentRuntime：多 Agent 运行时引擎¶

AgentRuntime 是整个 Swarm 体系的中枢，管理所有 AgentHost 实例的生命周期、消息路由和执行调度。

核心数据结构¶

classDiagram
    class AgentRuntime {
        _roles: dict[str, AgentRole]
        _hosts: dict[str, AgentHost]
        _attachments: dict[str, set[str]]
        _group_members: dict[str, set[str]]
        _topic_subscribers: dict[str, set[str]]
        _message_store: EnvelopeStoreProtocol
        _event_bus: InMemoryEventBus
        _tool_providers: list[ToolProvider]
        spawn(role, task, parent_id, mode, constraints) str
        kill(agent_id)
        send_message(...) int
        step_agent(agent_id) Status
        run_until_idle(max_steps) dict
        save_snapshot(thread_id) dict
    }
    class AgentHost {
        profile: AgentProfile
        role: AgentRole
        runner: ReactiveRunner
        state: StateManager
        inbox: deque[MessageEnvelope]
        mailbox_cursor_seq: int
        groups: set[str]
        topics: set[str]
        paused: bool
        attached_to: str?
        side_effects: dict
    }
    AgentRuntime *-- AgentHost

spawn 流程¶

runtime.spawn(role, task, parent_id, mode) 的执行路径如下：

角色解析：_resolve_role(role) 接受字符串（查 _roles 字典）或直接传递 AgentRole 对象
Profile 构建：生成 UUID 作为 agent_id，将 constraints 序列化后填入 metadata["constraints"]
行为树生产：调用 role_obj.build_tree(profile)，期间 profile 临时写入 _pending_profiles 供角色内部查询
校验：确保返回值是 py_trees.behaviour.Behaviour 实例
State 初始化：通过 _state_factory(profile) 创建 StateManager，注入 agent_id、role、task、session_id
RunContext 装配：注入 model_client、runtime 自身、constraints、sandbox、prompt_runtime、runtime_event_bus
AgentHost 创建：组装 AgentHost(profile, role, runner, state)，处理 attached 模式下的父子关联
事件发射：agent_created 事件通知所有订阅者
自动启动：若 auto_start=True，立即调用 _start_agent(agent_id) 启动后台异步任务

消息路由三通道¶

消息通过 send_message() 发送，必须指定且仅指定一个路由目标——同时提供多个目标或零目标均抛出 ValueError：

通道	参数	路由语义
直接消息	`to_agent_id`	精确投递到单个 Agent 的 inbox
群组广播	`group_id`	投递到群组内所有已加入 Agent
主题发布	`topic`	投递到所有订阅该主题的 Agent

每条消息经过 InMemoryMessageStore.append() 获得单调递增的序列号，然后通过 enqueue_hot_envelope() 实时推送到匹配的 AgentHost.inbox。消息存储支持基于序列号的光标消费和 GC 清理（message_store_gc_min_advance 控制最小步进间隔）。

Agent 调度模式¶

AgentRuntime 支持两种执行模式，二者互斥：

自动模式（auto_start=True）：spawn 后 AgentHost 启动后台 asyncio.Task，自主消费 mailbox 并驱动行为树
手动模式（auto_start=False）：外部通过 step_agent() / step_all() / step_all_concurrent() / run_until_idle() 精确控制每个 tick

手动模式提供了三个控制 API 形成递进式抽象：

API	粒度	适用场景
`step_agent(agent_id)`	单 Agent 单步	调试、人机交互挂起恢复
`step_all()` / `step_all_concurrent()`	全量单步	回合制模拟、测试
`run_until_idle(max_steps=200)`	自动推进至静止	批量任务、CI 集成

run_until_idle() 在每步后调用 is_idle() 检查所有 AgentHost：当所有 Agent 既不处于 RUNNING 状态也没有待处理的 mailbox 消息时，认为运行时进入静止态。

角色选择决策矩阵¶

三种角色定义方式的适用场景对比：

维度	SwarmRole	FunctionalRole (@role)	自定义 AgentRole 类
LLM 驱动	✅ ReAct 循环	❌	可选（手动嵌入 LLMNode）
工具/Skill 集成	✅ 内建	❌ 需手动	需手动构建
行为树控制力	低（黑盒 SwarmNode）	极低（单节点）	高（完全控制树结构）
开发复杂度	低（配置式）	极低（一个函数）	中等
典型场景	LLM Agent 对话、自主协作	判定器、数据管道	精确编排的多步工作流
子 Agent 创建	通过 `create` 工具	手动调用 runtime	通过 `SpawnAgent` 节点

完整协作流水线示例¶

以下时序图融合了三种角色定义方式在实际场景中的交互：

sequenceDiagram
    actor User
    participant AT as AgentTeam
    participant AR as AgentRuntime
    participant Leader as LeaderHost (SwarmRole)
    participant Scout as ScoutHost (Custom Role)
    participant Validator as ValidatorHost (@role)

    User->>AT: async with AgentTeam(roles={...}) as team
    AT->>AR: initialize_async()
    User->>AT: team.spawn("leader", task="...")
    AT->>AR: spawn("leader", task="...")
    AR->>AR: _resolve_role("leader") → SwarmRole
    AR->>Leader: role.build_tree(profile) → SwarmNode
    AR->>Leader: create AgentHost + ReactiveRunner
    Leader->>Leader: SwarmNode 内 ReAct 循环开始
    Leader->>AR: create(role="scout") via SwarmToolProvider
    AR->>Scout: spawn → build_tree → AgentHost
    Leader->>AR: create(role="validator") via SwarmToolProvider
    AR->>Validator: spawn → build_tree → _FunctionalNode
    Leader->>AR: send_direct_message(to=scout_id, ...)
    AR->>Scout: enqueue into inbox
    Scout->>Scout: WaitMessage → 处理 → SendMessage(validator)
    AR->>Validator: enqueue into inbox
    Validator->>Validator: handler() → dict → SUCCESS
    AR->>Leader: topic message → inbox
    Leader->>Leader: 完成 → swarm_success = True
    User->>AT: await leader_host.task
    AT->>AR: close_async()

阅读建议¶

深入理解 Swarm 的消息路由、事件总线与邮箱机制，请阅读 [Swarm 运行时](swarm_runtime_cn.md)
了解 SpawnAgent、SendMessage、WaitMessage 三个协作原语节点的细节，请参考 [Swarm 节点](swarm_nodes_cn.md)
关于 SwarmRole 依赖的 ReAct 循环与 SkillNode 机制，参见 [ReAct 节点工厂](react_cn.md) 和 [工具与技能节点](tool_nodes_cn.md)
AgentRuntime 的快照持久化机制详见 [Checkpoint 与长期记忆](memory_cn.md)