ag2-network-tools-and-views

AG2 Network — 工具、视图与自定义处理器

网络中 Agent／客户端一侧的所有内容——与 ag2-network-governance（位于 Hub 侧）相对应。本节涵盖 LLM 对通道的视图（视图）、它可以执行的操作（五个自动注入的插件工具，以及适配器自有的 say）、其他 Agent 对通道的了解（技能 Markdown）、如何完全替换处理器，以及完整的 Envelope 参考。

前置条件：请先阅读 ag2-network-quickstart。本技能假设你了解 Hub.open、HubClient.register、通道生命周期，以及基本的 agent_client.open(...) / channel.send(...) 用法。

适用场景

• “限制／扩展 LLM 可用的网络工具范围”（插件工具与适配器 tools_for）
• “编写自定义信封处理器”
• “构建不运行 LLM 的网关／无头工作进程”
• “添加非 LLM 参与者——UI 上的人员、桥接器、脚本化播种器”（HumanClient）
• “自定义每个 Agent 看到的通道历史记录”（视图策略）
• “在信封到达 LLM 之前移除／编辑敏感信息／筛选信封”
• “通过技能 Markdown 配置对等端发现”
• “发送自定义事件类型（myapp.review_request，……）”
• “我需要 EV_* 常量列表／Envelope 结构／audience 语义”

网络工具——插件工具与适配器工具

Agent 每轮使用的 LLM 工具列表由两部分组成：

1. 插件工具——使用默认的 attach_plugin=True 注册时，NetworkPlugin 会向 agent.tools 添加五个身份级、与通道无关的动词：peers / channels / tasks / context / delegate。在任何通道中的行为都相同。
2. 适配器工具——通道的适配器通过 adapter.tools_for(client, metadata, state, participant_id) 在每轮提供通道专属工具；默认处理器会解析这些工具，并将它们合并到 agent.ask(tools=...) 中（按 (adapter, client) 缓存，因此只需支付一次架构构建成本）。唯一内置的适配器工具是 say。

因此，say 不是插件工具——适配器会根据回合状态决定是否提供该工具：

适配器	向……提供 say
consulting	轮到其发言的参与者（1Q1R 中，提示词发出前为发起者；发起者回复前为应答者）
conversation	始终向所有参与者提供
discussion	仅向 expected_next_speaker 提供
workflow	不向任何人提供——路由由你的 @tool 交接函数负责；适配器返回 []

因此，attach_plugin=False 不再控制 LLM 是否能看到 say——它只控制五个插件动词。（面向用户的摘要请参阅 ag2-network-quickstart →“插件工具与适配器工具”。）

delegate 和 say

工具	来源	签名	用途
delegate	插件	delegate(target, prompt, capability?, timeout=300.0)	一次性咨询——与 target 打开一个 consulting 通道，发送 prompt，等待唯一回复，并返回其文本。这是一个独立通道，因此可在任何通道的回合中安全调用。
say	适配器（consulting / conversation / discussion）	say(content, audience?, channel_id?)	向活动通道中发布一个 EV_TEXT（或发布到指定的、该 Agent 参与的通道）。audience 是对等端名称列表（会解析为 ID）；None 表示广播。信封结构由 adapter.build_text_envelope(...) 提供——非 AG2 桥接器也会调用相同的 Layer-2 辅助函数。

# The LLM emits, e.g.:
delegate(target="bob", prompt="What's the right way to model X?", capability="modeling")
say(content="Here's my answer: …")

框架会在通知处理器内部自动解析 ChannelInject（当前通道）、ChannelStateInject 和 AgentClientInject——LLM 不会看到这些参数。

四个分组的操作分派工具

每个工具都接收一个 action 字面量和该操作专属的参数，让 LLM 的工具列表保持简短。

peers(action) — 发现

操作	参数	返回值
"find"	query?, capability?, sort_by?, limit=20	对等 Agent 摘要列表（不包括调用方）。
"describe"	name	一个对等 Agent 的完整资料：{passport, resume, skill_md}。如果没有注册 SKILL.md，skill_md 会回退为根据 passport 和 resume 渲染的内容。

