PyHydroGeophysX.agents.assistants package#

Subpackages#

Module contents#

Domain AI assistants: AQUAH for hydrogeophysics, GeoSAGE for geological modelling, and more.

An assistant is one complete AI agent for a domain - what the user talks to in the studio’s chat and what “Auto to report” runs. PyHydroGeophysX used to have exactly one, AQUAH, written into the desktop app, the workflow runner and the chat’s system prompt. This package makes an assistant a plug-in: each is a small package that describes itself with an Assistant, and the studio builds its chat, its Workflow page and its run from that description.

What an assistant supplies#

A workflow (workflow="pkg.module:function"): what “Auto to report” runs, in the run’s own process. It takes the run’s payload and returns its result; see WORKFLOW_CONTRACT. Most assistants build it on PyHydroGeophysX.agents.runtime.entry.drive(), the controller loop over a tool registry, which gives the studio’s live timeline, step approval and chat narration without any further work.
Tools (tools="pkg.module:attr"): the assistant’s own registry of Tool, each with what it requires, what it produces and the studio module that shows its work.
What it says and accepts: a chat persona, example requests, the kinds of input file the Workflow page offers, and the colours of its glow.

Adding one#

In this repository: add a subpackage beside aquah and geosage exposing ASSISTANT and list it in BUILTIN. From a separate package: declare an entry point, and nothing here needs editing:

[project.entry-points."pyhydrogeophysx.assistants"]
geosage = "geosage.pyhydrogeophysx:ASSISTANT"

The developer guide (docs/source/agents/adding_an_assistant.rst) walks through both, with GeoSAGE as the example.

class PyHydroGeophysX.agents.assistants.Assistant(key: str, name: str, domain: str, title: str, summary: str, workflow: str, tools: str = '', persona: str = '', examples: ~typing.Tuple[str, ...] = (), input_roles: ~typing.Tuple[~typing.Tuple[str, str], ...] = (), ordered_roles: ~typing.Tuple[str, ...] = (), folder_classifier: bool = False, studio_modules: ~typing.Tuple[str, ...] = (), colors: ~typing.Tuple[str, ...] = (), requires_packages: ~typing.Tuple[str, ...] = (), status: str = 'ready', status_note: str = '', providers: ~typing.Tuple[str, ...] = ('openai', 'anthropic'), offline_workflow: bool = False, workflow_setup: str = '', retrieval: ~typing.Tuple[str, ...] = ('rag', 'mcp'), focused_workspace: bool = False, _cache: ~typing.Dict[str, ~typing.Any] = <factory>)[source]#

Bases: object

One domain AI assistant, described for the studio.

Parameters:

key (str) – Identifier, lower case: 'aquah', 'geosage'.
name (str) – What the user sees: 'AQUAH'.
domain (str) – Its field in a few words: 'Hydrogeophysics'.
title (str) – The name spelt out.
summary (str) – One or two sentences: what it does and what it needs.
workflow (str) – 'module:function' of the run, per WORKFLOW_CONTRACT.
tools (str) – 'module:attribute' of its tool registry (a dict of name to Tool, or a function returning one). Optional; the studio uses it to put a name to the step running.
persona (str) – The part of the chat’s system prompt that is this assistant’s: who it is, and rules for the studio modules it works with.
examples (tuple of str) – Requests shown when a chat starts.
input_roles (tuple of (label, key)) – The kinds of input the Workflow page offers to add by hand. A key ending in _dir takes a folder.
ordered_roles (tuple of str) – Roles that take several files, kept in the order given (time-lapse surveys, say).
folder_classifier (bool) – Whether the workflow accepts mode='classify' to sort a chosen folder’s files into roles with the model.
studio_modules (tuple of str) – The studio modules its chat may drive (keys of qt_apps.modules.MODULE_SPECS); empty for all of them.
colors (tuple of str) – Four colours for its glow and orb; empty for the studio’s own.
requires_packages (tuple of str) – Importable packages it needs, checked by availability().
status (str) – 'ready', or 'in development' while it cannot run yet.
status_note (str) – Shown when it is not available.
providers (tuple of str) – Model providers its workflow supports.
offline_workflow (bool) – Whether the workflow can produce useful results without model access. Enables an explicit “Run without AI” action in the Workflow page.

availability() → Tuple[bool, str][source]#: Whether it can run here, and why not when it cannot.

colors: Tuple[str, ...] = ()#

domain: str#

examples: Tuple[str, ...] = ()#

focused_workspace: bool = False#

