Spaces:
Running
Running
File size: 5,048 Bytes
c57e87b | 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 | """User-owned facts and source exclusions, without extra model calls."""
from hashlib import sha256
from uuid import uuid4
from engine.relationship import get_default_relationship
def fingerprint(text):
return sha256(' '.join(text.casefold().split()).encode()).hexdigest()
def upgrade_conversation(state):
state.setdefault('relationship', get_default_relationship())
state.setdefault('excluded_sources', [])
state.setdefault('forgotten_facts', [])
for i, item in enumerate(state['long_term_memory']):
item.setdefault('id', fingerprint(f'{i}:{item["content"]}')[:32])
item.setdefault('origin', 'auto')
item.setdefault('start', 0)
item.setdefault('end', state['summarized_count'])
return state
def context_messages(state, start=0, end=None):
end = len(state['recent_messages']) if end is None else end
return [dict(message) if not any(a <= index < b for a, b in state['excluded_sources'])
else {'role': message['role'], 'content': '[Message excluded from memory by the user.]'}
for index, message in enumerate(state['recent_messages'][start:end], start)]
def _exclude_fact(state, item):
blocked = fingerprint(item['content'])
if blocked not in state['forgotten_facts']:
if len(state['forgotten_facts']) >= 1000:
raise ValueError('The forgotten-fact limit has been reached.')
state['forgotten_facts'].append(blocked)
if item['end'] > item['start']:
state['excluded_sources'].append([item['start'], item['end']])
# Also mask verbatim echoes of the fact later in the transcript.
needle = ' '.join(item['content'].casefold().split())
for index, message in enumerate(state['recent_messages']):
if needle in ' '.join(message['content'].casefold().split()):
state['excluded_sources'].append([index, index + 1])
merged = []
for start, end in sorted(state['excluded_sources']):
if merged and start <= merged[-1][1]:
merged[-1][1] = max(merged[-1][1], end)
else:
merged.append([start, end])
state['excluded_sources'] = merged
# The old summary may contain a paraphrase; rebuild it from unexcluded text.
state['summary'], state['summarized_count'] = '', 0
def overlaps(record):
return any(record['start'] < end and start < record['end'] for start, end in merged)
state['long_term_memory'] = [m for m in state['long_term_memory']
if m['id'] != item['id'] and not (m['origin'] == 'auto' and overlaps(m))]
state['relationship']['events'] = [e for e in state['relationship']['events'] if not overlaps(e)]
def change_memory(session_id, character_id, action, content='', memory_id=None):
from engine.memory import _LOCK, load_conversation, save_conversation
with _LOCK:
state = load_conversation(session_id, character_id)
if action not in ('remember', 'edit', 'forget'):
raise ValueError('Unknown memory action.')
if action != 'forget' and (not isinstance(content, str) or not content.strip() or len(content) > 2000):
raise ValueError('Enter a memory between 1 and 2000 characters.')
item = next((m for m in state['long_term_memory'] if m['id'] == memory_id), None)
if action in ('edit', 'forget'):
if item is None:
raise ValueError('Select an existing memory first.')
_exclude_fact(state, item)
if action != 'forget':
if sum(m['origin'] == 'manual' for m in state['long_term_memory']) >= 60:
raise ValueError('Up to 60 personal memories can be saved.')
content = content.strip()
state['forgotten_facts'] = [f for f in state['forgotten_facts'] if f != fingerprint(content)]
state['long_term_memory'] = [m for m in state['long_term_memory'] if fingerprint(m['content']) != fingerprint(content)]
state['long_term_memory'].append({'id': memory_id if action == 'edit' else uuid4().hex,
'content': content, 'importance': 'high', 'origin': 'manual', 'start': 0, 'end': 0})
save_conversation(session_id, character_id, state, state['revision'])
return state
def set_relationship_options(session_id, character_id, romance, attraction):
from engine.memory import _LOCK, load_conversation, save_conversation
if type(romance) is not bool or type(attraction) is not bool:
raise ValueError('Relationship options must be booleans.')
with _LOCK:
state = load_conversation(session_id, character_id)
enabled = {'romance': romance, 'attraction': attraction}
if state['relationship']['enabled'] != enabled:
state['relationship']['enabled'] = enabled
state['relationship']['events'] = [e for e in state['relationship']['events']
if e['axis'] not in enabled or enabled[e['axis']]]
save_conversation(session_id, character_id, state, state['revision'])
return state
|