Spaces:
Running on Zero
Running on Zero
File size: 4,878 Bytes
e726170 | 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 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 | """Cache identity and file-backed result cache for article classifications."""
from __future__ import annotations
import hashlib
import json
import os
from pathlib import Path
from typing import Any
from pydantic import BaseModel, ConfigDict, Field
from gcmd_classifier.config import ModelSettings
from gcmd_classifier.models import ArticleRecord, ArticleResult
from gcmd_classifier.vocabulary.index import VocabularyIndex
class CacheIdentity(BaseModel):
"""All inputs that can affect reuse of a cached article result."""
model_config = ConfigDict(extra="forbid", frozen=True)
DOI: str
article_fingerprint: str
vocabulary_version: str
model_provider: str
model_name: str
model_temperature: float
model_timeout_seconds: float
model_max_retries: int
prompt_version_topic: str
prompt_version_term: str
prompt_version_variable: str
application_version: str
configuration_hash: str
@property
def cache_key(self) -> str:
"""Stable cache key derived from the complete identity."""
payload = json.dumps(self.model_dump(mode="json"), sort_keys=True, separators=(",", ":"))
return hashlib.sha256(payload.encode("utf-8")).hexdigest()
class CachedArticleResult(BaseModel):
"""Persisted cache entry containing identity and article result."""
model_config = ConfigDict(extra="forbid", frozen=True)
cache_key: str = Field(min_length=1)
identity: CacheIdentity
result: ArticleResult
class ArticleResultCache:
"""Simple file-backed article result cache keyed by cache identity."""
def __init__(self, directory: str | Path) -> None:
self.directory = Path(directory)
self.directory.mkdir(parents=True, exist_ok=True)
def get(self, identity: CacheIdentity) -> ArticleResult | None:
"""Return a completed cached result for the exact identity, if present."""
path = self._path(identity.cache_key)
if not path.exists():
return None
entry = CachedArticleResult.model_validate(json.loads(path.read_text()))
if entry.cache_key != identity.cache_key or entry.identity != identity:
return None
return mark_cache_used(entry.result)
def put(self, identity: CacheIdentity, result: ArticleResult) -> None:
"""Persist a result for later exact-identity reuse."""
entry = CachedArticleResult(
cache_key=identity.cache_key,
identity=identity,
result=result,
)
_atomic_write_json(self._path(identity.cache_key), entry.model_dump(mode="json"))
def _path(self, cache_key: str) -> Path:
return self.directory / f"{cache_key}.json"
def article_fingerprint(article: ArticleRecord) -> str:
"""Hash exact source article values that influence classification."""
payload = json.dumps(article.model_dump(mode="json"), sort_keys=True, separators=(",", ":"))
return hashlib.sha256(payload.encode("utf-8")).hexdigest()
def configuration_hash(config: dict[str, Any] | None = None) -> str:
"""Hash relevant non-secret configuration values."""
payload = json.dumps({} if config is None else config, sort_keys=True, separators=(",", ":"))
return hashlib.sha256(payload.encode("utf-8")).hexdigest()
def build_cache_identity(
*,
article: ArticleRecord,
vocabulary: VocabularyIndex,
settings: ModelSettings,
application_version: str,
relevant_config: dict[str, Any] | None = None,
) -> CacheIdentity:
"""Build the complete cache identity for one article classification run."""
return CacheIdentity(
DOI=article.DOI,
article_fingerprint=article_fingerprint(article),
vocabulary_version=vocabulary.vocabulary_version,
model_provider=settings.provider,
model_name=settings.model_name,
model_temperature=settings.temperature,
model_timeout_seconds=settings.timeout_seconds,
model_max_retries=settings.max_retries,
prompt_version_topic=settings.prompt_version_topic,
prompt_version_term=settings.prompt_version_term,
prompt_version_variable=settings.prompt_version_variable,
application_version=application_version,
configuration_hash=configuration_hash(relevant_config),
)
def mark_cache_used(result: ArticleResult) -> ArticleResult:
"""Return a cached copy of a result marked as completed work from cache."""
metadata = result.processing_metadata.model_copy(update={"cache_used": True})
return result.model_copy(update={"processing_metadata": metadata})
def _atomic_write_json(path: Path, payload: dict[str, Any]) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
temp_path = path.with_name(f".{path.name}.tmp")
temp_path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n")
os.replace(temp_path, path)
|