dragon.ai.langgraph.DragonExecutor

class DragonExecutor[source]

Bases: object

Manages the lifecycle of Dragon AgentHost processes.

Owns a single DragonWatcher that resolves all pending task futures over one or more shared completion Queues (one recv thread per shard). All agent nodes dispatched through this executor share the same watcher — no per-node thread.

Use as a context manager:

with DragonExecutor() as executor:
    executor.launch_host(agents={...})
    ...

Or manage lifetime manually:

executor = DragonExecutor()
executor.launch_host(agents={...})
...
executor.shutdown()

__init__(*, max_concurrent_tasks: int = 64, num_shards: int = 1) → None[source]

Create the executor and its single shared-transport watcher.

This executor transports only agent results (inline, on Dragon Queues). It does not create or manage any DDict. Bulk scientific data (tensors, arrays, datasets) is the user’s responsibility: create your own Dragon DDict, pass it into your agent functions, store artifacts there, and return only a lightweight handle/key through LangGraph state.

Parameters:

max_concurrent_tasks (int, optional) – Cap on agent tasks in flight across the whole executor. Dispatching beyond it applies backpressure — graph.invoke blocks its calling thread, graph.ainvoke awaits the slot. Per-shard slots, completion-queue capacity and host input-queue size are all derived from this, so it is normally the only one you set. Distinct from a host’s max_threads, which caps sync node bodies inside one host; async nodes are bounded only by this cap. Defaults to 64.
num_shards (int, optional) – Independent completion lanes, each a Dragon Queue plus a watcher thread, sharing the max_concurrent_tasks budget evenly. Raise only if one watcher thread becomes the bottleneck. Defaults to 1.

Raises:

ValueError – If max_concurrent_tasks or num_shards is less than 1.

Methods

`__init__`(*[, max_concurrent_tasks, num_shards])	Create the executor and its single shared-transport watcher.
`launch_host`(agents, *[, policy, ...])	Spawn one AgentHost process containing multiple agent functions.
`launch_hosts`(hosts, *[, policies, ...])	Spawn multiple AgentHost processes, one per entry in hosts.
`node`(node_name)	Return the callable for node_name to pass to StateGraph.add_node.
`shutdown`(*[, graceful])	Stop all AgentHost processes and the watcher thread.

Attributes

`max_concurrent_tasks`	Maximum number of agent tasks allowed in flight at once.
`num_shards`	Number of independent completion lanes in the watcher.

__init__(*, max_concurrent_tasks: int = 64, num_shards: int = 1) → None[source]

Create the executor and its single shared-transport watcher.

This executor transports only agent results (inline, on Dragon Queues). It does not create or manage any DDict. Bulk scientific data (tensors, arrays, datasets) is the user’s responsibility: create your own Dragon DDict, pass it into your agent functions, store artifacts there, and return only a lightweight handle/key through LangGraph state.

Parameters:

max_concurrent_tasks (int, optional) – Cap on agent tasks in flight across the whole executor. Dispatching beyond it applies backpressure — graph.invoke blocks its calling thread, graph.ainvoke awaits the slot. Per-shard slots, completion-queue capacity and host input-queue size are all derived from this, so it is normally the only one you set. Distinct from a host’s max_threads, which caps sync node bodies inside one host; async nodes are bounded only by this cap. Defaults to 64.
num_shards (int, optional) – Independent completion lanes, each a Dragon Queue plus a watcher thread, sharing the max_concurrent_tasks budget evenly. Raise only if one watcher thread becomes the bottleneck. Defaults to 1.

Raises:

ValueError – If max_concurrent_tasks or num_shards is less than 1.

property max_concurrent_tasks: int: Maximum number of agent tasks allowed in flight at once.

property num_shards: int: Number of independent completion lanes in the watcher.

launch_host(agents: dict[str, Callable[[...], Any]], *, policy: Any | None = None, event_timeout: float | None = None, max_threads: int | None = None) → Any[source]

Spawn one AgentHost process containing multiple agent functions.

