Skip to content

Scatter-gather with Send

Three inbound tickets and three triage lenses (sentiment, routing, resolution) give nine analyst agents running in parallel, then one synthesizer collapses everything into a single Markdown triage summary.

The pattern is scatter-gather, also called map-reduce: a splitter node fans the work out as a list of Send directives, and a reducer node reads the results back from the merged graph state.

What this notebook covers:

  • Send(node, payload, metadata) is a first-class graph primitive: the splitter returns a list of Sends and the executor fans them out concurrently, without queues or a hand-written asyncio.gather.
  • Each analyst is a distinct Agent with a lens-specific system prompt. The graph orchestrates them, not a hand-written loop.
  • The synthesizer reads each Send's output back from the merged state and renders the final Markdown summary.
  • The whole pipeline is one StateGraph.execute call. The same graph streams through StateGraph.stream and checkpoints through GraphConfig with no change to the nodes, and because each analyst is an ordinary Agent, agent-level options such as GSAR typed grounding apply to each analyst as to any other agent.

Prerequisites

  • Basic graph.
  • Swarm, where agents claim tasks from a shared queue instead of each receiving one Send.

Inside the one StateGraph.execute call, the splitter returns a Send per ticket and lens. The executor runs the nine analysts concurrently and merges each successful result into the graph state under its Send's id:

Scatter-gather with Send Inside one StateGraph.execute call, the ticket splitter returns a list of Send directives, one per ticket and lens, each naming the analyst node and carrying its payload. The executor runs all of those analysts concurrently; the lenses are sentiment, routing and resolution, each run by a distinct Agent for every ticket. Every successful result is merged into the graph state under its Send's id. When all the Sends have finished, control moves to the synthesizer, the splitter's declared successor, which reads the merged state and renders the summary. One execute call In parallel a Send per ticket × lens Ticket splitter returns list[Send] Sentiment analyst one Agent per ticket Routing analyst one Agent per ticket Resolution analyst one Agent per ticket Merged state state[send_id] Synthesizer renders the summary No edge is declared from the splitter to the analysts: each Send names its target node and carries its payload. When every Send has finished, control moves to the synthesizer, the splitter's declared successor, which reads the merged state.

Key lines

The splitter returns one Send per ticket and lens, each naming the analyze_one node and carrying the ticket and role as its payload. Each analyst receives the graph state with that payload merged on top, so analyze_one reads the ticket and role from its Send and the shared model from __model__:

async def split_tickets(state: dict[str, Any]) -> list[Send]:
    """Fan out: one Send per (ticket, lens) — 3 × 3 = 9 analysts, all concurrent."""
    tickets: dict[str, str] = state["tickets"]
    roles = list(ANALYST_ROLES)
    return [
        Send(
            node="analyze_one",
            payload={"ticket_id": ticket_id, "message": message, "role": role},
            metadata={"ticket": ticket_id, "role": role},
        )
        for ticket_id, message in tickets.items()
        for role in roles
    ]

The graph declares no edge from split to analyze_one: each Send carries its own routing, and the edge from split to synthesize is where control goes once every Send has finished:

def build_triage_graph(model: Any) -> StateGraph:
    """Wire the three nodes: split → analyze_one (parallel) → synthesize → END.

    The model is threaded through state under ``__model__`` rather than
    captured by closure so the nodes stay plain module-level functions.
    """
    graph = StateGraph(name="support-triage-crew")
    graph.add_node("split", split_tickets)
    graph.add_node("analyze_one", analyze_one)
    graph.add_node("synthesize", synthesize)

    graph.add_edge(START, "split")
    # No explicit edge "split → analyze_one" — the Sends from ``split``
    # carry their own routing. Once every Send finishes, control returns
    # to ``split``'s adjacency, which points at ``synthesize``.
    graph.add_edge("split", "synthesize")
    graph.add_edge("synthesize", END)
    return graph

The synthesizer collects every analysis payload from the merged state, where each Send's result sits under its own id, groups them by ticket and renders the Markdown summary:

async def synthesize(state: dict[str, Any]) -> dict[str, Any]:
    """Reduce: walk the merged state, collect every ``analysis`` payload, render."""
    analyses = [v["analysis"] for v in state.values() if isinstance(v, dict) and "analysis" in v]
    by_ticket: dict[str, list[dict[str, Any]]] = {}
    for a in analyses:
        by_ticket.setdefault(a["ticket"], []).append(a)

    lines = ["# Customer-support triage — summary report", ""]
    for ticket_id in sorted(by_ticket):
        lines.append(f"## {ticket_id}")
        for a in sorted(by_ticket[ticket_id], key=lambda x: x["role"]):
            lines.append(f"### {a['role']}")
            lines.append(a["comments"])
            lines.append("")
    return {"report": "\n".join(lines), "analysis_count": len(analyses)}

