pycsamt.agents.batch_survey#

pycsamt.agents.batch_survey#

BatchSurveyAgent — Run the full processing pipeline over multiple MT profiles in parallel.

Each profile (EDI directory or file path) is processed independently through a configurable agent chain. Results are merged into a survey-level summary DataFrame and a multi-panel overview figure.

Parallelism is provided by joblib (soft dependency). When n_jobs=1 or joblib is unavailable, processing is sequential.

Classes

BatchSurveyAgent(*[, api_key, model, ...])

Process multiple MT profiles through a shared agent chain.
class pycsamt.agents.batch_survey.BatchSurveyAgent(*, api_key=None, model=None, llm_provider='claude', workflow='qc', n_jobs=1)[source]#

Bases: BaseAgent

Process multiple MT profiles through a shared agent chain.

Parameters:

  • api_key (str)

  • model (str)

  • llm_provider (str)

  • workflow (str, optional — override constructor default) – Pre-defined workflow key: 'qc', 'ai_inversion', 'phase_analysis', 'sensitivity', 'tipper'. Determines which agents are chained for each profile.

  • n_jobs (int, optional) – Parallel workers. -1 = all CPU cores. 1 = sequential (default).

  • keys (Output data)

  • ----------

  • profiles (dict {name: path} or list[str]) – Profile names to paths, or a list of paths (names auto-assigned).

  • workflow

  • n_jobs

  • output_dir (str, optional)

  • call. (Any additional keys are forwarded to each agent's execute())

  • keys

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

  • {name (profile_results dict)

  • metrics (summary_table pandas.DataFrame — per-profile)

  • int (n_failed)

  • int

  • dict (figure_paths)

  • dict

SYSTEM_PROMPT: str = 'You are an expert in large-scale MT survey processing and quality control.\nGiven a batch processing result, write 3-4 sentences that:\n1. Report how many profiles succeeded and how many failed.\n2. Identify which profiles have the worst QC scores or highest RMS.\n3. Highlight any systematic issues across profiles (e.g. dead band at same frequency).\n4. Recommend which profiles need re-processing or manual inspection.\nReply in plain English.\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