2.26. pycsamt.agents#

AI-assisted workflow agents for loading, quality control, processing, inversion preparation, interpretation, reporting, and orchestration.

AI-assisted MT, AMT, and CSAMT workflow automation.

All agents are lazy-loaded: importing this module costs nothing unless you actually instantiate an agent. This keeps the base pycsamt import fast even when LLM libraries (anthropic, openai, google-generativeai) are not installed.

2.26.1. Agent catalogue#

ContextInputAgent: Translate a natural-language request into a structured workflow config.
MTLoaderAgent: Load EDI / AVG / J files into a Sites object with a per-station QC report.
AgentCoordinator: Chain agents into named workflows with checkpointing and cost tracking.
DataQCAgent: SNR section, dead-band detection, per-station quality scores.
StaticShiftAgent: Static-shift detection and correction (AMA / LOESS / spatial median).
PhaseAnalysisAgent: Phase tensor, strike, dimensionality, Mohr circles, Argand diagrams.
ForwardModelAgent: 1-D, 2-D, and 3-D MT forward modelling.
InversionPrepAgent / Occam2DAgent / ModEmAgent / Mare2DEMAgent: Write Occam2D, ModEM3D, or MARE2DEM inversion input files; optionally run the binary and scan results.
InversionEvaluationAgent: Compute RMS, residual PT section, misfit pseudosection.
InterpretationAgent: Map resistivity ranges to lithology; correlate with borehole logs.
ReportAgent: Assemble all figures and tables into Markdown / HTML / PDF.
CodeGenerationAgent: Emit a standalone reproducible Python script from the workflow config.
WorkflowOrchestratorAgent: NL → classify workflow → build and run the correct agent chain.
AgentMaster: One-line front door over the orchestrator: AgentMaster(provider="anthropic").run("...").
DenoisingAgent: RPCA / Hampel / EMAP / AI-CAE denoising.
AnomalyDetectionAgent: Unsupervised CAE anomaly flagging per (station, frequency).
AIInversionAgent: End-to-end 1-D AI inversion (ResNet / CNN / FCN).
Inv2DAgent: U-Net 2-D profile inversion with lateral continuity.
Inv3DAgent: GCN 3-D spatial inversion using inter-station graph message-passing.
EnsembleAgent: Ensemble 1-D inversion with conformal uncertainty bands.
JointInversionAgent: DRCNN multi-modal joint inversion (MT + TEM / CSAMT / gravity).
ModelZooAgent: Browse, download, and deploy pre-trained EM inverter checkpoints.
PINNInversionAgent: Physics-informed 1-D/2-D/3-D MT inversion via Adam (no labelled training data required).
HybridInversionAgent: Two-stage AI warm-start + physics refinement for 1-D, 2-D, and 3-D MT inversion.
IoTFieldAgent: Monitor an IoT-enabled AMT/MT/CSAMT/CSEM field acquisition from its edge telemetry: monitoring status, sync/power summaries, provenance manifest, and dashboard figures.

2.26.2. Supporting classes#

AgentResult: Standardised return type for every agent.
BaseAgent: Abstract base class; inherit to build custom agents.

2.26.3. Quick start#

Without an LLM key (regex fallback, no cost):

from pycsamt.agents import (
    ContextInputAgent,
    MTLoaderAgent,
    AgentCoordinator,
)

ctx = ContextInputAgent()  # no api_key → pure regex
loader = MTLoaderAgent()

coord = AgentCoordinator("mt_qc")
coord.add_step("parse", ctx, description="Parse request into config")
coord.add_step(
    "load",
    loader,
    input_fn=lambda r: {"path": r["parse"]["config"]["data_path"]},
    description="Load EDI files and QC scan",
)

result = coord.execute(
    {"request": "Load /data/EDIs, QC, period range 1e-4 to 1 s"},
    dry_run=True,
)

With a Claude API key:

ctx = ContextInputAgent(api_key="sk-ant-…", llm_provider="claude")
res = ctx.execute({"request": "Load L22PLT EDIs, run PT analysis …"})
print(res["config"])
print(res.llm_interpretation)

2.26.4. LLM providers supported#

