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 import (
 26    Policy,
 27    System
 28)
 29
 30from dragon.ai.langgraph import DragonExecutor
 31from dragon.native.machine import Node
 32from dragon.native.process import current as current_process
 33
 34
 35# =============================================================================
 36# 1. Graph state
 37# =============================================================================
 38
 39class State(TypedDict):
 40    messages: Annotated[list[str], operator.add]
 41    research: str
 42    analysis: str
 43    report: str
 44
 45
 46# =============================================================================
 47# 2. Node functions
 48# =============================================================================
 49
 50def researcher(state: State) -> dict:
 51    """Research agent — runs on Node 0."""
 52    me = current_process()
 53    query = state["messages"][-1] if state["messages"] else "unknown"
 54    findings = f"Deep research on '{query}': discovered patterns X, Y, Z."
 55    print(f"  [researcher] puid={me.puid} node={me.node} findings ready")
 56    return {"messages": [findings], "research": findings}
 57
 58
 59def analyzer(state: State) -> dict:
 60    """Analysis agent — runs on Node 1, co-located with GPU resources."""
 61    me = current_process()
 62    research = state.get("research", "")
 63    analysis = f"Statistical analysis of: {research} → significance p<0.01"
 64    print(f"  [analyzer] puid={me.puid} node={me.node} analysis complete")
 65    return {"messages": [analysis], "analysis": analysis}
 66
 67
 68def writer(state: State) -> dict:
 69    """Writer agent — runs on Node 1 alongside analyzer (shared resources)."""
 70    me = current_process()
 71    research = state.get("research", "")
 72    analysis = state.get("analysis", "")
 73    report = f"Final Report:\n  Research: {research}\n  Analysis: {analysis}"
 74    print(f"  [writer] puid={me.puid} node={me.node} report drafted")
 75    return {"messages": [report], "report": report}
 76
 77
 78# =============================================================================
 79# 3. Build and run with multi-node placement
 80# =============================================================================
 81
 82def main() -> None:
 83    me = current_process()
 84    print(f"[orchestrator] main process puid={me.puid} node={me.node}")
 85
 86    # Discover cluster topology
 87    system = System()
 88    node_list = system.nodes
 89    assert len(node_list) >= 2, "This example requires at least 2 cluster nodes"
 90
 91    node0_hostname = Node(node_list[0]).hostname
 92    node1_hostname = Node(node_list[1]).hostname
 93
 94    print(f"Cluster: node0={node0_hostname}, node1={node1_hostname}")
 95
 96    # Policies for placement
 97    policy_node0 = Policy(
 98        placement=Policy.Placement.HOST_NAME,
 99        host_name=node0_hostname,
100    )
101    policy_node1 = Policy(
102        placement=Policy.Placement.HOST_NAME,
103        host_name=node1_hostname,
104    )
105
106    with DragonExecutor() as executor:
107        # Node 0: researcher agent
108        host0 = executor.launch_host(
109            agents={"researcher": researcher},
110            policy=policy_node0,
111        )
112        # Node 1: analyzer + writer (share one host process on the same node)
113        host1 = executor.launch_host(
114            agents={"analyzer": analyzer, "writer": writer},
115            policy=policy_node1,
116        )
117
118        # Alternatively, launch_hosts() does the same thing in one call:
119        #
120        #   hosts = executor.launch_hosts(
121        #       hosts=[
122        #           {"researcher": researcher},
123        #           {"analyzer": analyzer, "writer": writer},
124        #       ],
125        #       policies=[policy_node0, policy_node1],
126        #   )
127        #   host0, host1 = hosts
128
129        print(f"  host0 puid={host0.puid}, host1 puid={host1.puid}")
130
131        # Build graph
132        builder = StateGraph(State)
133        builder.add_node("researcher", executor.node("researcher"))
134        builder.add_node("analyzer", executor.node("analyzer"))
135        builder.add_node("writer", executor.node("writer"))
136
137        builder.add_edge(START, "researcher")
138        builder.add_edge("researcher", "analyzer")
139        builder.add_edge("analyzer", "writer")
140        builder.add_edge("writer", END)
141
142        graph = builder.compile()
143
144        # Run
145        print("\n=== Running multi-node graph ===")
146        result = graph.invoke(
147            {"messages": ["climate modeling at scale"], "research": "", "analysis": "", "report": ""},
148        )
149        print(f"\n=== Result ===\n{result['report']}")
150
151
152if __name__ == "__main__":
153    mp.set_start_method("dragon", force=True)
154    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