ag2-network-governance

AG2 Network — 治理

Hub 端的一切：身份、各 Agent 的规则、预期、审计和任务观测。Hub 是唯一可信来源——每次发送都经由它，每次观测都读取自它，每项策略都由它检查。

前置条件：请先阅读 ag2-network-quickstart。本技能假设你了解 Hub.open、Passport、Resume、通道生命周期以及 agent_client.register(...) 流程。

适用场景

当用户需要以下操作时，加载此技能：

• 限制哪些 Agent 可以相互通信（access / AccessBlock）
• 声明令牌桶速率意图（limits.rate / RateBlock）——保存在规则中，但进程内 Hub 不会强制执行
• 限制收件箱大小以防止消息泛滥（limits.inbox / InboxBlock）——并在达到上限前获取预警信号（on_inbox_pressure / high_water）
• 设置通道 TTL 默认值、并发上限或委派深度（limits / LimitsBlock）
• 接入自定义访问／路由逻辑——JWT 作用域、每租户配额、联邦（HubArbiter / RuleBasedArbiter）
• 在注册时对 Agent 进行身份验证（AuthAdapter）
• 调整通道关闭时间（acks_within、reply_within、max_silence、turn_within）
• 为合规目的读取或查询审计日志，或将实时状态变更流式传输到指标／告警系统（HubListener）
• 为 Hub 添加自定义定期任务（register_sweeper）
• 为每个 Agent 建立能力履历（Resume.observed）
• 根据哪些 Agent 已确实完成某项任务来路由（例如，“发送给 p50_latency_ms 最低的研究员”）

身份——每个 Agent 随身携带的信息

三个数据类用于描述网络中的 Agent。租户提供大多数字段；其余字段由 Hub 填入。

from ag2.network import Passport, Resume, ResumeExample

passport = Passport(
    name="alice",                # required, unique within the hub
    owner="acme",                # optional, tenant id for multi-tenant deployments
    model="claude-sonnet-4-6",   # optional, surfaces on peer-lookup results
)

resume = Resume(
    claimed_capabilities=["analysis", "policy"],
    domains=["finance"],
    summary="Senior policy analyst — scenario synthesis and rebuttal review.",
    examples=[ResumeExample(title="Q3 risk brief", note="…")],
)

字段	来源
Passport.name	租户提供（必填，且唯一）
Passport.agent_id	注册时由 Hub 填入；用于路由
Passport.created_at	由 Hub 填入的 ISO-Z 时间戳
Resume.claimed_capabilities	租户提供（自由格式字符串："research"、"summarisation"、……）
Resume.summary	租户提供——已为同伴查找建立索引
Resume.observed	由 Hub 修改的各能力 ObservedStat（n / completed / failed / expired / p50_latency_ms）
Resume.last_updated	由 Hub 填入的 ISO-Z 时间戳，在发生修改时刷新

observed 字段是 Agent 的履历。随着 Agent 执行带有能力标签的任务，该字段会自动增长（见下文“任务观测”）。

Per-agent 规则

注册时传入 Rule，以规范 Agent 在网络中的行为：

from ag2.network import (
    Rule, AccessBlock, LimitsBlock, RateBlock, InboxBlock, ChannelTypeAccess,
)

rule = Rule(
    access=AccessBlock(
        outbound_to=["bob", "carol"],           # whitelist of recipients (names or ids)
        channel_types=ChannelTypeAccess(
            initiate=["consulting", "discussion"],
            accept=["consulting", "discussion"],
        ),
    ),
    limits=LimitsBlock(                         # rate + inbox nest INSIDE limits
        channel_ttl_default="4h",               # default TTL for channels this agent creates
        delegation_depth=2,                     # max recursion through sub-task delegation
        rate=RateBlock(per_minute=60, burst=10),
        inbox=InboxBlock(max_pending=100),      # cap inbound queue depth
    ),
)

alice = await alice_hc.register(
    Agent("alice", config=config),
    Passport(name="alice"),
    Resume(),
    rule=rule,
)

Rule 有两个顶层区块：access（AccessBlock）和 limits（LimitsBlock）。RateBlock 和 InboxBlock 嵌套在 limits 内部（分别为 limits.rate 和 limits.inbox）。