Anthropic Claude (default) — llm_provider="claude"
OpenAI — llm_provider="openai"
Google Gemini — llm_provider="gemini"
DeepSeek — llm_provider="deepseek"
MiniMax — llm_provider="minimax"

Install the relevant client library before use: pip install anthropic / pip install openai / pip install google-generativeai DeepSeek and MiniMax reuse the openai package with a custom base_url; no extra install is needed.

class pycsamt.agents.AgentConfig#

Bases: object

Global LLM configuration singleton for all pycsamt agents.

BaseAgent calls resolve() on every instantiation and get_rate() + _add_spend() on every LLM call, so keys, rates, and the budget cap set here apply automatically to every agent in the session.

:param (none — use configure() after construction):

Variables:

provider (str or None) – Currently active LLM provider.
model (str or None) – Resolved model (explicit override or provider default).
api_key (str or None) – Resolved key for the active provider (stored key → env var → None).
is_configured (bool) – True when a provider and a resolvable key are both present.
spent_usd (float) – Accumulated LLM spend this session.
remaining_usd (float or None) – Remaining budget, or None when no cap is set.

Examples

from pycsamt.api.agents import AGENT_CONFIG

# one-call setup
AGENT_CONFIG.configure(provider="claude", api_key="sk-ant-…")

# pricing — override a rate (USD / 1 M tokens)
AGENT_CONFIG.set_rate(
    "claude", "claude-opus-4-8", input=12.0, output=60.0
)

# pricing — add a model not in the built-in table
AGENT_CONFIG.set_rate("openai", "gpt-5-turbo", input=5.0, output=20.0)

# budget cap
AGENT_CONFIG.set_budget(usd=5.0)
print(AGENT_CONFIG.remaining_usd)  # 5.0

# inspect
print(AGENT_CONFIG.info())

# reset
AGENT_CONFIG.reset()

configure(*, provider, api_key, model=None)#

Set the active provider, key, and optional model in one call.

Parameters:

provider ({"claude", "openai", "gemini"}) – LLM provider to activate.
api_key (str) – API key for provider. Stored per-provider so switching away and back does not require re-supplying the key.
model (str or None) – Model identifier override. None uses the provider default.

Returns:

self — allows chaining.

Return type:

AgentConfig

Examples

AGENT_CONFIG.configure(provider="claude", api_key="sk-ant-…")
AGENT_CONFIG.configure(
    provider="openai",
    api_key="sk-…",
    model="gpt-4o-mini",
)

set_key(provider, api_key)#

Store an API key for provider without changing the active provider.

Useful for pre-loading keys for multiple providers so you can switch() between them without re-supplying credentials.

Parameters:

provider (str)
api_key (str)

Returns:

self.

Return type:

AgentConfig

Examples

AGENT_CONFIG.set_key("claude", "sk-ant-…")
AGENT_CONFIG.set_key("openai", "sk-…")
AGENT_CONFIG.set_key("gemini", "AIza…")
AGENT_CONFIG.switch("claude")

switch(provider, *, model=None)#

Switch the active provider.

The key for provider must have been stored via configure() or set_key(), or be available as an environment variable.

Parameters:

provider (str)
model (str or None) – Override the model for this provider. None keeps the current model override (or uses the provider default).

Returns:

self.

Return type:

AgentConfig

Examples

AGENT_CONFIG.switch("openai")
AGENT_CONFIG.switch("gemini", model="gemini-2.0-flash")

reset(*, keys=True)#

Clear the active configuration, custom rates, and budget.

Parameters:: keys (bool) – When True (default) also wipe all stored per-provider keys. Pass False to clear only the active provider / model while keeping stored keys available for future switch() calls.
Returns:: self.
Return type:: AgentConfig

Notes

Custom rates and the budget cap/counter are always reset. Call reset_budget() to zero only the spend counter.

set_rate(provider, model, *, input, output)#

Override or add the cost rate for a specific provider + model.

Custom rates take precedence over the built-in table for every cost estimate made by agents in this session.

Parameters:

provider (str) – "claude" | "openai" | "gemini"
model (str) – Model identifier exactly as passed to the LLM client, e.g. "claude-opus-4-8" or "gpt-5-turbo".
input (float) – Cost per 1 000 000 input tokens in USD.
output (float) – Cost per 1 000 000 output tokens in USD.

