ag2-network-quickstart

AG2 Network — 快速入门

该网络可将一组 Agent 实例组成一个协调的多 Agent 系统。中央 Hub 保存权威状态（注册表、审计日志、每个通道的预写日志）；每个 Agent 都运行在精简的 AgentClient 后面，并通过 HubClient 连接。每条消息都会经过 Hub，因此每次交互都可重放且可观察。

该网络为可选启用。裸 Agent 仍可独立运行——只有从 ag2.network 导入时，网络才会启用。

适用场景

当用户希望两个或更多 Agent 进行交互时，加载此技能。明确的触发语句包括：

• “让两个 Agent 互相交谈”
• “设置多 Agent 系统／多 Agent 聊天／Agent 网络”
• “让 Agent 能够互相调用”
• “替换经典的 GroupChat／ConversableAgent.handoffs”
• “为我的 Agent 添加注册表／审计跟踪／共享收件箱”

**非网络任务：**如果用户要求一个 Agent生成自己的子任务（递归任务分发、并行执行子任务），请改用 ag2-subagent-delegation。网络适用于彼此独立且已注册的 Agent 协作。

概念模型

                        ┌────────────────┐
                        │      Hub       │  ←── audit log, registry, channels,
                        │ ── adapters ── │       WAL, sweepers, expectation
                        │ ── channels ── │       evaluators
                        │ ── audit log ──│
                        └────────┬───────┘
                                 │  in-process duplex (LocalLink)
              ┌──────────────────┼──────────────────┐
              ▼                  ▼                  ▼
        ┌──────────┐       ┌──────────┐       ┌──────────┐
        │AgentClient│      │AgentClient│      │AgentClient│
        │  alice    │      │   bob     │      │   carol   │
        │  Agent    │      │  Agent    │      │  Agent    │
        └──────────┘       └──────────┘       └──────────┘

概念	所在位置	用途
Hub	每个网络一个	权威状态——注册表、审计日志、通道表、WAL、清理器
Passport	Hub 注册表	稳定身份（name、由 Hub 分配的 agent_id、可选的 owner、model）
Resume	Hub 注册表	能力声明，以及由 Hub 更新的 observed 记录
HubClient	每个进程一个	每个进程的代理；管理注册
AgentClient	每个已注册 Agent 一个	包装 Agent；发送信封并运行通知处理程序
Envelope	线上格式	Hub 标记的单个通道中单个事件的记录
Channel	由 agent_client.open(...) 创建	由适配器管理的有界多方交互

Hub 在注册时分配 agent_id。路由时请使用该 ID（alice.agent_id），而非易读的名称。

最小网络示例

单进程 Hub、两个 Agent、一个 consulting 通道（严格 1Q1R，回复后自动关闭）：

import asyncio

from ag2 import Agent
from ag2.config import AnthropicConfig
from ag2.knowledge import MemoryKnowledgeStore
from ag2.network import (
    EV_CHANNEL_CLOSED,
    EV_TEXT,
    Hub,
    HubClient,
    LocalLink,
    Passport,
    Resume,
)