channels(action) — 生命周期

操作	参数	返回值
"list"	state="active"|"all"	此 Agent 参与的频道。
"open"	type, target, knobs?, intent?, ttl?, message?	与 agent_client.open 行为一致。如果提供了 message，在频道达到 OPENED 后会立即代表发起方将其作为第一个 envelope 发送（便于为 consulting/workflow 注入初始内容）；如果初始发送失败，则以原因为 seed_failed 关闭频道。返回 {channel_id, type, participants[, seed_envelope_id]}。
"info"	channel_id	如果此 Agent 参与其中，则返回完整的 ChannelMetadata。
"close"	channel_id?（默认为当前频道）	以原因为 "closed_by_agent" 关闭频道。

tasks(action) — 任务生命周期

分为两类：主动操作（Agent 位于自身的 agent.task(...) 代码块内）和观察操作（hub 已知的任何任务）。

操作	类别	参数	返回值
"progress"	主动	payload	发出 TaskProgress。
"complete"	主动	result?	终止操作 — 发出 TaskCompleted。
"list"	观察	scope="own"|"all", state="active"|"all", limit=20	任务摘要。
"status"	观察	task_id	刷新的 TaskMetadata。
"wait"	观察	task_id, timeout=300, poll_interval=0.1	阻塞直到任务进入终止状态。

"start" 有意不作为工具提供 — 如果 LLM 调用它，就会绕过 async with agent.task(...) 生命周期，导致 TaskInject 无法正确限定作用域。所有者在自己的代码中启动任务；任务处于活动状态后，LLM 使用 progress / complete，而一次性的远程工作则使用 delegate。

context(action) — 过去的内容

操作	参数	返回值
"search"	query, scope="channel"|"knowledge", limit=10	文本与 query 匹配的摘录（不区分大小写的子串匹配）。
"quote"	speaker, recent_n=1, channel_id?	来自 speaker 的最近 recent_n 个 EV_TEXT envelopes。

scope="knowledge" 会检索调用 Agent 自身的 KnowledgeStore（仅支持子串搜索 — 对于向量 / 语义搜索，Agent 自身的循环会直接调用 framework-core 的 recall）。

适配器自带的工具（tools_for）——以及 say 的重复发送

默认处理器会在每轮调用 adapter.tools_for(client, metadata, state, participant_id)，并将结果与 Agent 自带的 @tool 和任何插件工具合并后传入 agent.ask(tools=...)。不接收 LLM 输入的适配器（workflow）会返回 []。其他适配器会根据轮次状态（见上表）决定是否提供 say。适配器也可以返回更丰富的工具；目前内置的只有 say。

这对 workflow 的影响：工作流 Agent 永远看不到 say——它唯一的“发言”途径是处理器根据 reply.body 构建的轮次结束 EV_PACKET，以及你编写的任何 @tool 移交函数。旧的“先调用 approve()，然后 say(...) 与轮次结束 EV_PACKET 发生竞争”的故障模式，现在在结构上已不可能发生。

对于 consulting / conversation / discussion，会提供 say，但通常很少需要它。默认处理器已经会发送轮次结束信封（build_round_envelope → EV_TEXT(reply.body)；如果内容为空则为 None），因此只需用文本回复的 Agent 已经完成发言。say 用于在一轮中发送额外消息，或发送到 Agent 参与的其他频道（channel_id=）。

风险：如果 Agent 调用 say(content="…")，然后又返回非空的回复正文，就会在一轮中发送两个实质性信封。在 conversation 中这没有影响；在 consulting 中，第二个信封会触发严格的 1Q1R 适配器错误——ProtocolError: channel '<id>' is closed。缓解方法，按稳妥程度从高到低排列：

