01 — Quickstart: LangGraph Agents as Dragon Processes

The first example in the Dragon + LangGraph series. It runs an ordinary LangGraph graph on the Dragon runtime: each agent becomes a Dragon AgentHost process via executor.launch_host(...) and is wrapped with executor.node(name). The graph definition is unchanged — only execution moves onto Dragon.

What you’ll learn:

  • How to create a DragonExecutor and launch an AgentHost

  • How to wrap a LangGraph node with executor.node(name)

  • How agent execution moves onto the Dragon runtime with no change to the graph

Needs: 1 node.

Main Code

Listing 34 01_quickstart.py
  1"""01 - Quickstart: run LangGraph agents as Dragon processes.
  2
  3The new Dragon ability
  4----------------------
  5``launch_host(...)`` spawns one persistent Dragon **AgentHost process**, and
  6each agent you register runs inside it — a sync (``def``) node on the host's
  7thread pool, an async (``async def``) node on the host's event loop.  What
  8changes is *where* your agents run: instead of the coordinator's own Python
  9process, they run in a separate Dragon process on the runtime.  You wrap each
 10agent with ``executor.node(name)`` to dispatch to that host; the graph
 11definition itself is unchanged.  Because each ``launch_host`` call is its own
 12process, you can place hosts on different cluster nodes by calling it once per
 13node (see 02).
 14
 15Run
 16---
 17    dragon examples/dragon_ai/langgraph/01_quickstart.py
 18"""
 19
 20from __future__ import annotations
 21
 22import dragon
 23import multiprocessing as mp
 24import operator
 25from typing import Annotated, TypedDict
 26
 27from langgraph.graph import END, START, StateGraph
 28
 29from dragon.ai.langgraph import DragonExecutor
 30from dragon.native.process import current as current_process
 31
 32
 33class State(TypedDict):
 34    messages: Annotated[list[str], operator.add]
 35    research: str
 36    report: str
 37
 38
 39# =============================================================================
 40# 2. Node functions - ordinary agent bodies (swap in real LLM/tool calls)
 41# =============================================================================
 42
 43def researcher(state: State) -> dict:
 44    """Simulate a research agent. Replace with real LLM calls."""
 45    query = state["messages"][-1] if state["messages"] else "unknown"
 46    findings = f"Research findings on '{query}': key insight A, B, C."
 47    print(f"  [researcher] puid={current_process().puid} query='{query}'")
 48    return {"messages": [findings], "research": findings}
 49
 50
 51def writer(state: State) -> dict:
 52    """Simulate a writing agent that produces a report from research."""
 53    research = state.get("research", "")
 54    report = f"Report: Based on research — {research}"
 55    print(f"  [writer] puid={current_process().puid} drafting report")
 56    return {"messages": [report], "report": report}
 57
 58
 59# =============================================================================
 60# 3. Build and run
 61# =============================================================================
 62
 63def main() -> None:
 64    print(f"[orchestrator] main process puid={current_process().puid}")
 65    # DragonExecutor manages the AgentHost processes. A few optional knobs
 66    # control concurrency and memory — shown here at their DEFAULTS, so this
 67    # runs identically whether or not you pass them:
 68    #   max_concurrent_tasks (default 64) — max agent tasks in flight across the
 69    #       WHOLE system at once (global backpressure). Raise for more
 70    #       parallelism/pipelining; lower to use less memory. Everything internal
 71    #       (channel capacity, host queue sizes) is derived from this one number.
 72    #   num_shards (default 1) — parallel completion lanes on the coordinator.
 73    #       Raise ONLY if a single watcher thread can't keep up with a very high
 74    #       completion rate of small results.
 75    # You normally set only max_concurrent_tasks (or nothing). See the
 76    # "Sizing cheat-sheet" in doc/devguide/langgraph.rst and the README.
 77    with DragonExecutor(
 78        max_concurrent_tasks=64,
 79        num_shards=1,
 80    ) as executor:
 81        # launch_host spawns ONE AgentHost process; both agents run inside it.
 82        # max_threads bounds how many *sync* (def) node bodies run concurrently
 83        # inside this host (they run on a thread pool). Async (async def) nodes
 84        # run on the host's shared event loop and are NOT bounded by this.
 85        # Default: max(len(agents) * 4, 8). The agents print a different puid —
 86        # they live in the host process, not in this orchestrator process.
 87        executor.launch_host(
 88            agents={
 89                "researcher": researcher,
 90                "writer": writer,
 91            },
 92            max_threads=8,
 93        )
 94
 95        # Build the graph - executor.node(...) is the only Dragon-specific call.
 96        builder = StateGraph(State)
 97        builder.add_node("researcher", executor.node("researcher"))
 98        builder.add_node("writer", executor.node("writer"))
 99        builder.add_edge(START, "researcher")
100        builder.add_edge("researcher", "writer")
101        builder.add_edge("writer", END)
102
103        graph = builder.compile()
104
105        print("\n=== Running graph ===")
106        result = graph.invoke(
107            {"messages": ["quantum computing advances 2025"], "research": "", "report": ""},
108        )
109        print(f"\n=== Final report ===\n{result['report']}")
110
111
112if __name__ == "__main__":
113    mp.set_start_method("dragon", force=True)
114    main()

See Also