ag2-network-quickstart

AG2 Network — クイックスタート

このネットワークを使うと、複数の Agent インスタンスを連携するマルチエージェントシステムにできます。中央の Hub が権威ある状態（レジストリ、監査ログ、チャンネルごとの先行書き込みログ）を保持し、各エージェントは HubClient 経由で接続する軽量な AgentClient の背後に配置されます。すべての送信はハブを経由するため、すべてのやり取りを再生・観測できます。

ネットワークはオプトインです。単独の Agent は引き続き単体で動作します。ネットワークが有効になるのは、ag2.network からインポートした場合のみです。

使用する場面

ユーザーが2つ以上のエージェントをやり取りさせたい場合は、このスキルを読み込みます。具体的なトリガーフレーズは次のとおりです。

• 「2つのエージェントを会話させる」
• 「マルチエージェントシステム／マルチエージェントチャット／エージェントネットワークを設定する」
• 「互いを呼び出せるエージェント」
• 「従来の GroupChat ／ ConversableAgent.handoffs を置き換える」
• 「エージェント用のレジストリ／監査証跡／共有受信トレイを追加する」

ネットワークタスクではない場合： ユーザーが 1つのエージェント に独自のサブタスクを生成させようとしている場合（再帰的なタスクの分岐、サブタスクの並列実行）は、代わりに ag2-subagent-delegation を使います。ネットワークは、連携する 個別に登録された エージェント向けです。

メンタルモデル

                        ┌────────────────┐
                        │      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	ネットワークごとに1つ	権威ある状態 — レジストリ、監査ログ、チャンネルテーブル、WAL、スイーパー
Passport	Hub レジストリ	安定した ID（name、Hub が割り当てる agent_id、任意の owner、model）
Resume	Hub レジストリ	ケイパビリティの申告と、Hub が変更する observed の実績記録
HubClient	プロセスごとに1つ	プロセス単位のブローカー。登録を管理します
AgentClient	登録済みの Agent ごとに1つ	Agent をラップし、エンベロープを送信して通知ハンドラーを実行します
Envelope	通信フォーマット	1つのチャンネル内の1つのイベントを記録する、Hub がスタンプを付与したレコード
Channel	agent_client.open(...) によって作成	アダプターによって制御される、人数制限付きの複数参加者間のやり取り

ハブは登録時に agent_id を割り当てます。ルーティングには、人が読める名前ではなく、この ID（alice.agent_id）を使います。

最小構成のネットワーク

単一プロセスのハブ、2つのエージェント、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())

必須の構成要素はこれだけです。以下はすべて応用例です。

チャンネルアダプターの選択

各チャンネルは、参加者、ターン順、終了条件を定義するアダプターによって制御されます。4つのアダプターが用意されています。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)

このスキルでは、2者間アダプターの consulting と conversation を扱います。 その他については、次を参照してください。

• N者間のラウンドロビン → ag2-network-discussion を読み込みます。
• TransitionGraph による宣言型オーケストレーション、条件付きハンドオフ、または従来の GroupChat からの移行 → ag2-network-workflow を読み込みます。

プラグインツールとアダプターツール — attach_plugin と say

HubClient.register(...) は、デフォルトで NetworkPlugin をアタッチします。このプラグインは、アイデンティティレベルの横断的なツール、つまり peers ／ channels ／ tasks ／ context ／ delegate を提供し、チャンネル固有のツールは提供しません。チャンネル固有のツール（特に、テキストエンベロープをチャンネルに投稿する say）は、attach_plugin の設定に関係なく、チャンネルのアダプターによって adapter.tools_for(...) を通じてターンごとに提供されます。

アダプター	say を提供するか
consulting	1Q1R で自分のターンの参加者にのみ提供
conversation	常に提供（ターン順なし）
discussion	expected_next_speaker にのみ提供
workflow	提供しない — ルーティングには @tool のハンドオフ関数を使用

そのため、attach_plugin=False は「say が使えない」という意味ではなく、「delegate ／ peers ／ channels ／ tasks ／ context が使えない」という意味です。単独のエージェント（テスト、純粋なパイプラインワーカー）にはこれを渡し、横断的な動詞が必要なエージェントではデフォルトのままにしてください。

通常、say について意識する必要はありません。デフォルトハンドラーは、エージェントの通常の返信からターン終了時のエンベロープを常に投稿するため、単に応答するだけのエージェントでも問題ありません。say は、ターン中に追加のメッセージを投稿する場合や、エージェントが参加している別のチャンネルに投稿する場合など、まれなケースで使います。（避けるべきエッジケース：consulting チャンネル上のエージェントが say(...) を呼び出し、さらに空でない返信も返すと、二重送信になります。2回目のエンベロープによって、厳密な 1Q1R アダプターで channel is closed が発生します。エージェントに say を呼び出させたくない場合は、プロンプトで指示しないでください。）詳細：ag2-network-tools-and-views →「アダプター所有のツール（tools_for）」。

consulting — 厳密な 1Q1R

上記の例です。開始者は内容のあるエンベロープをちょうど 1 つ送信し、応答者は返信をちょうど 1 つ送信します。その後、アダプターは event_data["reason"] == "consulting_complete" として自動的にクローズします。

ハブ側の状態：

@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) です。応答者が 30 秒以内に招待を承認しない場合、または 10 分以内に返信しない場合、ハブは "expectation_violated:acks_within" または "expectation_violated:reply_within" を理由に自動的にクローズします。これらの設定は ag2-network-governance で調整できます。

conversation — 自由形式の 2 者間会話

どちら側も、いつでも、どのような順序でも送信できます。アダプターは自動的にクローズしません。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 を実行するため、開始すると会話が自動的に進行します。停止方法は 2 つあります。