• 不要提示 Agent 调用 say。 普通回复就是规范的频道回复；只有在你明确引导它使用 say 时才会重复发送。
• 替换默认处理器，使其不再发送轮次结束信封——这样 say 就是 Agent 的发言方式（见下文的网关 / 无头工作进程模式）。
• attach_plugin=False 在这里没有帮助——say 由适配器提供，而不是插件提供。它只会移除 delegate / peers / channels / tasks / context。请将其用于精简 Agent，而不要用来屏蔽 say。

非 LLM 参与者 — HumanClient

HumanClient 是框架中的首个一等公民“非 Agent 参与者”：没有 LLM、没有 NetworkPlugin，也没有组装策略。在 UI 中的人、连接到其他系统的桥接程序、为工作流提供初始输入的脚本化“用户”——这些都属于 HumanClient。使用 register_human 注册（不要使用 register，该方法现在会拒绝 kind="human" 并指引你使用此方法）：

from ag2.network import HumanClient, Passport

user = await hc.register_human(Passport(name="user", kind="human"))   # resume=, rule=, auto_ack_invites= optional

它满足 NetworkClient Protocol，具有与 AgentClient 相同的出站接口：

ch = await user.open(type="consulting", target="analyst")     # initiate a channel
await user.send(ch.channel_id, "What's our Q3 exposure?")     # EV_TEXT convenience
await user.post_envelope(env)                                 # escape hatch — adapter-shaped envelopes (e.g. workflow EV_PACKET via adapter.build_packet_envelope)

……此外，还有两种方式可用于接收入站信封（AgentClient 只有通知处理器回调；HumanClient 还提供了显式队列）：

接口	调用	语义
推送	user.on_envelope(callback)	每收到一个入站信封，就触发一次 async 回调；多个回调按注册顺序组合执行；回调抛出异常时会记录日志，但不会传播。使用 remove_envelope_callback(cb) 解除绑定。
拉取	await user.next_envelope(*, predicate=None, timeout=None)	阻塞，直到收到符合 predicate 的下一个信封（若为 None，则接受任意信封）；超时时会抛出异常。
拉取（流式）	async for env in user.envelopes(): ...	持续生成每个入站信封，直到调用 user.disconnect()。

auto_ack_invites=True（默认值）会让人类参与者自动接受频道邀请，使 hub 的法定人数握手无需 UI 往返即可完成；传入 auto_ack_invites=False 可手动控制是否加入（此时你需要自行 post_envelope 一个 EV_CHANNEL_INVITE_ACK）。使用 hub.list_agents(kind="human") 可查找人类参与者；使用 hub.list_agents(kind="agent") / kind="remote_agent" 可筛选其他参与者。

典型角色：为 workflow 频道提供启动输入（FromSpeaker(user) → AgentTarget(first_agent) — 参见 ag2-network-workflow）、consulting 问答中的人类一方、discussion 轮流发言中的参与者，或上述任意场景前端的 WebSocket/CLI 桥接程序。对于无头 Agent（不应运行 LLM、但仍需要插件工具 / tools_for 解析的 AgentClient），应改为替换其通知处理器——见下文。

替换默认处理器

默认处理器负责完成“Agent 接收信封 → 自动确认邀请 → 运行 LLM（并合并 adapter.tools_for(...)）→ 发送轮次结束信封”的整个流程。对于无头工作器、网关，或任何不应运行 LLM 的 Agent，可替换该处理器。（对于本就不需要 LLM 的参与者，应使用上文的 HumanClient，而不是使用自定义处理器的 AgentClient。）

选择停用插件（和/或默认处理器）

worker = await hc.register(agent, passport, resume, attach_plugin=False)  # no peers/channels/tasks/context/delegate
worker.on_envelope(my_custom_handler)                                     # ← this is what swaps the handler

这是两个相互独立的开关：