Returns:

self.

Return type:

AgentConfig

Examples

# provider lowered their price
AGENT_CONFIG.set_rate(
    "claude", "claude-opus-4-8", input=12.0, output=60.0
)

# a new model not yet in the built-in table
AGENT_CONFIG.set_rate(
    "openai", "gpt-5-turbo", input=5.0, output=20.0
)

get_rate(provider, model)#

Return the resolved {"input": …, "output": …} rate.

Resolution order: custom override → built-in table (exact match → prefix match) → provider default.

Parameters:

provider (str)
model (str)

Return type:

dict with keys "input" and "output" (USD / 1 M tokens).

Examples

rate = AGENT_CONFIG.get_rate("claude", "claude-sonnet-4-6")
# {"input": 3.0, "output": 15.0}

estimate_cost(provider, model, input_tokens, output_tokens)#

Compute the USD cost for one LLM call using the resolved rate.

Respects any rate overrides set via set_rate().

Parameters:

provider (str)
model (str)
input_tokens (int)
output_tokens (int)

Return type:

float — estimated cost in USD.

Examples

cost = AGENT_CONFIG.estimate_cost(
    "claude", "claude-sonnet-4-6", 500, 200
)

list_rates(provider=None)#

Return the effective rate table (custom overrides merged with built-in).

Parameters:: provider (str or None) – When given, return only that provider’s table. When None, return all providers.
Return type:: dict — same structure as the built-in table.

Examples

import pprint

pprint.pprint(AGENT_CONFIG.list_rates("claude"))

set_budget(*, usd)#

Set a session spend cap.

Once spent_usd reaches usd, any subsequent LLM call will raise BudgetExceededError before the API call is made.

Parameters:: usd (float) – Maximum spend in USD for this session.
Returns:: self.
Return type:: AgentConfig

Examples

AGENT_CONFIG.set_budget(usd=2.0)
# … after several agent calls …
print(
    f"${AGENT_CONFIG.spent_usd:.4f} used, "
    f"${AGENT_CONFIG.remaining_usd:.4f} left"
)

reset_budget(*, cap=False)#

Reset the session spend counter.

Parameters:: cap (bool) – When True also remove the budget cap (set_budget must be called again to re-enable it). Default False — zeroes the counter but keeps the existing cap.
Returns:: self.
Return type:: AgentConfig

Examples

AGENT_CONFIG.reset_budget()  # zero counter, keep cap
AGENT_CONFIG.reset_budget(cap=True)  # zero counter and remove cap

property spent_usd: float#: Accumulated LLM spend this session (USD).

property remaining_usd: float | None#: Remaining budget (USD), or None when no cap is set.

property provider: str | None#: Currently active provider, or None if unconfigured.

property api_key: str | None#

Resolved key for the active provider.

Resolution order: stored key → environment variable → None.

property model: str | None#: Active model (explicit override, or provider default).

property is_configured: bool#: True when a provider and a resolvable key are both present.

resolve(provider, api_key, model)#

Resolve the effective (provider, api_key, model) for an agent.

Called automatically by __init__. Users should not call this directly.

2.26. Resolution rules#

If api_key is explicitly given:

Use provider, api_key, and model (or provider default) exactly as supplied. The global config is ignored.

If api_key is None:

If provider equals the default "claude" and the global config has an active provider, inherit that provider.
Look up the key: stored key → env var.
Inherit the model from the global config when the effective provider matches the globally active provider.

Parameters:

provider (str)
api_key (str | None)
model (str | None)

Return type:

tuple[str, str | None, str | None]

info()#

Return a summary dict suitable for display or logging.

The API key is masked to its last four characters for safety.

Returns:: Keys: provider, model, has_key, key_source, stored_providers, custom_rate_models, budget_usd, spent_usd, remaining_usd.
Return type:: dict

using(*, provider=None, api_key=None, model=None)#

Temporarily override the global config inside a with block.

All arguments are optional; omitted values are left unchanged. The original config (including custom rates and budget) is restored on exit even if an exception occurs.

Parameters:

provider (str or None)
api_key (str or None)
model (str or None)

Yields:

AgentConfig – self with the override applied.

Return type:

Generator[AgentConfig, None, None]

Examples

with AGENT_CONFIG.using(provider="gemini", api_key="AIza…"):
    r = DataQCAgent().execute(data)
# original provider / key / model restored here

offline()#

Context manager: force truly offline mode in the current thread.

While active, _resolve_key() will not inspect environment variables, so agents created inside the block receive api_key=None even when ANTHROPIC_API_KEY (or similar) is set in the OS environment. Uses threading.local so concurrent requests in other threads are unaffected.

Examples

with AGENT_CONFIG.offline():
    agent = DataQCAgent()  # api_key will be None

Return type:: Generator[AgentConfig, None, None]

exception pycsamt.agents.BudgetExceededError(spent, budget)#

Bases: RuntimeError

Raised when an LLM call would be made after the session budget is used.

Variables:

spent_usd (float) – Accumulated spend so far this session.
budget_usd (float) – The cap that was set.

Parameters:

spent (float)
budget (float)

Return type:

None

pycsamt.agents.configure_agents(*, provider, api_key, model=None)#

Configure AGENT_CONFIG in one call and return it.

Equivalent to AGENT_CONFIG.configure(...).

Parameters:

provider ({"claude", "openai", "gemini"})
api_key (str)
model (str or None)

Return type:

AgentConfig

Examples

from pycsamt.agents import configure_agents

configure_agents(provider="claude", api_key="sk-ant-…")

pycsamt.agents.reset_agents(*, keys=True)#

Reset AGENT_CONFIG to its unconfigured state.

Equivalent to AGENT_CONFIG.reset(...).

Parameters:: keys (bool) – When True (default) also wipe stored per-provider keys.
Return type:: AgentConfig

class pycsamt.agents.AgentResult(status, summary, data=<factory>, warnings=<factory>, llm_interpretation=None, elapsed_seconds=0.0, cost_estimate_usd=0.0, error=None, error_fix_hint=None)#

Bases: object

Standardised output returned by every pycsamt agent.

Variables:

status (str) – "success" | "failed" | "needs_review"
summary (str) – One-sentence human-readable description of what happened.
data (dict) – Agent-specific outputs (arrays, paths, dataframes, figures …).
warnings (list of str) – Non-fatal issues encountered during execution.
llm_interpretation (str or None) – Free-text interpretation written by the LLM, when available.
elapsed_seconds (float)
cost_estimate_usd (float) – Estimated LLM API cost for this execution.
error (str or None) – Exception message if status == "failed".
error_fix_hint (str or None) – Suggested remediation for the error.

Parameters:

status (str)
summary (str)
data (dict[str, Any])
warnings (list[str])
llm_interpretation (str | None)
elapsed_seconds (float)
cost_estimate_usd (float)
error (str | None)
error_fix_hint (str | None)

Examples

>>> result = some_agent.execute({"path": "/data/EDIs"})
>>> result.status
'success'
>>> result["sites"]  # dict-like access to data
<Sites 25 stations>
>>> result.get("n_stations", 0)
25

status: str#

summary: str#

data: dict[str, Any]#

warnings: list[str]#

llm_interpretation: str | None = None#

elapsed_seconds: float = 0.0#

cost_estimate_usd: float = 0.0#

error: str | None = None#

error_fix_hint: str | None = None#

get(key, default=None)#

Parameters:

key (str)
default (Any)

Return type:

Any

classmethod failed(error, *, hint=None, elapsed=0.0)#

Convenience constructor for failure results.

Parameters:

error (str)
hint (str | None)
elapsed (float)

Return type:

AgentResult

class pycsamt.agents.BaseAgent(name, *, api_key=None, model=None, llm_provider='claude', section_preset='pseudosection', verbose=False)#

Bases: ABC

Abstract base class for all pycsamt agents.

Parameters:

name (str) – Human-readable agent name used in logs and reports.
api_key (str or None) – LLM API key. When None the agent runs without LLM support and every llm_interpretation field will be None.
model (str or None) – LLM model identifier. Defaults to the provider’s recommended model.
llm_provider ({"claude", "openai", "gemini", "deepseek", "minimax"}) – Which provider to use. Default "claude".
section_preset (str) – Which PYCSAMT_SECTION preset governs figures produced by this agent. Default "pseudosection".
verbose (bool, int, or str) – Progress verbosity forwarded to the underlying dataset-generation and training calls — see pycsamt.utils.progress.normalize_verbose(). Default False (silent), matching this agent framework’s historical behaviour of returning results/logs rather than printing.

