| import re |
| from typing import List |
|
|
| from schemas import MemoryWriteDecision |
|
|
|
|
| _SENSITIVE_PATTERNS = { |
| "credential": [ |
| r"\bpassword\b", |
| r"\bpasscode\b", |
| r"\bapi[ _-]?key\b", |
| r"\baccess[ _-]?token\b", |
| r"\brefresh[ _-]?token\b", |
| r"\bprivate[ _-]?key\b", |
| r"\bsecret[ _-]?key\b", |
| r"\bbearer[ _-]?token\b", |
| ], |
| "identity": [ |
| r"\bsocial security\b", |
| r"\bssn\b", |
| r"\bpassport\b", |
| ], |
| "financial": [ |
| r"\bcredit card\b", |
| r"\bcard number\b", |
| r"\brouting number\b", |
| r"\bbank account\b", |
| r"\bbrokerage account\b", |
| ], |
| "health": [ |
| r"\bmedical record\b", |
| r"\bmedical history\b", |
| r"\bdiagnosis\b", |
| ], |
| "confidential_client": [ |
| r"\bconfidential client\b", |
| r"\bclient account\b", |
| r"\bclient portfolio\b", |
| ], |
| "email_address": [ |
| r"\b[A-Z0-9._%+-]+@[A-Z0-9.-]+\.[A-Z]{2,}\b", |
| ], |
| } |
|
|
| _DURABLE_MARKERS = ( |
| "remember", |
| "we decided", |
| "the team decided", |
| "scheduled", |
| "deadline", |
| "due date", |
| "must", |
| "should", |
| "owns", |
| "owner", |
| "prefer", |
| "preference", |
| "launch date", |
| ) |
|
|
| _QUESTION_PREFIXES = ("what ", "when ", "who ", "where ", "why ", "how ", "can ", "do ", "does ") |
|
|
|
|
| def detect_sensitive_content(text: str) -> List[str]: |
| flags: List[str] = [] |
| for category, patterns in _SENSITIVE_PATTERNS.items(): |
| if any(re.search(pattern, text, flags=re.IGNORECASE) for pattern in patterns): |
| flags.append(category) |
| return flags |
|
|
|
|
| def _normalize_memory_statement(text: str) -> str: |
| statement = text.strip() |
| lowered = statement.lower() |
| prefix = "remember that " |
| if lowered.startswith(prefix): |
| statement = statement[len(prefix):].strip() |
| if statement and statement[-1] not in ".!?": |
| statement += "." |
| if statement: |
| statement = statement[0].upper() + statement[1:] |
| return statement |
|
|
|
|
| def evaluate_memory_write(user_query: str) -> MemoryWriteDecision: |
| query = (user_query or "").strip() |
| if not query: |
| return MemoryWriteDecision( |
| outcome="skip", |
| should_save=False, |
| reason="Empty input does not create a durable memory.", |
| ) |
|
|
| flags = detect_sensitive_content(query) |
| if flags: |
| return MemoryWriteDecision( |
| outcome="block", |
| should_save=False, |
| reason="The public demo does not store content that appears sensitive or credential-like.", |
| policy_flags=flags, |
| ) |
|
|
| lowered = query.lower() |
| explicit_memory_request = lowered.startswith("remember") |
| looks_like_question = lowered.startswith(_QUESTION_PREFIXES) or query.endswith("?") |
| durable_marker = any(marker in lowered for marker in _DURABLE_MARKERS) |
|
|
| if looks_like_question and not explicit_memory_request: |
| return MemoryWriteDecision( |
| outcome="skip", |
| should_save=False, |
| reason="A retrieval question is useful for this turn but is not itself durable project memory.", |
| ) |
|
|
| if not durable_marker: |
| return MemoryWriteDecision( |
| outcome="skip", |
| should_save=False, |
| reason="The interaction does not contain a clear durable decision, preference, owner, schedule, or explicit remember request.", |
| ) |
|
|
| memory_type = "semantic" if any(term in lowered for term in ("prefer", "preference", "owns", "owner")) else "episodic" |
| importance = 5 if any(term in lowered for term in ("must", "deadline", "launch date", "we decided", "the team decided")) else 4 |
|
|
| return MemoryWriteDecision( |
| outcome="save", |
| should_save=True, |
| memory_type=memory_type, |
| importance=importance, |
| reason="The interaction contains a durable, public-demo-safe project decision or preference.", |
| proposed_memory=_normalize_memory_statement(query), |
| ) |
|
|
