"""LangChain cache bootstrap helpers.
Purpose:
Provide small utilities for resolving a project-local LLM cache path and
configuring the global LangChain cache.
Design:
- Keep imports lazy so the module can be imported without LangChain.
- Default to SQLite because it is file-backed, simple, and local.
- Support networked caches through optional Redis and Upstash Redis clients.
- Reuse :class:`ooai_llm.settings.AppSettings` for path resolution.
Examples:
>>> from pathlib import Path
>>> settings = AppSettings(app_root=Path.cwd())
>>> resolve_llm_cache_path(settings).name
'langchain_llm_cache.sqlite3'
"""
from __future__ import annotations
from pathlib import Path
from typing import TYPE_CHECKING, Any
from .logging import get_logger, log_event
from .settings import AppSettings
try:
from langchain_core.caches import BaseCache
except ImportError: # pragma: no cover - langchain_core is a package dependency
[docs]
BaseCache = object # type: ignore[assignment,misc]
if TYPE_CHECKING:
from langchain_core.caches import BaseCache as BaseCacheType
else:
[docs]
BaseCacheType = BaseCache
[docs]
logger = get_logger(__name__)
[docs]
class NamespacedCache(BaseCache):
"""Wrap a LangChain cache and add namespace/profile material to cache keys.
The wrapper rewrites only LangChain's ``llm_string`` component. It does not
mutate prompt serialization, so chat prompt cache semantics remain delegated
to the underlying cache implementation.
"""
def __init__(
self,
cache: "BaseCacheType",
*,
namespace: str,
profile_key: str,
) -> None:
[docs]
self.namespace = namespace.strip() or "default"
[docs]
self.profile_key = profile_key.strip() or "default"
[docs]
def namespaced_llm_string(self, llm_string: str) -> str:
"""Return the LangChain cache key component with namespace material."""
return f"ooai:{self.namespace}:{self.profile_key}:{llm_string}"
[docs]
def lookup(self, prompt: str, llm_string: str) -> Any:
"""Look up a cached value using the namespaced key component."""
return self.cache.lookup(prompt, self.namespaced_llm_string(llm_string))
[docs]
def update(self, prompt: str, llm_string: str, return_val: Any) -> None:
"""Update a cached value using the namespaced key component."""
self.cache.update(prompt, self.namespaced_llm_string(llm_string), return_val)
[docs]
def clear(self, **kwargs: Any) -> None:
"""Clear the wrapped cache."""
self.cache.clear(**kwargs)
[docs]
async def alookup(self, prompt: str, llm_string: str) -> Any:
"""Async cache lookup when the wrapped cache supports it."""
return await self.cache.alookup(prompt, self.namespaced_llm_string(llm_string))
[docs]
async def aupdate(self, prompt: str, llm_string: str, return_val: Any) -> None:
"""Async cache update when the wrapped cache supports it."""
await self.cache.aupdate(prompt, self.namespaced_llm_string(llm_string), return_val)
[docs]
async def aclear(self, **kwargs: Any) -> None:
"""Async cache clear when the wrapped cache supports it."""
await self.cache.aclear(**kwargs)
[docs]
def resolve_llm_cache_path(settings: AppSettings, *, path: str | Path | None = None) -> Path:
"""Resolve the effective LLM cache path.
Args:
settings: Application settings.
path: Optional explicit cache path override.
Returns:
Resolved cache path.
"""
if path is None:
return settings.default_llm_cache_path
return Path(path).expanduser().resolve()
[docs]
def build_sqlite_cache(settings: AppSettings, *, path: str | Path | None = None) -> Any:
"""Build a SQLite-backed LangChain cache.
Args:
settings: Application settings.
path: Optional explicit cache path override.
Returns:
SQLite cache instance.
Raises:
ImportError: If ``langchain_community`` is not installed.
"""
from langchain_community.cache import SQLiteCache
resolved_path = resolve_llm_cache_path(settings, path=path)
if settings.llm.cache.create_dirs:
resolved_path.parent.mkdir(parents=True, exist_ok=True)
log_event(logger, "cache.build", backend="sqlite", path=str(resolved_path))
return SQLiteCache(database_path=str(resolved_path))
[docs]
def build_memory_cache() -> Any:
"""Build an in-memory LangChain cache.
Returns:
In-memory cache instance.
Raises:
ImportError: If ``langchain_community`` is not installed.
"""
from langchain_community.cache import InMemoryCache
log_event(logger, "cache.build", backend="memory")
return InMemoryCache()
[docs]
def build_sqlalchemy_cache(settings: AppSettings, *, path: str | Path | None = None) -> Any:
"""Build a SQLAlchemy-backed LangChain cache.
Args:
settings: Application settings.
path: Optional SQLite path used when ``sqlalchemy_url`` is unset.
Returns:
SQLAlchemy cache instance.
Raises:
ImportError: If SQLAlchemy or LangChain community caches are missing.
"""
from langchain_community.cache import SQLAlchemyCache
from sqlalchemy import create_engine
sqlalchemy_url = settings.llm.cache.sqlalchemy_url
if sqlalchemy_url is None:
resolved_path = resolve_llm_cache_path(settings, path=path)
if settings.llm.cache.create_dirs:
resolved_path.parent.mkdir(parents=True, exist_ok=True)
sqlalchemy_url = f"sqlite:///{resolved_path}"
log_event(logger, "cache.build", backend="sqlalchemy", explicit_url=settings.llm.cache.sqlalchemy_url is not None)
return SQLAlchemyCache(engine=create_engine(sqlalchemy_url))
[docs]
def build_redis_cache(settings: AppSettings) -> Any:
"""Build a Redis-backed LangChain cache.
Args:
settings: Application settings.
Returns:
Redis cache instance.
Raises:
ImportError: If ``redis`` or LangChain community caches are missing.
"""
from langchain_community.cache import RedisCache
try:
import redis
except ImportError as exc: # pragma: no cover - exercised with dependency absent
raise ImportError("Install `ooai-llm[redis]` or `redis` to use the Redis cache backend.") from exc
cache_settings = settings.llm.cache
connection_kwargs = dict(cache_settings.redis_connection_kwargs)
if cache_settings.redis_url:
client = redis.Redis.from_url(cache_settings.redis_url, **connection_kwargs)
else:
if cache_settings.redis_username is not None:
connection_kwargs.setdefault("username", cache_settings.redis_username)
if cache_settings.redis_password is not None:
connection_kwargs.setdefault("password", cache_settings.redis_password.get_secret_value())
client = redis.Redis(
host=cache_settings.redis_host,
port=cache_settings.redis_port,
db=cache_settings.redis_db,
ssl=cache_settings.redis_ssl,
**connection_kwargs,
)
log_event(logger, "cache.build", backend="redis", url_configured=bool(cache_settings.redis_url))
return RedisCache(redis_=client, ttl=cache_settings.ttl)
[docs]
def build_upstash_redis_cache(settings: AppSettings) -> Any:
"""Build an Upstash Redis-backed LangChain cache.
Args:
settings: Application settings.
Returns:
Upstash Redis cache instance.
Raises:
ValueError: If the Upstash URL or token is missing.
ImportError: If ``upstash_redis`` or LangChain community caches are missing.
"""
from langchain_community.cache import UpstashRedisCache
try:
from upstash_redis import Redis
except ImportError as exc: # pragma: no cover - exercised with dependency absent
raise ImportError(
"Install `ooai-llm[upstash]` or `upstash-redis` to use the Upstash Redis cache backend."
) from exc
cache_settings = settings.llm.cache
if not cache_settings.upstash_url:
raise ValueError("Upstash Redis cache backend requires `llm.cache.upstash_url`.")
if cache_settings.upstash_token is None:
raise ValueError("Upstash Redis cache backend requires `llm.cache.upstash_token`.")
client = Redis(url=cache_settings.upstash_url, token=cache_settings.upstash_token.get_secret_value())
log_event(logger, "cache.build", backend="upstash_redis")
return UpstashRedisCache(redis_=client, ttl=cache_settings.ttl)
[docs]
def build_llm_cache(settings: AppSettings, *, path: str | Path | None = None) -> Any:
"""Build the configured LangChain cache backend.
Args:
settings: Application settings.
path: Optional SQLite path override for file-backed backends.
Returns:
Cache object.
Raises:
ValueError: If the configured backend is unsupported.
ImportError: If backend-specific dependencies are missing.
"""
backend = settings.llm.cache.backend.strip().lower().replace("-", "_")
if backend in {"sqlite", "sqlite3"}:
return build_sqlite_cache(settings, path=path)
if backend in {"memory", "in_memory", "inmemory"}:
return build_memory_cache()
if backend == "sqlalchemy":
return build_sqlalchemy_cache(settings, path=path)
if backend == "redis":
return build_redis_cache(settings)
if backend in {"upstash", "upstash_redis"}:
return build_upstash_redis_cache(settings)
supported = "sqlite, memory, sqlalchemy, redis, upstash_redis"
raise ValueError(
f"Unsupported LLM cache backend: {settings.llm.cache.backend!r}. "
f"Supported backends: {supported}."
)
[docs]
def build_namespaced_cache(
settings: AppSettings,
*,
namespace: str,
profile_key: str,
path: str | Path | None = None,
) -> NamespacedCache:
"""Build the configured cache backend wrapped with namespace key material."""
log_event(logger, "cache.namespaced.build", namespace=namespace, profile_key=profile_key)
return NamespacedCache(
build_llm_cache(settings, path=path),
namespace=namespace,
profile_key=profile_key,
)
[docs]
def normalize_cache_argument(cache: "BaseCache | bool | None") -> "BaseCache | bool | None":
"""Normalize the per-model ``cache`` argument.
Args:
cache: ``True`` to force the global cache, ``False`` to disable,
``None`` to inherit the global setting, or a concrete cache object.
Returns:
Unchanged normalized cache value.
"""
return cache