ag2-network-workflow

AG2 Network — ワークフローアダプター

workflow は、複数の参加者をオーケストレーションするアダプターです。宣言型の TransitionGraph で、最初に発言する参加者、各ターンで発動する条件、チャネルが終了するタイミングを定義します。従来の GroupChat + Agent.handoffs パターンに代わる最新の仕組みで、ターン交代はプロセス内の話者セレクターではなく、ハブが管理します。

前提条件：まず ag2-network-quickstart を読んでください。このスキルでは、Hub.open、HubClient.register、チャネルのライフサイクル、基本的な agent_client.open(...) / channel.send(...) のフローを理解していることを前提とします。

使用する場面

次のような依頼があった場合は、このスキルを読み込んでください。

• 「条件付きのハンドオフを行う2人／3人／N人のエージェント」
• 「エージェントのパイプライン／順次処理（リサーチャー → ライター → エディター）」
• 「トリアージエージェントが専門エージェントに振り分ける」
• 「ドラフター／レビュアーのフィードバックループ」
• 「コーディネーター＋専門エージェント／階層型オーケストレーション」
• 「GroupChat / ConversableAgent.handoffs / ReplyResult(target=...) から移行したい」
• 「コンテキスト駆動のルーティング（OnContextCondition、StringContextCondition）を使いたい」

次の場合は workflow を使わないでください：

• 2つのエージェント間で厳密な1問1答を行う → consulting（ag2-network-quickstart を参照）。
• ターン順序を定めない自由形式の2者間チャット → conversation（ag2-network-quickstart を参照）。
• 条件のない固定のラウンドロビン → discussion（ag2-network-discussion を参照）。

ユーザーが迷っている場合は、次の経験則に従ってください。「固定リストの次の話者」より複雑な条件が少しでも必要なら、workflow を使います。

構成

参加者	2人以上
ターン順序	TransitionGraph の定義に従います
自動終了	はい — グラフが TerminateTarget を出力したとき、または max_turns に達したとき
デフォルトのビュー	NamedWindowedSummary(recent_n=N*2)
デフォルトの期待値	turn_within(120s, warn)、turn_within(600s, auto_close)
必須の設定項目	{"graph": <TransitionGraph.to_dict()>}

期待値は厳格です。応答が止まった話者がいると、10分後にチャネルが自動終了します。ag2-network-governance で調整してください。

ワークフローエージェントと NetworkPlugin

HubClient.register(...) は、デフォルトで NetworkPlugin をアタッチします。このプラグインが追加するのは、IDレベルで横断的に使えるツール（peers / channels / tasks / context / delegate）のみです。チャネル固有のツール（特に say と、ワークフローのハンドオフ用 @tool）は、adapter.tools_for(client, metadata, state, participant_id) から取得されます。これはデフォルトハンドラーによって各ターンで解決され、agent.ask(tools=...) にマージされます。ただし、workflow アダプターはこれらのツールを提供しません。 WorkflowAdapter.tools_for(...) は [] を返します。そのため、ワークフローエージェントに say は表示されません。ルーティングはすべて、Handoff を返す、コンテキスト変数を変更する、または channel.close() を呼び出す、@tool デコレーター付きのハンドオフ関数が担います。

その結果、従来の「ターン途中の say とラウンド終了時の EV_PACKET が競合する」問題は、ワークフローチャネルでは起こり得ません。正しく動作させるために attach_plugin=False を設定する必要はありません。機能に応じて選んでください。

• エージェントがピアに delegate する、peers / channels / tasks を参照する、またはグラフの外からチャネルの context を読み書きする必要がある場合は、（デフォルトのまま）プラグインを有効にします。（delegate / channels(action="open") は別のチャネルを作成します。ワークフローチャネルにはイベントを出力しないため、ターン管理に影響しません。）
• 本当に最小限のエージェントが必要な場合は、attach_plugin=False を設定します。自分で作成したドメイン固有の @tool のみを使います。純粋なパイプラインワーカーや大半のテストに適しています。