Examples

Subclass and implement execute():

class MyAgent(BaseAgent):
    SYSTEM_PROMPT = "You are an expert MT data analyst."

    def execute(self, input_data):
        t0 = time.time()
        # ... do work ...
        interp = self.query_llm("Interpret these results: ...")
        return AgentResult(
            status="success",
            summary="Analysis complete.",
            data={"result": 42},
            llm_interpretation=interp,
            elapsed_seconds=time.time() - t0,
            cost_estimate_usd=self._last_cost,
        )

SYSTEM_PROMPT: str = 'You are a geophysics expert specialising in magnetotelluric (MT/AMT/CSAMT) data processing and interpretation.'#: Override in subclasses to give the LLM its domain expertise.

abstractmethod execute(input_data)#

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

query_llm(prompt, system_message=None, *, temperature=0.2, max_tokens=1024)#

Send prompt to the configured LLM and return the response text.

Returns None when no API key is configured or all retries fail. Accumulates token cost into self._last_cost.

Parameters:

prompt (str)
system_message (str or None) – Overrides SYSTEM_PROMPT for this call.
temperature (float)
max_tokens (int)

Return type:

str or None

extract_json(text)#

Extract the first JSON object / array from text.

Returns None when no valid JSON is found.

Parameters:: text (str)
Return type:: dict | list | None

require_keys(input_data, *keys, agent_name=None)#

Return a list of missing required keys.

Parameters:

input_data (dict)
keys (str)
agent_name (str | None)

Return type:

list[str]

class pycsamt.agents.AgentCoordinator(workflow_name='pycsamt_workflow', *, checkpoint_dir=None, verbose=True)#

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)#

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)#

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)#

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()#

Delete all saved checkpoints for this workflow.

Return type:: None

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

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.

pycsamt.agents.estimate_cost(provider, model, input_tokens, output_tokens)#

Compute the USD cost for one LLM call.

Delegates to estimate_cost(), so any per-model overrides set via AGENT_CONFIG.set_rate(...) are used.

Parameters:

provider (str)
model (str)
input_tokens (int)
output_tokens (int)

Returns:

Estimated cost in USD.

Return type:

float

Examples

>>> estimate_cost("claude", "claude-sonnet-4-6", 500, 200)
0.00451...

pycsamt.agents.format_cost(usd)#

Return a human-readable cost string.

Examples

>>> format_cost(0.00012)
'$0.000120'
>>> format_cost(1.5)
'$1.5000'

Parameters:: usd (float)
Return type:: str

pycsamt.agents.get_rate(provider, model)#

Return the resolved {"input": …, "output": …} rate.

Delegates to get_rate(), so any per-model overrides set via AGENT_CONFIG.set_rate(...) take effect.

Parameters:

provider (str) – One of "claude", "openai", "gemini", "deepseek", "minimax".
model (str) – Model identifier, e.g. "claude-sonnet-4-6".

Return type:

dict with keys "input" and "output" (USD / 1 M tokens).

class pycsamt.agents.WorkflowPlan(request, workflow_type, data_path='', output_dir='', parameters=<factory>, risk_flags=<factory>, requires_human_review=False, expected_outputs=<factory>, provider='offline', grounding_citations=<factory>)#

Bases: object

Validated intermediate representation of a workflow request.

Produced by ContextInputAgent and consumed by WorkflowOrchestratorAgent.

Parameters:

request (str) – Original natural-language request.
workflow_type (str) – One of the recognised workflow identifiers (see VALID_WORKFLOWS).
data_path (str) – Path to EDI files or survey directory.
output_dir (str) – Destination for outputs.
parameters (dict) – Parsed parameters (period_range, component, etc.).
risk_flags (list of str) – Warnings or risks the user should review.
requires_human_review (bool) – True when ambiguous parameters were inferred.
expected_outputs (list of str) – Files / artefacts the workflow should produce.
provider (str) – LLM provider used for parsing, or 'offline'.
grounding_citations (list of str) – Source paths retrieved by RAG to ground this plan, when retrieval was available (empty offline or on a cold index).