区块	所在位置	控制内容	失败模式
AccessBlock	access（顶层）	此 Agent 可以寻址的对象（outbound_to）／可以接受消息的对象（inbound_from）；可创建／加入的频道类型	AccessDeniedError
LimitsBlock	limits（顶层）	TTL 默认值；delegation_depth；max_concurrent_channels／max_concurrent_tasks	AccessDeniedError
RateBlock	limits.rate（嵌套）	令牌桶值（per_minute、burst）	已存储，但进程内不会强制执行（默认情况下，per_minute=0 表示禁用）
InboxBlock	limits.inbox（嵌套）	入站队列深度（max_pending）	向发送方返回 InboxFull

规则检查失败时，中心会从 channel.send(...) 或 hc.register(...) 引发相应错误；信封不会写入 WAL。规则变更会记入审计日志（通过 set_rule 记录，类型为 AUDIT_KIND_RULE_SET），但发送遭拒不会写入内置审计日志——该事件会通过 on_envelope_rejected 监听器扇出机制传递（内置的 AuditLog 未实现 on_envelope_rejected 处理程序）。如果需要捕获拒绝事件，请注册 HubListener。负责运行这些检查的组件，以及可接入规则数据以外逻辑的扩展点，是下面介绍的裁决器。

注册后更新规则

new_rule = Rule(access=AccessBlock(outbound_to=["bob"]))
await hub.set_rule(alice.agent_id, new_rule)  # emits AUDIT_KIND_RULE_SET

解析时长字符串

LimitsBlock.channel_ttl_default 接受由 parse_duration 解析的字符串：

from ag2.network import parse_duration

parse_duration("30s")  # 30
parse_duration("4h")   # 14400
parse_duration("2d")   # 172800

s、m、h、d 后缀；纯整数（或 int）按秒处理；空字符串返回 0。返回以秒为单位的 int 数值。

仲裁器 — 可替换的访问与路由

枢纽不再通过内联的 if 检查来强制执行 Rule；它会将每项访问／路由决策委托给 HubArbiter — 一个 Protocol，每个决策点都有一个方法，每个方法都返回 Allow() 或 Deny(reason, error=<NetworkError subclass>)：

方法	在……之前调用	默认 Deny 错误
authorize_register(passport, resume, rule)	提交注册	AccessDeniedError
authorize_channel_open(manifest, creator, creator_rule, invitees, invitee_rules, active_creator_channels)	创建频道（受邀者的 inbound_from + 创建者的 max_concurrent_channels）	AccessDeniedError
authorize_send(envelope, sender, sender_rule, recipients)	将信封追加到 WAL（出站访问 + 委派深度）	AccessDeniedError
authorize_inbox(envelope, recipient, recipient_rule, current_pending)	将信封加入接收者的收件箱（容量）	InboxFull
authorize_dispatch(envelope, sender, recipient, recipient_rule)	分发一条投递	AccessDeniedError
resolve_unknown_audience(envelope, unknown_ids)	分发给枢纽不认识的 ID — 返回 None（静默丢弃 — 单枢纽默认行为）或替换后的 ID 列表（联邦钩子）	—

默认实现是 RuleBasedArbiter — 它会强制执行每个 Agent 的 Rule：access（出站／入站名称 glob、频道类型），以及它实际检查的 limits 上限（delegation_depth、max_concurrent_channels、inbox.max_pending）。它不会强制执行 limits.rate。这与该接缝出现之前枢纽内联执行的逻辑完全一致。如果你只使用 Rule，就无需接触仲裁器。

替换它即可叠加自定义逻辑 — 例如 JWT 作用域、每租户配额、联邦路由 — 以补充或取代规则数据。BaseHubArbiter 对所有检查都会返回 Allow()，因此只覆盖一个关卡的子类会对其余关卡一律放行 — 若要保留规则强制执行，请委托给一个 RuleBasedArbiter() 实例：

from ag2.network import HubArbiter, BaseHubArbiter, RuleBasedArbiter, Allow, Deny

