724 lines
27 KiB
Python
724 lines
27 KiB
Python
"""Memory updater for reading, writing, and updating memory data."""
|
|
|
|
import asyncio
|
|
import atexit
|
|
import concurrent.futures
|
|
import copy
|
|
import json
|
|
import logging
|
|
import math
|
|
import re
|
|
import uuid
|
|
from typing import Any
|
|
|
|
from deerflow.agents.memory.prompt import (
|
|
MEMORY_UPDATE_PROMPT,
|
|
format_conversation_for_update,
|
|
)
|
|
from deerflow.agents.memory.storage import (
|
|
create_empty_memory,
|
|
get_memory_storage,
|
|
utc_now_iso_z,
|
|
)
|
|
from deerflow.config.memory_config import get_memory_config
|
|
from deerflow.models import create_chat_model
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
# Thread pool for offloading sync memory updates when called from an async
|
|
# context. Unlike the previous asyncio.run() approach, this runs *sync*
|
|
# model.invoke() calls — no event loop is created, so the langchain async
|
|
# httpx client pool (globally cached via @lru_cache) is never touched and
|
|
# cross-loop connection reuse is impossible.
|
|
_SYNC_MEMORY_UPDATER_EXECUTOR = concurrent.futures.ThreadPoolExecutor(
|
|
max_workers=4,
|
|
thread_name_prefix="memory-updater-sync",
|
|
)
|
|
atexit.register(lambda: _SYNC_MEMORY_UPDATER_EXECUTOR.shutdown(wait=False))
|
|
|
|
|
|
def _create_empty_memory() -> dict[str, Any]:
|
|
"""Backward-compatible wrapper around the storage-layer empty-memory factory."""
|
|
return create_empty_memory()
|
|
|
|
|
|
def _save_memory_to_file(memory_data: dict[str, Any], agent_name: str | None = None, *, user_id: str | None = None) -> bool:
|
|
"""Backward-compatible wrapper around the configured memory storage save path."""
|
|
return get_memory_storage().save(memory_data, agent_name, user_id=user_id)
|
|
|
|
|
|
def get_memory_data(agent_name: str | None = None, *, user_id: str | None = None) -> dict[str, Any]:
|
|
"""Get the current memory data via storage provider."""
|
|
return get_memory_storage().load(agent_name, user_id=user_id)
|
|
|
|
|
|
def reload_memory_data(agent_name: str | None = None, *, user_id: str | None = None) -> dict[str, Any]:
|
|
"""Reload memory data via storage provider."""
|
|
return get_memory_storage().reload(agent_name, user_id=user_id)
|
|
|
|
|
|
def import_memory_data(memory_data: dict[str, Any], agent_name: str | None = None, *, user_id: str | None = None) -> dict[str, Any]:
|
|
"""Persist imported memory data via storage provider.
|
|
|
|
Args:
|
|
memory_data: Full memory payload to persist.
|
|
agent_name: If provided, imports into per-agent memory.
|
|
user_id: If provided, scopes memory to a specific user.
|
|
|
|
Returns:
|
|
The saved memory data after storage normalization.
|
|
|
|
Raises:
|
|
OSError: If persisting the imported memory fails.
|
|
"""
|
|
storage = get_memory_storage()
|
|
if not storage.save(memory_data, agent_name, user_id=user_id):
|
|
raise OSError("Failed to save imported memory data")
|
|
return storage.load(agent_name, user_id=user_id)
|
|
|
|
|
|
def clear_memory_data(agent_name: str | None = None, *, user_id: str | None = None) -> dict[str, Any]:
|
|
"""Clear all stored memory data and persist an empty structure."""
|
|
cleared_memory = create_empty_memory()
|
|
if not _save_memory_to_file(cleared_memory, agent_name, user_id=user_id):
|
|
raise OSError("Failed to save cleared memory data")
|
|
return cleared_memory
|
|
|
|
|
|
def _validate_confidence(confidence: float) -> float:
|
|
"""Validate persisted fact confidence so stored JSON stays standards-compliant."""
|
|
if not math.isfinite(confidence) or confidence < 0 or confidence > 1:
|
|
raise ValueError("confidence")
|
|
return confidence
|
|
|
|
|
|
def create_memory_fact(
|
|
content: str,
|
|
category: str = "context",
|
|
confidence: float = 0.5,
|
|
agent_name: str | None = None,
|
|
*,
|
|
user_id: str | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Create a new fact and persist the updated memory data."""
|
|
normalized_content = content.strip()
|
|
if not normalized_content:
|
|
raise ValueError("content")
|
|
|
|
normalized_category = category.strip() or "context"
|
|
validated_confidence = _validate_confidence(confidence)
|
|
now = utc_now_iso_z()
|
|
memory_data = get_memory_data(agent_name, user_id=user_id)
|
|
updated_memory = dict(memory_data)
|
|
facts = list(memory_data.get("facts", []))
|
|
facts.append(
|
|
{
|
|
"id": f"fact_{uuid.uuid4().hex[:8]}",
|
|
"content": normalized_content,
|
|
"category": normalized_category,
|
|
"confidence": validated_confidence,
|
|
"createdAt": now,
|
|
"source": "manual",
|
|
}
|
|
)
|
|
updated_memory["facts"] = facts
|
|
|
|
if not _save_memory_to_file(updated_memory, agent_name, user_id=user_id):
|
|
raise OSError("Failed to save memory data after creating fact")
|
|
|
|
return updated_memory
|
|
|
|
|
|
def delete_memory_fact(fact_id: str, agent_name: str | None = None, *, user_id: str | None = None) -> dict[str, Any]:
|
|
"""Delete a fact by its id and persist the updated memory data."""
|
|
memory_data = get_memory_data(agent_name, user_id=user_id)
|
|
facts = memory_data.get("facts", [])
|
|
updated_facts = [fact for fact in facts if fact.get("id") != fact_id]
|
|
if len(updated_facts) == len(facts):
|
|
raise KeyError(fact_id)
|
|
|
|
updated_memory = dict(memory_data)
|
|
updated_memory["facts"] = updated_facts
|
|
|
|
if not _save_memory_to_file(updated_memory, agent_name, user_id=user_id):
|
|
raise OSError(f"Failed to save memory data after deleting fact '{fact_id}'")
|
|
|
|
return updated_memory
|
|
|
|
|
|
def update_memory_fact(
|
|
fact_id: str,
|
|
content: str | None = None,
|
|
category: str | None = None,
|
|
confidence: float | None = None,
|
|
agent_name: str | None = None,
|
|
*,
|
|
user_id: str | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Update an existing fact and persist the updated memory data."""
|
|
memory_data = get_memory_data(agent_name, user_id=user_id)
|
|
updated_memory = dict(memory_data)
|
|
updated_facts: list[dict[str, Any]] = []
|
|
found = False
|
|
|
|
for fact in memory_data.get("facts", []):
|
|
if fact.get("id") == fact_id:
|
|
found = True
|
|
updated_fact = dict(fact)
|
|
if content is not None:
|
|
normalized_content = content.strip()
|
|
if not normalized_content:
|
|
raise ValueError("content")
|
|
updated_fact["content"] = normalized_content
|
|
if category is not None:
|
|
updated_fact["category"] = category.strip() or "context"
|
|
if confidence is not None:
|
|
updated_fact["confidence"] = _validate_confidence(confidence)
|
|
updated_facts.append(updated_fact)
|
|
else:
|
|
updated_facts.append(fact)
|
|
|
|
if not found:
|
|
raise KeyError(fact_id)
|
|
|
|
updated_memory["facts"] = updated_facts
|
|
|
|
if not _save_memory_to_file(updated_memory, agent_name, user_id=user_id):
|
|
raise OSError(f"Failed to save memory data after updating fact '{fact_id}'")
|
|
|
|
return updated_memory
|
|
|
|
|
|
def _extract_text(content: Any) -> str:
|
|
"""Extract plain text from LLM response content (str or list of content blocks).
|
|
|
|
Modern LLMs may return structured content as a list of blocks instead of a
|
|
plain string, e.g. [{"type": "text", "text": "..."}]. Using str() on such
|
|
content produces Python repr instead of the actual text, breaking JSON
|
|
parsing downstream.
|
|
|
|
String chunks are concatenated without separators to avoid corrupting
|
|
chunked JSON/text payloads. Dict-based text blocks are treated as full text
|
|
blocks and joined with newlines for readability.
|
|
"""
|
|
if isinstance(content, str):
|
|
return content
|
|
if isinstance(content, list):
|
|
pieces: list[str] = []
|
|
pending_str_parts: list[str] = []
|
|
|
|
def flush_pending_str_parts() -> None:
|
|
if pending_str_parts:
|
|
pieces.append("".join(pending_str_parts))
|
|
pending_str_parts.clear()
|
|
|
|
for block in content:
|
|
if isinstance(block, str):
|
|
pending_str_parts.append(block)
|
|
elif isinstance(block, dict):
|
|
flush_pending_str_parts()
|
|
text_val = block.get("text")
|
|
if isinstance(text_val, str):
|
|
pieces.append(text_val)
|
|
|
|
flush_pending_str_parts()
|
|
return "\n".join(pieces)
|
|
return str(content)
|
|
|
|
|
|
# Matches sentences that describe a file-upload *event* rather than general
|
|
# file-related work. Deliberately narrow to avoid removing legitimate facts
|
|
# such as "User works with CSV files" or "prefers PDF export".
|
|
_UPLOAD_SENTENCE_RE = re.compile(
|
|
r"[^.!?]*\b(?:"
|
|
r"upload(?:ed|ing)?(?:\s+\w+){0,3}\s+(?:file|files?|document|documents?|attachment|attachments?)"
|
|
r"|file\s+upload"
|
|
r"|/mnt/user-data/uploads/"
|
|
r"|<uploaded_files>"
|
|
r")[^.!?]*[.!?]?\s*",
|
|
re.IGNORECASE,
|
|
)
|
|
|
|
|
|
def _strip_upload_mentions_from_memory(memory_data: dict[str, Any]) -> dict[str, Any]:
|
|
"""Remove sentences about file uploads from all memory summaries and facts.
|
|
|
|
Uploaded files are session-scoped; persisting upload events in long-term
|
|
memory causes the agent to search for non-existent files in future sessions.
|
|
"""
|
|
# Scrub summaries in user/history sections
|
|
for section in ("user", "history"):
|
|
section_data = memory_data.get(section, {})
|
|
for _key, val in section_data.items():
|
|
if isinstance(val, dict) and "summary" in val:
|
|
cleaned = _UPLOAD_SENTENCE_RE.sub("", val["summary"]).strip()
|
|
cleaned = re.sub(r" +", " ", cleaned)
|
|
val["summary"] = cleaned
|
|
|
|
# Also remove any facts that describe upload events
|
|
facts = memory_data.get("facts", [])
|
|
if facts:
|
|
memory_data["facts"] = [f for f in facts if not _UPLOAD_SENTENCE_RE.search(f.get("content", ""))]
|
|
|
|
return memory_data
|
|
|
|
|
|
def _fact_content_key(content: Any) -> str | None:
|
|
if not isinstance(content, str):
|
|
return None
|
|
stripped = content.strip()
|
|
if not stripped:
|
|
return None
|
|
return stripped.casefold()
|
|
|
|
|
|
# Match a fenced code block, optionally with an info string (e.g. ```json).
|
|
# We deliberately use a non-greedy DOTALL pattern so that *only* the contents
|
|
# are kept — no leading prose like "Here is the updated memory:" and no
|
|
# trailing chatter that would break ``json.loads``.
|
|
_FENCED_BLOCK_RE = re.compile(r"```[a-zA-Z0-9_-]*\s*\n?(.*?)```", re.DOTALL)
|
|
|
|
# Match common "thinking" wrappers that some Chinese-domestic LLMs emit even
|
|
# when the system prompt asks for JSON only (GLM, Qwen, MindIE-served models,
|
|
# and others). Stripping these *before* JSON extraction prevents the body of
|
|
# the reasoning text from being mistakenly parsed as the payload.
|
|
_THINK_BLOCK_RE = re.compile(
|
|
r"<\s*(think|thinking|reasoning|analysis)\s*>.*?<\s*/\s*\1\s*>",
|
|
re.DOTALL | re.IGNORECASE,
|
|
)
|
|
|
|
|
|
def _extract_first_json_object(text: str) -> str | None:
|
|
"""Return the substring spanning the first balanced ``{...}`` JSON object.
|
|
|
|
Walks the text once tracking brace depth, ignoring braces that appear
|
|
inside double-quoted strings (with backslash-escape awareness). This is
|
|
sufficient for the JSON dialect the prompt asks for; we deliberately do
|
|
not handle single-quoted "JSON" because that would silently accept
|
|
invalid responses we'd rather reject and retry on.
|
|
"""
|
|
start = text.find("{")
|
|
if start == -1:
|
|
return None
|
|
|
|
depth = 0
|
|
in_string = False
|
|
escape = False
|
|
for index in range(start, len(text)):
|
|
char = text[index]
|
|
if in_string:
|
|
if escape:
|
|
escape = False
|
|
continue
|
|
if char == "\\":
|
|
escape = True
|
|
continue
|
|
if char == '"':
|
|
in_string = False
|
|
continue
|
|
if char == '"':
|
|
in_string = True
|
|
continue
|
|
if char == "{":
|
|
depth += 1
|
|
elif char == "}":
|
|
depth -= 1
|
|
if depth == 0:
|
|
return text[start : index + 1]
|
|
return None
|
|
|
|
|
|
def _parse_memory_update_response(response_text: str) -> dict[str, Any]:
|
|
"""Best-effort extraction of the JSON payload from a memory-update response.
|
|
|
|
Models served by ``vllm`` / ``MindIE`` (e.g. the GLM_5 endpoint at
|
|
``http://10.106.57.108:8010/GLM_5/v1/chat/completions`` users have hit
|
|
in production) routinely:
|
|
|
|
1. Wrap the JSON in a fenced code block (\\`\\`\\`json ... \\`\\`\\`)
|
|
even though the prompt forbids it.
|
|
2. Emit a ``<think>...</think>`` reasoning block before the answer.
|
|
3. Add prose ("Here is the updated memory:") before/after the JSON.
|
|
4. Return an empty body when the upstream model produced only reasoning
|
|
and no answer (the original cause of "Expecting value: line 1
|
|
column 1 (char 0)" reports).
|
|
|
|
This helper normalizes all four cases and surfaces a clear error
|
|
message containing a truncated response snippet so future failures are
|
|
diagnosable from the log alone.
|
|
"""
|
|
if not isinstance(response_text, str):
|
|
raise json.JSONDecodeError("Empty memory update response (non-string content)", "", 0)
|
|
|
|
cleaned = _THINK_BLOCK_RE.sub("", response_text).strip()
|
|
if not cleaned:
|
|
raise json.JSONDecodeError("Empty memory update response (only reasoning content)", response_text, 0)
|
|
|
|
# 1. Try fenced code block first — favours the model's explicit framing.
|
|
fenced_match = _FENCED_BLOCK_RE.search(cleaned)
|
|
if fenced_match:
|
|
candidate = fenced_match.group(1).strip()
|
|
if candidate:
|
|
try:
|
|
return json.loads(candidate)
|
|
except json.JSONDecodeError:
|
|
# Fall through to balanced-brace extraction; the fenced block
|
|
# may have contained truncated JSON or interleaved prose.
|
|
pass
|
|
|
|
# 2. Try balanced-brace JSON extraction anywhere in the text.
|
|
balanced = _extract_first_json_object(cleaned)
|
|
if balanced is not None:
|
|
try:
|
|
return json.loads(balanced)
|
|
except json.JSONDecodeError:
|
|
pass
|
|
|
|
# 3. Last resort: try parsing the whole cleaned text directly. This
|
|
# covers responses that already are pure JSON.
|
|
try:
|
|
return json.loads(cleaned)
|
|
except json.JSONDecodeError as exc:
|
|
snippet = response_text.strip()
|
|
if len(snippet) > 400:
|
|
snippet = snippet[:400] + "...[truncated]"
|
|
raise json.JSONDecodeError(
|
|
f"{exc.msg}; raw response head: {snippet!r}",
|
|
exc.doc,
|
|
exc.pos,
|
|
) from exc
|
|
|
|
|
|
class MemoryUpdater:
|
|
"""Updates memory using LLM based on conversation context."""
|
|
|
|
def __init__(self, model_name: str | None = None):
|
|
"""Initialize the memory updater.
|
|
|
|
Args:
|
|
model_name: Optional model name to use. If None, uses config or default.
|
|
"""
|
|
self._model_name = model_name
|
|
|
|
def _get_model(self):
|
|
"""Get the model for memory updates."""
|
|
config = get_memory_config()
|
|
model_name = self._model_name or config.model_name
|
|
return create_chat_model(name=model_name, thinking_enabled=False)
|
|
|
|
def _build_correction_hint(
|
|
self,
|
|
correction_detected: bool,
|
|
reinforcement_detected: bool,
|
|
) -> str:
|
|
"""Build optional prompt hints for correction and reinforcement signals."""
|
|
correction_hint = ""
|
|
if correction_detected:
|
|
correction_hint = (
|
|
"IMPORTANT: Explicit correction signals were detected in this conversation. "
|
|
"Pay special attention to what the agent got wrong, what the user corrected, "
|
|
"and record the correct approach as a fact with category "
|
|
'"correction" and confidence >= 0.95 when appropriate.'
|
|
)
|
|
if reinforcement_detected:
|
|
reinforcement_hint = (
|
|
"IMPORTANT: Positive reinforcement signals were detected in this conversation. "
|
|
"The user explicitly confirmed the agent's approach was correct or helpful. "
|
|
"Record the confirmed approach, style, or preference as a fact with category "
|
|
'"preference" or "behavior" and confidence >= 0.9 when appropriate.'
|
|
)
|
|
correction_hint = (correction_hint + "\n" + reinforcement_hint).strip() if correction_hint else reinforcement_hint
|
|
|
|
return correction_hint
|
|
|
|
def _prepare_update_prompt(
|
|
self,
|
|
messages: list[Any],
|
|
agent_name: str | None,
|
|
correction_detected: bool,
|
|
reinforcement_detected: bool,
|
|
user_id: str | None = None,
|
|
) -> tuple[dict[str, Any], str] | None:
|
|
"""Load memory and build the update prompt for a conversation."""
|
|
config = get_memory_config()
|
|
if not config.enabled or not messages:
|
|
return None
|
|
|
|
current_memory = get_memory_data(agent_name, user_id=user_id)
|
|
conversation_text = format_conversation_for_update(messages)
|
|
if not conversation_text.strip():
|
|
return None
|
|
|
|
correction_hint = self._build_correction_hint(
|
|
correction_detected=correction_detected,
|
|
reinforcement_detected=reinforcement_detected,
|
|
)
|
|
prompt = MEMORY_UPDATE_PROMPT.format(
|
|
current_memory=json.dumps(current_memory, indent=2),
|
|
conversation=conversation_text,
|
|
correction_hint=correction_hint,
|
|
)
|
|
return current_memory, prompt
|
|
|
|
def _finalize_update(
|
|
self,
|
|
current_memory: dict[str, Any],
|
|
response_content: Any,
|
|
thread_id: str | None,
|
|
agent_name: str | None,
|
|
user_id: str | None = None,
|
|
) -> bool:
|
|
"""Parse the model response, apply updates, and persist memory."""
|
|
response_text = _extract_text(response_content)
|
|
update_data = _parse_memory_update_response(response_text)
|
|
# Deep-copy before in-place mutation so a subsequent save() failure
|
|
# cannot corrupt the still-cached original object reference.
|
|
updated_memory = self._apply_updates(copy.deepcopy(current_memory), update_data, thread_id)
|
|
updated_memory = _strip_upload_mentions_from_memory(updated_memory)
|
|
return get_memory_storage().save(updated_memory, agent_name, user_id=user_id)
|
|
|
|
async def aupdate_memory(
|
|
self,
|
|
messages: list[Any],
|
|
thread_id: str | None = None,
|
|
agent_name: str | None = None,
|
|
correction_detected: bool = False,
|
|
reinforcement_detected: bool = False,
|
|
user_id: str | None = None,
|
|
) -> bool:
|
|
"""Update memory asynchronously by delegating to the sync path.
|
|
|
|
Uses ``asyncio.to_thread`` to run the *sync* ``model.invoke()`` path
|
|
in a worker thread so no second event loop is created and the
|
|
langchain async httpx client pool (shared with the lead agent) is
|
|
never touched. This eliminates the cross-loop connection-reuse bug
|
|
described in issue #2615.
|
|
"""
|
|
return await asyncio.to_thread(
|
|
self._do_update_memory_sync,
|
|
messages=messages,
|
|
thread_id=thread_id,
|
|
agent_name=agent_name,
|
|
correction_detected=correction_detected,
|
|
reinforcement_detected=reinforcement_detected,
|
|
user_id=user_id,
|
|
)
|
|
|
|
def _do_update_memory_sync(
|
|
self,
|
|
messages: list[Any],
|
|
thread_id: str | None = None,
|
|
agent_name: str | None = None,
|
|
correction_detected: bool = False,
|
|
reinforcement_detected: bool = False,
|
|
user_id: str | None = None,
|
|
) -> bool:
|
|
"""Pure-sync memory update using ``model.invoke()``.
|
|
|
|
Uses the *sync* LLM call path so no event loop is created. This
|
|
guarantees that the langchain provider's globally cached async
|
|
httpx ``AsyncClient`` / connection pool (the one shared with the
|
|
lead agent) is never touched — no cross-loop connection reuse is
|
|
possible.
|
|
"""
|
|
try:
|
|
prepared = self._prepare_update_prompt(
|
|
messages=messages,
|
|
agent_name=agent_name,
|
|
correction_detected=correction_detected,
|
|
reinforcement_detected=reinforcement_detected,
|
|
user_id=user_id,
|
|
)
|
|
if prepared is None:
|
|
return False
|
|
|
|
current_memory, prompt = prepared
|
|
model = self._get_model()
|
|
response = model.invoke(prompt, config={"run_name": "memory_agent"})
|
|
return self._finalize_update(
|
|
current_memory=current_memory,
|
|
response_content=response.content,
|
|
thread_id=thread_id,
|
|
agent_name=agent_name,
|
|
user_id=user_id,
|
|
)
|
|
except json.JSONDecodeError as e:
|
|
logger.warning("Failed to parse LLM response for memory update: %s", e)
|
|
return False
|
|
except Exception as e:
|
|
logger.exception("Memory update failed: %s", e)
|
|
return False
|
|
|
|
def update_memory(
|
|
self,
|
|
messages: list[Any],
|
|
thread_id: str | None = None,
|
|
agent_name: str | None = None,
|
|
correction_detected: bool = False,
|
|
reinforcement_detected: bool = False,
|
|
user_id: str | None = None,
|
|
) -> bool:
|
|
"""Synchronously update memory using the sync LLM path.
|
|
|
|
Uses ``model.invoke()`` (sync HTTP) which operates on a completely
|
|
separate connection pool from the async ``AsyncClient`` shared by
|
|
the lead agent. This eliminates the cross-loop connection-reuse
|
|
bug described in issue #2615.
|
|
|
|
When called from within a running event loop (e.g. from a LangGraph
|
|
node), the blocking sync call is offloaded to a thread pool so the
|
|
caller's loop is not blocked.
|
|
|
|
Args:
|
|
messages: List of conversation messages.
|
|
thread_id: Optional thread ID for tracking source.
|
|
agent_name: If provided, updates per-agent memory. If None, updates global memory.
|
|
correction_detected: Whether recent turns include an explicit correction signal.
|
|
reinforcement_detected: Whether recent turns include a positive reinforcement signal.
|
|
user_id: If provided, scopes memory to a specific user.
|
|
|
|
Returns:
|
|
True if update was successful, False otherwise.
|
|
"""
|
|
try:
|
|
loop = asyncio.get_running_loop()
|
|
except RuntimeError:
|
|
loop = None
|
|
|
|
if loop is not None and loop.is_running():
|
|
try:
|
|
future = _SYNC_MEMORY_UPDATER_EXECUTOR.submit(
|
|
self._do_update_memory_sync,
|
|
messages=messages,
|
|
thread_id=thread_id,
|
|
agent_name=agent_name,
|
|
correction_detected=correction_detected,
|
|
reinforcement_detected=reinforcement_detected,
|
|
user_id=user_id,
|
|
)
|
|
return future.result()
|
|
except Exception:
|
|
logger.exception("Failed to offload memory update to executor")
|
|
return False
|
|
|
|
return self._do_update_memory_sync(
|
|
messages=messages,
|
|
thread_id=thread_id,
|
|
agent_name=agent_name,
|
|
correction_detected=correction_detected,
|
|
reinforcement_detected=reinforcement_detected,
|
|
user_id=user_id,
|
|
)
|
|
|
|
def _apply_updates(
|
|
self,
|
|
current_memory: dict[str, Any],
|
|
update_data: dict[str, Any],
|
|
thread_id: str | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Apply LLM-generated updates to memory.
|
|
|
|
Args:
|
|
current_memory: Current memory data.
|
|
update_data: Updates from LLM.
|
|
thread_id: Optional thread ID for tracking.
|
|
|
|
Returns:
|
|
Updated memory data.
|
|
"""
|
|
config = get_memory_config()
|
|
now = utc_now_iso_z()
|
|
|
|
# Update user sections
|
|
user_updates = update_data.get("user", {})
|
|
for section in ["workContext", "personalContext", "topOfMind"]:
|
|
section_data = user_updates.get(section, {})
|
|
if section_data.get("shouldUpdate") and section_data.get("summary"):
|
|
current_memory["user"][section] = {
|
|
"summary": section_data["summary"],
|
|
"updatedAt": now,
|
|
}
|
|
|
|
# Update history sections
|
|
history_updates = update_data.get("history", {})
|
|
for section in ["recentMonths", "earlierContext", "longTermBackground"]:
|
|
section_data = history_updates.get(section, {})
|
|
if section_data.get("shouldUpdate") and section_data.get("summary"):
|
|
current_memory["history"][section] = {
|
|
"summary": section_data["summary"],
|
|
"updatedAt": now,
|
|
}
|
|
|
|
# Remove facts
|
|
facts_to_remove = set(update_data.get("factsToRemove", []))
|
|
if facts_to_remove:
|
|
current_memory["facts"] = [f for f in current_memory.get("facts", []) if f.get("id") not in facts_to_remove]
|
|
|
|
# Add new facts
|
|
existing_fact_keys = {fact_key for fact_key in (_fact_content_key(fact.get("content")) for fact in current_memory.get("facts", [])) if fact_key is not None}
|
|
new_facts = update_data.get("newFacts", [])
|
|
for fact in new_facts:
|
|
confidence = fact.get("confidence", 0.5)
|
|
if confidence >= config.fact_confidence_threshold:
|
|
raw_content = fact.get("content", "")
|
|
if not isinstance(raw_content, str):
|
|
continue
|
|
normalized_content = raw_content.strip()
|
|
fact_key = _fact_content_key(normalized_content)
|
|
if fact_key is not None and fact_key in existing_fact_keys:
|
|
continue
|
|
|
|
fact_entry = {
|
|
"id": f"fact_{uuid.uuid4().hex[:8]}",
|
|
"content": normalized_content,
|
|
"category": fact.get("category", "context"),
|
|
"confidence": confidence,
|
|
"createdAt": now,
|
|
"source": thread_id or "unknown",
|
|
}
|
|
source_error = fact.get("sourceError")
|
|
if isinstance(source_error, str):
|
|
normalized_source_error = source_error.strip()
|
|
if normalized_source_error:
|
|
fact_entry["sourceError"] = normalized_source_error
|
|
current_memory["facts"].append(fact_entry)
|
|
if fact_key is not None:
|
|
existing_fact_keys.add(fact_key)
|
|
|
|
# Enforce max facts limit
|
|
if len(current_memory["facts"]) > config.max_facts:
|
|
# Sort by confidence and keep top ones
|
|
current_memory["facts"] = sorted(
|
|
current_memory["facts"],
|
|
key=lambda f: f.get("confidence", 0),
|
|
reverse=True,
|
|
)[: config.max_facts]
|
|
|
|
return current_memory
|
|
|
|
|
|
def update_memory_from_conversation(
|
|
messages: list[Any],
|
|
thread_id: str | None = None,
|
|
agent_name: str | None = None,
|
|
correction_detected: bool = False,
|
|
reinforcement_detected: bool = False,
|
|
user_id: str | None = None,
|
|
) -> bool:
|
|
"""Convenience function to update memory from a conversation.
|
|
|
|
Args:
|
|
messages: List of conversation messages.
|
|
thread_id: Optional thread ID.
|
|
agent_name: If provided, updates per-agent memory. If None, updates global memory.
|
|
correction_detected: Whether recent turns include an explicit correction signal.
|
|
reinforcement_detected: Whether recent turns include a positive reinforcement signal.
|
|
user_id: If provided, scopes memory to a specific user.
|
|
|
|
Returns:
|
|
True if successful, False otherwise.
|
|
"""
|
|
updater = MemoryUpdater()
|
|
return updater.update_memory(messages, thread_id, agent_name, correction_detected, reinforcement_detected, user_id=user_id)
|