request: str#

workflow_type: str#

data_path: str = ''#

output_dir: str = ''#

parameters: dict[str, Any]#

risk_flags: list[str]#

requires_human_review: bool = False#

expected_outputs: list[str]#

provider: str = 'offline'#

grounding_citations: list[str]#

is_valid()#

Return True when the plan passes basic validation.

Return type:: bool

validation_errors()#

Return a list of validation error messages.

Return type:: list[str]

to_dict()#

Return a JSON-serialisable dict.

Return type:: dict[str, Any]

to_json(indent=2)#

Serialise to a JSON string.

Parameters:: indent (int)
Return type:: str

save(path)#

Write the plan to path as JSON.

Parameters:: path (str | Path)
Return type:: None

classmethod from_config(config, request='', provider='offline', citations=None)#

Build a WorkflowPlan from a config dict as returned by ContextInputAgent.

Parameters:

config (dict[str, Any])
request (str)
provider (str)
citations (list[str] | None)

Return type:

WorkflowPlan

pycsamt.agents.validate_workflow_plan(plan, *, raise_on_error=False)#

Validate plan and optionally raise on errors.

Parameters:

plan (WorkflowPlan)
raise_on_error (bool) – When True, raises ValueError if errors exist.

Returns:

errors – Empty when the plan is valid.

Return type:

list of str

class pycsamt.agents.AgentMaster(provider='claude', *, api_key=None, model=None, default_workflow='qc')#

Bases: PyCSAMTObject

Plain-language entry point to the agent workflows.

Parameters:

provider (str, default "claude") – LLM provider. Friendly aliases are accepted: "anthropic" → "claude" and "google" → "gemini". Without an API key the agents fall back to the rule-based (regex/keyword) path, so AgentMaster() works offline.
api_key (str, optional) – Provider API key. When omitted, the provider’s environment variable is used if set; otherwise the rule-based fallback runs at zero cost.
model (str, optional) – Provider model override (defaults per provider).
default_workflow (str, default "qc") – Workflow used when a request cannot be classified.

Examples

Plan first (no files touched), then execute:

>>> from pycsamt.agents import AgentMaster
>>> master = AgentMaster(provider="anthropic")
>>> plan = master.plan(
...     "QC the EDI files and prepare a short report",
...     data_path="data/edi/",
... )
>>> plan["workflow_type"]
'qc'
>>> report = master.run(
...     "Load data/edi/, flag stations with RMS > 2, build an Occam2D "
...     "input for profile L22, launch inversion, and produce a PDF "
...     "report."
... )

See also

pycsamt.agents.WorkflowOrchestratorAgent: The dispatcher this façade drives; use it directly for structured workflow configurations.

property orchestrator: WorkflowOrchestratorAgent#: The lazily-built orchestrator behind this façade.

run(request, *, data_path=None, output_dir=None, dry_run=False, **extra)#

Route request to the right specialist agent and run it.

request is first classified by IntentRouter — a question is answered by PackageQAAgent, a code request generates a script via CodeGenerationAgent, a metrics request is computed by MetricsAgent, and a workflow / plot request runs the full pipeline through WorkflowOrchestratorAgent. A meta (capability) or clarify (ambiguous) request returns immediately with no data path required.

Parameters:

request (str) – Plain-language description of what to do. Paths mentioned in the text are extracted when possible; pass data_path / output_dir explicitly for scripts and CI.
data_path (str, optional) – Survey input (EDI/AVG/J directory or file).
output_dir (str, optional) – Where products (figures, inputs, reports) are written.
dry_run (bool, default False) – For workflow / plot requests only: preview the selected chain without reading or writing. Ignored for questions, code, metrics, meta, and clarify requests, which never touch disk regardless.
**extra – Additional orchestrator payload fields, passed through for workflow / plot requests only.

Returns:

Status, per-step outputs, reasoning, and cost tracking.

Return type:

AgentResult

plan(request, **kwargs)#

Shortcut for run() with dry_run=True.

Parameters:

request (str)
kwargs (Any)

Return type:

AgentResult

