Spaces:
Sleeping
Sleeping
File size: 6,805 Bytes
13ceb89 | 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 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 | """
LifeOS — Memory Agent (File 7 of 15)
Stores past scheduling sessions and generates actionable insights
for the Planner via Groq LLM or a deterministic fallback.
Key design points:
- insights_cache is invalidated on every store() call
- retrieve() returns the cache if still valid (no redundant LLM calls)
- _generate_insights() is only called when cache is empty
- All Groq calls use os.getenv("GROQ_API_KEY") — never hardcoded
"""
from __future__ import annotations
import json
import os
from typing import Any, Dict, List
from dotenv import load_dotenv
load_dotenv()
try:
from groq import Groq
_HAS_GROQ = True
except ImportError:
_HAS_GROQ = False
from utils.prompts import MEMORY_RETRIEVAL_PROMPT
class MemoryAgent:
"""
Manages per-episode session history and LLM-powered scheduling insights.
Attributes
----------
sessions : list of stored session dicts
insights_cache : cached plain-text insights string (cleared on store)
"""
def __init__(self) -> None:
self.sessions: List[Dict[str, Any]] = []
self.insights_cache: str = ""
# ------------------------------------------------------------------
# Public API
# ------------------------------------------------------------------
def store(
self,
iteration: int,
plan: Dict[str, Any],
issues: List[str],
reward: float,
) -> None:
"""
Append one session record and invalidate the insights cache.
The cache is cleared so the next retrieve() generates fresh insights
that incorporate this new session.
"""
self.sessions.append({
"iteration": iteration,
"reward": reward,
"issue_count": len(issues),
"issues": issues,
"plan_summary": self._summarize_plan(plan),
})
self.insights_cache = "" # force fresh insights on next retrieve()
print(f"[Memory] Stored session {iteration}: reward={reward:.1f}, issues={len(issues)}")
def retrieve(self) -> str:
"""
Return insights string.
Returns cached value if available (avoids redundant LLM calls).
Generates fresh insights only when cache is empty.
"""
if not self.sessions:
return "- No history yet. Use default scheduling principles."
if self.insights_cache:
print("[Memory] Returning cached insights.")
return self.insights_cache
print("[Memory] Cache empty — generating fresh insights.")
self.insights_cache = self._generate_insights()
return self.insights_cache
# ------------------------------------------------------------------
# Internal helpers
# ------------------------------------------------------------------
def _generate_insights(self) -> str:
"""
Format MEMORY_RETRIEVAL_PROMPT and call Groq.
Falls back to _deterministic_insights() on any failure.
"""
history_json = json.dumps(self.sessions, indent=2)
prompt = MEMORY_RETRIEVAL_PROMPT.format(history=history_json)
if not _HAS_GROQ:
return self._deterministic_insights()
api_key = os.getenv("GROQ_API_KEY")
if not api_key:
return self._deterministic_insights()
try:
client = Groq(api_key=api_key)
response = client.chat.completions.create(
model="llama-3.1-8b-instant",
temperature=0.5,
messages=[
{"role": "system", "content": "You are LifeOS Memory Agent. Return plain text bullet points only."},
{"role": "user", "content": prompt},
],
)
result = (response.choices[0].message.content or "").strip()
if result:
print("[Memory] LLM insights generated successfully.")
return result
return self._deterministic_insights()
except Exception as e:
print(f"[Memory] LLM insight generation failed ({type(e).__name__}): {e}. Using deterministic fallback.")
return self._deterministic_insights()
def _deterministic_insights(self) -> str:
"""Produce bullet-point insights without any LLM call."""
if not self.sessions:
return "- No history yet. Use default scheduling principles."
lines: List[str] = []
rewards = [s["reward"] for s in self.sessions]
best = max(self.sessions, key=lambda s: s["reward"])
worst = min(self.sessions, key=lambda s: s["reward"])
lines.append(
f"- Pattern: Best reward {best['reward']:.1f} at iteration {best['iteration']} "
f"({best['plan_summary']}) -> Advice: Replicate this structure."
)
worst_issues = "; ".join(worst["issues"][:2]) or "none"
lines.append(
f"- Pattern: Worst iteration had issues: {worst_issues} "
f"-> Advice: Directly address these in the next plan."
)
# Recurring issues
all_issues: List[str] = [iss for s in self.sessions for iss in s["issues"]]
from collections import Counter
common = Counter(all_issues).most_common(2)
if common:
lines.append(
f"- Pattern: Recurring issues: {'; '.join(i for i, _ in common)} "
"-> Advice: Fix these first before optimizing elsewhere."
)
if len(rewards) >= 2:
trend = "improving" if rewards[-1] > rewards[0] else "declining"
lines.append(
f"- Pattern: Reward trend is {trend} ({rewards[0]:.1f} -> {rewards[-1]:.1f}) "
"-> Advice: Continue the current improvement strategy."
)
lines.append(
"- Pattern: Always — include breaks every 90 mins and at least one revision slot "
"-> Advice: Never omit these — they give +5 and +2 reward each."
)
return "\n".join(lines[:5])
def _summarize_plan(self, plan: Dict[str, Any]) -> str:
"""Return a short human-readable summary string of a plan dict."""
schedule = plan.get("schedule", {})
n_days = len([d for d, s in schedule.items() if s])
study_tasks = {
slot.get("task")
for slots in schedule.values()
for slot in (slots if isinstance(slots, list) else [])
if slot.get("type") == "study"
}
n_breaks = sum(
1 for slots in schedule.values()
for slot in (slots if isinstance(slots, list) else [])
if slot.get("type") == "break"
)
return f"{len(study_tasks)} tasks, {n_days} days, {n_breaks} breaks"
|