Caching & usage¶
Caching and usage tracking are per-model concerns. You wrap a language model in
a ManagedLanguageModel that adds a cache, a usage tracker, or both. There is no
global cache switch; you choose the behavior for each model you build.
Caching¶
ManagedLanguageModel takes a cache implementation. Pick one to match how long
results should live:
def uncached(lm : LanguageModel): LanguageModel = ManagedLanguageModel(lm, cache = Some(NoopLmCache))
def memoryCached(lm: LanguageModel): LanguageModel = ManagedLanguageModel(lm, cache = Some(new InMemoryLmCache()))
def diskCached(lm: LanguageModel, dir: Path): LanguageModel =
ManagedLanguageModel(lm, cache = Some(new DiskLmCache(dir)))
NoopLmCachedisables caching.InMemoryLmCachecaches for the life of the process.DiskLmCache(dir)persists results to a directory.
You can also implement LmCache yourself to control the cache key, for example
to key only on the messages and ignore the model name.
Tracking usage¶
Wrap a block in UsageTracking.withNewTracker and set trackUsage on the
context. The tracker accumulates token usage across the calls inside the block. A
cache hit contributes no new usage:
def usageAcrossCachedCalls(lm: LanguageModel, question: String)(using
RuntimeContext
)
: Either[DspyError, Map[String, LmUsage]] =
val managed = memoryCached(lm)
UsageTracking.withNewTracker { tracker =>
RuntimeEnvironment.withSettings(
summon[RuntimeContext].copy(lm = Some(managed), adapter = Some(ChatAdapter()), trackUsage = Some(true))
) {
given RuntimeContext = RuntimeEnvironment.current
for
_ <- Predict(qa)((question = question)) // miss: records usage
_ <- Predict(qa)((question = question)) // hit: fast, no new usage
yield tracker.totalUsage
}
}
When to use it¶
| You want | Approach |
|---|---|
| Avoid repeat model calls in a process | InMemoryLmCache |
| Persist results across runs | DiskLmCache(dir) |
| Measure token cost of a workload | UsageTracking.withNewTracker + trackUsage = Some(true) |
Next: Evaluation.