# Bare pipeline worker — no cross-cutting network tools:
worker = await wc.register(agent, Passport(name="worker"), Resume(), attach_plugin=False)

いずれの場合も、デフォルトハンドラーはLLMのプレーンテキストの応答をワークフローの EV_PACKET にラップします。以下の例では、純粋なパイプラインワーカーを想定して attach_plugin=False を指定しています。エージェントに delegate / peers などが必要な場合は、この指定を削除してください。詳細は ag2-network-tools-and-views の「アダプター所有のツール（tools_for）」を参照してください。

最小の例 — sequence によるパイプライン

import asyncio

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


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

    rhc, whc, ehc = (HubClient(link, hub=hub) for _ in range(3))

    researcher = await rhc.register(
        Agent("researcher", prompt="Bullet-point three facts on the topic.", config=config),
        Passport(name="researcher"), Resume(), attach_plugin=False,
    )
    writer = await whc.register(
        Agent("writer", prompt="Turn the bullets into one paragraph.", config=config),
        Passport(name="writer"), Resume(), attach_plugin=False,
    )
    editor = await ehc.register(
        Agent("editor", prompt="Tighten the paragraph. Reply with the final text.", config=config),
        Passport(name="editor"), Resume(), attach_plugin=False,
    )

    graph = TransitionGraph.sequence([
        researcher.agent_id, writer.agent_id, editor.agent_id,
    ])

    channel = await researcher.open(
        type="workflow",
        target=[writer.agent_id, editor.agent_id],
        knobs={"graph": graph.to_dict()},
    )
    await channel.send("Topic: how does HTTPS work?")

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

    for hc in (rhc, whc, ehc):
        await hc.close()
    await hub.close()


asyncio.run(main())

TransitionGraph.sequence([a, b, c]) は、FromSpeaker(x) → AgentTarget(next) の遷移を3つと、デフォルトの TerminateTarget("sequence_complete") を構築します。c が発話した後はどの遷移にも一致しないため、デフォルトが実行され、チャネルは自動的に閉じます。

TransitionGraph の構成

@dataclass(slots=True)
class TransitionGraph:
    initial_speaker: str            # agent_id of the first speaker (must be a participant)
    transitions: list[Transition]   # ordered list, walked in priority order on every fold
    default_target: TransitionTarget  # consulted if no transition matches
    max_turns: int | None = None    # hard turn cap; force-terminates on reaching


@dataclass(slots=True)
class Transition:
    when: TransitionCondition       # evaluated against the just-accepted envelope + state
    then: TransitionTarget          # if when() is True, this resolves the next speaker
    priority: int = 0               # LOWER runs first; ties break by insertion order

受け付けた実質的なエンベロープ（テキストまたはパケット）ごとに、アダプターは transitions のリストを順に確認します。最初に条件に一致したもののターゲットによって、次の話者が決まります。一致するものがなければ、default_target が参照されます。TerminateTarget は next_speaker=None として解決されるため、アダプターは AdapterResult(next_state=CLOSING, ...) を返し、ハブはそのターゲットの reason とともに EV_CHANNEL_CLOSED をポストします。max_turns は別個の、より厳しい上限です。turn_count がこの値に達すると、アダプターは "max_turns" を理由にチャネルをただちに閉じます。default_target にはフォールスルーしません（そのため、実際のターンで一致する遷移がなかった場合にのみ、default_target=TerminateTarget("…") の理由が表示されます）。

priority は昇順に並べ替えられます。つまり、数値が小さいほど先に確認されます。 アダプターは sorted(transitions, key=lambda t: t.priority) を実行してから、並べ替えた結果を順に確認します。そのため、priority=0（デフォルト）は priority=100 より優先されます。あるルールを他のルールより先に確認させたい場合は、リストの先頭に置くか（ソートは安定しているため、優先度が同じならリストの順序が維持されます）、負の優先度を付けてください。よくある間違いは、ContextEquals(...) → TerminateTarget(...) ルールに priority=100 を設定して、これを優先させようとすることです。ルールはリストの末尾に移動し、代わりに FromSpeaker(...) のフォールバックが先に実行されるため、チャネルが終了しません。