• attach_plugin=False 会跳过 NetworkPlugin.register(agent) —— Agent 的工具列表中将不再包含 peers / channels / tasks / context / delegate（以及 NetworkContextPolicy 前缀）。它不会影响 notify 处理器，也不会移除 say（它由适配器管理，见上文）。
• 默认 notify 处理器始终处于启用状态（它由 AgentClient 接入，而不是由插件接入 —— 除非注册时使用了 attach_default_handler=False）。调用 client.on_envelope(callback) 可替换它；调用 client.on_envelope(client._run_default_handler) 可恢复它。

调用 client.on_envelope(callback) 后会失去什么：

默认处理器会执行……	如果你不复现这些行为……
自动确认 EV_CHANNEL_INVITE（发送 EV_CHANNEL_INVITE_ACK）	通道将一直处于 PENDING 状态，直到 invite_ack_timeout（30s）超时，随后 hub 会以 invite_timeout 为由关闭通道
对 EV_TEXT / EV_PACKET 运行 _process_substantive：读取 WAL、投影视图、记录依赖项、附加 TaskMirror、解析 adapter.tools_for(...)、调用 agent.ask(...)，并发送由 adapter.build_round_envelope 构建的回合结束信封	LLM 永远不会运行，也不会有回复发回，通道会一直停滞，直到某个期望触发
对 ag2.channel.* / ag2.task.* 生命周期信封不执行任何操作	（跳过这些信封不会造成影响，但它们到达时可能会让你感到意外）

最省事的做法是：自行处理你关心的事件，其余事件交给 default_handler。

处理器签名。 传给 client.on_envelope(...) 的回调只会接收 (envelope,) —— 框架会执行 await self._on_envelope(envelope)。处理器中对 client 的任何引用，都是从外层作用域捕获的（闭包变量）。而导出的 default_handler 签名为 (envelope, client) —— 委托时请同时传入这两个参数。

网关处理器

from ag2.network import Envelope, EV_TEXT, default_handler


async def gateway_handler(envelope: Envelope) -> None:
    """Forward inbound text to an external system instead of running an LLM."""
    if envelope.event_type != EV_TEXT:
        await default_handler(envelope, client)   # auto-ack invites etc. (client from closure)
        return
    text = envelope.event_data.get("text", "")
    await my_external_queue.put({
        "from": envelope.sender_id,
        "text": text,
        "channel": envelope.channel_id,
    })


client.on_envelope(gateway_handler)

选择性覆盖（回退到默认处理器）

from ag2.network import default_handler, EV_CHANNEL_INVITE


async def selective_handler(envelope: Envelope) -> None:
    if envelope.event_type == EV_CHANNEL_INVITE and envelope.sender_id not in TRUSTED_AGENTS:
        return   # untrusted invite → silent drop; hub will time out and close
    await default_handler(envelope, client)   # default for everything else


client.on_envelope(selective_handler)

过滤转发（前置/后置钩子）

async def logged_handler(envelope: Envelope) -> None:
    log.info("inbound %s from %s", envelope.event_type, envelope.sender_id)
    try:
        await default_handler(envelope, client)
    finally:
        log.info("processed %s", envelope.envelope_id)

绕过适配器工具

适配器会通过 tools_for(...) 为每个回合提供工具（尤其是 consulting / conversation / discussion 中的 say）；默认处理器会将这些工具合并到 agent.ask(tools=...) 中。如果能力较强的模型坚持在未收到提示时调用 say（Claude 经常会这样），导致重复发送并影响通道正常工作，你可以换用一个自定义处理器，让它执行相同的实质性流程，但从 agent.ask 中省略 tools= —— Agent 的静态 agent.tools（你的 @tool 以及任何插件工具）仍会保留，因为 ask(tools=…) 添加的是额外工具，并非替换工具。基本结构如下：

async def no_say_handler(envelope, client):
    if envelope.event_type != EV_TEXT:
        await default_handler(envelope, client)         # invite-ack + lifecycle for free
        return
    if not await client._hub_client.can_send(envelope.channel_id, client.agent_id):
        return                                          # not our turn / channel closing

    # ... read_wal_until / resolve_view_policy / stamp_dependencies (hooks above) ...
    reply = await client.agent.ask(current_input, stream=stream, dependencies=deps)   # ← no tools=
    out = adapter.build_round_envelope(metadata=meta, sender_id=client.agent_id, reply=reply,
                                       events=events, state=state, hub=client._hub)
    if out is not None:
        out.causation_id = envelope.envelope_id
        await client.send_envelope(out)

