| """LLM-based result interpretation with structured output.""" |
| import json |
| import yaml |
| from pathlib import Path |
| from typing import Any, Dict |
| from agent.llm_provider import get_provider |
| from agent.schema import Interpretation |
|
|
|
|
| def load_interpret_prompt() -> str: |
| """Load interpreter system prompt.""" |
| prompt_path = Path("agent/prompts/system_interpret.txt") |
| return prompt_path.read_text() |
|
|
|
|
| def load_heuristics() -> Dict[str, Any]: |
| """Load interpretation thresholds from config/heuristics.yaml.""" |
| heuristics_path = Path("config/heuristics.yaml") |
| with open(heuristics_path) as f: |
| return yaml.safe_load(f) |
|
|
|
|
| def interpret_results( |
| tool_name: str, |
| sample_title: str, |
| metrics: Dict[str, float], |
| extras: Dict[str, Any] |
| ) -> Interpretation: |
| """ |
| Generate LLM interpretation of EO results. |
| |
| Args: |
| sample_title: Human-readable sample title |
| metrics: Computed metrics (min, max, mean, std, percentiles, etc.) |
| extras: Task-specific data (e.g., severity class counts) |
| |
| Returns: |
| Interpretation with headline, bullets, and caveats |
| """ |
| provider = get_provider() |
| system_prompt = load_interpret_prompt() |
| heuristics = load_heuristics() |
|
|
| tool_heuristic_map = { |
| "run_ndvi": "ndvi", |
| "run_lst_landsat": "lst", |
| "run_dnbr_s2pair": "dnbr" |
| } |
|
|
| thresholds = heuristics.get(tool_heuristic_map[tool_name], {}) |
|
|
| user_json = { |
| "sample_title": sample_title, |
| "metrics": metrics, |
| "extras": extras, |
| "thresholds": thresholds, |
| "style": { |
| "language": "en", |
| "max_sentences": 10 |
| } |
| } |
|
|
| try: |
| response_text = provider.interpret_call(system_prompt, user_json) |
|
|
| |
| |
| response_text = response_text.strip() |
| if response_text.startswith("```"): |
| response_text = response_text.split("```")[1] |
| if response_text.startswith("json"): |
| response_text = response_text[4:] |
|
|
| response_json = json.loads(response_text) |
| return Interpretation(**response_json) |
|
|
| except Exception as e: |
| |
| return Interpretation( |
| headline=f"Analysis completed for {sample_title}", |
| bullets=[f"Mean value: {metrics.get('mean', 0):.3f}"], |
| caveats=[f"Interpretation error: {str(e)}"] |
| ) |
|
|