注意点 — キックオフ時の channel.send() は initial_speaker の最初のターンです

agent.open(type="workflow", ...) を実行してから await channel.send(text) を呼び出すと、その最初の送信は initial_speaker のターンとして処理されます。つまり、そのエージェントからのエンベロープであり、expected_next_speaker は initial_speaker と同じ値で始まるため、validate_send はこれを受け付け、グラフはただちに次へルーティングします。そのターンでは、initial_speaker の Agent.ask は実行されません。送信したテキストが、そのエージェントの貢献内容になります。

したがって、キックオフプロンプトに対して実際に考えて応答する最初のエージェント（ドラフター、リサーチャー、プランナーなど）を想定している場合、そのエージェントを initial_speaker にしてはいけません。また、channel.send() を呼び出すエージェントにしてもいけません。すっきりした解決策は、シーダーとして HumanClient を使うことです。LLM を使わない参加者（Agent でもプラグインでもない）で、チャネルを開いて概要を投稿することだけを担当します。これを initial_speaker にして、FromSpeaker(user) → AgentTarget(drafter) とルーティングします。

from ag2.network import HumanClient, Passport  # plus the usual imports

user_hc = HubClient(link, hub=hub)
user = await user_hc.register_human(Passport(name="user", kind="human"))   # ← register_human, not register
drafter  = await drf_hc.register(drafter_agent,  Passport(name="drafter"),  Resume(), attach_plugin=False)
reviewer = await rev_hc.register(reviewer_agent, Passport(name="reviewer"), Resume(), attach_plugin=False)

graph = TransitionGraph(
    initial_speaker=user.agent_id,
    transitions=[
        Transition(when=ContextEquals("approved", value=True), then=TerminateTarget("approved")),
        Transition(when=FromSpeaker(user.agent_id),     then=AgentTarget(drafter.agent_id)),    # brief → drafter
        Transition(when=FromSpeaker(drafter.agent_id),  then=AgentTarget(reviewer.agent_id)),
        Transition(when=FromSpeaker(reviewer.agent_id), then=AgentTarget(drafter.agent_id)),    # revise loop
    ],
    default_target=TerminateTarget("max_revisions"),
    max_turns=12,
)

channel = await user.open(type="workflow", target=[drafter.agent_id, reviewer.agent_id], knobs={"graph": graph.to_dict()})
await channel.send("Brief: …")   # consumed as `user`'s turn → routes to `drafter`, who drafts
# Later, when the channel closes: await user.next_envelope(predicate=lambda e: e.event_type == EV_CHANNEL_CLOSED, timeout=300)

register_human(passport, *, resume=None, rule=None, auto_ack_invites=True) は HumanClient を返します（HubClient.register(...) は現在 kind="human" を拒否し、こちらを案内します）。招待を自動承認するため、UI を介した往復なしでクォーラムのハンドシェイクが完了します。参加を手動で制御するには、auto_ack_invites=False を渡してください。on_envelope(callback)（プッシュ）または next_envelope(...)／envelopes()（プル）で操作します。詳しくは ag2-network-quickstart →「HumanClient」を参照してください。

HumanClient を導入したくない場合は、以前からある回避策も引き続き使えます。initial_speaker として使い、Agent.ask が呼び出されない使い捨ての「intake」Agent を用意します（そこに戻る遷移はありません）。ただし、HumanClient のほうが軽量（モデル設定もプラグインも不要）で、フレームワーク独自のスモークテストでも使われています。どちらの場合も、TransitionGraph.sequence([a, b, c]) には同じことが当てはまります。channel.send(...) は a のターンなので、a にプロンプトを中継するだけでなく最初の成果物を作成させる場合は、シーダーを先頭に追加してください。つまり、sequence([user, a, b, c]) とします。