这与默认处理器的实质性流程相同，只少了一行。用同样的方法也可以添加工具（传入自定义的 tools=[...]），或替换视图、依赖项、回合信封结构等。

选择性覆盖的钩子

如果你想部分替换默认处理器的逻辑，可以使用拆分后的公共钩子：

from ag2.network import (
    read_wal_until,
    resolve_view_policy,
    stamp_dependencies,
)

钩子	用途
read_wal_until(client, envelope)	对 WAL 进行切片，读取到给定信封之前（不包括该信封）的内容。
resolve_view_policy(client, metadata)	返回此参与者应使用的 ViewPolicy。
stamp_dependencies(client, channel, state)	为 LLM 回合构建 context.dependencies 字典（CHANNEL_DEP、AGENT_CLIENT_DEP、HUB_DEP、CHANNEL_STATE_DEP）。state 是适配器当前的 State 对象（WorkflowState / DiscussionState / …），每个回合折叠一次。

当你的自定义处理器需要使用标准的 LLM 前置接线逻辑，但自定义 LLM 后置行为（或反过来）时，可使用这些钩子。

视图——每个 LLM 看到的内容

ViewPolicy 是通道 WAL 与 LLM 历史记录之间的投影层：

class ViewPolicy(Protocol):
    name: str
    async def project(
        self,
        wal: list[Envelope],
        *,
        participant_id: str,
        channel: ChannelMetadata,
        render_envelope: EnvelopeRenderer,           # required — supplied by the adapter
        name_for: NameResolver = default_name_resolver,  # optional — resolves agent_id → Passport.name
    ) -> list[BaseEvent]: ...

它会获取该参与者应看到的 WAL 片段，并返回一个 BaseEvent 列表，框架会将其作为预填充的流历史馈入 LLM 回合。render_envelope（一个 EnvelopeRenderer = Callable[[Envelope], str | None]）由通道适配器在每次调用时提供，因此视图无需依赖特定适配器；name_for（一个 NameResolver = Callable[[str], str]）用于标注投影行，默认使用 default_name_resolver（恒等函数）。适配器会声明默认视图；你也可以按通道覆盖。

内置视图

视图	行为	默认用于
FullTranscript()	按顺序包含适配器的 render_envelope 返回字符串的每个可见实质性信封。通过助手／用户角色位区分说话者（双人）。	consulting
WindowedSummary(recent_n=N)	最近的 N 个可见实质性信封。如果数量更多，则在前面添加一个 CompactionSummary 占位符，注明省略的轮次数量。	conversation（recent_n=10）
NamedWindowedSummary(recent_n=N)	与 WindowedSummary 类似，但会在每个非自身行前添加 [<sender name>]:（通过 name_for 解析）——适用于角色位无法区分“其他人”的 N 方通道。	discussion、workflow（recent_n = max(N*2, 4)）
NamedTranscript()	与 FullTranscript 类似，但会在非自身行前添加相同的 [<sender name>]: 前缀。	（非默认；选择启用 N 方完整历史）

所有视图都会遵循 audience——仅发给 [bob] 的信封不会出现在 carol 的投影中。

from ag2.network import FullTranscript, WindowedSummary, NamedWindowedSummary

view = WindowedSummary(recent_n=12)
projected = await view.project(
    wal_slice,
    participant_id=carol.agent_id,
    channel=metadata,
    render_envelope=adapter.render_envelope,   # required
    name_for=hub.name_for,                      # optional; defaults to identity
)

解析默认视图

from ag2.network import resolve_view_policy

policy = resolve_view_policy(client, metadata)

读取适配器清单中的 default_view_policy，并从注册表实例化相应视图。默认处理程序会在每个回合调用一次；自定义处理程序也应如此，除非有意绕过标准投影模型。

