917 lines
38 KiB
Python
917 lines
38 KiB
Python
"""Layered semantic memory and compaction for Space PMFlow AI.
|
||
|
||
The service keeps long-context behavior on a product/runtime contract:
|
||
thread, user, workspace and project memories are selected by explicit scope
|
||
anchors and generic relevance, not by business-scene keyword branches.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import re
|
||
import uuid
|
||
from datetime import datetime
|
||
from typing import Any
|
||
|
||
from sqlalchemy import and_, desc, select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from ..config.ai_prompts import PROMPT_PMFLOW_AI_CONTEXT_MEMORY_COMPACTOR
|
||
from ..models import PMFlowAIContextMemory, PMFlowAIThread
|
||
from .ai_function_runtime_profile_service import AIFunctionRuntimeProfileService
|
||
from .llm_service import llm_service
|
||
from .prompt_runtime_service import PromptRuntimeRenderResult, PromptRuntimeService
|
||
from .runtime_audit_context import bind_agent_audit_context
|
||
|
||
|
||
PROMPT_KEY_PMFLOW_AI_CONTEXT_MEMORY_COMPACTOR = "PROMPT_PMFLOW_AI_CONTEXT_MEMORY_COMPACTOR"
|
||
FUNCTION_NAME_PMFLOW_AI_CONTEXT_MEMORY_COMPACTOR = "pmflow_ai_context_memory_compactor"
|
||
SCENE_KEY_PMFLOW_AI_AGENT_RUNTIME = "space_pmflow_ai_agent_runtime"
|
||
MEMORY_KEY_CURRENT = "semantic_compaction.current"
|
||
MEMORY_PACKET_VERSION = "pmflow_ai_layered_memory_pack.v1"
|
||
|
||
_MEMORY_SCOPES = ("thread", "user", "workspace", "project")
|
||
_ACTIVE_MEMORY_STATUSES = ("active", "candidate")
|
||
|
||
|
||
class PMFlowAIContextMemoryService:
|
||
"""Build and update layered semantic memories for PMFlow AI."""
|
||
|
||
def __init__(self, db: AsyncSession, *, tenant_id: int, generator: Any | None = None) -> None:
|
||
self.db = db
|
||
self.tenant_id = int(tenant_id or 1)
|
||
self.generator = generator or llm_service
|
||
|
||
async def build_memory_context_pack(
|
||
self,
|
||
*,
|
||
thread: PMFlowAIThread | None,
|
||
current_user: Any | None = None,
|
||
conversation_context: dict[str, Any] | None = None,
|
||
tenant_project_context: dict[str, Any] | None = None,
|
||
user_question: str = "",
|
||
limit: int = 12,
|
||
) -> dict[str, Any]:
|
||
"""Load relevant long-context memory rows for the current turn."""
|
||
|
||
scope_refs = _scope_refs(
|
||
thread=thread,
|
||
current_user=current_user,
|
||
conversation_context=conversation_context,
|
||
tenant_project_context=tenant_project_context,
|
||
)
|
||
if not scope_refs or not callable(getattr(self.db, "execute", None)):
|
||
return {}
|
||
rows: list[PMFlowAIContextMemory] = []
|
||
for scope, refs in scope_refs.items():
|
||
if not refs:
|
||
continue
|
||
result = await self.db.execute(
|
||
select(PMFlowAIContextMemory)
|
||
.where(
|
||
and_(
|
||
PMFlowAIContextMemory.tenant_id == self.tenant_id,
|
||
PMFlowAIContextMemory.memory_scope == scope,
|
||
PMFlowAIContextMemory.scope_ref.in_(refs[:6]),
|
||
PMFlowAIContextMemory.status.in_(_ACTIVE_MEMORY_STATUSES),
|
||
)
|
||
)
|
||
.order_by(
|
||
desc(PMFlowAIContextMemory.confidence_score),
|
||
desc(PMFlowAIContextMemory.last_validated_at),
|
||
desc(PMFlowAIContextMemory.last_compacted_at),
|
||
desc(PMFlowAIContextMemory.updated_at),
|
||
)
|
||
.limit(max(1, limit))
|
||
)
|
||
rows.extend(list(result.scalars().all()))
|
||
ranked = _rank_memory_rows(question=user_question, rows=rows, limit=limit)
|
||
now = datetime.utcnow()
|
||
for row in ranked:
|
||
row.last_accessed_at = now
|
||
return _memory_pack_from_rows(ranked, scope_refs=scope_refs)
|
||
|
||
async def compact_and_persist_after_turn(
|
||
self,
|
||
*,
|
||
thread: PMFlowAIThread | None,
|
||
current_user: Any | None,
|
||
user_message: str,
|
||
assistant_answer: str,
|
||
context_pack: dict[str, Any] | None,
|
||
tenant_project_context: dict[str, Any] | None,
|
||
run_public_id: str | None = None,
|
||
user_message_index: int | None = None,
|
||
assistant_message_index: int | None = None,
|
||
) -> dict[str, Any]:
|
||
"""Compact the latest turn into layered long-context memories."""
|
||
|
||
if thread is None or not getattr(thread, "id", None) or not callable(getattr(self.db, "execute", None)):
|
||
return {}
|
||
scope_refs = _scope_refs(
|
||
thread=thread,
|
||
current_user=current_user,
|
||
conversation_context={},
|
||
tenant_project_context=tenant_project_context,
|
||
)
|
||
if not scope_refs:
|
||
return {}
|
||
previous_pack = await self.build_memory_context_pack(
|
||
thread=thread,
|
||
current_user=current_user,
|
||
conversation_context={},
|
||
tenant_project_context=tenant_project_context,
|
||
user_question=user_message,
|
||
limit=16,
|
||
)
|
||
compaction = await self._compact_turn(
|
||
user_message=user_message,
|
||
assistant_answer=assistant_answer,
|
||
previous_memory_pack=previous_pack,
|
||
context_pack=context_pack or {},
|
||
user_message_index=user_message_index,
|
||
assistant_message_index=assistant_message_index,
|
||
)
|
||
memories = _normalize_compaction_memories(compaction)
|
||
if not memories:
|
||
memories = _deterministic_compaction_memories(
|
||
user_message=user_message,
|
||
assistant_answer=assistant_answer,
|
||
user_message_index=user_message_index,
|
||
assistant_message_index=assistant_message_index,
|
||
)
|
||
persisted = []
|
||
for scope in _MEMORY_SCOPES:
|
||
refs = scope_refs.get(scope) or []
|
||
if not refs:
|
||
continue
|
||
memory = dict(memories.get(scope) or memories.get("thread") or {})
|
||
if not _memory_has_content(memory):
|
||
continue
|
||
row = await self._upsert_memory(
|
||
memory_scope=scope,
|
||
scope_ref=refs[0],
|
||
memory=memory,
|
||
thread=thread,
|
||
run_public_id=run_public_id,
|
||
user_message_index=user_message_index,
|
||
assistant_message_index=assistant_message_index,
|
||
compaction=compaction,
|
||
)
|
||
if row is None:
|
||
continue
|
||
persisted.append(
|
||
{
|
||
"memory_scope": row.memory_scope,
|
||
"scope_ref": row.scope_ref,
|
||
"status": row.status,
|
||
"confidence_score": float(row.confidence_score or 0.0),
|
||
"summary": row.summary[:500],
|
||
"token_estimate": int(row.token_estimate or 0),
|
||
}
|
||
)
|
||
return {
|
||
"packet_version": "pmflow_ai_memory_compaction_result.v1",
|
||
"persisted": persisted,
|
||
"compaction_source": str(compaction.get("compaction_source") or ""),
|
||
"scope_count": len(persisted),
|
||
}
|
||
|
||
async def _compact_turn(
|
||
self,
|
||
*,
|
||
user_message: str,
|
||
assistant_answer: str,
|
||
previous_memory_pack: dict[str, Any],
|
||
context_pack: dict[str, Any],
|
||
user_message_index: int | None,
|
||
assistant_message_index: int | None,
|
||
) -> dict[str, Any]:
|
||
generate = getattr(self.generator, "generate", None)
|
||
if not callable(generate) or not callable(getattr(self.db, "execute", None)):
|
||
return {
|
||
"compaction_source": "deterministic_fallback",
|
||
"memories": _deterministic_compaction_memories(
|
||
user_message=user_message,
|
||
assistant_answer=assistant_answer,
|
||
user_message_index=user_message_index,
|
||
assistant_message_index=assistant_message_index,
|
||
),
|
||
}
|
||
try:
|
||
prompt_result = await self._render_prompt(
|
||
user_message=user_message,
|
||
assistant_answer=assistant_answer,
|
||
previous_memory_pack=previous_memory_pack,
|
||
context_pack=context_pack,
|
||
user_message_index=user_message_index,
|
||
assistant_message_index=assistant_message_index,
|
||
)
|
||
runtime_profile = await AIFunctionRuntimeProfileService(self.db).resolve_profile(
|
||
function_name=FUNCTION_NAME_PMFLOW_AI_CONTEXT_MEMORY_COMPACTOR,
|
||
scene_key=SCENE_KEY_PMFLOW_AI_AGENT_RUNTIME,
|
||
prompt_key=PROMPT_KEY_PMFLOW_AI_CONTEXT_MEMORY_COMPACTOR,
|
||
)
|
||
with bind_agent_audit_context(agent_step_key="context_memory_compaction"):
|
||
raw = await generate(
|
||
prompt_result.prompt,
|
||
db=self.db,
|
||
tenant_id=self.tenant_id,
|
||
function_name=FUNCTION_NAME_PMFLOW_AI_CONTEXT_MEMORY_COMPACTOR,
|
||
use_prompt_profile=False,
|
||
temperature=float(runtime_profile.temperature),
|
||
response_mode=runtime_profile.response_mode,
|
||
max_tokens=int(runtime_profile.max_tokens),
|
||
routing_meta={
|
||
"scene_key": SCENE_KEY_PMFLOW_AI_AGENT_RUNTIME,
|
||
"runtime": "pmflow_ai_context_memory",
|
||
"scene_contract": "semantic_memory_compaction",
|
||
**runtime_profile.routing_meta(),
|
||
**prompt_result.routing_meta(),
|
||
"allow_low_to_high_escalation": False,
|
||
"disable_thinking": True,
|
||
},
|
||
)
|
||
parsed = _extract_json_object(raw)
|
||
parsed["compaction_source"] = "llm_semantic_compaction"
|
||
parsed["prompt_meta"] = prompt_result.routing_meta()
|
||
return parsed
|
||
except Exception as exc:
|
||
return {
|
||
"compaction_source": "deterministic_fallback",
|
||
"compaction_error": exc.__class__.__name__,
|
||
"memories": _deterministic_compaction_memories(
|
||
user_message=user_message,
|
||
assistant_answer=assistant_answer,
|
||
user_message_index=user_message_index,
|
||
assistant_message_index=assistant_message_index,
|
||
),
|
||
}
|
||
|
||
async def _render_prompt(
|
||
self,
|
||
*,
|
||
user_message: str,
|
||
assistant_answer: str,
|
||
previous_memory_pack: dict[str, Any],
|
||
context_pack: dict[str, Any],
|
||
user_message_index: int | None,
|
||
assistant_message_index: int | None,
|
||
) -> PromptRuntimeRenderResult:
|
||
return await PromptRuntimeService(self.db).render(
|
||
tenant_id=self.tenant_id,
|
||
prompt_key=PROMPT_KEY_PMFLOW_AI_CONTEXT_MEMORY_COMPACTOR,
|
||
fallback_template=PROMPT_PMFLOW_AI_CONTEXT_MEMORY_COMPACTOR,
|
||
latest_turn_json=_dump_json(
|
||
{
|
||
"user_message_index": user_message_index,
|
||
"assistant_message_index": assistant_message_index,
|
||
"user_message": _clean_text(user_message, limit=1600),
|
||
"assistant_answer": _clean_text(assistant_answer, limit=2400),
|
||
},
|
||
limit=6000,
|
||
),
|
||
previous_memory_pack_json=_dump_json(previous_memory_pack, limit=12000),
|
||
context_pack_json=_dump_json(context_pack, limit=16000),
|
||
)
|
||
|
||
async def _upsert_memory(
|
||
self,
|
||
*,
|
||
memory_scope: str,
|
||
scope_ref: str,
|
||
memory: dict[str, Any],
|
||
thread: PMFlowAIThread,
|
||
run_public_id: str | None,
|
||
user_message_index: int | None,
|
||
assistant_message_index: int | None,
|
||
compaction: dict[str, Any],
|
||
) -> PMFlowAIContextMemory | None:
|
||
result = await self.db.execute(
|
||
select(PMFlowAIContextMemory).where(
|
||
and_(
|
||
PMFlowAIContextMemory.tenant_id == self.tenant_id,
|
||
PMFlowAIContextMemory.memory_scope == memory_scope,
|
||
PMFlowAIContextMemory.scope_ref == scope_ref,
|
||
PMFlowAIContextMemory.memory_key == MEMORY_KEY_CURRENT,
|
||
)
|
||
)
|
||
)
|
||
row = result.scalar_one_or_none()
|
||
now = datetime.utcnow()
|
||
gate = _memory_write_gate(memory=memory, existing=row)
|
||
if gate["action"] == "skip":
|
||
if row is not None:
|
||
row.last_validated_at = now
|
||
row.compaction_payload = {
|
||
**(dict(row.compaction_payload or {}) if isinstance(row.compaction_payload, dict) else {}),
|
||
"last_gate": gate,
|
||
}
|
||
return None
|
||
if row is None:
|
||
row = PMFlowAIContextMemory(
|
||
tenant_id=self.tenant_id,
|
||
public_memory_id=f"ctxmem_{uuid.uuid4().hex[:16]}",
|
||
memory_scope=memory_scope,
|
||
scope_ref=scope_ref,
|
||
memory_key=MEMORY_KEY_CURRENT,
|
||
)
|
||
self.db.add(row)
|
||
merged_memory = _merge_memory(existing=row if gate["action"] != "create" else None, incoming=memory)
|
||
row.source_thread_id = int(thread.id)
|
||
row.source_run_public_id = str(run_public_id or "")[:80] or None
|
||
row.status = str(gate["status"])[:32]
|
||
row.confidence_score = float(gate["confidence_score"])
|
||
row.supersedes_memory_id = int(gate["supersedes_memory_id"]) if gate.get("supersedes_memory_id") else None
|
||
row.invalidation_reason = str(gate.get("invalidation_reason") or "")[:1000] or None
|
||
row.summary = _clean_text(merged_memory.get("summary") or "", limit=4000)
|
||
row.durable_facts = _safe_list(merged_memory.get("durable_facts"), limit=24, item_limit=900)
|
||
row.open_loops = _safe_list(merged_memory.get("open_loops"), limit=16, item_limit=700)
|
||
row.decisions = _safe_list(merged_memory.get("decisions"), limit=16, item_limit=700)
|
||
row.preferences = _safe_list(merged_memory.get("preferences"), limit=16, item_limit=700)
|
||
row.object_refs = _safe_list(merged_memory.get("object_refs"), limit=16, item_limit=500)
|
||
row.source_message_range = {
|
||
"start": _safe_int(user_message_index),
|
||
"end": _safe_int(assistant_message_index),
|
||
}
|
||
row.confidence = str(memory.get("confidence") or "medium")[:32]
|
||
row.token_estimate = _estimated_tokens(
|
||
{
|
||
"summary": row.summary,
|
||
"durable_facts": row.durable_facts,
|
||
"open_loops": row.open_loops,
|
||
"decisions": row.decisions,
|
||
"preferences": row.preferences,
|
||
}
|
||
)
|
||
row.compaction_payload = {
|
||
"packet_version": "pmflow_ai_context_memory_compaction.v1",
|
||
"source": str(compaction.get("compaction_source") or "")[:80],
|
||
"prompt_meta": compaction.get("prompt_meta") if isinstance(compaction.get("prompt_meta"), dict) else {},
|
||
"write_gate": gate,
|
||
"conflict_resolution": gate.get("conflict_resolution") if isinstance(gate.get("conflict_resolution"), dict) else {},
|
||
"raw_memory_keys": sorted([str(key) for key in dict(memory or {}).keys()])[:32],
|
||
}
|
||
row.last_compacted_at = now
|
||
row.last_validated_at = now
|
||
row.last_accessed_at = now
|
||
return row
|
||
|
||
|
||
def _scope_refs(
|
||
*,
|
||
thread: PMFlowAIThread | None,
|
||
current_user: Any | None,
|
||
conversation_context: dict[str, Any] | None,
|
||
tenant_project_context: dict[str, Any] | None,
|
||
) -> dict[str, list[str]]:
|
||
context = conversation_context if isinstance(conversation_context, dict) else {}
|
||
tenant = tenant_project_context if isinstance(tenant_project_context, dict) else {}
|
||
refs: dict[str, list[str]] = {scope: [] for scope in _MEMORY_SCOPES}
|
||
if thread is not None:
|
||
public_thread_id = str(getattr(thread, "public_thread_id", "") or "").strip()
|
||
thread_id = str(getattr(thread, "id", "") or "").strip()
|
||
if public_thread_id:
|
||
refs["thread"].append(f"thread:{public_thread_id}")
|
||
elif thread_id:
|
||
refs["thread"].append(f"thread_id:{thread_id}")
|
||
user_id = _safe_int(getattr(current_user, "id", None)) or _safe_int(tenant.get("user_id"))
|
||
if user_id:
|
||
refs["user"].append(f"user:{user_id}")
|
||
workspace_scope = str(
|
||
context.get("conversation_scope")
|
||
or context.get("workspace_scope")
|
||
or tenant.get("workspace_scope")
|
||
or context.get("surface")
|
||
or "space"
|
||
).strip()
|
||
if workspace_scope:
|
||
refs["workspace"].append(f"workspace:{workspace_scope[:120]}")
|
||
project_id = _safe_int(context.get("current_project_id")) or _safe_int(tenant.get("current_project_id"))
|
||
if project_id:
|
||
refs["project"].append(f"project:{project_id}")
|
||
else:
|
||
for item in _project_refs_from_context(context, tenant):
|
||
if item not in refs["project"]:
|
||
refs["project"].append(item)
|
||
return {scope: values for scope, values in refs.items() if values}
|
||
|
||
|
||
def _project_refs_from_context(context: dict[str, Any], tenant: dict[str, Any]) -> list[str]:
|
||
refs: list[str] = []
|
||
roots = [context, tenant.get("surface_context") if isinstance(tenant.get("surface_context"), dict) else {}]
|
||
for root in roots:
|
||
for key in ("current_focus_object", "object_ledger"):
|
||
value = root.get(key) if isinstance(root, dict) else None
|
||
if isinstance(value, dict):
|
||
project_id = _safe_int(value.get("project_id") or value.get("id") or value.get("object_id"))
|
||
object_type = str(value.get("object_type") or value.get("type") or "").strip()
|
||
if project_id and (not object_type or object_type == "project"):
|
||
refs.append(f"project:{project_id}")
|
||
for item in list(root.get("visible_objects") or [])[:12] if isinstance(root, dict) else []:
|
||
if not isinstance(item, dict):
|
||
continue
|
||
project_id = _safe_int(item.get("project_id") or item.get("id") or item.get("object_id"))
|
||
object_type = str(item.get("object_type") or item.get("type") or "").strip()
|
||
if project_id and (not object_type or object_type == "project"):
|
||
refs.append(f"project:{project_id}")
|
||
out: list[str] = []
|
||
for ref in refs:
|
||
if ref not in out:
|
||
out.append(ref)
|
||
return out[:4]
|
||
|
||
|
||
def _memory_pack_from_rows(rows: list[PMFlowAIContextMemory], *, scope_refs: dict[str, list[str]]) -> dict[str, Any]:
|
||
if not rows:
|
||
return {}
|
||
return {
|
||
"packet_version": MEMORY_PACKET_VERSION,
|
||
"scope_refs": scope_refs,
|
||
"memories": [_memory_row_packet(row) for row in rows],
|
||
"context_policy": {
|
||
"long_term_memory_is_semantic_compaction_not_raw_transcript": True,
|
||
"same_thread_recent_messages_override_memory_when_conflicting": True,
|
||
"memory_scope_binding_is_anchor_based_not_scene_keyword_based": True,
|
||
},
|
||
}
|
||
|
||
|
||
def _memory_row_packet(row: PMFlowAIContextMemory) -> dict[str, Any]:
|
||
return {
|
||
"memory_id": str(row.public_memory_id or "")[:80],
|
||
"memory_scope": str(row.memory_scope or "")[:40],
|
||
"scope_ref": str(row.scope_ref or "")[:160],
|
||
"memory_key": str(row.memory_key or "")[:120],
|
||
"status": str(getattr(row, "status", "") or "active")[:32],
|
||
"summary": _clean_text(row.summary, limit=1800),
|
||
"durable_facts": _safe_list(row.durable_facts, limit=10, item_limit=700),
|
||
"open_loops": _safe_list(row.open_loops, limit=8, item_limit=500),
|
||
"decisions": _safe_list(row.decisions, limit=8, item_limit=500),
|
||
"preferences": _safe_list(row.preferences, limit=8, item_limit=500),
|
||
"object_refs": _safe_list(row.object_refs, limit=8, item_limit=300),
|
||
"confidence": str(row.confidence or "")[:32],
|
||
"confidence_score": float(getattr(row, "confidence_score", 0.0) or 0.0),
|
||
"token_estimate": int(row.token_estimate or 0),
|
||
"last_compacted_at": _iso(row.last_compacted_at),
|
||
}
|
||
|
||
|
||
def _rank_memory_rows(*, question: str, rows: list[PMFlowAIContextMemory], limit: int) -> list[PMFlowAIContextMemory]:
|
||
if not rows:
|
||
return []
|
||
question_terms = _text_terms(question)
|
||
|
||
def score(row: PMFlowAIContextMemory) -> tuple[int, str]:
|
||
text = " ".join(
|
||
[
|
||
str(row.summary or ""),
|
||
_dump_json(row.durable_facts, limit=3000),
|
||
_dump_json(row.open_loops, limit=2000),
|
||
_dump_json(row.decisions, limit=2000),
|
||
]
|
||
)
|
||
terms = _text_terms(text)
|
||
overlap = len(question_terms & terms)
|
||
scope_weight = {"thread": 8, "project": 6, "workspace": 4, "user": 3}.get(str(row.memory_scope or ""), 1)
|
||
confidence_weight = int(float(getattr(row, "confidence_score", 0.0) or 0.0) * 4)
|
||
status_weight = 2 if str(getattr(row, "status", "") or "active") == "active" else 0
|
||
return (overlap + scope_weight + confidence_weight + status_weight, str(row.last_compacted_at or row.updated_at or ""))
|
||
|
||
ranked = sorted(rows, key=score, reverse=True)
|
||
out: list[PMFlowAIContextMemory] = []
|
||
seen: set[str] = set()
|
||
for row in ranked:
|
||
key = f"{row.memory_scope}:{row.scope_ref}:{row.memory_key}"
|
||
if key in seen:
|
||
continue
|
||
seen.add(key)
|
||
out.append(row)
|
||
if len(out) >= max(1, limit):
|
||
break
|
||
return out
|
||
|
||
|
||
def _normalize_compaction_memories(value: dict[str, Any]) -> dict[str, dict[str, Any]]:
|
||
raw = value.get("memories") if isinstance(value, dict) else {}
|
||
if not isinstance(raw, dict):
|
||
return {}
|
||
out: dict[str, dict[str, Any]] = {}
|
||
for scope in _MEMORY_SCOPES:
|
||
item = raw.get(scope)
|
||
if isinstance(item, dict):
|
||
out[scope] = _sanitize_memory(item)
|
||
return out
|
||
|
||
|
||
def _memory_write_gate(*, memory: dict[str, Any], existing: PMFlowAIContextMemory | None) -> dict[str, Any]:
|
||
"""Decide whether a semantic memory should be written or merged.
|
||
|
||
The gate only uses lifecycle signals, generic text novelty and structured
|
||
field deltas. It deliberately avoids business-domain keyword rules.
|
||
"""
|
||
|
||
sanitized = _sanitize_memory(memory)
|
||
status = _normalize_memory_status(sanitized.get("status"))
|
||
confidence_score = _confidence_score(sanitized.get("confidence"))
|
||
incoming_text = _memory_text(sanitized)
|
||
if not incoming_text:
|
||
return _gate_result(
|
||
"skip",
|
||
"empty_memory",
|
||
status=status,
|
||
confidence_score=confidence_score,
|
||
conflict_resolution=_resolve_memory_conflict(memory=sanitized, existing=existing),
|
||
)
|
||
if status in {"stale", "invalidated", "rejected"}:
|
||
if existing is None:
|
||
return _gate_result(
|
||
"skip",
|
||
"cannot_invalidate_missing_memory",
|
||
status=status,
|
||
confidence_score=confidence_score,
|
||
conflict_resolution=_resolve_memory_conflict(memory=sanitized, existing=None),
|
||
)
|
||
return _gate_result(
|
||
"update",
|
||
"lifecycle_status_update",
|
||
status=status,
|
||
confidence_score=max(confidence_score, float(getattr(existing, "confidence_score", 0.0) or 0.0)),
|
||
supersedes_memory_id=int(existing.id) if getattr(existing, "id", None) is not None else None,
|
||
invalidation_reason=sanitized.get("invalidation_reason") or sanitized.get("summary") or "",
|
||
conflict_resolution=_resolve_memory_conflict(memory=sanitized, existing=existing),
|
||
)
|
||
if existing is None:
|
||
create_status = "candidate" if confidence_score < 0.5 else "active"
|
||
return _gate_result(
|
||
"create",
|
||
"new_scope_memory",
|
||
status=create_status,
|
||
confidence_score=confidence_score,
|
||
conflict_resolution=_resolve_memory_conflict(memory=sanitized, existing=None),
|
||
)
|
||
|
||
existing_text = _memory_text(_row_memory_dict(existing))
|
||
incoming_terms = _text_terms(incoming_text)
|
||
existing_terms = _text_terms(existing_text)
|
||
new_terms = incoming_terms - existing_terms
|
||
novelty_ratio = len(new_terms) / max(1, len(incoming_terms))
|
||
structured_delta = _structured_delta_count(sanitized, _row_memory_dict(existing))
|
||
existing_score = float(getattr(existing, "confidence_score", 0.0) or 0.0)
|
||
supersedes = bool(memory.get("supersedes_current") or memory.get("supersedes_existing"))
|
||
conflict_resolution = _resolve_memory_conflict(
|
||
memory={**sanitized, "supersedes_current": supersedes},
|
||
existing=existing,
|
||
novelty_ratio=novelty_ratio,
|
||
structured_delta=structured_delta,
|
||
)
|
||
if not supersedes and novelty_ratio < 0.08 and structured_delta <= 0 and confidence_score <= existing_score + 0.05:
|
||
return _gate_result(
|
||
"skip",
|
||
"duplicate_or_lower_confidence",
|
||
status=str(getattr(existing, "status", "") or "active"),
|
||
confidence_score=existing_score,
|
||
conflict_resolution=conflict_resolution,
|
||
)
|
||
return _gate_result(
|
||
"update",
|
||
"merge_new_semantic_delta",
|
||
status="active",
|
||
confidence_score=max(confidence_score, existing_score),
|
||
supersedes_memory_id=int(existing.id) if supersedes and getattr(existing, "id", None) is not None else None,
|
||
invalidation_reason=sanitized.get("invalidation_reason") if supersedes else "",
|
||
novelty_ratio=round(novelty_ratio, 4),
|
||
structured_delta=structured_delta,
|
||
conflict_resolution=conflict_resolution,
|
||
)
|
||
|
||
|
||
def _resolve_memory_conflict(
|
||
*,
|
||
memory: dict[str, Any],
|
||
existing: PMFlowAIContextMemory | None,
|
||
novelty_ratio: float | None = None,
|
||
structured_delta: int | None = None,
|
||
) -> dict[str, Any]:
|
||
sanitized = _sanitize_memory(memory)
|
||
status = _normalize_memory_status(sanitized.get("status"))
|
||
incoming_score = _confidence_score(sanitized.get("confidence"))
|
||
if existing is None:
|
||
return {
|
||
"packet_version": "pmflow_ai_memory_conflict_resolution.v1",
|
||
"resolution": "create",
|
||
"reason": "no_existing_memory",
|
||
"confidence_movement": {"incoming": incoming_score, "existing": None},
|
||
}
|
||
existing_memory = _row_memory_dict(existing)
|
||
existing_score = float(getattr(existing, "confidence_score", 0.0) or 0.0)
|
||
if status in {"stale", "invalidated", "rejected"}:
|
||
return {
|
||
"packet_version": "pmflow_ai_memory_conflict_resolution.v1",
|
||
"resolution": "invalidate",
|
||
"reason": "incoming_lifecycle_status_invalidates_existing",
|
||
"old_memory_id": str(getattr(existing, "public_memory_id", "") or "")[:80],
|
||
"confidence_movement": {"incoming": incoming_score, "existing": existing_score},
|
||
}
|
||
if bool(memory.get("supersedes_current") or memory.get("supersedes_existing")):
|
||
return {
|
||
"packet_version": "pmflow_ai_memory_conflict_resolution.v1",
|
||
"resolution": "supersede",
|
||
"reason": "incoming_memory_declares_supersession",
|
||
"old_memory_id": str(getattr(existing, "public_memory_id", "") or "")[:80],
|
||
"confidence_movement": {"incoming": incoming_score, "existing": existing_score},
|
||
}
|
||
delta = int(structured_delta if structured_delta is not None else _structured_delta_count(sanitized, existing_memory))
|
||
novelty = float(novelty_ratio if novelty_ratio is not None else 0.0)
|
||
if delta > 0 and novelty >= 0.35:
|
||
resolution = "keep_both"
|
||
reason = "incoming_adds_distinct_structured_context"
|
||
elif delta > 0 or novelty >= 0.08:
|
||
resolution = "merge"
|
||
reason = "incoming_extends_existing_memory"
|
||
else:
|
||
resolution = "no_change"
|
||
reason = "incoming_memory_duplicates_existing"
|
||
return {
|
||
"packet_version": "pmflow_ai_memory_conflict_resolution.v1",
|
||
"resolution": resolution,
|
||
"reason": reason,
|
||
"old_memory_id": str(getattr(existing, "public_memory_id", "") or "")[:80],
|
||
"novelty_ratio": round(novelty, 4),
|
||
"structured_delta": delta,
|
||
"confidence_movement": {"incoming": incoming_score, "existing": existing_score},
|
||
}
|
||
|
||
|
||
def _gate_result(
|
||
action: str,
|
||
reason: str,
|
||
*,
|
||
status: str,
|
||
confidence_score: float,
|
||
supersedes_memory_id: int | None = None,
|
||
invalidation_reason: str = "",
|
||
**extra: Any,
|
||
) -> dict[str, Any]:
|
||
return {
|
||
"packet_version": "pmflow_ai_memory_write_gate.v1",
|
||
"action": action,
|
||
"reason": reason,
|
||
"status": _normalize_memory_status(status),
|
||
"confidence_score": max(0.0, min(float(confidence_score or 0.0), 1.0)),
|
||
"supersedes_memory_id": supersedes_memory_id,
|
||
"invalidation_reason": _clean_text(invalidation_reason, limit=1000),
|
||
**extra,
|
||
}
|
||
|
||
|
||
def _merge_memory(*, existing: PMFlowAIContextMemory | None, incoming: dict[str, Any]) -> dict[str, Any]:
|
||
new_memory = _sanitize_memory(incoming)
|
||
if existing is None:
|
||
return new_memory
|
||
old_memory = _row_memory_dict(existing)
|
||
old_summary = _clean_text(old_memory.get("summary") or "", limit=3000)
|
||
new_summary = _clean_text(new_memory.get("summary") or "", limit=3000)
|
||
if not old_summary:
|
||
summary = new_summary
|
||
elif not new_summary or _text_terms(new_summary).issubset(_text_terms(old_summary)):
|
||
summary = old_summary
|
||
else:
|
||
summary = _clean_text(f"{old_summary}\n{new_summary}", limit=4000)
|
||
return {
|
||
"summary": summary,
|
||
"durable_facts": _merge_list_rows(old_memory.get("durable_facts"), new_memory.get("durable_facts"), limit=24),
|
||
"open_loops": _merge_list_rows(old_memory.get("open_loops"), new_memory.get("open_loops"), limit=16),
|
||
"decisions": _merge_list_rows(old_memory.get("decisions"), new_memory.get("decisions"), limit=16),
|
||
"preferences": _merge_list_rows(old_memory.get("preferences"), new_memory.get("preferences"), limit=16),
|
||
"object_refs": _merge_list_rows(old_memory.get("object_refs"), new_memory.get("object_refs"), limit=16),
|
||
"confidence": _higher_confidence(old_memory.get("confidence"), new_memory.get("confidence")),
|
||
"status": new_memory.get("status") or old_memory.get("status") or "active",
|
||
"invalidation_reason": new_memory.get("invalidation_reason") or old_memory.get("invalidation_reason") or "",
|
||
}
|
||
|
||
|
||
def _sanitize_memory(value: dict[str, Any]) -> dict[str, Any]:
|
||
summary = _clean_text(value.get("summary") or "", limit=4000)
|
||
return {
|
||
"summary": summary,
|
||
"durable_facts": _safe_list(value.get("durable_facts"), limit=24, item_limit=900),
|
||
"open_loops": _safe_list(value.get("open_loops"), limit=16, item_limit=700),
|
||
"decisions": _safe_list(value.get("decisions"), limit=16, item_limit=700),
|
||
"preferences": _safe_list(value.get("preferences"), limit=16, item_limit=700),
|
||
"object_refs": _safe_list(value.get("object_refs"), limit=16, item_limit=500),
|
||
"confidence": str(value.get("confidence") or "medium")[:32],
|
||
"status": _normalize_memory_status(value.get("status")),
|
||
"invalidation_reason": _clean_text(value.get("invalidation_reason") or "", limit=1000),
|
||
}
|
||
|
||
|
||
def _deterministic_compaction_memories(
|
||
*,
|
||
user_message: str,
|
||
assistant_answer: str,
|
||
user_message_index: int | None,
|
||
assistant_message_index: int | None,
|
||
) -> dict[str, dict[str, Any]]:
|
||
user_text = _clean_text(user_message, limit=900)
|
||
answer_text = _clean_text(assistant_answer, limit=1400)
|
||
summary = "；".join(item for item in [f"用户提出：{user_text}" if user_text else "", f"PMFlow AI 回答：{answer_text}" if answer_text else ""] if item)
|
||
base = {
|
||
"summary": summary[:3600],
|
||
"durable_facts": [
|
||
{
|
||
"kind": "turn_summary",
|
||
"summary": summary[:900],
|
||
"source_message_indexes": [item for item in [user_message_index, assistant_message_index] if item is not None],
|
||
}
|
||
]
|
||
if summary
|
||
else [],
|
||
"open_loops": [],
|
||
"decisions": [],
|
||
"preferences": [],
|
||
"object_refs": [],
|
||
"confidence": "medium",
|
||
}
|
||
return {scope: dict(base) for scope in _MEMORY_SCOPES}
|
||
|
||
|
||
def _row_memory_dict(row: PMFlowAIContextMemory) -> dict[str, Any]:
|
||
return {
|
||
"summary": str(getattr(row, "summary", "") or ""),
|
||
"durable_facts": _safe_list(getattr(row, "durable_facts", None), limit=24, item_limit=900),
|
||
"open_loops": _safe_list(getattr(row, "open_loops", None), limit=16, item_limit=700),
|
||
"decisions": _safe_list(getattr(row, "decisions", None), limit=16, item_limit=700),
|
||
"preferences": _safe_list(getattr(row, "preferences", None), limit=16, item_limit=700),
|
||
"object_refs": _safe_list(getattr(row, "object_refs", None), limit=16, item_limit=500),
|
||
"confidence": str(getattr(row, "confidence", "") or "medium")[:32],
|
||
"status": str(getattr(row, "status", "") or "active")[:32],
|
||
"invalidation_reason": str(getattr(row, "invalidation_reason", "") or "")[:1000],
|
||
}
|
||
|
||
|
||
def _memory_text(memory: dict[str, Any]) -> str:
|
||
return " ".join(
|
||
item
|
||
for item in [
|
||
str(memory.get("summary") or ""),
|
||
_dump_json(memory.get("durable_facts") or [], limit=5000),
|
||
_dump_json(memory.get("open_loops") or [], limit=3000),
|
||
_dump_json(memory.get("decisions") or [], limit=3000),
|
||
_dump_json(memory.get("preferences") or [], limit=3000),
|
||
]
|
||
if item
|
||
)
|
||
|
||
|
||
def _merge_list_rows(left: Any, right: Any, *, limit: int) -> list[Any]:
|
||
out: list[Any] = []
|
||
seen: set[str] = set()
|
||
for item in [*list(left or []), *list(right or [])]:
|
||
key = _dedupe_key(item)
|
||
if not key or key in seen:
|
||
continue
|
||
seen.add(key)
|
||
out.append(item)
|
||
if len(out) >= limit:
|
||
break
|
||
return out
|
||
|
||
|
||
def _structured_delta_count(left: dict[str, Any], right: dict[str, Any]) -> int:
|
||
delta = 0
|
||
for key in ("durable_facts", "open_loops", "decisions", "preferences", "object_refs"):
|
||
right_keys = {_dedupe_key(item) for item in list(right.get(key) or [])}
|
||
for item in list(left.get(key) or []):
|
||
item_key = _dedupe_key(item)
|
||
if item_key and item_key not in right_keys:
|
||
delta += 1
|
||
return delta
|
||
|
||
|
||
def _dedupe_key(value: Any) -> str:
|
||
if isinstance(value, dict):
|
||
text = str(value.get("summary") or value.get("display_name") or value.get("object_id") or value)
|
||
else:
|
||
text = str(value or "")
|
||
return re.sub(r"\s+", "", text.strip().lower())[:240]
|
||
|
||
|
||
def _normalize_memory_status(value: Any) -> str:
|
||
status = str(value or "").strip().lower()
|
||
if status in {"active", "candidate", "stale", "invalidated", "rejected"}:
|
||
return status
|
||
return "active"
|
||
|
||
|
||
def _confidence_score(value: Any) -> float:
|
||
if isinstance(value, (int, float)):
|
||
return max(0.0, min(float(value), 1.0))
|
||
label = str(value or "").strip().lower()
|
||
if label == "high":
|
||
return 0.85
|
||
if label == "low":
|
||
return 0.35
|
||
return 0.6
|
||
|
||
|
||
def _higher_confidence(left: Any, right: Any) -> str:
|
||
return str(left or "medium") if _confidence_score(left) >= _confidence_score(right) else str(right or "medium")
|
||
|
||
|
||
def _memory_has_content(memory: dict[str, Any]) -> bool:
|
||
return bool(str(memory.get("summary") or "").strip() or memory.get("durable_facts") or memory.get("open_loops"))
|
||
|
||
|
||
def _extract_json_object(raw: Any) -> dict[str, Any]:
|
||
if isinstance(raw, dict):
|
||
return raw
|
||
text = str(raw or "").strip()
|
||
if not text:
|
||
return {}
|
||
try:
|
||
parsed = json.loads(text)
|
||
return parsed if isinstance(parsed, dict) else {}
|
||
except Exception:
|
||
pass
|
||
match = re.search(r"\{.*\}", text, flags=re.S)
|
||
if not match:
|
||
return {}
|
||
try:
|
||
parsed = json.loads(match.group(0))
|
||
return parsed if isinstance(parsed, dict) else {}
|
||
except Exception:
|
||
return {}
|
||
|
||
|
||
def _text_terms(value: Any) -> set[str]:
|
||
text = str(value or "").lower()
|
||
alnum = {item for item in re.findall(r"[a-z0-9_]{2,}", text)}
|
||
cjk = re.findall(r"[\u4e00-\u9fff]", text)
|
||
cjk_terms = {"".join(cjk[index : index + 2]) for index in range(max(0, len(cjk) - 1))}
|
||
return {item for item in [*alnum, *cjk_terms] if item}
|
||
|
||
|
||
def _safe_list(value: Any, *, limit: int, item_limit: int) -> list[Any]:
|
||
rows = value if isinstance(value, list) else []
|
||
out: list[Any] = []
|
||
for item in rows[: max(0, limit)]:
|
||
if isinstance(item, dict):
|
||
cleaned = {
|
||
str(key)[:80]: _clean_text(val, limit=item_limit) if not isinstance(val, (dict, list)) else val
|
||
for key, val in item.items()
|
||
if val not in (None, "", [], {})
|
||
}
|
||
if cleaned:
|
||
out.append(cleaned)
|
||
continue
|
||
text = _clean_text(item, limit=item_limit)
|
||
if text:
|
||
out.append(text)
|
||
return out
|
||
|
||
|
||
def _dump_json(value: Any, *, limit: int) -> str:
|
||
try:
|
||
text = json.dumps(value, ensure_ascii=False, default=str)
|
||
except Exception:
|
||
text = str(value or "")
|
||
return text[: max(0, limit)]
|
||
|
||
|
||
def _clean_text(value: Any, *, limit: int) -> str:
|
||
text = " ".join(str(value or "").replace("\x00", " ").split())
|
||
return text[: max(0, limit)]
|
||
|
||
|
||
def _safe_int(value: Any) -> int | None:
|
||
try:
|
||
if value is None or value == "":
|
||
return None
|
||
return int(value)
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def _estimated_tokens(value: Any) -> int:
|
||
text = _dump_json(value, limit=60000)
|
||
return max(1, int(len(text) / 3.5))
|
||
|
||
|
||
def _iso(value: Any) -> str:
|
||
if value is None:
|
||
return ""
|
||
try:
|
||
return value.isoformat()
|
||
except Exception:
|
||
return str(value or "")
|
||
|
||
|
||
__all__ = [
|
||
"FUNCTION_NAME_PMFLOW_AI_CONTEXT_MEMORY_COMPACTOR",
|
||
"MEMORY_PACKET_VERSION",
|
||
"PMFlowAIContextMemoryService",
|
||
"PROMPT_KEY_PMFLOW_AI_CONTEXT_MEMORY_COMPACTOR",
|
||
"_resolve_memory_conflict",
|
||
]
|