（これを誤った場合の症状：チェーンの2番目のエージェントは、最初のエージェントの出力であるかのように、キックオフプロンプトだけを受け取ります。存在しない内容に「応答」し、パイプラインはしばしば成果物が存在すると幻覚した結果、でたらめな出力を生成します。）

組み込みターゲット

ターゲット	動作
AgentTarget(agent_id)	名前を指定した特定のエージェントに引き継ぎます。
RoundRobinTarget()	参加者の順序に沿って進めます。
StayTarget()	同じ話者が続けます（まれなケース。「詳しく説明します」のようなパターン）。
RevertToInitiatorTarget()	チャネルを開始した相手に引き継ぎます。
TerminateTarget(reason="…")	チャネルを終了します。理由は EV_CHANNEL_CLOSED.event_data["reason"] に格納されます。

組み込み条件

条件	発火するタイミング
Always()	受け入れられたすべてのエンベロープ。
FromSpeaker(agent_id)	直前に受け入れられたエンベロープが、このエージェントから送信されたとき。
ToolCalled(tool_name)	直前のターンで、この名前のツールが呼び出されたとき（パケットの routing.tool によって照合されます）。
ContextEquals(key, value)	チャネルスコープの context_vars[key] が value と等しいとき。キーがない場合は None と比較されます。

これらの組み合わせでは表現できない条件については、以下の「カスタムターゲット／条件」を参照してください。

便利なファクトリー

# Pipeline: a → b → c → terminate.
graph = TransitionGraph.sequence([alice.agent_id, bob.agent_id, carol.agent_id])

# Round-robin with a cap: a → b → c → a → b → c → terminate after 6 turns.
graph = TransitionGraph.round_robin(
    participants=[alice.agent_id, bob.agent_id, carol.agent_id],
    max_turns=6,
)

round_robin は Always() → RoundRobinTarget() と TerminateTarget("round_robin_complete") のデフォルトを使用します。ただし、Always() は毎ターン一致するため、デフォルトには到達せず、上限に達するとチャネルは実際には理由 "max_turns" で終了します（max_turns を設定しないと、無限に繰り返します）。sequence は FromSpeaker(steps[i]) → AgentTarget(steps[i+1]) の連鎖と TerminateTarget("sequence_complete") のデフォルトを使用します。こちらは到達可能です。最後のステップが発話した後は、どの FromSpeaker ルールにも一致しなくなるため、デフォルトが発火して "sequence_complete" になります。

条件付きハンドオフ（最も一般的な手動グラフ）

トリアージエージェントが各リクエストを調べ、ツール呼び出しを通じて専門エージェントにルーティングします。

from ag2.network import (
    AgentTarget, FromSpeaker, RevertToInitiatorTarget,
    TerminateTarget, ToolCalled, Transition, TransitionGraph,
)

graph = TransitionGraph(
    initial_speaker=triage.agent_id,
    transitions=[
        # Triage's `escalate` tool fired → route to security.
        Transition(when=ToolCalled("escalate"),
                   then=AgentTarget(security.agent_id)),
        # Once security has spoken, hand back to whoever opened the channel.
        Transition(when=FromSpeaker(security.agent_id),
                   then=RevertToInitiatorTarget()),
        # Otherwise triage's reply goes to the generic responder.
        Transition(when=FromSpeaker(triage.agent_id),
                   then=AgentTarget(general.agent_id)),
    ],
    default_target=TerminateTarget(reason="triage_complete"),
    max_turns=20,
)

ToolCalled(name) → AgentTarget(agent) の各遷移に対して、話者の Agent に @tool でデコレートされた関数を追加します。最もシンプルな形式では、型付きの Handoff を返します。フレームワークはツールの結果からそれを読み取り、ルーティングします。