async def main() -> None:
    config = AnthropicConfig(model="claude-sonnet-4-6")

    # 1. Boot the hub. MemoryKnowledgeStore for in-process; DiskKnowledgeStore(path)
    #    for durability across restarts.
    hub = await Hub.open(MemoryKnowledgeStore(), ttl_sweep_interval=0)
    link = LocalLink(hub)  # in-process duplex transport

    # 2. One HubClient per process boundary; here we use two for clarity.
    alice_hc = HubClient(link, hub=hub)
    bob_hc = HubClient(link, hub=hub)

    # 3. Register each Agent. The hub stamps the agent_id and returns an AgentClient.
    #    attach_plugin=False gives a bare agent (no peers/channels/tasks/context/
    #    delegate tools) — fine here since alice & bob just send/reply. The `say`
    #    tool is offered per-turn by the adapter, not the plugin (see below).
    alice = await alice_hc.register(
        Agent("alice", prompt="Ask one focused question and stop.", config=config),
        Passport(name="alice"),
        Resume(),
        attach_plugin=False,
    )
    bob = await bob_hc.register(
        Agent("bob", prompt="Answer in one short sentence.", config=config),
        Passport(name="bob"),
        Resume(),
        attach_plugin=False,
    )

    # 4. Open a consulting channel. The adapter posts EV_CHANNEL_INVITE to bob,
    #    bob's default handler auto-acks, channel goes ACTIVE.
    channel = await alice.open(type="consulting", target="bob")
    await channel.send(
        "What's the single most important property of a distributed system?",
        audience=[bob.agent_id],
    )

    # 5. Bob's default handler probes can_send, runs Agent.ask on the inbound
    #    text, sends the reply. ConsultingAdapter sees both flags set and
    #    auto-closes with reason "consulting_complete".
    close_env = await alice.wait_for_channel_event(
        channel_id=channel.channel_id,
        predicate=lambda e: e.event_type == EV_CHANNEL_CLOSED,
        timeout=60.0,
    )
    print(f"closed: {close_env.event_data.get('reason')!r}")

    # 6. Replay the conversation from the hub's WAL.
    wal = await hub.read_wal(channel.channel_id)
    for env in wal:
        if env.event_type == EV_TEXT:
            speaker = "alice" if env.sender_id == alice.agent_id else "bob"
            print(f"{speaker}: {env.event_data['text']}")

    await alice_hc.close()
    await bob_hc.close()
    await hub.close()


asyncio.run(main())

必需内容就这么多。以下内容都是变体。

选择通道适配器

每个通道都由一个适配器管理，用于定义参与者、轮次顺序和终止条件。系统提供四种适配器；调用 agent_client.open(type=..., target=...) 时选择一种。

适配器	参与者	轮次顺序	终止条件	默认视图
consulting	恰好 2 个	严格 1Q1R（发起方 → 响应方）	响应方回复后自动关闭	FullTranscript()
conversation	恰好 2 个	自由形式（任一方均可随时发言）	显式调用 channel.close() 或 TTL 到期	WindowedSummary(recent_n=10)
discussion	2 个或更多	通过 ordering="round_robin" 进行轮询	显式关闭或 TTL 到期	WindowedSummary(recent_n=N*2)
workflow	2 个或更多	声明式 TransitionGraph	图终止（TerminateTarget／max_turns）	WindowedSummary(recent_n=N*2)

**此技能涵盖 consulting 和 conversation（两方适配器）。**其他适配器请参阅：

• N 方轮询 → 加载 ag2-network-discussion。
• 使用 TransitionGraph 进行声明式编排、条件移交，或从经典 GroupChat 迁移 → 加载 ag2-network-workflow。

插件工具与适配器工具 — attach_plugin 和 say

HubClient.register(...) 默认会附加 NetworkPlugin。该插件提供身份级别的横切工具 — peers / channels / tasks / context / delegate — 不包含任何特定于频道的工具。特定于频道的工具（尤其是 say，用于向频道发布文本信封）由频道适配器通过 adapter.tools_for(...) 在每轮中提供，与 attach_plugin 无关：

适配器	是否提供 say？
consulting	仅提供给在 1Q1R 中轮到发言的参与者
conversation	始终提供（无轮次顺序）
discussion	仅提供给 expected_next_speaker
workflow	从不提供 — 路由由你的 @tool 交接函数完成

因此，attach_plugin=False 不再意味着“没有 say” — 而是意味着“没有 delegate / peers / channels / tasks / context”。对于纯 Agent（测试、纯流水线工作进程），请传入此参数；如果 Agent 需要这些横切动词，则保留默认值。

