2.26.5.1. pycsamt.agents._base

pycsamt.agents._base

Abstract foundation for all pycsamt agents.

Every agent:

• Inherits BaseAgent and implements execute().

• Returns an AgentResult dataclass.

• Has access to pycsamt’s full API (PYCSAMT_SECTION, PYCSAMT_STYLE, PYCSAMT_STATION_RENDERING, PYCSAMT_CONTROL, PLOT_CONFIG) so all figures produced by agents are indistinguishable from hand-crafted pycsamt plots.

• Supports five LLM providers: Anthropic Claude (default), OpenAI, Google Gemini, DeepSeek, MiniMax. When no API key is supplied every LLM-dependent method degrades gracefully.

Classes

AgentResult(status, summary[, data, ...])	Standardised output returned by every pycsamt agent.
BaseAgent(name, *[, api_key, model, ...])	Abstract base class for all pycsamt agents.

class pycsamt.agents._base.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._base.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]