Run it

python examples/notebook_20_map_reduce_support_triage.py

Model choice comes from examples/config.py: OCI Generative AI when ~/.oci/config has a usable profile, otherwise the bundled mock model, so the example also runs offline with no credentials. Set LOCUS_MODEL_PROVIDER (oci, openai, anthropic or mock) to force one.

See also

Source

Full source — examples/notebook_20_map_reduce_support_triage.py 
# Copyright (c) 2025, 2026 Oracle and/or its affiliates.
# Licensed under the Universal Permissive License v1.0 as shown at
# https://oss.oracle.com/licenses/upl/

"""Notebook 20: customer-support triage at scale — scatter-gather with Send.

Three inbound tickets, three triage lenses (sentiment, routing, resolution)
= nine analyst agents running in parallel, then one synthesizer collapses
everything into a single triage summary. Each lens emits a structured tag —
SENTIMENT, QUEUE, PRIORITY — so the output lands in vocabulary a helpdesk
or routing pipeline already speaks.

Key ideas:
- ``Send(node, payload, metadata)`` is a first-class graph primitive.
  The splitter node returns a list of Sends; the executor fans them out
  and runs them concurrently. No queues, no manual ``asyncio.gather``.
- Each analyst is a distinct ``Agent`` with its own lens-specific
  system prompt — the graph orchestrates them, not a hand-rolled loop.
- The synthesizer reads each Send's output back from merged state and
  emits a single Markdown triage summary.
- The whole pipeline is one ``StateGraph.execute`` call. The same graph
  streams through ``StateGraph.stream`` and checkpoints through
  ``GraphConfig`` with no change to the nodes; each analyst is an ordinary
  ``Agent``, so agent-level options such as GSAR typed grounding apply to
  it unchanged.

Run it:
    .venv/bin/python examples/notebook_20_map_reduce_support_triage.py

The default provider is OCI GenAI when ``~/.oci/config`` holds a usable
profile, otherwise the bundled mock model; set ``LOCUS_MODEL_PROVIDER``
(``oci``, ``openai``, ``anthropic``, ``mock``) to choose. Set
``LOCUS_MODEL_PROVIDER=mock`` for offline runs.

Prerequisites:
- Notebook 12 (basic graph).
- Notebook 24 (Swarm) for the dynamic-claim counterpoint.
"""

from __future__ import annotations

import asyncio
from typing import Any

from config import get_model

from locus.agent import Agent, AgentConfig
from locus.core.send import Send
from locus.multiagent.graph import END, START, StateGraph


# ----------------------------------------------------------------------------
# Three independent tickets to triage in parallel. Each message carries a
# concrete, quotable issue so the analysts have something specific to cite.
#   T-4821 — duplicate billing charge, refund request (billing)
#   T-4822 — login failure / app hang, time-sensitive (technical)
#   T-4823 — cancellation / churn risk, open to a downgrade (account)
# ----------------------------------------------------------------------------

SAMPLE_TICKETS = {
    "T-4821": (
        "I was charged twice for my June subscription — $29.99 on the 3rd and "
        "again on the 5th. I only have one account. Please refund the duplicate "
        "charge today; this is the second time this has happened."
    ),
    "T-4822": (
        "I've been trying to log in for the past hour and the app just spins on a "
        "blank screen. I'm on the latest iOS build. I have a client demo in 30 "
        "minutes and I'm locked out — terrible timing."
    ),
    "T-4823": (
        "I'd like to cancel my plan. The product is fine but I'm not using it "
        "enough to justify the cost, and the renewal hit before I could downgrade. "
        "If there's a cheaper tier I might stay; otherwise please close the account."
    ),
}


# ----------------------------------------------------------------------------
# Triage lenses — each runs as its own Agent with a lens-specific prompt
# ----------------------------------------------------------------------------

ANALYST_ROLES = {
    "sentiment": (
        "You are a customer-sentiment analyst. Read the support ticket and "
        "gauge the customer's emotional state and frustration level. Quote the "
        "specific phrases that signal how the customer feels. Do not invent "
        "emotions the text does not show. End with a single "
        "line: SENTIMENT=<calm|frustrated|angry|at-risk>."
    ),
    "routing": (
        "You are a support-routing analyst. Classify the ticket into the right "
        "queue (billing, technical, account, or shipping) and name the single "
        "most likely root cause you can support from the message. Quote the line "
        "that decides the category. Do not invent details. End "
        "with: QUEUE=<billing|technical|account|shipping>."
    ),
    "resolution": (
        "You are a resolution analyst. Propose the single best next action to "
        "resolve the ticket — a concrete step the agent can take. Quote the part "
        "of the ticket that drives your recommendation. Be terse. End "
        "with: PRIORITY=<low|medium|high|urgent>."
    ),
}


