ag2-network-discussion

AG2 Network — ディスカッションアダプター

discussion は、N 人の参加者が順番に発言するチャネルです。参加者は固定された順序で発言し、チャネルを閉じるまで順番が循環し続けます。アダプターは validate_send によって「自分の番を待つ」ことを強制します。ハブの can_send プローブを使うと、各エージェントのデフォルトハンドラーは、自分の番ではないときに無駄な LLM 呼び出しを省略できます。

前提条件：まず ag2-network-quickstart をお読みください。このスキルでは、Hub.open、HubClient.register、チャネルのライフサイクル、Envelope の基本をすでに理解していることを前提としています。

使用する場面

• 「3 つのエージェントが順番に議論する」
• 「固定メンバーによるパネルディスカッション／ブレインストーミング」
• 「ラウンドロビン形式で草稿にコメントする査読者」
• 各参加者がサイクルごとに発言枠を得る、固定順序の N 人による順番制のあらゆるケース

次の用途には discussion を使用しないでください：

• 条件付きの引き継ぎ（「alice がセキュリティに言及したら、セキュリティの専門家に引き継ぐ」）→ ag2-network-workflow を使用します。
• 各ステップを一度だけ実行するパイプライン → TransitionGraph.sequence([...]) とともに ag2-network-workflow を使用します。
• 固定順序のない 2 人の参加者 → conversation を使用します（ag2-network-quickstart で説明しています）。
• 厳密な 1Q1R → consulting を使用します（ag2-network-quickstart で説明しています）。

仕様

参加者	2 人以上
発言順	ラウンドロビン（作成者が最初で、その後は登録順）
自動クローズ	なし
終了条件	明示的な channel.close() または TTL（以下の「終了」を参照）
デフォルトのビュー	NamedWindowedSummary(recent_n=N*2)。N は参加者数です
デフォルトの期待値	turn_within(120s, warn)、turn_within(600s, hide)
設定項目	{"ordering": ORDERING_ROUND_ROBIN}（現時点で提供されている順序設定はこれのみ）

ビューの最近のウィンドウは N*2 に設定されているため、各エージェントの投影にはおよそ直近 2 サイクル分が含まれます。議論が進んでもプロンプトが肥大化せず、筋の通った返信を作成するのに十分なコンテキストが得られます。

ディスカッションエージェント、NetworkPlugin、順番制

HubClient.register(...) はデフォルトで NetworkPlugin をアタッチします。これにより、アイデンティティレベルの動詞である peers／channels／tasks／context／delegate が追加されます。DiscussionAdapter.tools_for(...) は [] を返します。ディスカッションアダプターには、チャネル固有のツール（say など）はありません。参加者は返信を返すだけで自分の順番に貢献します。デフォルトハンドラーは、ラウンド終了時に EV_TEXT(reply.body) をそのターンの投稿として送信します。

discussion 固有の say がないため、これに起因する「二重送信」の危険はありません。順番制は validate_send のみで強制されます。state.expected_next_speaker 以外の参加者からの送信は ProtocolError（expects <other> to speak）で拒否されます。後述の can_send プローブを使うと、各ハンドラーは自分の番ではないときに LLM 呼び出しを省略できます。そのため、通常の運用では順番を無視した送信は起こらないはずです。

attach_plugin=False にすると、NetworkPlugin の 5 つの動詞（delegate／peers／channels／tasks／context）のみが削除されます。アダプターツールには影響しません（discussion にアダプターツールはありません）。必要最小限の参加者を使う場合に指定してください。

# Bare discussion participant (no plugin verbs):
alice = await alice_hc.register(alice_agent, Passport(name="alice"), Resume(), attach_plugin=False)

以下の例では、簡潔にするため attach_plugin=False を指定しています。参加者に delegate／peers などが必要な場合は、この指定を削除してください。詳細については、ag2-network-tools-and-views →「アダプター所有のツール（tools_for）」を参照してください。

最小の例

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,
    ORDERING_ROUND_ROBIN,
    Hub,
    HubClient,
    LocalLink,
    Passport,
    Resume,
)


