pandeydigant31's picture
Initial deploy: DELPHI Pump Analytics (renamed from MURPHY)
199bfa3 verified
Raw
History Blame Contribute Delete
8.42 kB
"""
Natural Language Query Parser for Data Explorer
Uses Claude API to parse engineer queries into structured sensor + time range selections.
Adapted from clearskies_agent_v1.1/llm_parser.py
"""
import json
from typing import Dict, Optional, List
from datetime import datetime
from core import config
class NLQueryParser:
"""Parses natural language queries into structured sensor + time range selections"""
def __init__(self):
self.api_key = config.ANTHROPIC_API_KEY
self.api_available = bool(self.api_key and self.api_key != 'your_api_key_here')
self.model = "claude-sonnet-4-20250514"
self._client = None
@property
def client(self):
if self._client is None and self.api_available:
import anthropic
self._client = anthropic.Anthropic(api_key=self.api_key)
return self._client
def parse(
self,
user_query: str,
available_sensors: List[str],
) -> Optional[Dict]:
"""
Parse a natural language query into structured parameters.
Args:
user_query: Natural language query from the engineer
available_sensors: List of available sensor tag names in the database
Returns:
Dict with keys:
sensors: List[str] - matched sensor tag names
start_time: datetime - query start time
end_time: datetime - query end time
explanation: str - human-readable interpretation
Or None if parsing fails.
"""
if not self.api_available:
return None
static_instructions = self._build_static_instructions(available_sensors)
dynamic_query = f'User query: "{user_query}"\n\nReturn ONLY the JSON object, no other text.'
try:
message = self.client.messages.create(
model=self.model,
max_tokens=1000,
system=[
{
"type": "text",
"text": static_instructions,
"cache_control": {"type": "ephemeral"},
}
],
messages=[{"role": "user", "content": dynamic_query}],
)
response_text = message.content[0].text
parsed = self._extract_json(response_text)
if parsed:
return self._process_result(parsed, available_sensors)
return None
except Exception as e:
return {'error': str(e)}
def _build_static_instructions(self, available_sensors: List[str]) -> str:
"""Build the cacheable static instruction block with sensor context.
This is separated from the dynamic user query so the Anthropic API
can cache this large, stable prefix across repeated calls within
the same session (5-minute TTL).
"""
# Build sensor reference with descriptions
sensor_lines = []
for tag_name, info in config.SENSOR_TAGS.items():
desc = info.get('description', '')
units = info.get('units', '')
sensor_lines.append(f" {tag_name}: {desc} ({units})")
sensor_reference = "\n".join(sensor_lines)
# Build sensor group reference
group_lines = []
for group_name, tags in config.SENSOR_GROUPS.items():
group_lines.append(f" {group_name}: {', '.join(tags)}")
group_reference = "\n".join(group_lines)
current_date = datetime.now()
return f"""You are a query parser for a ClearSkies Hydrogen (CSH2) cryopump testing data system. Parse the engineer's natural language query into structured JSON for data retrieval.
Current date/time: {current_date.strftime('%Y-%m-%d %H:%M:%S')}
Data is available from March 2025 to September 2025.
AVAILABLE SENSORS (tag name: description (units)):
{sensor_reference}
SENSOR GROUPS (for category-based queries):
{group_reference}
COMMON ALIASES:
- "pressure" or "pressures" -> PT130 (Discharge), PT01T (Cryotank), PT110 (Inlet)
- "temperature" or "temperatures" -> TT110 (Pump Feed Line), TT130 (Discharge)
- "flow" -> FT140 (Flow Transmitter)
- "motor" or "speed" or "rpm" -> M130_Speed (Motor Speed)
- "power" -> AvgPower, PeakPower, PowerConsumption
- "discharge pressure" -> PT130
- "inlet pressure" or "supply pressure" -> PT110
- "tank pressure" or "cryotank pressure" -> PT01T
- "discharge temperature" -> TT130
- "feed temperature" or "pump temperature" -> TT110
- "valve" -> AOV140
- "density" -> N2Density, H2Density
- "level" -> LT200
- "all sensors" or "everything" -> PT130, PT01T, PT110, TT110, TT130, FT140, M130_Speed
Parse queries into a JSON object with these fields:
{{
"sensors": ["PT130", "TT110"],
"start_time": "2025-09-25T10:00:00",
"end_time": "2025-09-25T14:00:00",
"explanation": "Discharge pressure and pump feed temperature from Sept 25, 10 AM to 2 PM"
}}
RULES:
- Match sensor names to the available sensors list (case-insensitive)
- Use the aliases above to resolve common terms like "pressure", "temperature" etc.
- For relative dates like "yesterday", "last week", calculate based on current date
- For date-only queries without specific times, use 00:00:00 for start and 23:59:59 for end
- For "between X and Y" time references, parse both times
- Use ISO 8601 format for timestamps (YYYY-MM-DDTHH:MM:SS)
- If a sensor group is mentioned (e.g., "all pressures"), include all sensors from that group
- The explanation should be a concise human-readable summary of what was parsed
- If no sensors are explicitly mentioned, try to infer from context, or default to PT130, TT110, FT140
- If no time range is mentioned, default to the most recent day (today)"""
def _build_prompt(self, user_query: str, available_sensors: List[str]) -> str:
"""Build the full Claude prompt (used by fallback path).
The main parse() method uses _build_static_instructions() + dynamic query
separately for prompt caching. This method is kept for backward compatibility.
"""
static = self._build_static_instructions(available_sensors)
return f"""{static}
User query: "{user_query}"
Return ONLY the JSON object, no other text."""
def _extract_json(self, text: str) -> Optional[Dict]:
"""Extract JSON object from LLM response"""
try:
start = text.find('{')
end = text.rfind('}') + 1
if start != -1 and end > start:
json_str = text[start:end]
parsed = json.loads(json_str)
if 'sensors' in parsed:
return parsed
return None
except json.JSONDecodeError:
return None
def _process_result(self, parsed: Dict, available_sensors: List[str]) -> Optional[Dict]:
"""Validate and process the parsed result"""
try:
# Convert time strings to datetime
start_time = self._convert_to_datetime(parsed.get('start_time'))
end_time = self._convert_to_datetime(parsed.get('end_time'))
if not start_time or not end_time:
return {'error': 'Could not parse time range from query.'}
# Validate sensors exist in database
raw_sensors = parsed.get('sensors', [])
valid_sensors = []
# Case-insensitive match against available sensors
sensor_map = {s.upper(): s for s in available_sensors}
for s in raw_sensors:
matched = sensor_map.get(s.upper())
if matched:
valid_sensors.append(matched)
if not valid_sensors:
return {'error': f'No matching sensors found. Requested: {", ".join(raw_sensors)}'}
return {
'sensors': valid_sensors,
'start_time': start_time,
'end_time': end_time,
'explanation': parsed.get('explanation', ''),
}
except Exception as e:
return {'error': f'Processing failed: {e}'}
def _convert_to_datetime(self, iso_string: Optional[str]) -> Optional[datetime]:
"""Convert ISO string to datetime object"""
if not iso_string:
return None
try:
return datetime.fromisoformat(iso_string)
except (ValueError, TypeError):
return None