pycsamt.agents.router#

pycsamt.agents.router#

IntentRouter — the true “master” entry point.

The orchestrator classifies which geophysical workflow a request maps to. But that is only the right question once we already know the user wants to run a workflow at all. Many messages are not workflow requests:

  • a question about the package (“what does StaticShiftAgent do?”),

  • a code request (“write a script to load EDIs”),

  • a plot request (“show me the pseudosection”),

  • a meta request (“what can you do?”, “hi”),

The IntentRouter sits above the workflow classifier and decides the intent first, then lets each intent dispatch to the right specialist agent.

Design#

  • LLM-first when a key is available — a single structured tool-style call returns {intent, workflow, confidence, clarification, reasoning}.

  • Deterministic offline fallback — classify_intent_offline() is a fast, pure-function heuristic used when no key is configured and as the cheap pre-check the GUI uses to decide whether a message even needs an EDI dataset loaded.

  • Low-confidence → clarify — rather than guessing, the router can return intent == CLARIFY so the caller asks one disambiguating question.

The router never executes anything; it only decides where a message goes.

Functions

classify_intent_offline(text)

Classify text into an intent without any LLM call.

Classes

IntentRouter(*[, api_key, model, llm_provider])

Classify a chat message into a RouterDecision.

RouterDecision(intent[, workflow, ...])

Structured routing decision returned by IntentRouter.route().
class pycsamt.agents.router.IntentRouter(*, api_key=None, model=None, llm_provider='claude')[source]#

Bases: BaseAgent

Classify a chat message into a RouterDecision.

Parameters:

Examples

Offline:

router = IntentRouter()
d = router.route("what does StaticShiftAgent do?")
d.intent          # 'question'
d.needs_data      # False

Online:

router = IntentRouter(api_key="sk-ant-...", llm_provider="claude")
d = router.route("run QC on /data/willy")
d.intent          # 'workflow'
SYSTEM_PROMPT: str = 'You are the top-level intent router for the pycsamt v2 magnetotelluric (MT)\nassistant. Classify the user\'s message into exactly one INTENT.\n\nINTENTS:\n- "question": the user asks ABOUT pycsamt — concepts, what a class/function\n  does, how something works, which method to use, definitions. They want an\n  explanation, not execution.\n- "code": the user wants a Python script / function / notebook generated.\n- "plot": the user wants a figure/plot/visualisation produced from data.\n- "workflow": the user wants to RUN a processing pipeline on their data\n  (QC, static-shift, phase analysis, an inversion, report, etc.).\n- "meta": greetings, "what can you do", "introduce yourself", capability\n  questions about the assistant itself, and requests to LIST/ENUMERATE the\n  agents, tasks or workflows the assistant can run ("list the agents", "which\n  workflows are available", "what tasks can you perform"). A question about\n  what ONE named agent/function does is a "question", not "meta".\n- "metrics": the user asks for a COMPUTED VALUE of their survey line(s) and\n  wants the number back inline — strike, azimuth/bearing, dimensionality,\n  skew, station count, period/frequency range, coordinates/length, quality\n  score, or a one-line summary ("what\'s the strike of L22PLT?", "azimuth of\n  all lines", "how many stations", "tell me about this line"). This is NOT a\n  plot/figure request and NOT "run an analysis".\n\nReturn ONLY a JSON object:\n{\n  "intent": one of question|code|plot|workflow|meta|metrics,\n  "workflow": one of [qc, static_shift, phase_analysis, forward, pre_inversion, inversion_eval, interpretation, report, full, ai_inversion, inv2d, inv3d, ensemble_inversion, joint_inversion, pinn_inversion, hybrid_inversion, modem, occam2d, tipper, sensitivity, rotation, freq_decimation, batch, comparison, code_gen, denoise, rhophi, phase_psection, pt_psection, tipper_plot, phase_tensor_map, pt_strip, pt_strip_grid, station_response, strike_profile, strike, dimensionality, validator, coords, elevation, converter, batch_export, freq_editor, layered_model, corr_ss_ama, corr_ss_loess, corr_ss_bilateral, corr_ss_refmedian, corr_ss_emap, corr_notch, corr_smooth_logfreq, corr_smooth_rho_phase, corr_rotate_angle, corr_rotate_strike, corr_rotate_pt_strike, corr_rotate_profile, corr_antisymmetrize, corr_coord_projection, corr_coord_spacing, corr_coord_snap, corr_coord_elevation, corr_coord_shift, corr_coord_interpolate, corr_near_field, corr_strat_qc, corr_strat_static_shift, corr_strat_noise, corr_strat_freq_filter, corr_strat_full] or null (only for workflow/plot/code),\n  "confidence": float 0..1,\n  "clarification": a single question to ask IF the request is too ambiguous\n     to route, else null,\n  "reasoning": one short sentence\n}\n\nRules:\n- "How do I run an inversion?" is a QUESTION (asking for guidance).\n- "Run an inversion on /data/x" is a WORKFLOW (asking to execute).\n- "Write code to run an inversion" is CODE.\n- Prefer "question" when the user clearly wants to learn, not execute.\n- Set a low confidence (<0.5) and provide "clarification" when genuinely\n  ambiguous.\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

route(text, *, history=None)[source]#

Return a RouterDecision for text.

Uses the LLM when a key is configured, falling back to the offline heuristic on any failure or when offline.

Parameters:
Return type:

RouterDecision

class pycsamt.agents.router.RouterDecision(intent, workflow=None, confidence=0.0, clarification=None, reasoning='', source='offline')[source]#

Bases: object

Structured routing decision returned by IntentRouter.route().

Variables:

  • intent (str) – One of INTENTS.

  • workflow (str or None) – Workflow slot, only meaningful when intent is WORKFLOW or PLOT.

  • confidence (float) – 0.0–1.0 self-reported confidence.

  • clarification (str or None) – A question to ask the user when intent == CLARIFY.

  • reasoning (str) – One-sentence rationale (useful for logs / debugging).

  • source (str) – "llm" or "offline" — which path produced the decision.

Parameters:

  • intent (str)

  • workflow (str | None)

  • confidence (float)

  • clarification (str | None)

  • reasoning (str)

  • source (str)

intent: str#
workflow: str | None = None#
confidence: float = 0.0#
clarification: str | None = None#
reasoning: str = ''#
source: str = 'offline'#
property needs_data: bool[source]#

True when this intent requires a loaded EDI dataset.

pycsamt.agents.router.classify_intent_offline(text)[source]#

Classify text into an intent without any LLM call.

Returns (intent, confidence). Fast and deterministic; used both as the offline router backend and as the GUI pre-check that decides whether a message needs an EDI dataset (so questions/code/help never trip the data guard).

Examples

>>> classify_intent_offline("what does StaticShiftAgent do?")[0]
'question'
>>> classify_intent_offline("run AI inversion on /data/willy")[0]
'workflow'
>>> classify_intent_offline("write a script to load EDIs")[0]
'code'
>>> classify_intent_offline("hi")[0]
'meta'
Parameters:

text (str)

Return type:

tuple[str, float]