dragon.ai.agent.orchestrator.orchestrator.DAGOrchestrator

class DAGOrchestrator

Bases: object

Build and execute multi-agent DAG workflows on Dragon Batch.

The orchestrator is the request-side entry point. It assumes all agents are already running persistently (started externally via their listen() loops).

DAGOrchestrator owns the lifecycle of the DDict. The caller creates a Batch instance and passes it to run(); the orchestrator uses it to build and execute the DAG. The caller is responsible for calling batch.join() when it is done with that client handle; if the Batch runtime was created with managed_lifecycle=True, some client must later call batch.destroy() to shut down the Batch instance shared by those clients.

Construction sets up all shared infrastructure (DDict, HITL bridge, trace bridge) so that hitl_address, trace_address, and serialized_ddict are available immediately — before run() is called.

Parameters

config:

OrchestratorConfig with agent declarations, poll interval, timeout, and optional ddict_kwargs for DDict tuning.

pipeline:

A Pipeline defining the DAG topology — nodes and their dependencies.

Example

orchestrator = DAGOrchestrator(
    config=OrchestratorConfig(
        agents=[...],
        poll_interval=0.5,
        poll_timeout=120.0,
        ddict_kwargs={"managers_per_node": 2, "total_mem": 1 << 30},
    ),
    pipeline=pipeline,
)

if orchestrator.hitl_address:
    host, port = orchestrator.hitl_address
    print(f"HITL client:  python -m dragon.ai.agent.hitl --tcp {host}:{port}")
if orchestrator.trace_address:
    host, port = orchestrator.trace_address
    print(f"Trace viewer: python -m dragon.ai.agent.observability --tcp {host}:{port} -i")

# External clients can attach to the shared DDict for debugging:
print(f"DDict descriptor: {orchestrator.serialized_ddict}")

batch = Batch()
try:
    result = orchestrator.run("Write a report on quantum computing.", batch=batch)
finally:
    orchestrator.destroy()
    batch.join()

__init__(config: OrchestratorConfig, pipeline: Pipeline) → None

Methods

__init__(config, pipeline)
destroy()	Tear down all orchestrator-owned resources.
run(user_input, batch)	Execute the pipeline and return the final result.

Attributes

ddict	Return the shared DDict instance for this run.
hitl_address	Return (host, port) of the HITL TCP bridge, or None.
serialized_ddict	Return the serialized DDict descriptor for this run.
trace_address	Return (host, port) of the Trace TCP bridge, or None.

__init__(config: OrchestratorConfig, pipeline: Pipeline) → None

property hitl_address: tuple [str , int ] | None

Return (host, port) of the HITL TCP bridge, or None.

Available immediately after construction. Returns None if no agent in the pipeline uses an approval_filter.

property trace_address: tuple [str , int ] | None

Return (host, port) of the Trace TCP bridge, or None.

Available immediately after construction. Returns None when tracing is disabled (OrchestratorConfig.tracing=False).

property ddict: DDict

Return the shared DDict instance for this run.

property serialized_ddict: str

Return the serialized DDict descriptor for this run.

External clients can use this to attach to the shared working memory for debugging or monitoring:

from dragon.data.ddict import DDict
ddict = DDict.attach(orchestrator.serialized_ddict)

run(user_input: str , batch: Batch) → Any

Execute the pipeline and return the final result.

Parameters

user_input:

The end-user’s request string.

batch:

Caller-owned Batch instance used to build and execute the DAG. The caller is responsible for batch.join() after this method returns. If the runtime uses managed_lifecycle=True, some client must later call batch.destroy() to stop the Batch instance shared by those clients.

Returns

Any

If the pipeline has a single terminal node, returns that node’s result dict directly. If there are multiple terminal nodes, returns dict[agent_id, result].

destroy() → None

Tear down all orchestrator-owned resources.

Stops the HITL and trace TCP bridges, destroys the HITL queue and the shared DDict. Batch teardown remains the caller’s responsibility: callers should batch.join() their client handles, and only managed-lifecycle runtimes need a later batch.destroy().

Safe to call multiple times — each resource is cleaned up at most once.