class ScopedArbiter(BaseHubArbiter):
    def __init__(self, inner: HubArbiter) -> None:
        self._inner = inner                           # the rule checks
    async def authorize_send(self, envelope, sender, sender_rule, recipients):
        if not _token_has_scope(sender, "net.send"):
            return Deny("missing net.send scope")     # → AccessDeniedError back to the caller
        return await self._inner.authorize_send(envelope, sender, sender_rule, recipients)
    # authorize_register / _channel_open / _inbox / _dispatch / resolve_unknown_audience
    # all fall through to BaseHubArbiter's Allow() — explicitly re-delegate any you want enforced.

hub.register_arbiter(ScopedArbiter(RuleBasedArbiter()))   # one active arbiter; replaces the prior one
# hub.arbiter  → the active instance (read-only; handy in tests)

Deny.error 用于选择枢纽要引发的 NetworkError 子类（默认为 AccessDeniedError）— 使用 Deny(..., error=InboxFull) 等来控制它。仲裁器是守门人（在状态变更之前调用）；HubListener（稍后介绍）是观察者（在状态变更之后收到通知）。这与下文的 AuthAdapter 职责不同 — 后者在注册时验证一次凭据；仲裁器则在频道的整个生命周期内授权操作。

身份验证

默认情况下，枢纽使用 AuthRegistry.default() — 一个仅包含 NoAuth 的注册表（方案 "none"），接受每项声明，因此无需凭据即可成功注册。方案通过 passport.auth.scheme 按 Passport 选择（一个 AuthBlock 字段，默认为 "none"）；凭据存放在 passport.auth.claim（一个 dict）中。用于生产环境：

from ag2.network import AuthAdapter, AuthRegistry, NoAuth, ApiKeyAuth, AuthError, Hub
from ag2.network import AuthBlock, Passport
from ag2.knowledge import MemoryKnowledgeStore

import hmac
from typing import Any


class HMACAuth:
    scheme = "hmac"                                       # class-level scheme label

    async def validate(self, passport: Passport, claim: dict[str, Any]) -> None:
        expected = self._sign(passport.name)
        token = claim.get("token", "")
        if not hmac.compare_digest(expected, token):
            raise AuthError(f"bad hmac for {passport.name}")


# AuthRegistry takes a LIST of adapters at construction (keyed by .scheme).
registry = AuthRegistry([NoAuth(), HMACAuth()])

hub = await Hub.open(MemoryKnowledgeStore(), auth=registry)

AuthAdapter 是一个 Protocol，包含 scheme 属性和 validate 方法：

class AuthAdapter(Protocol):
    scheme: str
    async def validate(self, passport: Passport, claim: dict[str, Any]) -> None: ...

引发 AuthError 即可拒绝。在注册时，枢纽会根据 passport.auth.scheme 查找适配器，调用 adapter.validate(passport, passport.auth.claim)，并在成功后记录 AUDIT_KIND_AGENT_REGISTERED。远程 Agent 的 Passport 会跳过本地身份验证。此库内置 NoAuth（接受所有请求，默认）和 ApiKeyAuth(keys=..., resolver=...)（以恒定时间比较 claim["token"] 中的令牌）。

预期 — 频道级 SLA

每个适配器都会在其清单中附带默认预期。预期检查任务会按 expectation_sweep_interval（默认为 10s）定期评估这些预期，并将违规情况分发给处理程序。

内置评估器

default_evaluators() 恰好包含 三个评估器：

名称	类	默认 seconds	阈值
"acks_within"	AcksWithinEvaluator	30	所有仍待确认的受邀者都必须在频道创建后的 params["seconds"] 时间内确认（仅当频道处于 PENDING 状态时）。
"reply_within"	ReplyWithinEvaluator	600	收到 EV_TEXT 定向消息的参与者必须在 params["seconds"] 时间内回复（仅当频道处于 ACTIVE 状态时）。
"max_silence"	MaxSilenceEvaluator	3600	在 params["seconds"] 时间内，频道内任何人都没有发送内容信封（适用于整个频道）。

discussion 和 workflow 适配器会在其清单中声明 turn_within 预期，但不会提供内置的 turn_within 评估器；"warn" / "hide" 也不是内置处理器。对于名称没有已注册评估器，或 on_violation 没有已注册处理器的任何预期，清理器都会静默跳过（_expectation_tick 会调用 .get(...)，并在结果为 None 时执行 continue）。若要启用 turn_within / warn / hide，请注册自己的评估器（register_expectation_evaluator）和处理器（register_violation_handler）。