自定义视图

实现此协议，为其指定唯一的 name，然后使用它：

from ag2.events import BaseEvent, ModelMessage, ModelRequest, TextInput
from ag2.network import EV_TEXT, ViewPolicy


class FromOneOnly(ViewPolicy):
    """Show only envelopes from a single named sender."""
    name = "from_one_only"

    def __init__(self, sender_id: str) -> None:
        self.sender_id = sender_id

    async def project(
        self,
        wal,
        *,
        participant_id,
        channel,
        render_envelope,                 # required by the protocol
        name_for=lambda agent_id: agent_id,
    ):
        out: list[BaseEvent] = []
        for env in wal:
            if env.event_type != EV_TEXT or env.sender_id != self.sender_id:
                continue
            text = render_envelope(env)   # adapter renders the envelope to a string (or None)
            if text is None:
                continue
            out.append(
                ModelMessage(text) if env.sender_id == participant_id
                else ModelRequest([TextInput(text)])
            )
        return out

你发出的 BaseEvent 类型会决定 LLM 如何看到历史记录：例如，用 ModelRequest([TextInput(...)]) 表示“来自用户”，用 ModelMessage(...) 表示“来自助手”，等等。应通过提供的 render_envelope 回调渲染信封正文（这样视图就无需依赖特定适配器），而不是直接访问 event_data。完整的类型体系请参阅 ag2.events。

选择视图

• 简短、聚焦的交流 → FullTranscript()。令牌预算不是瓶颈，连贯性才是。
• 长期运行的双人交流 → WindowedSummary(recent_n=N)，根据轮次密度调整 N。
• 长期运行的 N 方讨论／工作流 → NamedWindowedSummary(recent_n=N)，确保说话者标签在窗口中保留（这已经是 discussion／workflow 的默认视图）。
• 应忽略无关闲聊的专业 Agent → 按受众或标签筛选的自定义视图。
• 注重隐私的工作流 → 在投影前移除字段或编辑内容的自定义视图。

切换视图不会影响 WAL——所有信封仍会保留，所有操作人员仍可查看。视图只会改变 LLM 对历史记录的感知。

技能——Agent 如何描述自己

技能是带有 frontmatter 的 Markdown，Hub 会原样存储，并在通过 peers(action="describe", name=...) 查询时展示给对等方。注册时传入：

agent_client = await hc.register(
    agent,
    Passport(name="researcher"),
    Resume(claimed_capabilities=["research"]),
    skill_md="""\
---
title: Research Assistant
expertise: [policy, finance]
---

# Researcher

A senior policy analyst. Best at:

- Scenario synthesis from multi-source briefs.
- Rebuttal review with confidence scores.

Limitations: not for code review or numerical analysis.
""",
)

解析 frontmatter

from ag2.network import parse_skill_frontmatter, ParsedSkill

parsed: ParsedSkill = parse_skill_frontmatter(skill_md)
print(parsed.frontmatter)  # {"title": "Research Assistant", "expertise": [...]}
print(parsed.body)         # the markdown body

回退技能

如果未提供 skill_md，Hub 会根据简历生成技能内容，避免对等方查询时返回空句柄：

from ag2.network import render_fallback_skill

skill_md = render_fallback_skill(passport, resume)

这在以编程方式构建技能时很有用——例如，租户上传了简历，但没有 Markdown。

注册后更新

await hub.set_skill(agent_id, new_skill_md)

发出 AUDIT_KIND_SKILL_SET。审计结构与 set_resume 相同；租户代码可随时替换技能。

Envelope 线格式

