Skip to content

Cache Utils

summ.cache.CacheDocument

Bases: EmbeddedJsonModel

A serializable version of a Document.

Source code in summ/cache/cacher.py 
13
14
15
16
17
18
19
20
21
22
23
24
class CacheDocument(EmbeddedJsonModel):
    """A serializable version of a Document."""

    page_content: str
    metadata: dict = Field(default_factory=dict)

    @classmethod
    def from_doc(cls, doc: Document):
        for k, v in doc.metadata.items():
            if isinstance(v, types.GeneratorType):
                doc.metadata[k] = list(v)
        return cls(**doc.dict())

summ.cache.CacheItem

Bases: JsonModel

A base class for cached responses.

Source code in summ/cache/cacher.py 
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
class CacheItem(JsonModel):
    """A base class for cached responses."""

    @classmethod
    def passthrough(cls, **kwargs) -> Self:
        instance = cls.construct(**kwargs)
        instance.pk = cls.make_pk(instance)
        if cached := cls.safe_get(instance.pk):
            return cached

        for k in cls.__fields__.keys() - kwargs.keys():
            setattr(instance, k, None)

        try:
            return cast(Self, instance.save())
        except Exception:
            return instance

    @classmethod
    def safe_get(cls, pk: Optional[str]) -> Optional[Self]:
        try:
            return cast(Self, cls.get(pk))
        except NotFoundError:
            return None

    @staticmethod
    def _hash(s: str):
        return metrohash.hash64(s, seed=0).hex()

    @classmethod
    @abstractmethod
    def make_pk(cls, instance: Self) -> str:
        raise NotImplementedError

    @override
    def save(self, *args, **kwargs) -> "JsonModel":
        self.pk = self.make_pk(self)
        return super().save(*args, **kwargs)

summ.cache.ChainCacheItem

Bases: CacheItem

A base class for cached langchain LLM responses.

Source code in summ/cache/cacher.py 
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
class ChainCacheItem(CacheItem):
    """A base class for cached langchain LLM responses."""

    klass: str
    name: str
    document: Union[CacheDocument, list[CacheDocument]]
    result: str
    meta: dict = Field(default_factory=dict)

    def page_contents(self) -> list[str]:
        if isinstance(self.document, list):
            return [doc.page_content for doc in self.document]
        return [self.document.page_content]

    @classmethod
    def make_pk(cls, instance: Self) -> str:
        return cls._hash(
            ":".join(
                [
                    instance.klass,
                    instance.name,
                    *instance.page_contents(),
                    json.dumps(instance.meta, sort_keys=True),
                ]
            )
        )