"""
Domain-Specific Few-Shot Examples Cache Manager
This module manages caching of domain-specific few-shot examples generated by the
DomainGraphExtractor. Examples are saved to and loaded from JSON files organized by domain.
The cache helps avoid regenerating examples for the same domain multiple times,
improving performance when processing multiple documents from the same domain.
"""
import json
import logging
from pathlib import Path
from typing import Dict, Optional
logger = logging.getLogger(__name__)
# Mapping of example types to variable names for consistency
EXAMPLE_TYPE_MAPPING = {
'entity_extraction': 'examples_for_entity_extraction',
'triples_extraction': 'examples_for_triples_extraction',
'type_assertion': 'examples_for_type_assertion',
'type_generation': 'examples_for_type_generation',
'literal_extraction': 'examples_for_literal_extraction',
'triples_with_numeric_literals_extraction': 'examples_for_spl_triples_extraction',
}
# Reverse mapping
VARIABLE_NAME_TO_TYPE = {v: k for k, v in EXAMPLE_TYPE_MAPPING.items()}
[docs]
class DomainExamplesCache:
"""
Manages caching of domain-specific few-shot examples.
Examples are stored in JSON format with the filename pattern:
domain_examples_{domain_name}.json
The cache directory defaults to the current working directory but can be configured.
"""
def __init__(self, cache_dir: Optional[str] = None):
"""
Initialize the cache manager.
Args:
cache_dir: Directory to store cached examples. If None, uses current working directory.
"""
self.cache_dir = Path(cache_dir) if cache_dir else Path.cwd()
self.cache_dir.mkdir(parents=True, exist_ok=True)
def _get_cache_file(self, domain: str) -> Path:
"""
Get the cache file path for a given domain.
Args:
domain: The domain name.
Returns:
Path object for the cache file.
"""
# Sanitize domain name for filename
safe_domain = self._sanitize_domain_name(domain)
return self.cache_dir / f"domain_examples_{safe_domain}.json"
@staticmethod
def _sanitize_domain_name(domain: str) -> str:
"""
Sanitize domain name for use in filenames.
Args:
domain: The domain name to sanitize.
Returns:
Sanitized domain name.
"""
# Replace spaces and special characters with underscores
safe_name = domain.lower().strip()
safe_name = ''.join(c if c.isalnum() else '_' for c in safe_name)
# Remove consecutive underscores
while '__' in safe_name:
safe_name = safe_name.replace('__', '_')
return safe_name
[docs]
def save_examples(self, domain: str, examples: Dict[str, str]) -> bool:
"""
Save generated examples for a domain.
Args:
domain: The domain name.
examples: Dictionary mapping example types to example strings.
Expected keys are from EXAMPLE_TYPE_MAPPING.keys()
Returns:
True if save was successful, False otherwise.
"""
try:
cache_file = self._get_cache_file(domain)
# Validate that all example types are present
for example_type in EXAMPLE_TYPE_MAPPING.keys():
if example_type not in examples:
raise ValueError(f"Missing example type: {example_type}")
# Save to JSON
with open(cache_file, 'w', encoding='utf-8') as f:
json.dump(examples, f, indent=2, ensure_ascii=False)
return True
except Exception:
logger.exception(f"Failed to save examples for domain '{domain}'")
return False
[docs]
def load_examples(self, domain: str) -> Optional[Dict[str, str]]:
"""
Load cached examples for a domain.
Args:
domain: The domain name.
Returns:
Dictionary mapping example types to example strings, or None if not cached.
"""
try:
cache_file = self._get_cache_file(domain)
if not cache_file.exists():
return None
with open(cache_file, 'r', encoding='utf-8') as f:
examples = json.load(f)
# Validate structure
for example_type in EXAMPLE_TYPE_MAPPING.keys():
if example_type not in examples:
logger.warning(f"Missing example type '{example_type}' in cached examples for domain '{domain}'")
return None
return examples
except Exception:
logger.exception(f"Failed to load examples for domain '{domain}'")
return None
[docs]
def examples_exist(self, domain: str) -> bool:
"""
Check if cached examples exist for a domain.
Args:
domain: The domain name.
Returns:
True if examples are cached, False otherwise.
"""
cache_file = self._get_cache_file(domain)
return cache_file.exists()
[docs]
def get_cache_file_path(self, domain: str) -> str:
"""
Get the full path to the cache file for a domain.
Args:
domain: The domain name.
Returns:
String path to the cache file.
"""
return str(self._get_cache_file(domain))
[docs]
def clear_domain_cache(self, domain: str) -> bool:
"""
Delete the cache file for a domain.
Args:
domain: The domain name.
Returns:
True if deletion was successful, False otherwise.
"""
try:
cache_file = self._get_cache_file(domain)
if cache_file.exists():
cache_file.unlink()
return True
except Exception:
logger.exception(f"Failed to clear cache for domain '{domain}'")
return False
[docs]
def clear_all_caches(self) -> bool:
"""
Delete all domain example cache files.
Returns:
True if all deletions were successful, False otherwise.
"""
try:
cache_files = list(self.cache_dir.glob("domain_examples_*.json"))
for cache_file in cache_files:
cache_file.unlink()
return True
except Exception:
logger.exception("Failed to clear all domain example caches")
return False
[docs]
def list_cached_domains(self) -> list:
"""
List all domains that have cached examples.
Returns:
List of domain names with cached examples.
"""
try:
cache_files = list(self.cache_dir.glob("domain_examples_*.json"))
domains = []
for cache_file in cache_files:
# Extract domain name from filename
filename = cache_file.stem # e.g., "domain_examples_biology"
domain_name = filename.replace("domain_examples_", "", 1)
domains.append(domain_name)
return sorted(domains)
except Exception:
logger.exception("Failed to list cached domains")
return []