from ag2.network import Handoff


# `.tool` lives on `Agent`. The `AgentClient` returned by `hc.register(...)`
# is the network handle (used for `.open()`, `.send()`, `.agent_id`) and does
# NOT expose `.tool`. So: build the Agent, attach tools, *then* register.
triage_agent = Agent("triage", prompt="…", config=config)


@triage_agent.tool
async def escalate(reason: str = "") -> Handoff:
    """Escalate this ticket to the security reviewer."""
    # Handoff.target is the participant's Passport NAME, not a raw agent_id —
    # the framework resolves name → id at packet-finalize via the hub's name directory.
    return Handoff(target="security", reason=reason)


triage = await triage_hc.register(
    triage_agent, Passport(name="triage"), Resume(), attach_plugin=False,
)

両方が設定されている場合、型付きの Handoff 戻り値は、一致する ToolCalled ルールより優先されます。これは、ターゲット が実行時の状態に依存する場合に便利です。グラフの ToolCalled ルールは、ドキュメントとして、またフォールバックとして引き続き有用です。

コンテキスト変数 — 読み書きループ

チャネルスコープの可変状態は WorkflowState.context_vars: dict[str, Any] に格納され、1つのワークフローチャネルにスコープされます。また、WAL 上の EV_CONTEXT_SET エンベロープとして永続化されます。

コンテキストの書き込み（ツールから）

from ag2.network import ChannelInject, EV_CONTEXT_SET
from ag2.network.workflow_helpers import set_context


async def set_route(route: str, channel: ChannelInject) -> str:
    """Record the routing decision for this channel."""
    if channel is None:
        return "no active channel"
    await set_context(channel, "route", route)
    return f"route set to {route!r}"

set_context(channel, key, value) は、デフォルトのブロードキャスト対象（audience=None — すべての参加者から見えます）で EV_CONTEXT_SET を発行する薄いラッパーです。変更はフォールド後に WorkflowState.context_vars に反映されます。同等の生の形式は次のとおりです。

await channel.send(
    "",
    event_type=EV_CONTEXT_SET,
    event_data={"set": {"route": route}},
)

完全な event_data の形式は、設定と削除の両方をサポートしています。{"set": {...}, "delete": [...]} のいずれかは省略可能です。1つのエンベロープ内では、delete が set より先に実行されます。

コンテキストの読み取り（遷移内）

Transition(
    when=ContextEquals(key="route", value="security"),
    then=AgentTarget(security.agent_id),
)

ContextEquals(key, value=None) は、キーが未設定または明示的に削除されたときに発火します。変更は、実質的なターンのチェックより前にフォールドされるため、同じフォールド内の ContextEquals ルールは、設定されたばかりの値を参照します。

コンテキストの読み取り（ツール内）

from ag2.network import ChannelStateInject


async def increment_counter(channel: ChannelInject, state: ChannelStateInject) -> str:
    if state is None or channel is None:
        return "no active channel"
    current = state.context_vars.get("counter", 0)
    await set_context(channel, "counter", current + 1)
    return f"counter now {current + 1}"

初期値

channel = await alice.open(
    type="workflow",
    target=[bob.agent_id, carol.agent_id],
    knobs={
        "graph": graph.to_dict(),
        "context_vars": {"escalation_level": 0, "ticket_id": ticket_id},
    },
)

ルーティングが停止する落とし穴

ContextEquals は持続的です。キーが設定されると、その後のフォールドのたびに再評価されます。ルールリストの上位付近に ContextEquals("route", "security") → AgentTarget(security) がある場合、セキュリティに永遠に戻ることになります（または max_turns に達するまで）。対処方法は2つあります。

1. 終了ルールをコンテキスト条件より先に並べます。 FromSpeaker(security) → TerminateTarget(...) を先に置くと、security の発話後にループが短絡されます。
2. ルーティング先のエージェントにキーをクリアさせます。 security エージェントのツールは、完了時に {"delete": ["route"]} を指定した EV_CONTEXT_SET を発行します。

