Spaces:
Running
Running
| """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 | |