通常完全不需要考虑 say：默认处理程序总是会根据 Agent 的普通回复发布轮次结束信封，因此只需回答即可。say 用于少数情况，例如在一轮中发布额外消息，或发布到 Agent 也参与的其他频道。（需要避免的边缘情况：在 consulting 频道上的 Agent 调用 say(...) 并且返回非空回复，会发送两次 — 第二个信封会在严格的 1Q1R 适配器上触发 channel is closed。如果不希望 Agent 调用 say，只需不要在提示词中要求它这样做。）完整详情：ag2-network-tools-and-views →“适配器拥有的工具（tools_for）”。

consulting — 严格的 1Q1R

即上面的示例。发起方恰好发送一个实质性信封；响应方恰好发送一条回复；适配器会自动关闭，并将 event_data["reason"] 设为 "consulting_complete"。

Hub 端的状态：

@dataclass(slots=True)
class ConsultingState:
    initiator_sent: bool = False
    respondent_replied: bool = False

ConsultingAdapter.validate_send 会拒绝：

• 顺序错误的发送（响应方在发起方发送第一个信封前发言）。
• 两个标志均已设置后发送的任何内容 — 会引发 ProtocolError，并传播回调用方的 channel.send(...)。

默认预期严格：acks_within(30s, auto_close) 和 reply_within(600s, auto_close)。如果响应方未在 30s 内确认邀请，或未在 10 分钟内回复，Hub 会自动关闭，并使用 "expectation_violated:acks_within" 或 "expectation_violated:reply_within"。可通过 ag2-network-governance 调整这些设置。

conversation — 自由形式的双方对话

任何一方都可以随时以任意顺序发送消息。适配器绝不会自动关闭；你可以通过 channel.close() 停止，也可以依赖 TTL。

channel = await alice.open(type="conversation", target="bob")
await channel.send("Hi bob, what's a good first ML concept to learn?")

两个默认处理程序都会对每个入站 EV_TEXT 运行 Agent.ask，因此对话启动后会自动持续进行。两种停止模式：

1. 应用端限制 — 轮询 WAL，直到看到 N 个文本信封，然后执行 await channel.close()。
2. 空回复作为停止信号 — 默认处理程序会将空正文视为“不发送”，因此回复 "" 的 LLM 会结束链条。可搭配类似 "reply with empty string when you have nothing useful to add" 的提示词。

ConversationAdapter.validate_send 只会检查“发送方是否为参与者？” — 允许同一方连续发送，因为适配器不会建模轮次顺序。

默认预期较宽松：max_silence(3600s, audit)（会将日志写入审计类型 AUDIT_KIND_EXPECTATION_VIOLATED，但不会关闭）。

适用场景： 两个专业 Agent 在没有固定顺序的情况下进行实际对话（分析师 ↔ 评论者）；由应用控制停止信号的聊天 UI。

不适用场景： 严格的 1Q1R（使用 consulting）；多参与者聊天（使用 discussion 或 workflow）。

非大模型参与者 — HumanClient

并非所有参与者都是 Agent。HumanClient 是网络成员，不使用大模型、没有 NetworkPlugin，也没有组装策略——它可以是 UI 上的用户、连接到其他系统的桥梁，或用于向频道填入初始内容的脚本化“用户”。使用 register_human 注册它（而不是 register——现在这条路径会拒绝 kind="human" 并引导你使用此方法）：

from ag2.network import HumanClient, Passport

user_hc = HubClient(link, hub=hub)
user = await user_hc.register_human(Passport(name="user", kind="human"))   # resume=, rule=, auto_ack_invites= optional

它在出站方面实现了与 AgentClient 相同的 NetworkClient 接口——user.open(type=..., target=...)、user.send(channel_id, text, audience=...)、user.post_envelope(env)（用于发送类似工作流 EV_PACKET 这类适配器格式信封的备用方法）——并提供两种方式来使用入站信封：

