02 — Multinode Placement

This example pins each AgentHost to a chosen cluster node. A single launch_host call runs its agents as threads in one process; to spread agents across nodes you issue one ``launch_host`` per node, each with its own Policy.

What you’ll learn:

  • How a Policy places an AgentHost on a specific node

  • Why distributing agents means one launch_host per node

  • How to confirm placement from the runtime (node name + host PID)

  • How launch_hosts places several hosts in one call

Needs: 2+ nodes.

Main Code

Listing 35 02_multinode_placement.py
  1"""02 - Multi-node placement: pin agents to specific cluster nodes.
  2
  3The new Dragon ability
  4----------------------
  5A ``Policy`` places each AgentHost on a chosen node.  Call ``launch_host``
  6once per node, each with its own ``Policy``, and your agents run on different
  7machines in the allocation - co-located with the data or GPUs they need.
  8Plain LangGraph has no concept of node placement; this is a Dragon runtime
  9capability surfaced through the integration.
 10
 11Run (requires 2+ nodes)
 12---
 13    dragon examples/dragon_ai/langgraph/02_multinode_placement.py
 14"""
 15
 16from __future__ import annotations
 17
 18import dragon
 19import multiprocessing as mp
 20import operator
 21from typing import Annotated, TypedDict
 22
 23from langgraph.graph import END, START, StateGraph
 24
 25from dragon.ai.langgraph import DragonExecutor
 26from dragon.infrastructure.policy import Policy
 27from dragon.native.machine import Node, System
 28from dragon.native.process import current as current_process
 29
 30
 31# =============================================================================
 32# 1. Graph state
 33# =============================================================================
 34
 35class State(TypedDict):
 36    messages: Annotated[list[str], operator.add]
 37    research: str
 38    analysis: str
 39    report: str
 40
 41
 42# =============================================================================
 43# 2. Node functions
 44# =============================================================================
 45
 46def researcher(state: State) -> dict:
 47    """Research agent — runs on Node 0."""
 48    me = current_process()
 49    query = state["messages"][-1] if state["messages"] else "unknown"
 50    findings = f"Deep research on '{query}': discovered patterns X, Y, Z."
 51    print(f"  [researcher] puid={me.puid} node={me.node} findings ready")
 52    return {"messages": [findings], "research": findings}
 53
 54
 55def analyzer(state: State) -> dict:
 56    """Analysis agent — runs on Node 1, co-located with GPU resources."""
 57    me = current_process()
 58    research = state.get("research", "")
 59    analysis = f"Statistical analysis of: {research} → significance p<0.01"
 60    print(f"  [analyzer] puid={me.puid} node={me.node} analysis complete")
 61    return {"messages": [analysis], "analysis": analysis}
 62
 63
 64def writer(state: State) -> dict:
 65    """Writer agent — runs on Node 1 alongside analyzer (shared resources)."""
 66    me = current_process()
 67    research = state.get("research", "")
 68    analysis = state.get("analysis", "")
 69    report = f"Final Report:\n  Research: {research}\n  Analysis: {analysis}"
 70    print(f"  [writer] puid={me.puid} node={me.node} report drafted")
 71    return {"messages": [report], "report": report}
 72
 73
 74# =============================================================================
 75# 3. Build and run with multi-node placement
 76# =============================================================================
 77
 78def main() -> None:
 79    me = current_process()
 80    print(f"[orchestrator] main process puid={me.puid} node={me.node}")
 81
 82    # Discover cluster topology
 83    system = System()
 84    node_list = system.nodes
 85    assert len(node_list) >= 2, "This example requires at least 2 cluster nodes"
 86
 87    node0_hostname = Node(node_list[0]).hostname
 88    node1_hostname = Node(node_list[1]).hostname
 89
 90    print(f"Cluster: node0={node0_hostname}, node1={node1_hostname}")
 91
 92    # Policies for placement
 93    policy_node0 = Policy(
 94        placement=Policy.Placement.HOST_NAME,
 95        host_name=node0_hostname,
 96    )
 97    policy_node1 = Policy(
 98        placement=Policy.Placement.HOST_NAME,
 99        host_name=node1_hostname,
100    )
101
102    with DragonExecutor() as executor:
103        # Node 0: researcher agent
104        host0 = executor.launch_host(
105            agents={"researcher": researcher},
106            policy=policy_node0,
107        )
108        # Node 1: analyzer + writer (share one host process on the same node)
109        host1 = executor.launch_host(
110            agents={"analyzer": analyzer, "writer": writer},
111            policy=policy_node1,
112        )
113
114        # Alternatively, launch_hosts() does the same thing in one call:
115        #
116        #   hosts = executor.launch_hosts(
117        #       hosts=[
118        #           {"researcher": researcher},
119        #           {"analyzer": analyzer, "writer": writer},
120        #       ],
121        #       policies=[policy_node0, policy_node1],
122        #   )
123        #   host0, host1 = hosts
124
125        print(f"  host0 puid={host0.puid}, host1 puid={host1.puid}")
126
127        # Build graph
128        builder = StateGraph(State)
129        builder.add_node("researcher", executor.node("researcher"))
130        builder.add_node("analyzer", executor.node("analyzer"))
131        builder.add_node("writer", executor.node("writer"))
132
133        builder.add_edge(START, "researcher")
134        builder.add_edge("researcher", "analyzer")
135        builder.add_edge("analyzer", "writer")
136        builder.add_edge("writer", END)
137
138        graph = builder.compile()
139
140        # Run
141        print("\n=== Running multi-node graph ===")
142        result = graph.invoke(
143            {"messages": ["climate modeling at scale"], "research": "", "analysis": "", "report": ""},
144        )
145        print(f"\n=== Result ===\n{result['report']}")
146
147
148if __name__ == "__main__":
149    mp.set_start_method("dragon", force=True)
150    main()

Placing several hosts in one call

When you have several hosts to place, launch_hosts is a convenience wrapper that mirrors the Dragon convention of passing a list of policies — one per host, applied in order:

hosts = executor.launch_hosts(
    hosts=[
        {"researcher": researcher},
        {"analyzer": analyzer, "writer": writer},
    ],
    policies=[policy_node0, policy_node1],
)

This is equivalent to calling launch_host once per entry. policies must be the same length as hosts (or None to use default placement), and the returned list of processes is in the same order as hosts.

See Also