より一般的なパターンは修正方法 #1 です。

フィードバックループ — コンテキスト駆動の終了

graph = TransitionGraph(
    initial_speaker=intake.agent_id,
    transitions=[
        # Approve sets done=True → terminate.
        Transition(when=ContextEquals("done", value=True),
                   then=TerminateTarget("approved")),
        # intake → drafter.
        Transition(when=FromSpeaker(intake.agent_id),
                   then=AgentTarget(drafter.agent_id)),
        # Alternate drafter ↔ reviewer.
        Transition(when=FromSpeaker(drafter.agent_id),
                   then=AgentTarget(reviewer.agent_id)),
        Transition(when=FromSpeaker(reviewer.agent_id),
                   then=AgentTarget(drafter.agent_id)),
    ],
    default_target=TerminateTarget("max_iterations"),
    max_turns=10,
)


# `reviewer_agent` is the `Agent`; `reviewer` is the `AgentClient` returned
# by `register()` and used in the graph above (for `reviewer.agent_id`).
# Tools must be attached to the Agent before registration.
@reviewer_agent.tool
async def approve(reason: str, channel: ChannelInject) -> str:
    """Mark the draft approved; the graph terminates with reason='approved'."""
    if channel is None:
        return "no channel"
    await set_context(channel, "done", True)
    return f"approved: {reason}"

レビュアーの approve 呼び出しによって done=True が書き込まれ、レビュアーの返信テキストが同じパケットのフォールドに格納されます。ContextEquals("done", True) が一致すると、チャネルは "approved" で終了します。approve が呼び出されない場合、turn_count が max_turns に達するまで、ドラフターとレビュアーの交互の処理が続き、チャネルは理由 "max_turns" で終了します。（上記の default_target=TerminateTarget("max_iterations") は、いずれのターンでも一致する遷移が生成されなかった場合にのみ実行されます。ここでは FromSpeaker ルールのいずれかが常に一致するため、このグラフでは実質的に使われません。到達する可能性のあるグラフでは、適切なフォールバックとして残しておいてください。）

8 つのクックブックパターン

AG2 のドキュメントには、実行可能な例を含むパターンクックブックがあります。1 行でまとめると、それぞれ特定のグラフを持つ workflow チャネルです。

パターン	グラフ	用途
パイプライン	TransitionGraph.sequence([a, b, c])	A→B→C→終了の直線的な処理（リサーチ → 執筆 → 編集）。
階層型	Coordinator + FromSpeaker(specialist) → RevertToInitiatorTarget()	コーディネーターが割り当て、スペシャリストがコーディネーターに戻ります。
スター型	ハブが N 個のスポークに問い合わせ、ToolCalled("synthesise") → TerminateTarget	複数ソースの集約、並列データ収集。
エスカレーション	階層化された Handoff + ToolCalled("resolve") → TerminateTarget("resolved")	サポートの階層、複数レベルの承認。
冗長型	N 人の提案者 + 評価者による sequence	ニッチな用途。比較を伴う新規性の追求やアイデア創出。
フィードバックループ	ContextEquals("done", True) → TerminateTarget + max_turns	承認されるまでドラフター ↔ レビュアーで処理を続けます。
コンテキスト対応ルーティング	トリアージが context_vars["category"] を設定し、ContextEquals(category, X) → AgentTarget(specialist_X)	サポートのトリアージ、スキルに基づく割り当て。
タスク付きトリアージ	TransitionGraph.sequence + 初期値の knobs["context_vars"]	トリアージで計画を作成し、シーケンスで計画を実行します。

パイプライン、エスカレーション、コンテキスト対応ルーティング、フィードバックループを組み合わせれば、実際のニーズの大部分に対応できます。クックブックのページには、それぞれの実行可能なエンドツーエンドのコードがあります。