各适配器的默认预期

适配器	默认值
consulting	acks_within(30s, auto_close)、reply_within(600s, auto_close)
conversation	max_silence(3600s, audit)
discussion	turn_within(120s, warn)、turn_within(600s, hide)
workflow	turn_within(120s, warn)、turn_within(600s, auto_close)

违规处理器

from ag2.network import Expectation

Expectation(name="acks_within", on_violation="auto_close", params={"seconds": 30})

default_handlers() 恰好包含三个处理器，以 on_violation 为键：

on_violation	处理器类	效果
"audit"	AuditHandler	空操作处理器。 实际的审计记录（AUDIT_KIND_EXPECTATION_VIOLATED）由 AuditLog 监听器通过 on_expectation_fired 写入；在调用任何处理器之前，中心会将事件分发给该监听器，因此 AuditHandler 本身不执行任何操作。频道继续运行。
"notify_channel"	NotifyChannelHandler	向频道的每位参与者发布 EV_EXPECTATION_VIOLATED。频道继续运行。（审计仍由 AuditLog 监听器写入。）
"auto_close"	AutoCloseHandler	以 reason="expectation_violated:<name>" 关闭频道。（审计仍由 AuditLog 监听器写入。）

没有内置的 "warn" 或 "hide" 处理器——这些名称仅出现在 discussion / workflow 清单中；在为其注册处理器之前，它们都不会执行任何操作。

覆盖适配器默认值

在频道配置项中传入 expectations，即可替换适配器的默认值：

channel = await alice.open(
    type="conversation",
    target=bob.agent_id,
    knobs={
        "expectations": [
            {"name": "max_silence", "on_violation": "auto_close",
             "params": {"seconds": 600}},
        ],
    },
)

自定义评估器

from ag2.network import EV_TEXT, Expectation
from ag2.network import ExpectationContext, Violation


class TooManyMessagesEvaluator:
    name = "too_many_messages"

    # Signature: evaluate(self, expectation, context) -> Violation | None
    def evaluate(self, expectation: Expectation, context: ExpectationContext) -> Violation | None:
        threshold = int(expectation.params["max"])
        text_count = sum(1 for e in context.wal if e.event_type == EV_TEXT)
        if text_count > threshold:
            return Violation(
                expectation=expectation,                 # the Expectation object, not a string
                violator_ids=[],                          # channel-wide
                detail={"text_count": text_count, "threshold": threshold},
            )
        return None

Violation 的定义为 Violation(expectation: Expectation, violator_ids: list[str] = [], detail: dict = {})——expectation 是 Expectation 对象（包含 name、on_violation、params）；它没有 channel_id 字段（中心已知该字段）。评估器是针对频道状态的纯函数——不执行 I/O，也不进行修改——因此可以轻松测试。通过 hub.register_expectation_evaluator(TooManyMessagesEvaluator()) 注册。

确定性测试

hub = await Hub.open(MemoryKnowledgeStore(), expectation_sweep_interval=0)

# Manually advance state and tick:
clock.advance(45)
await hub._expectation_tick()  # operator API (leading underscore by convention)

审计日志

中心维护一个只追加的审计日志（AuditLog 实例），通过公开的 hub.audit_log 属性提供访问（内部属性为 hub._audit_log；可通过 hub.replace_audit_log(...) 替换该实例）：

records = await hub.audit_log.read_all()
for r in records:
    print(r["kind"], r["at"], r)

每条记录都是一个普通字典，至少包含 kind 和 at（ISO-Z 时间戳）；具体类别的字段会一并列出。

审计类型

from ag2.network import (
    AUDIT_KIND_AGENT_REGISTERED,
    AUDIT_KIND_AGENT_UNREGISTERED,
    AUDIT_KIND_RESUME_SET,
    AUDIT_KIND_SKILL_SET,
    AUDIT_KIND_RULE_SET,
    AUDIT_KIND_CHANNEL_CREATED,
    AUDIT_KIND_CHANNEL_CLOSED,
    AUDIT_KIND_CHANNEL_EXPIRED,
    AUDIT_KIND_TASK_TERMINATED,
    AUDIT_KIND_EXPECTATION_VIOLATED,
)

