| """ |
| 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] |
|
|