class pycsamt.agents.ContextInputAgent(*, api_key=None, model=None, llm_provider='claude', use_rag=True)#

`pycsamt.agents._base`	pycsamt.agents._base
`pycsamt.agents.context`	pycsamt.agents.context
`pycsamt.agents.loader`	pycsamt.agents.loader
`pycsamt.agents.coordinator`	pycsamt.agents.coordinator
`pycsamt.agents.orchestrator`	pycsamt.agents.orchestrator
`pycsamt.agents.qc`	pycsamt.agents.qc
`pycsamt.agents.static_shift`	pycsamt.agents.static_shift
`pycsamt.agents.phase_analysis`	pycsamt.agents.phase_analysis
`pycsamt.agents.tensor_rotation`	pycsamt.agents.tensor_rotation
`pycsamt.agents.tipper_analysis`	pycsamt.agents.tipper_analysis
`pycsamt.agents.freq_decimation`	pycsamt.agents.freq_decimation
`pycsamt.agents.denoising`	pycsamt.agents.denoising
`pycsamt.agents.forward`	pycsamt.agents.forward
`pycsamt.agents.inversion_prep`	pycsamt.agents.inversion_prep
`pycsamt.agents.inversion_eval`	pycsamt.agents.inversion_eval
`pycsamt.agents.inversion_backend`	pycsamt.agents.inversion_backend
`pycsamt.agents.inversion_comparison`	pycsamt.agents.inversion_comparison
`pycsamt.agents.occam2d_agent`	pycsamt.agents.occam2d_agent
`pycsamt.agents.modem_agent`	pycsamt.agents.modem_agent
`pycsamt.agents.inv2d_agent`	pycsamt.agents.inv2d_agent
`pycsamt.agents.inv3d_agent`	pycsamt.agents.inv3d_agent
`pycsamt.agents.ai_inversion`	pycsamt.agents.ai_inversion
`pycsamt.agents.ensemble_agent`	pycsamt.agents.ensemble_agent
`pycsamt.agents.joint_agent`	pycsamt.agents.joint_agent
`pycsamt.agents.model_zoo_agent`	pycsamt.agents.model_zoo_agent
`pycsamt.agents.anomaly_agent`	pycsamt.agents.anomaly_agent
`pycsamt.agents.interpretation`	pycsamt.agents.interpretation
`pycsamt.agents.iot_agent`	pycsamt.agents.iot_agent
`pycsamt.agents.resistivity_map`	pycsamt.agents.resistivity_map
`pycsamt.agents.sensitivity`	pycsamt.agents.sensitivity
`pycsamt.agents.edi_export`	pycsamt.agents.edi_export
`pycsamt.agents.report`	pycsamt.agents.report
`pycsamt.agents.code_gen`	pycsamt.agents.code_gen
`pycsamt.agents.pipeline_agent`	pycsamt.agents.pipeline_agent
`pycsamt.agents.batch_survey`	pycsamt.agents.batch_survey
`pycsamt.agents.hybrid_agent`	pycsamt.agents.hybrid_agent
`pycsamt.agents.mare2dem_agent`	pycsamt.agents.mare2dem_agent
`pycsamt.agents.master`	One-line front door to the pyCSAMT agent stack.
`pycsamt.agents.metrics`	pycsamt.agents.metrics
`pycsamt.agents.package_qa`	pycsamt.agents.package_qa PackageQAAgent — answer free-form questions about the pycsamt v2 package.
`pycsamt.agents.pinn_agent`	pycsamt.agents.pinn_agent
`pycsamt.agents.plotting`	pycsamt.agents.plotting
`pycsamt.agents.router`	pycsamt.agents.router
`pycsamt.agents.tooling`	pycsamt.agents.tooling
`pycsamt.agents.web`	pycsamt.agents.web

2.26. pycsamt.agents#

2.26.1. Agent catalogue#

2.26.2. Supporting classes#

2.26.3. Quick start#

2.26.4. LLM providers supported#

2.26. Resolution rules#

2.26. Currently supported output formats#

2.26. Input keys#

2.26. Output data keys#

2.26. Input keys#

2.26. Output data keys#

2.26.5. Agent Modules#

2.26.6. IoT Field Agent#