Deployment note: the agent layer runs in the AI bundled (edge) and AI separate (cloud) topologies; Historian-only installs run no AI service (see Overview § Technical topologies).
The Pulse Agentic AI layer is a Google ADK multi-agent system. The root pulse_manager agent orchestrates 6 specialized sub-agents via tool delegation.
index.pySource: pulse_multi_agents:index.py
google.adk.cli.fast_api.get_fast_api_appPORT, default 8001)agents_dir: directory of index.py (so ADK discovers pulse_manager/ as the agent package)sqlite:///<abs_db_path>) or PostgreSQL (postgresql+asyncpg://...) based on USE_POST_GRESQL_DB env var["http://localhost", "http://localhost:8080", "*"]pulse_manager.plugins.incident_retry_plugin.IncidentValidationRetryPlugin/run_sse endpoint overridden: validates RunAgentUserRequest schema before delegating to ADK handler (pulse_multi_agents:index.py:234-266)/health: returns {"status": "ok"}PureASGIHeartbeatMiddleware wraps all /run_sse connections:
ACTIVITY_SNAPSHOT type)pulse_multi_agents:index.py:78-187pulse_multi_agents:index.py:47-69:
pulse_manager_path (default ./sessions.db) → PAPA_AGENT_DB_PATHsqlite:///<abs_db_path>POSTGRES_HOST/PORT/DB/USER/PASSWORD env varsclarity:backend/src-tauri/src/api/google_adk.rs adds warp routes that forward POST requests to localhost:8000 (where the bundled papa_agent_app binary listens).
pulse_managerSource: pulse_multi_agents:pulse_manager/agent.py
root_agent = Agent(
name="pulse_manager",
model=model,
instruction=get_root_instruction, # dynamic: prepends current UTC time
tools=valid_tools,
before_agent_callback=inject_auth_credentials,
before_tool_callback=handle_incident_reset,
)
pulse_multi_agents:pulse_manager/agent.py:316-324
| Tool | Type | Purpose |
|---|---|---|
meta_data_tool |
AgentTool(meta_data_agent) |
Tag/unit lookup, data fetching, CSV output |
data_analysis_tool |
AgentTool(data_analysis_agent) |
Statistical analysis of CSV data |
question_tool |
AgentTool(question_agent) |
Generate diagnostic questions |
dashboard_creation_tool |
AgentTool(dashboard_creation_agent) |
Create dashboard section definitions |
system_health_tool |
AgentTool(system_health_agent) |
Diagnose a specific system by unit_id + system_name |
incident_summary_tool |
AgentTool(incident_agent) |
Summarize a specific incident |
resolve_unit_tool |
FunctionTool(resolve_unit) |
Map unit name → unit_id |
resolve_unit_site_choice_tool |
FunctionTool(resolve_unit_site_choice) |
Disambiguate site for unit resolution |
fetch_unit_systems_tool |
FunctionTool(fetch_unit_systems) |
List systems for a resolved unit_id |
pulse_multi_agents:pulse_manager/agent.py:52-80
CLARITY_APP mode (standalone binary): system_health_tool, incident_summary_tool, and fetch_unit_systems_tool are excluded from valid_tools.
inject_auth_credentials (before_agent): on the first invocation — guarded by the module-global HEADERS_STATUS — calls initialize_headers(...) to set up the BASE_URL/AUTH_TOKEN used for backend calls, then short-circuits on later invocations. The docstring describes two intended modes — credentials from session state (Kubernetes/prod) vs .env fallback (local dev) — but the session-state force-reinit branch is currently commented out (agent.py:257-271); the active path initializes once via initialize_headers(None, callback_context).
pulse_multi_agents:pulse_manager/agent.py:197-277
handle_incident_reset (before_tool): resets incident retry counter when the invocation_id changes (new user message), preserving retry state within a single turn.
pulse_multi_agents:pulse_manager/agent.py:291-313
pulse_manager_inst.pySource: pulse_multi_agents:pulse_manager/pulse_manager_inst.py
The instruction set enforces strict orchestration discipline. Key rules:
Intent classification: L1 (conversational/meta) → answer directly; L2 (system diagnosis) → resolve unit → fetch systems → call system_health_agent; L4 (incident diagnosis) → resolve unit → fetch incidents → call incident_agent.
Unit resolution: resolve_unit(unit_name) must be called before any unit-dependent action. Never guess unit_id. If resolve_unit returns a clarification question, relay verbatim to user and stop.
Delegation rules: meta_data_agent fetches data; data_analysis_agent analyzes CSV; incident_agent summarizes incidents; system_health_agent takes (unit_id, system_name) only. Never perform computation in the root agent.
Anti-hallucination rules: no pre-existing plant data; every factual statement about an incident or system must come from a tool response in the current turn.
Dashboard suggestion mode: retrieves last 30 days of data via meta_data_agent, analyzes via data_analysis_agent, then recommends from 5 viz types: lines, histogram, scatter, table, comparison.
State column grouping: when state columns (state__1, state__2, etc.) are present, tags must be analyzed only within their validity mask group (rows where state_col == 1).
pulse_multi_agents:pulse_manager/pulse_manager_inst.py:349-714
Source: pulse_multi_agents:pulse_manager/sub_agents/meta_data_agent/agent.py
Registers four FunctionTools (agent.py:1471-1476):
| Tool | Purpose | Source |
|---|---|---|
fetch_data |
Fetch time-series for tags over a natural-language time range → session-isolated CSV; returns the CSV path | agent.py:349-515 / 521-735 |
df_operations_tool |
Apply filter conditions to the session CSV and overwrite it in place |
agent.py:773-829 |
getTagMeta |
Look up tag metadata for a unitsId/assetId/description list via LoopBack like queries against {meta}/tagmeta; returns ≤10 rows and injects stateDataTagId |
agent.py:890-1052 |
getIncidents |
List incidents for a unitsId ({meta}/incidents), dropping noise=true entries |
agent.py:1066-1117 |
NOTE:
getDeviations(agent.py:1120-1167) is defined but not registered as a tool.
CLARITY_APP data-source switch (agent.py:1262-1265): fetch_data is bound to fetch_data_clarity when CLARITY_APP is set — posts a per-tag pipeline payload to Clarity's query API, auth via login_clarity() — otherwise to fetch_data_prod, which posts to KairosDB (config["api"]["query"]) with HTTP Basic auth from KAIROS_USERNAME/KAIROS_PASSWORD.
Per-session isolation: a before_agent_callback=inject_session_id (agent.py:833-861) stores the real _session_id in tool state. The fetch tools write {session_id}_datafetch.csv and reuse it across invocations via the session_data_file / last_csv_path state keys (agent.py:435-510), so data never mixes across sessions.
NOTE: The
from .tag_resolver import get_tag_resolverimport is commented out (agent.py:1170); the registeredgetTagMetatool resolves tags through metadatalikequeries, not the sentence-transformers resolver. The agent's instruction prose still references "AI embeddings" / "semantic search". Whethertag_resolver.pyis invoked elsewhere is not confirmed here — see Tag Resolver.
Source: pulse_multi_agents:pulse_manager/sub_agents/data_analysis_agent/agent.py
Responsibilities:
NOTE: This agent does NOT fetch data. It only analyzes CSV files produced by meta_data_agent.
Source: pulse_multi_agents:pulse_manager/sub_agents/incident_agent/agent.py
Pipeline: IncidentInitializer → analyzer_agent → assessment_agent
IncidentValidationRetryPlugin)analyzer_agent: examines deviating tags, generates base64 diagnostic plots, includes historical recurrence context. Prompt now describes the long-term plots as 8 months (was 1 year). (pulse_multi_agents:pulse_manager/sub_agents/incident_agent/analyzer_agent.py:104,112)assessment_agent: synthesizes analyzer output into executive diagnostic text (pulse_multi_agents:pulse_manager/sub_agents/incident_agent/assessment_agent.py)pulse_multi_agents:pulse_manager/sub_agents/incident_agent/utils.py)Bulk metadata prefetch (perf, 7cdfec7). build_incident_context now issues one bulk fetch_unit_tagmeta(unit_id, deviating_tags, merged_fields) call (inq over all deviating tags) into a tagmeta_lookup, and fetches equipment once (fetch_equipment(equipment_ids[0])). These are passed into generate_base64_plot(... tagmeta=, equipment=), replacing the previous per-tag fetch_tag_limits calls — limits/units/description now come straight from the prefetched tagmeta. get_deviating_tags also gained an incident= param so the already-fetched incident is reused instead of re-fetched. pulse_multi_agents:pulse_manager/sub_agents/incident_agent/utils.py:367-525
Plot window. The incident long-term plot is now 8 months = 350400 minutes (was 525600 = 1 year). pulse_multi_agents:pulse_manager/sub_agents/incident_agent/utils.py:436
Plot generation goes through the shared
generate_base64_plothelper — see Diagnostic plots. Artifacts are saved as.webp/image/webp.pulse_multi_agents:pulse_manager/sub_agents/incident_agent/utils.py:511-516
Source: pulse_multi_agents:pulse_manager/sub_agents/system_health_agent/agent.py
Responsibilities:
(unit_id, system_name) — no free-form input525600 min) diagnostic plots per incident, capped at 5 concurrent plot callsBulk metadata prefetch (perf, 7cdfec7). Like the incident agent, build_system_health_context collects the primary tag of every current-system incident, issues one bulk fetch_unit_tagmeta(unit_id, primary_tags, merged_fields) into a tagmeta_lookup, and prefetches equipment per incident into an equipment_lookup. These are injected into _bounded_plot_gen → generate_base64_plot(... tagmeta=, equipment=) and used directly as inc["tagLimits"], replacing the previous concurrent fetch_tag_limits gather. Plot artifacts are saved as .webp / image/webp. pulse_multi_agents:pulse_manager/sub_agents/system_health_agent/agent.py:109-253
Source: pulse_multi_agents:pulse_manager/sub_agents/dashboard_creation_agent/agent.py
Responsibilities:
lines, histogram, scatter, table, comparisonSource: pulse_multi_agents:pulse_manager/sub_agents/question_agent/agent.py
Responsibilities:
generate_base64_plotSource: pulse_multi_agents:pulse_manager/utilis.py
Shared by the incident and system-health agents. generate_base64_plot(incident_id, ct, pad_minutes, context, incident_data=None, tagmeta=None, equipment=None) builds an SPC payload via build_spc_payload_from_incident and POSTs it to the clarity datacenter to render a chart, returning the base64 image.
7cdfec7). The plot endpoint is now {public_datacenter_url}/sensordata/spcplot/matplotlib (was /sensordata/spcplot) and the response is expected as image/webp (was image/png); the returned dict carries mime_type: image/webp. A non-image/webp content type is treated as an error. pulse_multi_agents:pulse_manager/utilis.py:884-958generate_base64_plot and build_spc_payload_from_incident accept optional tagmeta= and equipment= arguments; when supplied (by the agents' bulk prefetch) the per-plot fetch_unit_tagmeta / fetch_equipment lookups are skipped. pulse_multi_agents:pulse_manager/utilis.py:721-749, 884fetch_unit_tagmetafetch_unit_tagmeta(units_id, data_tag_id, fields=None, context=None) now accepts either a single dataTagId (returns one dict, as before) or a list of IDs — in which case it queries {"dataTagId": {"inq": [...]}} against {meta}/units/{units_id}/tagmeta and returns a list ([] on miss). The LoopBack filter is now passed as a request param rather than interpolated into the URL. This is what lets the agents replace N per-tag lookups with one call. pulse_multi_agents:pulse_manager/utilis.py:570-600
get_modelget_model(fast=False) builds the Gemini model (gemini-3-flash-preview) with 3 retry attempts. As of 7cdfec7 it also sets http_options=types.HttpOptions(timeout=600_000) (600 s) to tolerate long agent turns. pulse_multi_agents:pulse_manager/utilis.py:444-470
Important: Two files are named
tag_resolverbut serve entirely different purposes.
| File | Purpose | Mechanism |
|---|---|---|
clarity:backend/src-tauri/src/api/tag_resolver.rs |
Filesystem scope mapper | Scans data/ directory structure to build TAG_SCOPE_MAP (tagname → Arc<Scope>) |
pulse_multi_agents:pulse_manager/sub_agents/meta_data_agent/tag_resolver.py |
Semantic NLP tag resolver | Uses sentence-transformers to match natural language descriptions to tag names; 24h TTL cache |
These are unrelated despite the same filename. See Tag Resolver for full details.
Source: pulse_multi_agents:pulse_manager/paths_config.py
Manages runtime directory paths for CSV data, plot images, and code logs. Handles both dev mode and PyInstaller binary mode. Implements TTL-based cleanup of old session files.
Session cleanup functions cleanup_session_files() and cleanup_old_csv_files() are called at startup to remove orphaned files from previous sessions.
pulse_multi_agents:pulse_manager/agent.py:83-194
Source: pulse_multi_agents:pulse_manager/plugins/incident_retry_plugin.py
ADK plugin that validates incident summaries for hallucinations (e.g., fabricated incident details). Manages per-invocation retry-count state. Triggers retry when validation fails, up to max_retries within a single user turn.
HTTP request to SSE response, including sub-agent dispatch and incident retry path.
Sources: pulse_multi_agents:index.py, pulse_multi_agents:pulse_manager/agent.py
Note on CLARITY_APP mode: When running as standalone binary, system_health_tool, incident_summary_tool, and fetch_unit_systems_tool are excluded from valid_tools. The incident path and system health path in the diagram are not available in that mode.
Diagram generated from source. If implementation has changed, update the Mermaid source in this file directly.
Last updated: 2026-06-22 from pulse_multi_agents@7cdfec7