カスタムターゲットと条件

組み込みのものが適合しない場合は、Protocol を実装し、クラスを登録すると、TransitionGraph.to_dict() で往復変換できるようになります。

from typing import ClassVar
from dataclasses import dataclass, field

from ag2.network import (
    Envelope,
    TransitionDecision,
    TransitionTarget,
    register_target,
)


@dataclass(slots=True)
class HighestRankedReviewer(TransitionTarget):
    name: ClassVar[str] = "highest_ranked_reviewer"
    role_priority: list[str] = field(default_factory=list)

    def resolve(self, state, envelope: Envelope) -> TransitionDecision:
        # ... look up the next reviewer based on domain logic ...
        return TransitionDecision(next_speaker=chosen_id)


register_target(HighestRankedReviewer)

条件も同じパターンです。evaluate(state, envelope) -> bool を実装し、ClassVar[str] name を設定して、register_condition(MyCondition) を呼び出します。どちらのカスタマイズも to_dict() で保持されます。name フィールドがキーとなり、データクラスのフィールドが loads(...) で再構成される args dict になります。

パケット実行モデルと冪等なツール

ワークフローチャネルでの各 Agent.ask ラウンドは、単一の EV_PACKET エンベロープとして WAL にアトミックにコミットされます。パケットには、エージェントのルーティング判断（ToolCalled ルールと照合される routing.tool、または型付き Handoff から事前に解決された routing.target）、ラウンドの本文テキスト、予約済みの context_updates スロットが含まれます。ツール呼び出しによる EV_CONTEXT_SET エンベロープはパケットよりも前に格納されるため、同じフォールド上の ContextEquals ルールは設定されたばかりの値を確認できます。

アトミック性には、次の影響があります。エージェントがパケットの途中でクラッシュすると、チャネルはパケット前の状態に戻り、入力が再ディスパッチされます。再試行時には、そのパケット内のツール呼び出しも再実行されます。

外部システムに作用するツールは、再試行時にも冪等でなければなりません。

• 利用できる場合は、外部サービスの冪等性キー機能（Stripe、S3）を使用します。(channel_id, round_counter, tool_name) から安定したキーを導出してください。
• データベースへの書き込みには、無条件の insert ではなく upsert（INSERT ... ON CONFLICT）を使用します。
• 本質的に冪等にできないツールは、HITL の背後に置くか、パケット境界をより細かくできるように、単一ツールのラウンドで実行します。

classic GroupChat からの移行

移行の大部分は機械的に行えます。2つのシステムでは、話者、条件、ターゲット、終了という用語が共通していますが、新しいシステムはデータファーストです（TransitionGraph は JSON にシリアライズ可能で、Hub.hydrate() 後も維持されます）。

概念の対応

AG2 Classic	AG2 network	備考
GroupChat(agents=[...])	参加者を含む workflow チャネル	ハブが GroupChatManager の役割を担います。
GroupChatManager	WorkflowAdapter + Hub	ターン交代の処理はハブに移ります。
Agent.handoffs	TransitionGraph（エージェント単位ではなく、チャネル単位）	ハンドオフはチャネルレベルで一度だけ記述します。
AgentTarget(agent)	AgentTarget(agent_id)	参照ではなく ID を受け取ります。
RevertToInitiator()	RevertToInitiatorTarget()	セマンティクスは同じです。
Stay()	StayTarget()	同じです。
Terminate()	TerminateTarget(reason="…")	理由を含めるようになりました。
OnContextCondition(...)	ContextEquals またはカスタムの TransitionCondition	register_condition で登録します。
OnCondition(...)	カスタムの TransitionCondition	同じです。
ツールからの ReplyResult(target=...)	型指定された Handoff の戻り値	Handoff(target=..., reason=...)。
FunctionTarget(fn)	カスタムの TransitionTarget	resolve() を実装し、register_target(...) で登録します。
max_round=N	グラフの max_turns=N	同じハード上限です。

