pycsamt.agents.coordinator#

pycsamt.agents.coordinator#

AgentCoordinator orchestrates multi-agent workflows:

  • Registers agents by name.

  • Executes ordered workflow steps with dependency tracking.

  • Checkpoints each step to disk so workflows can be resumed after failure.

  • Provides a dry_run / preview mode that estimates cost and prints the execution plan without running any agents.

  • Aggregates per-step costs into a workflow-level total.

Classes

AgentCoordinator([workflow_name, ...])

Orchestrate a sequence of pycsamt agents as a named workflow.

WorkflowStep(name, agent, *[, input_fn, ...])

One step in an AgentCoordinator workflow.
class pycsamt.agents.coordinator.AgentCoordinator(workflow_name='pycsamt_workflow', *, checkpoint_dir=None, verbose=True)[source]#

Bases: object

Orchestrate a sequence of pycsamt agents as a named workflow.

Parameters:

  • workflow_name (str) – Name used for log messages and checkpoint directory.

  • checkpoint_dir (str or Path or None) – Where to save step checkpoints. Defaults to ./pycsamt_agent_checkpoints/<workflow_name>/.

  • verbose (bool) – Print step-level progress to stdout.

Examples

Build and run a QC workflow:

from pycsamt.agents import AgentCoordinator, MTLoaderAgent, DataQCAgent

coord = AgentCoordinator("mt_qc")
coord.add_step("load",  MTLoaderAgent(...))
coord.add_step("qc",    DataQCAgent(...),
               input_fn=lambda r: {"sites": r["load"]["sites"]})

result = coord.execute({"path": "/data/EDIs"})
print(result.summary)

Dry-run preview:

result = coord.preview({"path": "/data/EDIs"})
print(result["plan"])
add_step(name, agent, *, input_fn=None, description='', required=True)[source]#

Append a workflow step. Returns self for chaining.

Parameters:

  • name (str)

  • agent (BaseAgent)

  • input_fn (callable(prev_results_dict) → dict, optional) – Maps accumulated results to this step’s input dict. prev_results_dict keys are earlier step names; each value is the step’s AgentResult.

  • description (str)

  • required (bool)

Returns:

self for method chaining.

Return type:

AgentCoordinator

preview(config)[source]#

Return an execution plan and estimated cost without running anything.

Parameters:

config (dict) – The workflow configuration that would be passed to execute().

Returns:

data["plan"] contains the formatted plan string. data["steps"] is a list of step metadata dicts.

Return type:

AgentResult

execute(config, *, dry_run=False, resume=False)[source]#

Run all workflow steps sequentially.

Parameters:

  • config (dict) – Top-level workflow configuration forwarded to every step’s input_fn (or directly to agent.execute() when no input_fn is set).

  • dry_run (bool) – When True return a preview without executing any agents.

  • resume (bool) – When True skip steps whose checkpoint already exists on disk.

Returns:

data contains one key per step name with its AgentResult.

Return type:

AgentResult

reset_checkpoints()[source]#

Delete all saved checkpoints for this workflow.

Return type:

None

class pycsamt.agents.coordinator.WorkflowStep(name, agent, *, input_fn=None, description='', required=True)[source]#

Bases: object

One step in an AgentCoordinator workflow.

Parameters:

  • name (str) – Unique identifier used for checkpointing and logging.

  • agent (BaseAgent) – The agent instance that runs this step.

  • input_fn (callable or None) – input_fn(prev_results) → dict fed to agent.execute(). When None the coordinator passes the raw workflow config.

  • description (str) – Human-readable description shown in the dry-run preview.

  • required (bool) – When False a failure skips the step rather than aborting.