async def main() -> None:
    config = AnthropicConfig(model="claude-sonnet-4-6")
    hub = await Hub.open(MemoryKnowledgeStore(), ttl_sweep_interval=0)
    link = LocalLink(hub)

    alice_hc, bob_hc, carol_hc = (HubClient(link, hub=hub) for _ in range(3))

    alice = await alice_hc.register(
        Agent("alice", prompt="The optimist. One short sentence.", config=config),
        Passport(name="alice"), Resume(), attach_plugin=False,
    )
    bob = await bob_hc.register(
        Agent("bob", prompt="The realist. One short sentence.", config=config),
        Passport(name="bob"), Resume(), attach_plugin=False,
    )
    carol = await carol_hc.register(
        Agent("carol", prompt="The skeptic. One short sentence.", config=config),
        Passport(name="carol"), Resume(), attach_plugin=False,
    )

    channel = await alice.open(
        type="discussion",
        target=[bob.agent_id, carol.agent_id],
        knobs={"ordering": ORDERING_ROUND_ROBIN},
    )
    await channel.send("Topic: should every developer learn Rust?")

    # Halt after 6 EV_TEXT envelopes (= 2 full alice→bob→carol cycles).
    await wait_for_text_count(hub, channel.channel_id, expected=6)
    await channel.close(reason="cap_reached")

    for hc in (alice_hc, bob_hc, carol_hc):
        await hc.close()
    await hub.close()


async def wait_for_text_count(hub, channel_id, expected, *, timeout=180.0):
    deadline = asyncio.get_event_loop().time() + timeout
    while asyncio.get_event_loop().time() < deadline:
        wal = await hub.read_wal(channel_id)
        if sum(1 for e in wal if e.event_type == EV_TEXT) >= expected:
            return
        await asyncio.sleep(0.05)
    raise asyncio.TimeoutError("did not reach expected count")


asyncio.run(main())

target=[bob.agent_id, carol.agent_id] の順序は重要です。作成者の後の巡回順になります。そのため、上限に達するまでは alice → bob → carol → alice → bob → carol → … の順に循環します。

ターンスキップの仕組み（can_send プローブ）

alice が最初のエンベロープを送信すると、ハブは EV_TEXT を bob と carol に配信します。両方のデフォルトハンドラーが並行して実行されます。

• bob のハンドラー — await hc.can_send(channel_id, bob.agent_id)。アダプターは「はい、bob は expected_next_speaker です」と返します。ハンドラーは Agent.ask を実行し、bob の返信を送信します。
• carol のハンドラー — await hc.can_send(channel_id, carol.agent_id)。アダプターは「いいえ、expected_next_speaker は bob です」と返します。ハンドラーは LLM を呼び出さずに 戻ります。

bob の返信が届くと、同じ配信が繰り返されます。今度は expected_next_speaker = carol なので、carol のハンドラーが呼び出され、bob はスキップします。LLM 呼び出しの無駄は発生しません。

ディスカッションチャンネル用にカスタムハンドラーを書く場合は、このパターンにならい、テキスト以外のエンベロープは default_handler に委譲してください。これにより、EV_CHANNEL_INVITE の自動 ACK を維持できます（そうしないと、チャンネルは invite_ack_timeout まで PENDING のままになり、その後ハブによって閉じられます）。

from ag2.network import Envelope, EV_TEXT, default_handler


async def my_handler(envelope: Envelope) -> None:
    if envelope.event_type != EV_TEXT:
        await default_handler(envelope, client)   # invite-ack + lifecycle bookkeeping
        return
    if not await hc.can_send(envelope.channel_id, my_agent_id):    # async — must await
        return  # not our turn — skip the LLM call
    # ... read WAL, project view, run Agent.ask, send reply ...

ラウンドロビンにおける HumanClient

ディスカッションの参加者は Agent である必要はありません。HumanClient を登録し（hc.register_human(Passport(name="user", kind="human")) — ag2-network-quickstart → 「HumanClient」を参照）、ほかの参加者と同様に参加者順序に追加できます。自動応答する通知ハンドラーはないため、アプリケーションがそのターンを制御します。on_envelope(...) で購読するか（または next_envelope(...) で取得し）、エンベロープが届いたら await hc.can_send(channel_id, user.agent_id) を確認します（非同期なので await が必要です）。人間のターンであれば、その人にプロンプトを表示して await user.send(channel_id, text) を実行し、そうでなければ待機します。（フレームワーク自体のディスカッションテストもまさにこの方法で、人間のターンになるまでアダプターの状態を確認しています。）

状態オブジェクト

@dataclass(slots=True)
class DiscussionState:
    participant_order: list[str]
    expected_next_speaker: str
    turn_count: int = 0

participant_order は、作成時に Participant.order で参加者を並べ替えて固定されます。受理された EV_TEXT ごとに、ラウンドロビンは (current_index + 1) % len(participant_order) で進みます。確認やテストのために状態を読み取るには、hub._adapter_states[channel_id] を使用します（アンダースコアは「運用者向け API」を示しており、本番環境でのエージェントロジックには使用しないでください）。

