File size: 1,844 Bytes
66cfaef | 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 | """
DeAbstractionLayer: Privileged operation OUTSIDE the model.
Maps abstract reasoning outputs back to concrete entities.
"""
from typing import Dict, Any, Optional
class DeAbstractionLayer:
"""
Converts abstract text back to concrete text using entity vault mappings.
This is a PRIVILEGED operation that runs OUTSIDE the model.
The de-abstraction keys are NEVER part of model weights or training data.
"""
def __init__(self, vault_store: Optional[Dict[str, Any]] = None):
self.vault_store = vault_store or {}
def register_vault(self, vault_id: str, mapping: Dict[str, Any]):
self.vault_store[vault_id] = mapping
def deabstract(self, abstract_text: str, vault_id: str) -> str:
if vault_id not in self.vault_store:
raise ValueError(f"Vault {vault_id} not found. Cannot de-abstract securely.")
mapping = self.vault_store[vault_id]
concrete_text = abstract_text
sorted_tokens = sorted(mapping.keys(), key=len, reverse=True)
for token in sorted_tokens:
entity_info = mapping[token]
concrete_value = entity_info.get("surface", token)
concrete_text = concrete_text.replace(token, concrete_value)
return concrete_text
def deabstract_batch(self, abstract_texts, vault_ids):
return [self.deabstract(text, vid) for text, vid in zip(abstract_texts, vault_ids)]
def get_entity_metadata(self, vault_id: str, abstract_token: str) -> Optional[Dict]:
if vault_id not in self.vault_store:
return None
return self.vault_store[vault_id].get(abstract_token)
def list_vaults(self) -> list:
return list(self.vault_store.keys())
def delete_vault(self, vault_id: str):
if vault_id in self.vault_store:
del self.vault_store[vault_id]
|