比較例：ラウンドロビン

# Classic
groupchat = GroupChat(agents=[alice, bob, carol],
                     speaker_selection_method="round_robin", max_round=6)
manager = GroupChatManager(groupchat=groupchat, llm_config=llm_config)
alice.initiate_chat(manager, message="Topic: …")

# AG2 network
graph = TransitionGraph.round_robin(
    participants=[alice.agent_id, bob.agent_id, carol.agent_id],
    max_turns=6,
)
channel = await alice.open(
    type="workflow",
    target=[bob.agent_id, carol.agent_id],
    knobs={"graph": graph.to_dict()},
)
await channel.send("Topic: …")

比較例：条件付きハンドオフ（ReplyResult → Handoff）

# Classic
@triage.register_for_llm(description="Escalate to security review.")
def escalate(reason: str) -> ReplyResult:
    return ReplyResult(target=AgentTarget(security_reviewer), message=...)

# AG2 network — attach to the Agent *before* registering with the hub
@triage_agent.tool
async def escalate(reason: str = "") -> Handoff:
    # Handoff.target is the Passport NAME (resolved name → id by the framework),
    # whereas AgentTarget in the graph takes the raw agent_id.
    return Handoff(target="security_reviewer", reason=reason)

# Plus the matching transition in the graph (as documentation/fallback):
Transition(when=ToolCalled("escalate"), then=AgentTarget(security_reviewer.agent_id))

移行チェックリスト

1. ハブを起動します（Hub.open(MemoryKnowledgeStore())）。
2. 既存の各 Agent を、Passport と Resume を指定して hc.register(...) でラップします。
3. Handoffs の設定を TransitionGraph に変換します。適用できる場合は sequence / round_robin ファクトリーを使用し、条件付きロジックの場合は手動で構築します。
4. initiate_chat(...) を、alice.open(type="workflow", target=[...], knobs={"graph": graph.to_dict()}) の後に channel.send(text) を呼び出す形に置き換えます。
5. wait_for_channel_event(...) を使用して EV_CHANNEL_CLOSED を待ちます。
6. hub.read_wal(channel_id) と hub._audit_log.read_all() で確認します。

State オブジェクト

@dataclass(slots=True)
class WorkflowState:
    participant_order: list[str]
    expected_next_speaker: str | None
    last_speaker_id: str | None = None
    last_envelope_id: str | None = None
    turn_count: int = 0
    pending_close_reason: str = ""
    creator_id: str = ""
    graph_data: dict = field(default_factory=dict)
    context_vars: dict[str, Any] = field(default_factory=dict)

expected_next_speaker = None は「チャネルを終了する必要がある」ことを示します。アダプターの on_accepted がこれを読み取り、AdapterResult(next_state=CLOSING, ...) を返します。graph_data はシリアライズされたグラフです。アダプターはフォールドのたびに TransitionGraph を再構築するため、メモリー内に可変のグラフ状態はありません。

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

from ag2.network import (
    # Graph and transitions
    TransitionGraph,
    Transition,
    TransitionTarget,
    TransitionCondition,
    TransitionDecision,
    register_target,
    register_condition,
    # Built-in targets
    AgentTarget,
    RoundRobinTarget,
    StayTarget,
    RevertToInitiatorTarget,
    TerminateTarget,
    # Built-in conditions
    Always,
    FromSpeaker,
    ToolCalled,
    ContextEquals,
    # Typed handoff return for tools
    Handoff,
    # Context variables
    EV_CONTEXT_SET,
    ChannelInject,
    ChannelStateInject,
    # Channel types
    WORKFLOW_TYPE,
    # Seeding a workflow (kickoff participant)
    HumanClient,  # via HubClient.register_human(Passport(name=..., kind="human"))
)
from ag2.network.workflow_helpers import set_context, delete_context