• 推送 — user.on_envelope(callback)；每收到一个入站信封，协程就会触发一次。多个回调可以组合；若回调抛出异常，异常会被记录，但不会传播。
• 拉取 — await user.next_envelope(predicate=..., timeout=...) 会等待下一个匹配项；async for env in user.envelopes(): ... 会持续传输所有信封，直到调用 user.disconnect()。

它会自动确认频道邀请（默认 auto_ack_invites=True），使 hub 的加入握手无需 UI 往返即可完成；传入 auto_ack_invites=False，即可自行控制加入流程。使用 hub.list_agents(kind="human") 查找人类参与者。

常见用途：为 workflow 频道填入初始内容的启动者（FromSpeaker(user) → AgentTarget(first_agent)——参见 ag2-network-workflow）、consulting 问答中的人类参与者，或 discussion 轮流发言中的参与者。Agent 端的接口（自定义处理器、无头工作进程、网关）见 ag2-network-tools-and-views。

身份 — Passport 和 Resume

三个数据类用于描述网络中的 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
    kind="agent",          # optional: "agent" (default / None) | "human" | "remote_agent"
)

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

hub 会在注册时填入 Passport.agent_id 和 Passport.created_at。Passport.kind（类型别名 PassportKind）用于区分参与者类型——常规的、由大模型驱动的 AgentClient 使用 None/"agent"，HumanClient 使用 "human"（请使用 register_human，而不是 register），"remote_agent" 保留给 A2A/联邦场景。hub.list_agents(kind=...) 会按此字段筛选。Agent 执行带有能力标签的任务时，hub 会更新 Resume.observed 字段（按能力统计的 ObservedStat 计数）——如何进行配置，请参见 ag2-network-governance。

最简单的情况下，你可以传入 Passport(name="alice") 和 Resume()，即可完成设置。

频道生命周期

agent_client.open(type=..., target=...)
    │
    ▼
PENDING ──┬─ all targets ack ──→ ACTIVE ──┬─ adapter terminates ──→ CLOSED
          │                                │
          └─ ack timeout ──→ CLOSED       └─ explicit channel.close() ──→ CLOSING ──→ CLOSED
                  (invite_timeout)               or TTL expired

Hub.open(...) 上 invite_ack_timeout 的默认值为 30s；如果受邀目标中有任何一个未在此时间窗口内确认，create_channel 就会引发 ProtocolError，并将频道直接关闭，状态变为 CLOSED，原因是 "invite_timeout"。（另一条独立的期望检查器路径——acks_within 期望——会改为自动关闭频道，原因是 "expectation_violated:acks_within"；参见下文的 consulting 部分。）

发送信封

await channel.send(text, audience=[bob.agent_id])

audience 控制哪些人可以看到该信封（默认：所有参与者）。hub 会填入 Envelope.envelope_id、sender_id 和 created_at，并将信封写入 WAL。

对于自定义事件类型，请通过 agent_client.send_envelope(envelope) 发送原始信封。Envelope 的完整结构和 EV_* 事件常量见 ag2-network-tools-and-views。

五种通道关闭路径

每个通道都会以 EV_CHANNEL_CLOSED 信封终止。有五种路径可以到达该状态；根据由谁决定来选择：

模式	决定者	最适合
channel.close(reason=...)	你的编排代码	自定义上限（轮数、时间、谓词）
Agent 侧工具（ChannelInject）	大模型	“由 Agent 决定何时结束”
适配器哨兵（子类）	框架	基于内容的停止（“TERMINATE”关键字）
工作流 TerminateTarget	声明式图	多步骤编排
TTL / 预期	Hub 的扫描器	基于时间或预期的安全网

适配器兼容性：

	自动关闭	应用关闭	Agent 工具	哨兵	工作流图
consulting	是（回复后）	是（提前退出）	是	子类	不适用
conversation	从不	是（典型用法）	是	规范实现	不适用
discussion	从不	是（典型用法）	是	子类	不适用
workflow	是（图）	是（覆盖）	是（通过 ToolCalled）	不适用	规范实现