类型	时机	常见字段
AUDIT_KIND_AGENT_REGISTERED	hc.register(...)	agent_id、name、owner
AUDIT_KIND_AGENT_UNREGISTERED	hc.unregister(agent_id)	agent_id
AUDIT_KIND_RESUME_SET	hub.set_resume(...)	来源：RESUME_SOURCE_TENANT 或 RESUME_SOURCE_OBSERVED
AUDIT_KIND_SKILL_SET	hub.set_skill(...)	更新后的技能 Markdown
AUDIT_KIND_RULE_SET	hub.set_rule(...)	新的 Rule
AUDIT_KIND_CHANNEL_CREATED	alice.open(...)	creator_id、清单类型/版本、参与者
AUDIT_KIND_CHANNEL_CLOSED	任意关闭路径	reason
AUDIT_KIND_CHANNEL_EXPIRED	TTL 清理器	TTL 详情
AUDIT_KIND_TASK_TERMINATED	agent.task(...) 通过 TaskMirror 达到终止状态	owner_id、capability、outcome、latency_ms
AUDIT_KIND_EXPECTATION_VIOLATED	期望评估器的阈值已过	expectation、channel_id、评估器详情

常见查询

# All violations on the system.
violations = [r for r in await hub.audit_log.read_all()
              if r["kind"] == AUDIT_KIND_EXPECTATION_VIOLATED]

# Everything that happened on one channel.
channel_records = [r for r in await hub.audit_log.read_all()
                   if r.get("channel_id") == channel_id]

# All registrations for one tenant.
acme_agents = [r for r in await hub.audit_log.read_all()
               if r["kind"] == AUDIT_KIND_AGENT_REGISTERED
               and r.get("owner") == "acme"]

使用 DiskKnowledgeStore 时，审计日志具有持久性；使用 MemoryKnowledgeStore 时，日志仅在 hub 的生命周期内存在。

Hub 监听器 — 实时程序化可观测性

审计日志是持久记录。若要对 Hub 状态变更做出实时响应 — 推送到指标后端、流式传输到仪表板、向值班人员发送警报 — 请注册一个 HubListener：这是一个只读 Protocol，Hub 会在每次状态转换提交后向其分发事件。（内置审计日志本身也是这类监听器之一。）

方法（精确签名）	触发时机
on_envelope_posted(envelope, metadata)	信封已被接受、追加到 WAL、折叠并分发
on_envelope_rejected(envelope, reason)	仲裁器 / 验证拒绝了发送（reason 是类型化的 NetworkError）
on_dispatch_failed(envelope, recipient_id, reason)	向某个收件人投递时引发异常（reason 是一个 BaseException）
on_channel_event(channel_id, kind, payload)	kind ∈ opened / closed / expired / participant_removed / participant_hidden
on_agent_event(agent_id, kind, payload)	kind ∈ registered / unregistered / resume_set / skill_set / rule_set / observation_recorded
on_expectation_fired(channel_id, expectation, violation)	期望评估器发出一个 Violation
on_turn_failed(channel_id, agent_id, envelope_id, exc)	Agent 的通知处理器轮次引发异常（默认处理器会将失败路由到这里）
on_task_event(task_id, kind, payload)	观察到 ag2.task.* 生命周期事件（kind ∈ started / progress / completed / failed / expired / cancelled / mirror_failed）
on_inbox_pressure(agent_id, pending, cap)	收件人的收件箱首次超过 limits.inbox.high_water（每次越过阈值时触发一次，而非每个信封都触发）

所有方法都是 async；Hub 会按注册顺序依次等待它们，每个方法都由 try/except 包裹 — 有问题的监听器不会阻塞分发。请保持方法快速执行（将 I/O 操作排入你自己的任务）。继承 BaseHubListener（其中每个方法都是 pass），只重写你需要的方法：

from ag2.network import BaseHubListener

