ag2-network-workflow

# AG2 Network — Workflow Adapter `workflow` is the orchestrated multi-party adapter. A declarative `TransitionGraph` describes who speaks first, what conditions fire on each turn, and when the channel terminates. It's the modern replacement for the classic `GroupChat + Agent.handoffs` pattern — turn-taking lives in the hub, not in an in-process speaker selector. > Prerequisite: read `ag2-network-quickstart` first. This skill assumes you know `Hub.open`, `HubClient.register`, the channel lifecycle, and the basic `agent_client.open(...)` / `channel.send(...)` flow. ## When to use Load this skill when the user asks for any of: - "Two/three/N agents with conditional handoffs" - "Pipeline / sequence of agents (researcher → writer → editor)" - "Triage agent routes to specialists" - "Drafter / reviewer feedback loop" - "Coordinator + specialists / hierarchical orchestration" - "Migrate from `GroupChat` / `ConversableAgent.handoffs` / `ReplyResult(target=...)`" - "I want context-driven routing (`OnContextCondition`, `StringContextCondition`)" **Don't use `workflow` for:** - Strict 1Q1R between two agents → `consulting` (see `ag2-network-quickstart`). - Free-form 2-party chat with no turn order → `conversation` (see `ag2-network-quickstart`). - Fixed round-robin with no conditions → `discussion` (see `ag2-network-discussion`). If the user is unsure, the rule of thumb: **need any condition more complex than "next speaker in a fixed list"? Use `workflow`.** ## Shape | | | |---|---| | Participants | 2+ | | Turn order | Whatever `TransitionGraph` says | | Auto-close | Yes — when graph emits `TerminateTarget` or `max_turns` is hit | | Default view | `NamedWindowedSummary(recent_n=N*2)` | | Default expectations | `turn_within(120s, warn)`, `turn_within(600s, auto_close)` | | Required knob | `{"graph": <TransitionGraph.to_dict()>}` | The expectation is strict: a stuck speaker auto-closes the channel after 10 minutes. Tune via `ag2-network-governance`. ## Workflow agents and the `NetworkPlugin` `HubClient.register(...)` attaches `NetworkPlugin` by default. The plugin adds only the *identity-level cross-cutting* tools — `peers` / `channels` / `tasks` / `context` / `delegate`. Channel-specific tools (notably `say`, plus your workflow handoff `@tool`s) come from `adapter.tools_for(client, metadata, state, participant_id)`, resolved per turn by the default handler and merged into `agent.ask(tools=...)` — and **the `workflow` adapter offers none**: `WorkflowAdapter.tools_for(...)` returns `[]`. So a workflow agent never sees `say`; routing is entirely your `@tool`-decorated handoff functions (returning `Handoff`, mutating context vars, or calling `channel.close()`). Consequence: the old "mid-turn `say` races the round-end `EV_PACKET`" hazard **can't happen** on a workflow channel — you do **not** need `attach_plugin=False` for correctness. Choose by capability instead: - **Keep the plugin** (default) if the agent should also `delegate` to a peer, look up `peers` / `channels` / `tasks`, or read/write channel `context` from outside the graph. (`delegate` / `channels(action="open")` spin up *separate* channels — they never emit on the workflow channel, so they don't disturb its turn machinery.) - **`attach_plugin=False`** for a genuinely bare agent — only the domain `@tool`s you wrote. Tidy for pure pipeline workers and most tests. ```python # Bare pipeline worker — no cross-cutting network tools: worker = await wc.register(agent, Passport(name="worker"), Resume(), attach_plugin=False) ``` The default handler wraps the LLM's plain-text reply into the workflow `EV_PACKET` either way. The examples below pass `attach_plugin=False` because they're pure pipeline workers; drop it if your agents need `delegate` / `peers` / etc. Full detail: `ag2-network-tools-and-views` → "Adapter-owned tools (`tools_for`)". ## Smallest example — pipeline via `sequence` ```python 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])` builds three `FromSpeaker(x) → AgentTarget(next)` transitions plus a `TerminateTarget("sequence_complete")` default. After `c` speaks no transition matches, the default fires, and the channel auto-closes. ## `TransitionGraph` anatomy ```python @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 ``` On every accepted substantive envelope (text or packet), the adapter walks the `transitions` list. The **first** matching condition's target resolves the next speaker. If none match, `default_target` is consulted. A `TerminateTarget` resolves with `next_speaker=None`, which makes the adapter return `AdapterResult(next_state=CLOSING, ...)` and the hub posts `EV_CHANNEL_CLOSED` with that target's `reason`. `max_turns` is a separate, harder cap: once `turn_count` reaches it the adapter closes the channel immediately with reason **`"max_turns"`** — it does *not* fall through to `default_target` (so a `default_target=TerminateTarget("…")` reason only appears when an actual turn produced no matching transition). **`priority` sorts ascending — a *lower* number is checked first.** The adapter does `sorted(transitions, key=lambda t: t.priority)` then walks the result, so `priority=0` (the default) beats `priority=100`. If you want a rule consulted before the others, either put it at the top of the list (the sort is stable, so equal priorities keep list order) or give it a *negative* priority. A common mistake: writing `priority=100` on a `ContextEquals(...) → TerminateTarget(...)` rule expecting it to win — it gets pushed to the *end* and a `FromSpeaker(...)` fallback fires first instead, so the channel never terminates. ## Gotcha — the kickoff `channel.send()` is the `initial_speaker`'s first turn When you do `agent.open(type="workflow", ...)` then `await channel.send(text)`, that first send is folded as `initial_speaker`'s turn — it's an envelope from that agent, and `expected_next_speaker` starts equal to `initial_speaker`, so `validate_send` accepts it and the graph immediately routes onward. **The `initial_speaker`'s `Agent.ask` never runs for that turn** — the text you sent *is* its contribution. So if your design wants the *first agent that should actually think* (the drafter, the researcher, the planner) to respond to the kickoff prompt, that agent must **not** be `initial_speaker` and must **not** be the one calling `channel.send()`. The clean answer is a **`HumanClient`** as the seeder — a non-LLM participant (no `Agent`, no plugin) whose only job is to open the channel and post the brief. Make it `initial_speaker` and route `FromSpeaker(user) → AgentTarget(drafter)`: ```python 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)` returns a `HumanClient` (`HubClient.register(...)` now *rejects* `kind="human"` and points you here). It auto-acks invites so the quorum handshake completes without UI round-trips; pass `auto_ack_invites=False` to gate joins by hand. Drive it with `on_envelope(callback)` (push) or `next_envelope(...)` / `envelopes()` (pull) — see `ag2-network-quickstart` → "`HumanClient`". If you'd rather not introduce a `HumanClient`, the older workaround still works: a throwaway "intake" `Agent` as `initial_speaker` whose `Agent.ask` is never invoked (no transition routes back to it) — but a `HumanClient` is cheaper (no model config, no plugin) and is what the framework's own smoke tests use. Either way, the same applies to `TransitionGraph.sequence([a, b, c])`: `channel.send(...)` is `a`'s turn, so if `a` is supposed to *produce* the first artifact (not just relay a prompt), prepend a seeder — `sequence([user, a, b, c])`. (Symptom when you get this wrong: the second agent in the chain receives the bare kickoff prompt as if it were the first agent's output, "responds" to something that isn't there, and the pipeline produces nonsense — often hallucinating that an artifact exists.) ## Built-in targets | Target | Decision | |---|---| | `AgentTarget(agent_id)` | Hand off to a specific named agent. | | `RoundRobinTarget()` | Advance through the participant order. | | `StayTarget()` | Same speaker continues (rare; "let me elaborate" patterns). | | `RevertToInitiatorTarget()` | Hand back to whoever opened the channel. | | `TerminateTarget(reason="…")` | End the channel; reason flows on `EV_CHANNEL_CLOSED.event_data["reason"]`. | ## Built-in conditions | Condition | Fires when | |---|---| | `Always()` | Every accepted envelope. | | `FromSpeaker(agent_id)` | The just-accepted envelope was sent by this agent. | | `ToolCalled(tool_name)` | The previous turn called this tool by name (matched via `routing.tool` on the packet). | | `ContextEquals(key, value)` | Channel-scoped `context_vars[key]` equals `value`. Missing keys compare as `None`. | For combinations beyond what these compose to, see "Custom targets / conditions" below. ## Convenience factories ```python # 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` uses `Always() → RoundRobinTarget()` and a `TerminateTarget("round_robin_complete")` default — but since `Always()` matches every turn, the default is unreachable and the channel actually closes with reason `"max_turns"` once the cap is hit (set `max_turns` or it cycles forever). `sequence` uses `FromSpeaker(steps[i]) → AgentTarget(steps[i+1])` chains with a `TerminateTarget("sequence_complete")` default — *that* one is reachable: after the last step speaks, no `FromSpeaker` rule matches, so the default fires and you get `"sequence_complete"`. ## Conditional handoff (the most common manual graph) A triage agent inspects each request and routes to a specialist via a tool call: ```python 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.