skip_provisioning: bool = False,
    ):
        """
        Initialize with a ready session and a keyspace name.
        Args:
            session (cassandra.cluster.Session): an open Cassandra session
            keyspace (str): the keyspace to use for storing the cache
            table_name (str): name of the Cassandra table to use as cache
            ttl_seconds (optional int): time-to-live for cache entries
                (default: None, i.e. forever)
        """
        try:
            from cassio.table import ElasticCassandraTable
        except (ImportError, ModuleNotFoundError):
            raise ValueError(
                "Could not import cassio python package. "
                "Please install it with `pip install cassio`."
            )
        self.session = session
        self.keyspace = keyspace
        self.table_name = table_name
        self.ttl_seconds = ttl_seconds
        self.kv_cache = ElasticCassandraTable(
            session=self.session,
            keyspace=self.keyspace,
            table=self.table_name,
            keys=["llm_string", "prompt"],
            primary_key_type=["TEXT", "TEXT"],
            ttl_seconds=self.ttl_seconds,
            skip_provisioning=skip_provisioning,
        )
[docs]    def lookup(self, prompt: str, llm_string: str) -> Optional[RETURN_VAL_TYPE]:
        """Look up based on prompt and llm_string."""
        item = self.kv_cache.get(
            llm_string=_hash(llm_string),
            prompt=_hash(prompt),
        )
        if item is not None:
            generations = _loads_generations(item["body_blob"])
            # this protects against malformed cached items:
            if generations is not None:
                return generations
            else:
                return None