pycsamt.agents.orchestrator#

pycsamt.agents.orchestrator#

WorkflowOrchestratorAgent — Intelligent workflow dispatcher.

Given a natural-language MT processing request the agent:

  1. Uses the LLM (or a rule-based fallback) to classify the workflow type.

  2. Builds an AgentCoordinator with the appropriate agent chain.

  3. Executes (or previews) the full workflow, returning a consolidated result.

Supported workflow types#

"qc"

QC-only pipeline: load → QC → static-shift → report

"phase_analysis"

Tensor analysis: load → QC → static-shift → phase analysis → report

"pre_inversion"

Pre-inversion: load → QC → static-shift → phase analysis → Occam2D prep → codegen

"ai_inversion"

AI 1-D end-to-end: load → QC → denoising → AI inversion → interpretation → report

"inv3d"

GCN 3-D spatial: load → QC → static-shift → Inv3DAgent → interpretation → report

"inv2d"

U-Net 2-D: load → QC → denoising → Inv2DAgent → interpretation → report

"ensemble_inversion"

Ensemble uncertainty: load → QC → denoising → EnsembleAgent → interpretation → report

"joint_inversion"

Multi-modal DRCNN: load → QC → static-shift → JointInversionAgent → interpretation → report

"modem"

ModEM prep: load → QC → static-shift → ModEM file → report

"full"

Everything: load → QC → static-shift → phase analysis → denoising → AI inversion → Occam2D prep → report

"denoise"

Denoising only: load → QC → denoising → report

"static_shift"

Static-shift only: load → QC → static-shift → report

"inversion_eval"

Evaluate result: load → QC → static-shift → eval → report

"code_gen"

Code generation: load → QC → code generation → report

"report"

Report only: load → QC → report

"forward"

Forward modelling: load → QC → forward model → report

"interpretation"

Interpretation: load → QC → static-shift → geological interpretation → report

Classes

WorkflowOrchestratorAgent(*[, api_key, ...])

Intelligently route an NL request to the correct agent chain.
class pycsamt.agents.orchestrator.WorkflowOrchestratorAgent(*, api_key=None, model=None, llm_provider='claude', default_workflow='qc')[source]#

Bases: BaseAgent

Intelligently route an NL request to the correct agent chain.

Parameters:

  • api_key (str)

  • model (str)

  • llm_provider (str)

  • default_workflow (str) – Fallback when NL classification is ambiguous (default "qc").

  • keys (Output data)

  • ----------

  • request (str — natural-language processing request)

  • config (dict, optional — pre-built config (skips NL parsing))

  • dry_run (bool — preview without executing (default False))

  • output_dir (str)

  • data_path (str — EDI path (overrides extracted path))

  • keys

  • ----------------

  • str (reasoning)

  • str

  • instance (coordinator AgentCoordinator)

  • coordinator (result AgentResult from the)

  • metadata (steps list of step)

Examples

Dry-run preview:

agent = WorkflowOrchestratorAgent()
r = agent.execute({
    "request": "Load L22PLT EDIs, run full phase tensor analysis",
    "dry_run": True,
})
print(r["workflow_type"])  # "phase_analysis"

Full run with LLM:

agent = WorkflowOrchestratorAgent(api_key="sk-ant-…")
r = agent.execute({
    "request": "Clean and denoise the WILLY data, then run AI inversion",
    "data_path": "/data/WILLY_DATA",
})
SYSTEM_PROMPT: str = 'You are a pycsamt MT workflow routing expert.\nGiven a natural-language MT processing request,\nreturn a JSON object with:\n{\n  "workflow_type": one of:\n    "qc", "phase_analysis", "pre_inversion",\n    "ai_inversion", "inv2d", "inv3d",\n    "ensemble_inversion", "joint_inversion",\n    "modem", "mare2dem", "full", "full_ai_workflow",\n    "pinn_inversion", "hybrid_inversion",\n    "tipper", "sensitivity", "rotation",\n    "freq_decimation", "batch", "comparison",\n    "denoise", "static_shift", "inversion_eval",\n    "code_gen", "report", "forward",\n    "interpretation",\n  "reasoning": one sentence explaining the choice\n}\n\nRules:\n- "qc" if quality, cleaning, or flagging only.\n- "denoise" if denoising or noise removal only.\n- "static_shift" if static-shift or galvanic\n  distortion only (no full inversion).\n- "phase_analysis" if phase tensor, strike,\n  dimensionality, or Mohr circles.\n- "pre_inversion" if Occam2D, mesh preparation.\n- "inversion_eval" if evaluating an existing\n  inversion result, RMS, misfit, or residuals.\n- "ai_inversion" if 1-D AI or neural inversion.\n- "inv3d" if GCN or 3-D AI inversion.\n- "inv2d" if U-Net or 2-D profile AI inversion.\n- "ensemble_inversion" if ensemble or uncertainty.\n- "joint_inversion" if multi-modal or TEM+MT.\n- "modem" if ModEM or 3-D conventional inversion.\n- "mare2dem" if MARE2DEM or 2.5-D FEM inversion.\n- "pinn_inversion" if PINN or physics-informed.\n- "hybrid_inversion" if two-stage or AI warm-start.\n- "tipper" if tipper or induction arrows.\n- "sensitivity" if sensitivity kernels or DOI.\n- "rotation" if tensor rotation or strike frame.\n- "freq_decimation" if period selection/decimation.\n- "batch" if batch processing multiple profiles.\n- "comparison" if comparing inversion results.\n- "code_gen" if generating a Python script.\n- "report" if generating a survey report only.\n- "forward" if forward modelling or synthetic data.\n- "interpretation" if geological interpretation.\n- "full" if complete pipeline or multiple methods.\nDefault to "qc" when uncertain.\nReturn ONLY the JSON.\n'#

Override in subclasses to give the LLM its domain expertise.

execute(input_data)[source]#

Run this agent on input_data and return an AgentResult.

Subclasses must implement this method. The contract:

  • Reset self._last_cost = 0.0 at the top.

  • Record wall-clock time with t0 = time.time().

  • Return AgentResult(elapsed_seconds=time.time()-t0, cost_estimate_usd=self._last_cost, ...).

Parameters:

input_data (dict[str, Any])

Return type:

AgentResult