検証ルール

DiscussionAdapter.validate_send は以下を拒否します。

• state.expected_next_speaker 以外からの EV_TEXT — ProtocolError を発生させます。
• 参加者以外からの送信。
• 閉じたチャンネルへの送信。

プロトコルエンベロープ（EV_CHANNEL_*、ag2.task.*）はターンチェックを回避します。これらはターン交代ではなく、記録管理のためのものです。

順序のカスタマイズ

現在提供されているのは ORDERING_ROUND_ROBIN のみです。

from ag2.network import ORDERING_ROUND_ROBIN

channel = await alice.open(
    type="discussion",
    target=[bob.agent_id, carol.agent_id],
    knobs={"ordering": ORDERING_ROUND_ROBIN},  # equivalent to "round_robin"
)

ほかの順序を渡すと、作成時にエラーが発生します。今後の順序方式（動的、重み付き、優先度）は、ラウンドロビンの契約を壊すことなく、ここに追加できます。

条件付きターン交代が今すぐ必要な場合は、アダプターを workflow に切り替えてください。TransitionGraph.round_robin(participants, max_turns=N) を使えば、同じ構成で、後から条件を追加できます。ag2-network-workflow を参照してください。

終了

discussion は自動的に終了しません。次の 4 つの方法があります。終了を誰が決めるかに応じて選んでください。

方法	決定する主体	タイミング
アプリケーション側の上限 → channel.close()	オーケストレーションコード	ターン数、時間、またはその他の外部条件
エージェント側のツール（ChannelInject 経由）	LLM	参加者が「終了です」と判断したとき
カスタムアダプターのサブクラス	フレームワーク	コンテンツに基づく停止（ディスカッションではまれ）
TTL / turn_within の期待値	ハブのスイーパー	セーフティネットとしてのみ

アプリケーション側の上限（標準的な方法）

await wait_for_text_count(hub, channel.channel_id, expected=6)
await channel.close(reason="cap_reached")

終了理由は EV_CHANNEL_CLOSED.event_data["reason"] で確認できます。監視する側が、上限到達による正常終了と TTL や期待値違反による終了を区別できるよう、具体的な理由を選んでください。

エージェント側のツール

ほかのアダプターと同じパターンです。アクティブな Channel を注入し、close() を呼び出します。

from ag2.network.client.inject import ChannelInject


async def end_discussion(reason: str, channel: ChannelInject) -> str:
    if channel is None:
        return "no active channel"
    await channel.close(reason=f"agent_close:{reason}")
    return f"closed: {reason}"


for agent in (alice_agent, bob_agent, carol_agent):
    agent.tool(end_discussion)

どの参加者でも終了できます。全体像については、ag2-network-quickstart の「5 つの終了方法」を参照してください。

discussion では不十分な場合

「bob が期限に言及したら deadlines_expert に切り替える」のような条件付きのターン交代が必要になったら、discussion は適切なアダプターではありません。workflow に移行してください。N 者間という同じ形を維持しながら、ターン交代は FromSpeaker、ToolCalled、ContextEquals の条件を持つ TransitionGraph で管理されます。移行手順については ag2-network-workflow を参照してください。

デフォルトの動作

期待値	閾値	ハンドラー	動作
turn_within	120s	warn	AUDIT_KIND_EXPECTATION_VIOLATED をログに記録します。チャネルは開いたままです。
turn_within	600s	hide	ビューのプロジェクションで、遅れている話者を1サイクルの間「スキップ」としてマークします。チャネルは開いたままです。

discussion は意図的に、期待値を理由として自動で終了することはありません。応答の遅い話者によって、パネル全体が終了すべきではないためです。サイクルが停止したときにチャネルを終了させたい場合は、ag2-network-governance を使って期待値を auto_close に変更してください。

クイックリファレンス — インポート

from ag2.network import (
    DISCUSSION_TYPE,           # = "discussion"
    ORDERING_ROUND_ROBIN,      # = "round_robin"
    DiscussionAdapter,         # the adapter class itself (rarely instantiated directly)
    DiscussionState,           # state dataclass
    HumanClient,               # a non-LLM participant in the rotation — HubClient.register_human(...)
)

文字列形式の type="discussion" と ordering="round_robin" も使用できます。定数は同じ値であり、大規模なコードベースで型安全性を確保するために用意されています。