Summarize stage traces into research-facing direction/strategy memories.
| 8 | |
| 9 | |
| 10 | class MemoryEvolutionAnalyzer: |
| 11 | """Summarize stage traces into research-facing direction/strategy memories.""" |
| 12 | |
| 13 | def __init__(self, store: MemoryStore) -> None: |
| 14 | self.store = store |
| 15 | |
| 16 | @staticmethod |
| 17 | def _compact(text: str, *, limit: int = 320) -> str: |
| 18 | text = " ".join((text or "").strip().split()) |
| 19 | if len(text) <= limit: |
| 20 | return text |
| 21 | return text[: limit - 3].rstrip() + "..." |
| 22 | |
| 23 | @staticmethod |
| 24 | def _ensure_dict(payload: Any) -> dict[str, Any]: |
| 25 | if payload is None: |
| 26 | return {} |
| 27 | if isinstance(payload, dict): |
| 28 | return payload |
| 29 | if hasattr(payload, "model_dump"): |
| 30 | return payload.model_dump(mode="json") |
| 31 | return {} |
| 32 | |
| 33 | @staticmethod |
| 34 | def _first_non_empty(*values: Any) -> str: |
| 35 | for value in values: |
| 36 | if isinstance(value, str) and value.strip(): |
| 37 | return value.strip() |
| 38 | return "" |
| 39 | |
| 40 | @staticmethod |
| 41 | def _top_strings(items: list[Any], key: str, *, limit: int = 3) -> list[str]: |
| 42 | results: list[str] = [] |
| 43 | for item in items[:limit]: |
| 44 | if isinstance(item, dict): |
| 45 | value = str(item.get(key, "")).strip() |
| 46 | if value: |
| 47 | results.append(value) |
| 48 | return results |
| 49 | |
| 50 | def _summarize_round_trajectory(self, rounds: list[Any], *, limit: int = 4) -> list[str]: |
| 51 | trajectory: list[str] = [] |
| 52 | for round_data in rounds[:limit]: |
| 53 | if not isinstance(round_data, dict): |
| 54 | continue |
| 55 | round_number = round_data.get("round_number", "?") |
| 56 | preflight = self._ensure_dict(round_data.get("preflight")) |
| 57 | metrics = self._ensure_dict(round_data.get("metrics")) |
| 58 | analysis = self._ensure_dict(round_data.get("analysis")) |
| 59 | pieces = [f"round {round_number}"] |
| 60 | if preflight.get("overall_status"): |
| 61 | pieces.append(f"preflight={preflight.get('overall_status')}") |
| 62 | if metrics: |
| 63 | metric_pairs = ", ".join(f"{key}={value}" for key, value in list(metrics.items())[:2]) |
| 64 | pieces.append(f"metrics[{metric_pairs}]") |
| 65 | if analysis.get("attribution"): |
| 66 | pieces.append(f"attribution={analysis.get('attribution')}") |
| 67 | if analysis.get("recommended_action"): |