class MetricsListener(BaseHubListener):
    async def on_envelope_posted(self, envelope, metadata):
        statsd.incr(f"net.envelope.{envelope.event_type}")
    async def on_inbox_pressure(self, agent_id, pending, cap):
        statsd.gauge(f"net.inbox.{agent_id}", pending / cap)
    async def on_turn_failed(self, channel_id, agent_id, envelope_id, exc):
        sentry.capture_exception(exc)

hub.register_listener(MetricsListener())     # hub.unregister_listener(inst) to detach

与 Hub 子类相关的两个扩展接口：

• Hub 自身的 on_* 钩子 — Hub 上也有同一组空方法；Hub 子类可以直接重写它们（分发时会在调用已注册监听器的同时调用绑定方法）。如果可观测性就是你要发布的 Hub 变体，请使用子类；如果要添加可插拔的扩展，请使用 register_listener。
• hub.register_sweeper(name, interval_seconds, fn) / unregister_sweeper(name) — 将你自己的定期协程添加到 Hub 的间隔扫描器机制中（与内置 TTL 和期望扫描器并行）。如果 Hub.start() 已经运行，则通过子类注册的扫描器会立即启动；否则会排队等待启动。

on_inbox_pressure 由 limits.inbox.high_water（InboxBlock 字段）控制 — 这是一个绝对待处理数量阈值（int | None）。None（默认值）会自动解析为 int(limits.inbox.max_pending * 0.8)；0 会禁用此信号。它是硬性限制 InboxFull 的早期预警机制（上限为 limits.inbox.max_pending，由仲裁器的 authorize_inbox 强制执行）。

任务观察 — 建立履历记录

带能力标签的任务会自动更新 Agent 的 Resume.observed[capability]。网络由此得知“bob 已完成 47 个研究任务，中位延迟为 4.2 秒”。

为任务添加标签

agent.task(..., capability="X") 接受一个自由格式的能力字符串：

# `.tool` and `.task(...)` live on `Agent`, not on the `AgentClient` returned
# by `hc.register(...)`. So decorate the Agent before registering.
@worker_agent.tool
async def research(topic: str, ctx: Context) -> str:
    async with worker_agent.task(
        f"research: {topic}",
        capability="research",
        context=ctx,
    ) as task:
        await task.progress({"step": "gather"})
        # ... do work ...
        await task.complete({"items_found": 7})
    return "researched"


worker = await worker_hc.register(worker_agent, Passport(name="worker"), Resume())

传入 context=ctx，让任务在 LLM 轮次的流上触发事件 — TaskMirror 就附加在该流上。如果不传入，事件永远不会到达 Hub。

如果 capability=None（默认值），生命周期事件仍会写入 Hub 的审计日志，但不会更新 Resume.observed。履历记录是可选启用的。

读取履历记录

resume = await hub.get_resume(bob.agent_id)
stat = resume.observed.get("research")
if stat:
    print(f"completed={stat.completed}/{stat.n}  "
          f"failed={stat.failed}  "
          f"p50_latency={stat.p50_latency_ms}ms")

@dataclass(slots=True)
class ObservedStat:
    n: int = 0                        # total terminal events
    completed: int = 0
    failed: int = 0
    expired: int = 0
    p50_latency_ms: int | None = None  # rolling median of started_at → completed_at

延迟根据 task_meta.started_at 到终止事件时间计算，使用 Hub 的时钟。在测试中使用 MockClock，可以构造确定性的延迟。

TaskMirror 的作用

默认处理程序会为每个轮次自动附加一个 TaskMirror，其作用范围限定为当前活跃频道。该镜像会订阅 LLM 轮次流中的 TaskStarted / TaskProgress / TaskCompleted / TaskFailed / TaskExpired 事件，将每个事件作为 ag2.task.* 信封转发给 hub，并在带有 capability 标签的终态事件发生时调用 Hub.record_observation(...)，以更新 Resume.observed。

只有在编写自定义处理程序时，才需要手动附加 TaskMirror：

from ag2.network import TaskMirror
from ag2.stream import MemoryStream

mirror = TaskMirror(
    hub_client=client._hub_client,
    owner_id=client.agent_id,
    channel_id=metadata.channel_id,
)
stream = MemoryStream()
sub_ids = mirror.attach(stream)
try:
    await client.agent.ask(text, stream=stream)
