ag2-network-workflow

AG2 Network — 工作流适配器

workflow 是用于编排多方协作的适配器。声明式 TransitionGraph 描述谁先发言、每轮触发哪些条件，以及何时终止通道。它是经典 GroupChat + Agent.handoffs 模式的现代替代方案——轮流发言由 Hub 管理，而不是由进程内的发言者选择器管理。

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

使用场景

用户提出以下任一需求时，加载此技能：

• “带条件交接的两个／三个／N 个 Agent”
• “Agent 管线／序列（研究员 → 撰稿人 → 编辑）”
• “分诊 Agent 将任务路由给专家”
• “起草者／审核者反馈循环”
• “协调者 + 专家／分层编排”
• “从 GroupChat / ConversableAgent.handoffs / ReplyResult(target=...) 迁移”
• “我想要基于上下文的路由（OnContextCondition、StringContextCondition）”

以下情况不要使用 workflow：

• 两个 Agent 之间严格的一问一答 → consulting（参阅 ag2-network-quickstart）。
• 没有轮次顺序的自由形式双人聊天 → 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 调整。

工作流 Agent 与 NetworkPlugin

默认情况下，HubClient.register(...) 会附加 NetworkPlugin。该插件仅添加身份级的跨切面工具——peers / channels / tasks / context / delegate。特定通道的工具（尤其是 say，以及你自己的工作流交接 @tool）由 adapter.tools_for(client, metadata, state, participant_id) 提供，并由默认处理程序按轮次解析，再合并到 agent.ask(tools=...) 中——而且 workflow 适配器不提供任何此类工具：WorkflowAdapter.tools_for(...) 返回 []。因此，工作流 Agent 看不到 say；路由完全由你使用 @tool 装饰的交接函数负责（这些函数可以返回 Handoff、修改上下文变量，或调用 channel.close()）。

因此，旧有的“轮次中途的 say 与轮次结束时的 EV_PACKET 发生竞争”的隐患不会在工作流通道上出现——为保证正确性，无需设置 attach_plugin=False。应根据能力需求选择：

• 如果 Agent 还需要 delegate 给其他 Agent、查询 peers / channels / tasks，或在图外读写通道 context，则保留插件（默认设置）。（delegate / channels(action="open") 会启动单独的通道——它们不会在工作流通道上发出任何内容，因此不会干扰其轮次机制。）
• 对于确实不需要其他功能的 Agent，设置 attach_plugin=False——它只能使用你编写的领域专用 @tool。适合纯管线工作者和大多数测试。

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

无论采用哪种设置，默认处理程序都会将大模型的纯文本回复封装到工作流 EV_PACKET 中。下方示例设置了 attach_plugin=False，因为它们是纯管线工作者；如果你的 Agent 需要 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) 转换，以及一个默认的 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, ...)，而 hub 会使用该目标的 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 的一轮——这是来自该 Agent 的信封，而 expected_next_speaker 起始时等于 initial_speaker，因此 validate_send 会接受它，并且图会立即路由到下一位。initial_speaker 的 Agent.ask 在这一轮中绝不会运行——你发送的文本就是它的贡献。

因此，如果你的设计希望第一个真正需要思考的 Agent（起草者、研究员、规划者）响应启动提示词，那么该 Agent 不能是 initial_speaker，也不能是调用 channel.send() 的参与者。简洁的做法是使用 HumanClient 作为初始发送者——一个非大模型参与者（没有 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，旧的变通办法仍然可用：将一个一次性的“接收” Agent 设为 initial_speaker，其 Agent.ask 不会被调用（没有转换会路由回它）——不过 HumanClient 开销更小（无需模型配置，也无需插件），也是框架自身冒烟测试所用的方式。无论哪种方式，TransitionGraph.sequence([a, b, c]) 也同样适用：channel.send(...) 是 a 的一轮，因此，如果 a 应该生成第一个产物（而不只是转发提示词），就要在前面加上一个初始发送者——sequence([user, a, b, c])。