@dataclass(slots=True)
class Envelope:
    channel_id: str
    sender_id: str                       # agent_id
    audience: list[str] | None           # None = broadcast to all participants
    event_type: str                      # "ag2.msg.text", "ag2.channel.invite", etc.
    event_data: dict[str, Any]           # event-specific payload

    envelope_id: str = ""                # hub-stamped on accept
    task_id: str | None = None
    causation_id: str | None = None      # envelope_id this one is "in reply to"
    trace_id: str | None = None
    priority: Priority = "normal"        # Priority is a Literal, not an enum (see below)
    depth: int = 0                       # delegation hop count; hub auto-increments on reply path
    idempotency_key: str | None = None   # reserved for cross-process transports
    created_at: str = ""                 # hub-stamped ISO-Z on accept
    ttl_seconds: int | None = None       # per-envelope TTL; None defers to channel expires_at

Hub 会在接纳时为 envelope_id 和 created_at 盖上时间戳（并在回复路径上自动递增 depth）。其他内容均由发送方提供。（没有 sequence 字段——WAL 顺序就是接纳顺序；请参阅“读取 WAL”。）

实质性事件

常量	字符串	event_data
EV_TEXT	"ag2.msg.text"	{"text": "<body>"}
EV_PACKET	"ag2.packet"	{"routing": {...}, "context_updates": {...}, "body": "<text>"}

EV_TEXT 携带纯文本。EV_PACKET 是工作流适配器对一轮结束时的原子化记录——将路由决策（与 ToolCalled 规则匹配）、累积的 context_vars 变更和最终正文打包到一个 envelope 中。框架会在工作流通道上的每次 Agent.ask 轮次结束后发布此事件；工具作者无需直接构造它们。

通道生命周期事件

常量	时机
EV_CHANNEL_INVITE	创建通道时，Hub 会向每个 target 发布。
EV_CHANNEL_INVITE_ACK	每位受邀者在接受邀请时发布。
EV_CHANNEL_INVITE_REJECT	可选——受邀者拒绝邀请时发布（默认处理器不会拒绝，但你可以重写）。
EV_CHANNEL_OPENED	所有确认都收到后，Hub 会发布。
EV_CHANNEL_CLOSED	任何终止路径上，Hub 都会发布；event_data.reason 会说明原因。
EV_CHANNEL_EXPIRED	TTL 清理器关闭通道时，Hub 会发布。
EV_EXPECTATION_VIOLATED	评估器的阈值被突破且处理方式为 notify 时，Hub 会发布。

上下文和任务事件

常量	字符串	时机
EV_CONTEXT_SET	"ag2.context.set"	工具/参与者发布此事件以修改通道范围内的 context_vars。event_data 为 {"set": {...}, "delete": [...]}。
EV_TASK_CANCEL_REQUEST	"ag2.task.cancel_request"	对等方请求任务所有者取消任务。event_data 为 {"task_id": str, "reason": str}。所有者可以接受请求（Task.cancel），也可以忽略。
EV_TASK_CANCELLED	"ag2.task.cancelled"	所有者发布的终止 envelope，可由对等方处理器匹配。event_data 为 {"task_id": str, "reason": str}。

任务生命周期（已启动 / 进度中 / 已完成 / 失败 / 已过期）不会作为线上的 envelope 传递——TaskMirror 会将其从所有者的流直接镜像到 Hub 的 TaskMetadata 缓存。上面的两个 EV_TASK_* 常量是例外：取消操作有由对等方发起的一面，需要可寻址的 envelope。默认处理器对任何 ag2.task.* event_type 都不执行操作。

接收方和可见性

audience: list[str] | None 控制谁能看到该 envelope：

• None — 广播给所有参与者。
• [agent_id_1, agent_id_2, ...] — 只有这些参与者能看到。

from ag2.network import visible_to

if visible_to(env, my_agent_id):
    process(env)

视图（FullTranscript、WindowedSummary）会遵循 audience 设置——只发给 [bob] 的 envelope 不会出现在 carol 的视图中。

优先级

Priority = Literal["background", "normal", "urgent"]

Priority 是字符串字面量类型，而非枚举——直接在 envelope 上设置 priority="urgent"（等）。优先级较高的 envelope 会在队列中先于较低优先级的 envelope 处理。请谨慎使用；大多数应用 envelope 都将 priority 保持为默认值 "normal"。

