""" 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)