def _make_analyst(role: str, model: Any) -> Agent:
    return Agent(
        config=AgentConfig(
            agent_id=f"analyst-{role}",
            model=model,
            system_prompt=ANALYST_ROLES[role],
            # One model call is enough to triage a short ticket.
            max_iterations=2,
            max_tokens=400,
        )
    )


# ----------------------------------------------------------------------------
# Graph nodes
# ----------------------------------------------------------------------------


async def split_tickets(state: dict[str, Any]) -> list[Send]:
    """Fan out: one Send per (ticket, lens) — 3 × 3 = 9 analysts, all concurrent."""
    tickets: dict[str, str] = state["tickets"]
    roles = list(ANALYST_ROLES)
    return [
        Send(
            node="analyze_one",
            payload={"ticket_id": ticket_id, "message": message, "role": role},
            metadata={"ticket": ticket_id, "role": role},
        )
        for ticket_id, message in tickets.items()
        for role in roles
    ]


async def analyze_one(state: dict[str, Any]) -> dict[str, Any]:
    """One analyst Agent against one (ticket, lens).

    ``arun`` is a coroutine, so the nine instances overlap inside the
    graph's ``asyncio.gather`` instead of waiting on each other.
    """
    role: str = state["role"]
    ticket_id: str = state["ticket_id"]
    message: str = state["message"]
    model = state["__model__"]
    agent = _make_analyst(role, model)
    prompt = (
        f"Ticket: {ticket_id}\n"
        f"Lens: {role}\n\n"
        f'"""\n{message}\n"""\n\n'
        f"Analyze the ticket through the {role} lens."
    )
    result = await agent.arun(prompt)
    return {
        "analysis": {
            "ticket": ticket_id,
            "role": role,
            "comments": (result.message or "").strip(),
            "iterations": result.metrics.iterations,
        }
    }


async def synthesize(state: dict[str, Any]) -> dict[str, Any]:
    """Reduce: walk the merged state, collect every ``analysis`` payload, render."""
    analyses = [v["analysis"] for v in state.values() if isinstance(v, dict) and "analysis" in v]
    by_ticket: dict[str, list[dict[str, Any]]] = {}
    for a in analyses:
        by_ticket.setdefault(a["ticket"], []).append(a)

    lines = ["# Customer-support triage — summary report", ""]
    for ticket_id in sorted(by_ticket):
        lines.append(f"## {ticket_id}")
        for a in sorted(by_ticket[ticket_id], key=lambda x: x["role"]):
            lines.append(f"### {a['role']}")
            lines.append(a["comments"])
            lines.append("")
    return {"report": "\n".join(lines), "analysis_count": len(analyses)}


# ----------------------------------------------------------------------------
# Build the graph
# ----------------------------------------------------------------------------


def build_triage_graph(model: Any) -> StateGraph:
    """Wire the three nodes: split → analyze_one (parallel) → synthesize → END.

    The model is threaded through state under ``__model__`` rather than
    captured by closure so the nodes stay plain module-level functions.
    """
    graph = StateGraph(name="support-triage-crew")
    graph.add_node("split", split_tickets)
    graph.add_node("analyze_one", analyze_one)
    graph.add_node("synthesize", synthesize)

    graph.add_edge(START, "split")
    # No explicit edge "split → analyze_one" — the Sends from ``split``
    # carry their own routing. Once every Send finishes, control returns
    # to ``split``'s adjacency, which points at ``synthesize``.
    graph.add_edge("split", "synthesize")
    graph.add_edge("synthesize", END)
    return graph


# ----------------------------------------------------------------------------
# Driver
# ----------------------------------------------------------------------------


async def main() -> None:
    print("Notebook 20: customer-support triage at scale — scatter-gather with Send")
    print("=" * 60)

    model = get_model()
    graph = build_triage_graph(model)

    initial = {"tickets": SAMPLE_TICKETS, "__model__": model}

    print(
        f"\nFanning out {len(SAMPLE_TICKETS)} tickets × {len(ANALYST_ROLES)} lenses "
        f"= {len(SAMPLE_TICKETS) * len(ANALYST_ROLES)} analyst agents in parallel...\n"
    )

    result = await graph.execute(initial)

    print(
        f"Graph completed in {result.duration_ms:.0f} ms across "
        f"{result.iterations} graph iteration(s)"
    )
    print(f"Analyses collected: {result.final_state.get('analysis_count', 0)}")
    print()
    print(result.final_state.get("report", "(no report)"))


if __name__ == "__main__":
    asyncio.run(main())