（如果设置有误，可能会出现这样的症状：链中的第二个 Agent 收到原封不动的启动提示词，却以为这是第一个 Agent 的输出，于是对并不存在的内容作出“响应”，导致流水线产出一堆无意义结果——通常还会幻想某个产物已经存在。）

内置目标

目标	决策
AgentTarget(agent_id)	移交给指定名称的 Agent。
RoundRobinTarget()	按参与者顺序轮流推进。
StayTarget()	由同一发言者继续（较少见；用于“让我进一步说明”模式）。
RevertToInitiatorTarget()	交还给开启此通道的参与者。
TerminateTarget(reason="…")	结束通道；原因会传递到 EV_CHANNEL_CLOSED.event_data["reason"]。

内置条件

条件	触发时机
Always()	每个已接受的信封。
FromSpeaker(agent_id)	刚刚接受的信封由此 Agent 发送。
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"。

条件移交（最常见的手动图）

一个分诊 Agent 会检查每个请求，并通过工具调用将其路由给专门处理的 Agent：

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] 中，作用范围为单个工作流通道，并作为 EV_CONTEXT_SET 信封持久化到 WAL。

写入上下文（从工具中）

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": [...]}。两者均为可选；在同一个信封中，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)，就会一直跳回 security（或直到达到 max_turns）。有两种解决办法：

1. 将终止规则放在上下文条件之前： 将 FromSpeaker(security) → TerminateTarget(...) 放在前面，security 发言后即可跳出循环。
2. 让被路由到的 Agent 清除键： security Agent 的工具在完成时发出 EV_CONTEXT_SET，并携带 {"delete": ["route"]}。

更常见的做法是解决办法 #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 规则总有一条匹配，因此在此图中它实际上不会生效；对于可能会触发它的图，可以将它保留为合理的后备方案。）

八种实用模式

AG2 文档提供了一个模式手册，其中包含可运行的示例。每种模式用一行概括——每种都是带有特定图的 workflow 通道：

模式	图	适用场景
流水线	TransitionGraph.sequence([a, b, c])	线性执行 A→B→C→终止（研究 → 撰写 → 编辑）。
分层	协调者 + 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(...) 重新构建的参数字典。

数据包执行模型和幂等工具

在工作流通道中，每一轮 Agent.ask 都会作为单个 EV_PACKET 信封原子性地提交到 WAL。数据包包含 Agent 的路由决策（与 ToolCalled 规则匹配的 routing.tool，或由类型化 Handoff 预先解析的 routing.target）、本轮正文文本，以及预留的 context_updates 槽位。工具调用产生的 EV_CONTEXT_SET 信封会在数据包之前写入，因此同一折叠上的 ContextEquals 规则可以看到刚设置的值。

原子性会带来一个后果：如果 Agent 在数据包处理中途崩溃，通道会恢复到数据包处理前的状态，并重新分派输入。重试时，该数据包中的工具调用会再次执行。

涉及外部系统的工具必须支持重试时幂等：

• 在可用时，使用外部服务的幂等键功能（Stripe、S3）。根据 (channel_id, round_counter, tool_name) 生成稳定的键。
• 对于数据库写入，使用 upsert（INSERT ... ON CONFLICT），而不是直接插入。
• 对于确实无法做到幂等的工具，将它们置于 HITL 之后，或让它们在单工具轮次中运行，以缩小数据包边界。

从经典 GroupChat 迁移

迁移过程大多是机械性的。两个系统使用相同的术语——发言者、条件、目标、终止——但新系统以数据为先（TransitionGraph 可序列化为 JSON，并能在 Hub.hydrate() 后保留）。

概念映射

AG2 Classic	AG2 network	说明
GroupChat(agents=[...])	包含参与者的 workflow 通道	Hub 承担 GroupChatManager 的职责。
GroupChatManager	WorkflowAdapter + Hub	轮流发言的逻辑移至 Hub。
Agent.handoffs	TransitionGraph（按通道配置，而非按 Agent 配置）	在通道级别统一描述交接。
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（Hub.open(MemoryKnowledgeStore())）。
2. 使用 Passport 和 Resume，通过 hc.register(...) 封装每个现有的 Agent。
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() 检查。

状态对象

@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