Spaces:
Sleeping
Sleeping
File size: 8,379 Bytes
199bfa3 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 | """
Post-Execution Result Interpreter — "Murphy's Take"
After the Advanced Explorer executes an analytical plan and produces raw
tables/metrics, this module calls the LLM to generate a plain-English
interpretation grounded in cryogenic physics context.
"""
import pandas as pd
from typing import Optional
from core import config
from analysis.llm_analyzer import MURPHY_SYSTEM_PROMPT, sanitize_llm_output
class ResultInterpreter:
"""Generate plain-English interpretation of Advanced Explorer results.
Uses the shared Murphy system prompt with Anthropic prompt caching.
The system prompt is cached for 5 minutes across repeated calls,
so interpreting multiple query results in a session is fast.
"""
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 interpret(self, user_query: str, result) -> str:
"""Generate Murphy's interpretation of execution results.
Args:
user_query: The original natural language query from the user.
result: An ExecutionResult from the execution engine.
Returns:
Plain-English interpretation string, or an error message.
"""
if not self.api_available:
return "LLM interpretation unavailable. Add ANTHROPIC_API_KEY to .env."
prompt = self._build_interpretation_prompt(user_query, result)
try:
message = self.client.messages.create(
model=self.model,
max_tokens=1500,
system=[
{
"type": "text",
"text": MURPHY_SYSTEM_PROMPT,
"cache_control": {"type": "ephemeral"},
}
],
messages=[{"role": "user", "content": prompt}],
)
return sanitize_llm_output(message.content[0].text)
except Exception as e:
return f"Interpretation failed: {e}"
def _build_interpretation_prompt(self, user_query: str, result) -> str:
"""Build the prompt from execution results.
Serializes operation results compactly — key metrics, first N rows,
summary stats — to keep token usage reasonable while giving the LLM
enough context for a physics-grounded interpretation.
"""
sections = []
# Header
sections.append(f'The engineer asked: "{user_query}"')
sections.append(f"Query type: {result.query_type}")
sections.append(f"Plan explanation: {result.explanation}")
# Data context
sections.append(
f"\nData fetched: {result.row_count:,} rows from {result.table_used or 'unknown table'}"
)
# Source data summary (compact)
if result.source_data is not None and not result.source_data.empty:
src = result.source_data
if "timestamp" in src.columns:
ts = src["timestamp"]
sections.append(
f"Time range: {ts.min()} to {ts.max()}"
)
numeric_cols = [
c for c in src.columns
if c != "timestamp" and pd.api.types.is_numeric_dtype(src[c])
]
if numeric_cols:
stats_lines = []
for col in numeric_cols[:8]: # Cap at 8 sensors
col_data = src[col].dropna()
if len(col_data) > 0:
stats_lines.append(
f" {col}: min={col_data.min():.3f}, "
f"max={col_data.max():.3f}, "
f"mean={col_data.mean():.3f}, "
f"std={col_data.std():.3f}"
)
if stats_lines:
sections.append("Source data summary:\n" + "\n".join(stats_lines))
# Operation results (compact serialization)
for i, op_result in enumerate(result.operation_results):
op_section = [f"\n--- Step {i + 1}: {op_result.label} ---"]
if op_result.metadata.get("error"):
op_section.append(f" ERROR: {op_result.metadata['error']}")
sections.append("\n".join(op_section))
continue
# Key metadata (non-error)
meta = {k: v for k, v in op_result.metadata.items() if k != "error"}
if meta:
for k, v in meta.items():
op_section.append(f" {k}: {v}")
data = op_result.data
# Dict result (pump strokes, statistics, blocks summary)
if isinstance(data, dict):
for k, v in data.items():
if isinstance(v, (int, float, str, bool)):
op_section.append(f" {k}: {v}")
elif isinstance(v, list) and len(v) > 0:
op_section.append(f" {k}: {len(v)} items")
# Show first few items if they're dicts
if isinstance(v[0], dict):
for item in v[:5]:
op_section.append(f" {item}")
# List-of-dicts result (fills, periods, ramps)
elif isinstance(data, list) and data and isinstance(data[0], dict):
op_section.append(f" Found {len(data)} results")
# Show first 10 rows
for row in data[:10]:
row_str = ", ".join(
f"{k}={v}" for k, v in row.items()
)
op_section.append(f" {row_str}")
if len(data) > 10:
op_section.append(f" ... and {len(data) - 10} more")
# List-of-DataFrames result (extract_windows)
elif isinstance(data, list) and data and isinstance(data[0], pd.DataFrame):
op_section.append(f" Extracted {len(data)} windows")
for j, window_df in enumerate(data[:5]):
op_section.append(
f" Window {j + 1}: {len(window_df)} rows, "
f"cols: {', '.join(window_df.columns[:6])}"
)
# Single DataFrame result
elif isinstance(data, pd.DataFrame):
op_section.append(f" DataFrame: {len(data)} rows × {len(data.columns)} cols")
numeric = [
c for c in data.columns
if c != "timestamp" and pd.api.types.is_numeric_dtype(data[c])
]
for col in numeric[:5]:
col_data = data[col].dropna()
if len(col_data) > 0:
op_section.append(
f" {col}: min={col_data.min():.3f}, "
f"max={col_data.max():.3f}, "
f"mean={col_data.mean():.3f}"
)
sections.append("\n".join(op_section))
# Warnings
if result.warnings:
sections.append("\nWarnings: " + "; ".join(result.warnings))
# Errors
if result.errors:
sections.append("\nErrors: " + "; ".join(result.errors))
# Benchmarks reference
sections.append("""
REFERENCE BENCHMARKS (CSH2 Phase 1B):
- Specific energy: 0.26 kWh/kg (vs 4.5-5 kWh/kg conventional)
- Flow rate: 1.13-1.25 kg/min at 65% motor power
- Peak pressure: 370 bar achieved
- Delivered temperature: 55K minimum
- Uptime: 100% across 6 fills""")
# Final instruction
sections.append("""
Based on the above results, provide Murphy's interpretation in 3-5 concise sentences:
1. What do the results mean in physical/engineering terms?
2. Flag anything unusual or noteworthy.
3. Compare to Phase 1B benchmarks where applicable.
Be direct and technical. Focus on the "so what" — what should the engineer take away from these results.""")
return "\n".join(sections)
|