| import json |
| import re |
| from pathlib import Path |
| from typing import List, Optional, Union |
|
|
|
|
| def _extract_recommendation_items(text: str) -> List[str]: |
| """Extract compact recommendation items from numbered or bulleted text.""" |
| if not isinstance(text, str): |
| return [] |
|
|
| items: List[str] = [] |
| for raw_line in text.splitlines(): |
| line = raw_line.strip() |
| if not line: |
| continue |
| if re.match(r"^\d+\.\s+", line) or line.startswith("- ") or line.startswith("•"): |
| items.append(line) |
| return items |
|
|
|
|
| def build_meta_recommendation_context_lines( |
| results_dir: Union[str, Path], |
| max_current_items: int = 5, |
| max_delta_items: int = 3, |
| ) -> Optional[List[str]]: |
| """ |
| Build prompt lines describing current evolution-agent strategy from meta memory. |
| |
| Reads `meta_memory.json` under results_dir and returns a compact section with: |
| - current recommendations used by the evolution loop |
| - best-effort delta vs the previous recommendation refresh |
| """ |
| base_path = Path(results_dir) |
| if base_path.name == "meta_memory.json": |
| meta_path = base_path |
| else: |
| meta_path = base_path / "meta_memory.json" |
| if not meta_path.exists(): |
| return None |
|
|
| try: |
| with open(meta_path, "r", encoding="utf-8") as f: |
| meta_data = json.load(f) |
| except Exception: |
| return None |
|
|
| current_recs = meta_data.get("meta_recommendations") |
| history = meta_data.get("meta_recommendations_history", []) |
| if not isinstance(current_recs, str) or not current_recs.strip(): |
| return None |
|
|
| latest_items = _extract_recommendation_items(current_recs) |
| previous_recs = history[-2] if isinstance(history, list) and len(history) >= 2 else "" |
| previous_items = set(_extract_recommendation_items(previous_recs)) |
| improvement_items = [item for item in latest_items if item not in previous_items] |
|
|
| context_lines: List[str] = [ |
| "", |
| "🧭 Evolution Agent Strategy Context (from meta memory):", |
| f"- Source: {meta_path}", |
| "- Current recommendations used by evolution agent:", |
| ] |
|
|
| if latest_items: |
| for item in latest_items[:max_current_items]: |
| context_lines.append(f" {item}") |
| else: |
| trimmed = current_recs.strip().replace("\n", " ") |
| if len(trimmed) > 1200: |
| trimmed = trimmed[:1200] + "..." |
| context_lines.append(f" {trimmed}") |
|
|
| context_lines.append("- Latest recommendation updates since previous refresh:") |
| if improvement_items: |
| for item in improvement_items[:max_delta_items]: |
| context_lines.append(f" {item}") |
| else: |
| context_lines.append( |
| " - No explicit delta detected (recommendations likely unchanged or reformatted)." |
| ) |
|
|
| return context_lines |
|
|