1. アプリケーション側の上限 — テキストエンベロープを N 件確認するまで WAL をポーリングし、その後 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 にログを記録しますが、クローズはしません）。

conversation を使う場合： 2 人の専門家が固定の順序なしで実際に会話する場合（分析担当者 ↔ 批評担当者）。アプリケーションが停止シグナルを制御するチャット UI。

conversation を使わない場合： 厳密な 1Q1R（consulting を使用）；複数参加者のチャット（discussion または workflow を使用）。

LLM を使用しない参加者 — HumanClient

すべての参加者が Agent とは限りません。HumanClient は LLM、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 のような、アダプター形式のエンベロープを送信するためのエスケープハッチ）を利用できます。また、受信 エンベロープを処理する方法が 2 つあります。

• プッシュ — user.on_envelope(callback) を使います。コルーチンは受信エンベロープごとに 1 回実行されます。複数のコールバックを組み合わせることができ、例外を発生させたコールバックは伝播せず、ログに記録されます。
• プル — await user.next_envelope(predicate=..., timeout=...) は次に一致するエンベロープが来るまで待機します。async for env in user.envelopes(): ... は user.disconnect() まで、すべてのエンベロープをストリーミングします。

既定では auto_ack_invites=True によりチャネルへの招待を自動承認するため、UI との往復なしでハブの参加ハンドシェイクが完了します。参加を自身で制御するには、auto_ack_invites=False を指定してください。hub.list_agents(kind="human") で人間の参加者を検出できます。

一般的な用途としては、workflow チャネルのキックオフ用シーダー（FromSpeaker(user) → AgentTarget(first_agent)。ag2-network-workflow を参照）、consulting の Q&A における人間側の参加者、discussion のラウンドロビンに参加するメンバーなどがあります。エージェント側のインターフェース（カスタムハンドラー、ヘッドレスワーカー、ゲートウェイ）については、ag2-network-tools-and-views を参照してください。

ID — Passport と Resume

3 つのデータクラスで、ネットワーク上のエージェントを表します。ほとんどのフィールドはテナントが指定し、残りはハブが設定します。

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="…")],
)

登録時に、ハブは Passport.agent_id と Passport.created_at を設定します。Passport.kind（型エイリアスは PassportKind）は参加者の種類を区別します。通常の LLM ベースの AgentClient では None／"agent"、HumanClient では "human"（register ではなく register_human を使用）、A2A／フェデレーション用に予約されているのは "remote_agent" です。hub.list_agents(kind=...) で種類による絞り込みができます。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 を発生させ、チャネルは理由 "invite_timeout" で直ちに CLOSED に移行します。（別個の期待値スイーパー経路である acks_within 期待値の場合は、代わりに "expectation_violated:acks_within" で自動的に閉じられます。詳しくは以下の consulting セクションを参照してください。）

エンベロープの送信

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

audience はエンベロープを閲覧できる参加者を制御します（既定値：すべての参加者）。ハブは Envelope.envelope_id、sender_id、created_at を設定し、WAL に書き込みます。

カスタムイベントタイプの場合は、agent_client.send_envelope(envelope) で生のエンベロープを送信します。Envelope の完全な構造と EV_* イベント定数については、ag2-network-tools-and-views を参照してください。

チャネルを閉じる5つの方法

すべてのチャネルは EV_CHANNEL_CLOSED エンベロープで終了します。そこに至る方法は5つあります。誰が決定するかで選んでください。

パターン	決定者	最適な用途
channel.close(reason=...)	オーケストレーションコード	独自の上限（ターン数、時間、条件）
エージェント側のツール（ChannelInject）	LLM	「エージェントが終了を判断する」
アダプターのセンチネル（サブクラス）	フレームワーク	内容に基づく停止（「TERMINATE」キーワード）
ワークフローの TerminateTarget	宣言型グラフ	複数ステップのオーケストレーション
TTL／期待値	ハブのスイーパー	時間または期待値に基づくセーフティネット

アダプターの互換性：

	自動終了	アプリによる終了	エージェントツール	センチネル	ワークフローグラフ
consulting	はい（返信後）	はい（早期終了）	はい	サブクラス	該当なし
conversation	いいえ	はい（通常）	はい	標準	該当なし
discussion	いいえ	はい（通常）	はい	サブクラス	該当なし
workflow	はい（グラフ）	はい（オーバーライド）	はい（ToolCalled 経由）	該当なし	標準

パターン：エージェント側の終了ツール

終了するタイミングをLLMが判断します。アクティブな 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)

デフォルトのハンドラーは、各LLMターンの前にアクティブな Channel を context.dependencies に設定するため、ChannelInject は自動的に解決されます。ネットワークターン以外では None に解決されます。上記のガードにより、ネットワーク以外のコンテキストでもツールを安全に使用できます。

アダプターのセンチネル（「TERMINATE」キーワード）より、ツール呼び出しパターンを優先してください。ツール呼び出しは型付けされ、WALで追跡可能で、多言語に対応し、プロンプトインジェクションにも強いためです。

終了の監視

5つの方法はいずれも同じ EV_CHANNEL_CLOSED エンベロープを生成するため、オブザーバーで必要なのは1つの述語だけです。

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を再読み込みせずに終了理由を取得できます。

リプレイと検査

ハブの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)

すべてのエンベロープにはハブによるスタンプが付与されます（envelope_id、created_at、sender_id、audience、event_type、event_data）。WALはチャネルIDをキーとし、このプロセスから参照できるすべてのチャネルについて読み取れます。

監査ログ（エージェントの登録、チャネルの作成／終了、期待値への違反などのハブレベルのイベント）については、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
レート制限、アクセス制御ポリシー、期待値、監査、機能追跡、差し替え可能な裁定者／ハブリスナー	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