File size: 2,842 Bytes
6f90f5c | 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 | 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
|