folder_classifier: bool = False#

input_roles: Tuple[Tuple[str, str], ...] = ()#

key: str#

load_tools() → Dict[str, Any][source]#: Its tool registry, name to Tool; empty when it declares none.

load_workflow() → Callable[[...], Dict[str, Any]][source]#: The workflow function, imported on first use.

name: str#

offline_workflow: bool = False#

ordered_roles: Tuple[str, ...] = ()#

persona: str = ''#

providers: Tuple[str, ...] = ('openai', 'anthropic')#

requires_packages: Tuple[str, ...] = ()#

retrieval: Tuple[str, ...] = ('rag', 'mcp')#

status: str = 'ready'#

status_note: str = ''#

studio_modules: Tuple[str, ...] = ()#

summary: str#

title: str#

tool_for_label(label: str) → str[source]#: The name of the tool whose label (or name) is label, or “”.

tools: str = ''#

workflow: str#

workflow_setup: str = ''#

PyHydroGeophysX.agents.assistants.BUILTIN: Tuple[str, ...] = ('PyHydroGeophysX.agents.assistants.aquah:ASSISTANT', 'PyHydroGeophysX.agents.assistants.geosage:ASSISTANT')#: The assistants that ship with PyHydroGeophysX, as module:attribute.

PyHydroGeophysX.agents.assistants.DEFAULT_KEY = 'aquah'#: The assistant the studio starts with.

PyHydroGeophysX.agents.assistants.ENTRY_POINT_GROUP = 'pyhydrogeophysx.assistants'#: Where installed packages register further assistants.

PyHydroGeophysX.agents.assistants.WORKFLOW_CONTRACT = "run(payload, progress, *, approve=None, on_event=None, events=None, **hooks) -> dict\n\npayload the run as the studio starts it:\n request the user's goal, in their words\n inputs {role: path or [paths]}, roles from Assistant.input_roles\n provider, model, api_key, reasoning_effort model access\n output_dir where everything the run writes goes\n step_mode True to pause before each step (pass approve on to drive)\n use_rag, use_mcp\n mode 'classify' with data_folder, when the assistant offers\n folder classification\nprogress progress(step, fraction, details, module='') for the status line\napprove approve(event) -> 'proceed' | 'skip' | 'stop', or a question's option\n id; blocks until the user answers. Pass to drive() as on_step via\n runtime.step_by_step when payload['step_mode'] is set, and as ask_user.\non_event structured step events; pass straight to drive(on_event=...)\nevents list the progress events are being collected in (for an audit)\n\nreturns {'status': 'success' | 'needs_review' | 'incomplete',\n 'interpretation': str, first lines shown in the chat\n 'warnings': [str],\n 'report_files': {'report_markdown': path, ...},\n 'output_dir': str, 'execution_plan': [...], 'workflow_config': {...}}\n or, for mode='classify', {'status': 'classified', 'catalog': {...}}\nraises for a run that cannot start or produced nothing; the message is shown\n"#: What a workflow function is given and must return. Written out here because it is the one interface an assistant has to honour exactly.

PyHydroGeophysX.agents.assistants.accepted_hooks(fn: Callable[[...], Any], **hooks: Any) → Dict[str, Any][source]#

The hooks that fn takes, by name or through **kwargs.

A run function that predates them - an older caller, or a stub in a test - still runs; it simply cannot pause or ask. One whose signature cannot be read is given them all.

Examples

>>> accepted_hooks(lambda a, on_step=None: a, on_step=1, ask_user=2)
{'on_step': 1}
>>> accepted_hooks(lambda a, **kw: a, on_step=1, ask_user=2)
{'on_step': 1, 'ask_user': 2}

PyHydroGeophysX.agents.assistants.active() → Assistant[source]#: The assistant the user has chosen in this process.

PyHydroGeophysX.agents.assistants.assistants() → List[Assistant][source]#: Every registered assistant, built-ins first.

PyHydroGeophysX.agents.assistants.get_assistant(key: str | None = None) → Assistant[source]#

The assistant key (the active one when omitted).

Raises:: KeyError – For a key nobody registered, naming the ones that were.

PyHydroGeophysX.agents.assistants.load_errors() → List[str][source]#: Plug-ins that failed to load, with the reason.

PyHydroGeophysX.agents.assistants.register_assistant(assistant: Assistant) → Assistant[source]#: Add assistant to the registry, replacing one with the same key.

PyHydroGeophysX.agents.assistants.set_active(key: str) → Assistant[source]#: Choose the assistant this process works with; returns it.