模式：Agent 侧关闭工具

由大模型决定何时停止。将活动的 Channel 注入工具：

from ag2.network.client.inject import ChannelInject


async def end_conversation(reason: str, channel: ChannelInject) -> str:
    """Close the active channel. The reason flows on EV_CHANNEL_CLOSED."""
    if channel is None:
        return "no active channel"
    await channel.close(reason=f"agent_close:{reason}")
    return f"closed: {reason}"


alice_agent.tool(end_conversation)
bob_agent.tool(end_conversation)

默认处理程序会在每轮大模型调用前将活动的 Channel 写入 context.dependencies，因此 ChannelInject 会自动解析。在网络轮次之外，它会解析为 None —— 上述保护逻辑可确保该工具在非网络上下文中安全运行。

优先使用工具调用模式，而非适配器哨兵（“TERMINATE”关键字）——工具调用有类型、可在 WAL 中追踪、支持多语言，且能抵御提示词注入。

监听关闭

这五种路径都会生成相同的 EV_CHANNEL_CLOSED 信封，因此观察者只需一个谓词：

close_env = await alice.wait_for_channel_event(
    channel_id=channel.channel_id,
    predicate=lambda e: e.event_type == EV_CHANNEL_CLOSED,
    timeout=180.0,
)
print(f"reason: {close_env.event_data.get('reason')!r}")

ChannelMetadata.close_reason 也会存储在通道记录中——await hub.get_channel(channel_id) 无需重新读取 WAL 即可返回原因。

重放与检查

Hub 的 WAL 是记录“此通道中发生了什么”的唯一真实来源：

wal = await hub.read_wal(channel.channel_id)
for env in wal:
    print(env.created_at, env.event_type, env.sender_id, env.event_data)

每个信封都会由 Hub 写入戳记（envelope_id、created_at、sender_id、audience、event_type、event_data）。WAL 以通道 ID 为键；对于此进程能够访问的任何通道，都可以读取其 WAL。

如需查看审计日志（Hub 级事件：Agent 已注册、通道已创建/关闭、预期未满足），请参阅 ag2-network-governance。

关闭

await alice_hc.close()
await bob_hc.close()
await hub.close()

始终将 Hub.open(...) 与 hub.close() 配对使用（通常放在 try/finally 中）。hub.close() 会取消扫描器任务、关闭底层存储并清空待处理的 I/O。HubClient.close() 会取消链接的监听任务，并取消订阅客户端。

接下来加载什么

用户目标	技能
N 方轮询 / 固定轮次顺序	ag2-network-discussion
声明式编排 / TransitionGraph / GroupChat 迁移	ag2-network-workflow
速率限制、访问策略、预期、审计、能力追踪、可替换仲裁器 / Hub 监听器	ag2-network-governance
适配器拥有的工具（say / tools_for）、插件工具（delegate / peers / …）、自定义处理程序、HumanClient 内部实现、视图、对等方发现	ag2-network-tools-and-views

快速参考 — 导入

from ag2.network import (
    # Hub + transport
    Hub,
    HubClient,
    LocalLink,
    # Participants
    HumanClient,           # register via HubClient.register_human(...)
    # Identity
    Passport,
    PassportKind,          # Literal["agent", "human", "remote_agent"] | None
    Resume,
    ResumeExample,
    # Envelopes + events
    Envelope,
    EV_TEXT,
    EV_CHANNEL_INVITE,
    EV_CHANNEL_INVITE_ACK,
    EV_CHANNEL_OPENED,
    EV_CHANNEL_CLOSED,
    EV_CHANNEL_EXPIRED,
    Priority,
    # Errors
    AccessDeniedError,
    InboxFull,
    ProtocolError,
)
from ag2.knowledge import MemoryKnowledgeStore  # or DiskKnowledgeStore