finally:
    mirror.detach(stream, sub_ids)

镜像是按轮次附加的，而不是按 Agent 附加——每个入站信封都会创建一个新的镜像。它还会静默吞掉转发到 hub 时出现的错误；hub 连接不稳定不应导致 LLM 轮次崩溃。

何时不添加 capability 标签

仅在以下情况下添加标签：

• 任务代表你想要在 Agent 的简历中跟踪的能力。
• 失败 / 延迟信号具有运维意义（会影响路由、告警或同伴排名）。

未添加标签的任务仍会生成完整的生命周期审计记录——只是不会更新 Resume.observed。对于不代表外部可见能力的内部记账任务，请使用未添加标签的任务。

跨领域模式：多能力工作者

@worker_agent.tool
async def research(topic: str, ctx: Context) -> str:
    async with worker_agent.task(f"research: {topic}", capability="research", context=ctx) as t:
        # ...
    return "..."


@worker_agent.tool
async def summarise(text: str, ctx: Context) -> str:
    async with worker_agent.task("summarise", capability="summarisation", context=ctx) as t:
        # ...
    return "..."

经过几个频道后，worker.resume.observed 会同时保存 "research" 和 "summarisation" 的 ObservedStat，并分别跟踪它们。随后，同伴发现查询（peers(action="find", capability="research")——参见 ag2-network-tools-and-views）即可按延迟或完成率进行排名。

读取 hub 状态

调用	返回值
await hub.get_channel(channel_id)	ChannelMetadata 快照（状态、参与者、关闭原因）
await hub.get_resume(agent_id)	当前 Resume（包括 observed）
await hub.get_passport(agent_id)	当前 Passport
await hub.list_agents(kind=None)	已注册的 passport；kind="agent" / "human" / "remote_agent" 按 Passport.kind 筛选
await hub.read_wal(channel_id)	该频道中按顺序排列的 Envelope 列表
await hub.audit_log.read_all()	所有审计记录
hub.arbiter	当前活跃的 HubArbiter（只读）

hub 会在每次变更时更新 Resume.last_updated，因此你可以通过比较时间戳来检测视图是否过时。若要实现推送（而非使用这些拉取调用），请注册一个 HubListener。

快速参考 — 导入

from ag2.network import (
    # Identity
    Passport, Resume, ResumeExample, ObservedStat,
    # Rules
    Rule, AccessBlock, LimitsBlock, RateBlock, InboxBlock,
    ChannelTypeAccess, parse_duration,
    # Arbiter (swappable access / routing seam)
    HubArbiter, BaseHubArbiter, RuleBasedArbiter, Allow, Deny,
    # Listeners (live observability)  — hub.register_listener(...)
    HubListener, BaseHubListener,
    # Auth
    AuthAdapter, AuthRegistry, AuthBlock, NoAuth, ApiKeyAuth,
    # Expectations
    Expectation,
    ExpectationEvaluator, ExpectationContext,
    AcksWithinEvaluator, ReplyWithinEvaluator, MaxSilenceEvaluator,
    AuditHandler, NotifyChannelHandler, AutoCloseHandler,
    Violation, ViolationHandler,
    default_evaluators, default_handlers,
    # Audit kinds
    AUDIT_KIND_AGENT_REGISTERED,
    AUDIT_KIND_AGENT_UNREGISTERED,
    AUDIT_KIND_RESUME_SET,
    AUDIT_KIND_SKILL_SET,
    AUDIT_KIND_RULE_SET,
    AUDIT_KIND_CHANNEL_CREATED,
    AUDIT_KIND_CHANNEL_CLOSED,
    AUDIT_KIND_CHANNEL_EXPIRED,
    AUDIT_KIND_TASK_TERMINATED,
    AUDIT_KIND_EXPECTATION_VIOLATED,
    RESUME_SOURCE_OBSERVED, RESUME_SOURCE_TENANT,
    # Task observation
    TaskMirror,
    # Errors — the full family (no `LimitsExceeded`; limit/access denials raise AccessDeniedError, inbox-full raises InboxFull)
    NetworkError, AccessDeniedError, AuthError, InboxFull, NotFoundError, ProtocolError,
)