Parameters:

agents (dict[str, Callable]) – Mapping of node_name to fn(state) -> dict. All functions in this dict run in a single Dragon process. Both plain def and async def node functions are supported — sync nodes run on a bounded thread pool; async nodes run as concurrent tasks on the host’s shared event loop, so an existing LangGraph graph with async nodes runs unchanged.
policy (dragon.infrastructure.policy.Policy, optional) – Optional dragon.infrastructure.policy.Policy to pin this host to a specific cluster node, NUMA zone, or GPU. Defaults to None.
event_timeout (float, optional) – Seconds before a task is considered timed out. None means no limit. Defaults to None.
max_threads (int, optional) – Maximum concurrent sync node bodies in this host — the size of the thread pool that runs plain def nodes. None defaults to max(len(agents) * 4, 8). Async (async def) nodes are not bounded by this; they run as tasks on the host’s shared event loop (bounded only by max_concurrent_tasks). See __init__() (the max_concurrent_tasks note) for how the two relate. Defaults to None.

Returns:

The Dragon Process running this AgentHost. Useful for monitoring, joining, or killing the host externally.

Return type:

dragon.native.process.Process

Raises:

ValueError – If an agent node_name was already registered by a previous launch_host() call.
RuntimeError – If called after shutdown(), or if the host process exits before completing its startup handshake.

node(node_name: str) → Any[source]

Return the callable for node_name to pass to StateGraph.add_node.

The returned RunnableCallable carries both a sync and an async implementation. The choice of graph.invoke vs graph.ainvoke only changes how the coordinator waits — the node runs on the host the same way either way (sync def on the thread pool, async def on the host loop) and returns the identical result:

graph.invoke → sync path (a BackgroundExecutor thread parks on result()): up to one parked thread per in-flight node. Simplest for a synchronous driver with a chain or modest fan-out.
graph.ainvoke → async path (awaits an asyncio future): N concurrent agents cost 0 extra threads — only the event loop and the watcher thread(s). Prefer it as you scale the number of concurrent agents.

Parameters:: node_name (str) – The name of a registered agent node.
Returns:: A LangGraph RunnableCallable wrapping the agent’s sync and async entry points.
Return type:: langgraph.utils.runnable.RunnableCallable
Raises:: KeyError – If node_name was not included in any launch_host() call.

shutdown(*, graceful: bool = True) → None[source]

Stop all AgentHost processes and the watcher thread.

Parameters:: graceful (bool, optional) – If True, send shutdown sentinels and wait for hosts to exit. If False, kill immediately. Defaults to True.

launch_hosts(hosts: list[dict[str, Callable[[...], Any]]], *, policies: list[Any] | None = None, event_timeout: float | None = None, max_threads: int | None = None) → list[Any][source]

Spawn multiple AgentHost processes, one per entry in hosts.

This is a convenience wrapper around launch_host() that mirrors the Dragon convention of passing a list of policies (one per item) to cover multi-node placement in a single call.

Parameters:

hosts (list[dict[str, Callable]]) – List of agent dicts. Each dict maps node_name to fn(state) -> dict and becomes one AgentHost process. Node functions may be plain def or async def.
policies (list[dragon.infrastructure.policy.Policy], optional) – Optional list of dragon.infrastructure.policy.Policy objects, one per host. Must be the same length as hosts when provided. None means all hosts use default (unspecified) placement. Defaults to None.
event_timeout (float, optional) – Seconds before a task is considered timed out. None means no limit. Applied to every host. Defaults to None.
max_threads (int, optional) – Maximum concurrent sync node bodies inside each host (thread-pool size for def nodes). None uses the per-host default max(len(agents) * 4, 8). Async nodes run on the host’s shared event loop and are not bounded by this (see the __init__ note). Defaults to None.

Returns:

The Dragon Processes running each AgentHost, in the same order as hosts.

Return type:

list[dragon.native.process.Process]

Raises:

ValueError – If policies is provided but its length differs from hosts.

Example usage:

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