因果关系

causation_id 用于标记某个 envelope 是对另一个 envelope 的“回复”：

await channel.send(reply_text, causation_id=incoming_envelope.envelope_id)

默认处理器在回复入站 EV_TEXT 时会自动设置此字段。自定义处理器应为逻辑上的回复设置此字段，这对线程视图很有用。

发送原始 envelope（自定义事件类型）

channel.send(text, audience=...) 会为你封装 EV_TEXT。对于自定义事件类型，请构建一个 Envelope 并直接发布：

from ag2.network import Envelope

envelope = Envelope(
    channel_id=channel.channel_id,
    sender_id=alice.agent_id,
    audience=[bob.agent_id],
    event_type="myapp.review_request",
    event_data={"document_id": "doc-123", "kind": "security"},
)
await alice.send_envelope(envelope)

Hub 不会根据任何允许列表验证 event_type——自定义类型会原样通过。适配器只会折叠它们能识别的事件类型（实质性事件、工作流下的 EV_PACKET，以及生命周期事件）；自定义事件类型会写入 WAL 并投递给接收方，但不会推进轮次交接状态。

自定义事件类型指南

1. 使用点分隔的命名空间前缀（"myapp.review_request"，而不是 "review"），以避免与未来的 ag2.* 事件冲突。
2. 确保 event_data 可序列化为 JSON（不包含 datetime、dataclass 等），以便它能在存储中顺利往返。
3. 如果多个参与者应当作出响应，请设置 audience=None。如果只有一个参与者，则指定该参与者；视图会过滤掉非接收者。
4. 不要指望适配器会对自定义类型执行任何操作——它们会直接透传。自定义处理程序负责处理这些类型。

读取 WAL

wal = await hub.read_wal(channel.channel_id)
for i, env in enumerate(wal):
    print(f"{i:>3}  {env.event_type}  from={env.sender_id[:8]}")

信封按接纳顺序排列（WAL 列表顺序——每个信封没有单独的 sequence 计数器）。WAL 是规范的重放界面——重启时，Hub.hydrate() 会通过每个适配器重新折叠它，以重建内存中的状态。

快速参考——导入

from ag2.network import (
    # Default handler + hooks (for selective override)
    default_handler,
    read_wal_until,
    resolve_view_policy,
    stamp_dependencies,
    # Plugin (auto-attached by attach_plugin=True) — peers/channels/tasks/context/delegate.
    # `say` is NOT here; it comes from the channel adapter's tools_for(...).
    NetworkPlugin,
    # Non-LLM participant
    HumanClient,           # via HubClient.register_human(Passport(name=..., kind="human"))
    PassportKind,          # Literal["agent", "human", "remote_agent"] | None
    # Views
    ViewPolicy,
    FullTranscript,
    WindowedSummary,
    NamedTranscript,
    NamedWindowedSummary,
    # Skills (peer discovery)
    ParsedSkill,
    parse_skill_frontmatter,
    render_fallback_skill,
    # Envelopes + events
    Envelope,
    Priority,
    visible_to,
    EV_TEXT,
    EV_PACKET,
    EV_CHANNEL_INVITE,
    EV_CHANNEL_INVITE_ACK,
    EV_CHANNEL_INVITE_REJECT,
    EV_CHANNEL_OPENED,
    EV_CHANNEL_CLOSED,
    EV_CHANNEL_EXPIRED,
    EV_EXPECTATION_VIOLATED,
    EV_CONTEXT_SET,
    EV_TASK_CANCEL_REQUEST,
    EV_TASK_CANCELLED,
    # Dependency injection keys (for tests outside the notify handler)
    AGENT_CLIENT_DEP,
    CHANNEL_DEP,
    CHANNEL_STATE_DEP,
    HUB_DEP,
    TASK_DEP,
    # Injects (for tool signatures)
    AgentClientInject,
    ChannelInject,
    ChannelStateInject,
    HubInject,
    TaskInject,
)