File size: 1,532 Bytes
cdc87cb 4129a71 daca5bd cdc87cb 4129a71 cdc87cb 4129a71 daca5bd cdc87cb | 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 | """update_memory — deterministic rolling-window summary (no LLM call)."""
from typing import List, Optional, Tuple
from src.graph.config import MEMORY_WINDOW
from src.graph.utils import traced
_TURN_LIMIT = 200 # chars per query/answer kept in summary
_ANSWER_PREVIEW = 80 # answer preview length in summary line
def _pair_turns(history: List[dict]) -> List[Tuple[str, str]]:
"""Walk history into (user, assistant) pairs."""
pairs: List[Tuple[str, str]] = []
pending_user: Optional[str] = None
for msg in history or []:
role = msg.get("role")
content = (msg.get("content") or "")[:_TURN_LIMIT]
if role == "user":
pending_user = content
elif role == "assistant" and pending_user is not None:
pairs.append((pending_user, content))
pending_user = None
return pairs
@traced("memory")
def update_memory(state: dict) -> dict:
history = state.get("history", []) or []
query = (state.get("query") or "")[:_TURN_LIMIT]
answer = (state.get("answer") or "")[:_TURN_LIMIT]
pairs = _pair_turns(history)
if query and answer:
pairs.append((query, answer))
pairs = pairs[-MEMORY_WINDOW:]
summary = " | ".join(
f"Q: {q} → A: {a[:_ANSWER_PREVIEW]}{'…' if len(a) > _ANSWER_PREVIEW else ''}"
for q, a in pairs
)
return {
"memory_summary": summary,
"_summary": f"{len(pairs)} turns",
"_payload": {"turns": len(pairs), "summary_preview": summary[:200]},
}
|