Implementation:LMCache LMCache Chunk Statistics Lookup Client

Knowledge Sources	LMCache
Domains	Lookup Client, Observability
Last Updated	2026-02-09 00:00 GMT

Overview

A decorator lookup client that wraps another lookup client to collect chunk reuse statistics, including timing data, unique/duplicate chunk counts, and automatic stop conditions.

Description

ChunkStatisticsLookupClient implements the decorator pattern around any LookupClientInterface to track KV cache chunk reuse statistics without modifying the underlying lookup logic. It records each unique request's token chunks via an asynchronous recorder (AsyncRecorder) backed by a configurable RecordStrategy (e.g., bloom filter, file hash). The client tracks cumulative timing for lookup, recording, and exit-condition checking. It supports start/stop/reset lifecycle for statistics collection, automatic stop conditions based on elapsed time (timeout_hours) or unique chunk count targets (target_unique_chunks), and Prometheus metric integration. Statistics are collected per-request (deduplicated by lookup_id) and the asynchronous recording pipeline minimizes overhead on the critical lookup path.

Usage

Use this client as a wrapper around any lookup client when chunk reuse analysis is needed. It is automatically applied by the LookupClientFactory when enable_chunk_statistics is set in the configuration. Call get_statistics to retrieve current statistics, start_statistics/stop_statistics to control collection, and wait_for_async_processing to ensure all queued records are processed before reading results.

Code Reference

Source Location

• Repository: LMCache
• File: lmcache/v1/lookup_client/chunk_statistics_lookup_client.py
• Lines: 1-200

Signature

class ChunkStatisticsLookupClient(LookupClientInterface):
    def __init__(self, actual_lookup_client: LookupClientInterface,
                 config: LMCacheEngineConfig) -> None: ...
    def lookup_cache(self, lookup_id: str) -> Optional[int]: ...
    def lookup(self, token_ids: Union[torch.Tensor, list[int]],
               lookup_id: str,
               request_configs: Optional[dict] = None) -> Optional[int]: ...
    def start_statistics(self) -> None: ...
    def stop_statistics(self) -> None: ...
    def reset_statistics(self) -> None: ...
    def get_statistics(self) -> dict: ...
    def wait_for_async_processing(self, timeout: float = 5.0) -> bool: ...
    def clear_lookup_status(self, lookup_id: str) -> None: ...
    def supports_producer_reuse(self) -> bool: ...
    def close(self) -> None: ...

Import

from lmcache.v1.lookup_client.chunk_statistics_lookup_client import ChunkStatisticsLookupClient

I/O Contract

Inputs

Name	Type	Required	Description
actual_lookup_client	LookupClientInterface	Yes	The underlying lookup client to delegate actual lookups to
config	LMCacheEngineConfig	Yes	Configuration with chunk_size, auto_exit settings, and extra config for async queue
token_ids	Union[torch.Tensor, list[int]]	Yes	Token IDs for lookup (passed through to underlying client)
lookup_id	str	Yes	Unique request identifier used for deduplication
timeout	float	No	Timeout in seconds for waiting on async processing (default: 5.0)

Outputs

Name	Type	Description
hit_tokens	Optional[int]	Number of cached tokens from the underlying lookup client (pass-through)
statistics	dict	Dictionary with enabled, total_requests, timing breakdown, total/unique/duplicate chunks, and reuse_rate
completed	bool	Whether async processing completed within the timeout (from wait_for_async_processing)

Usage Examples

from lmcache.v1.lookup_client.chunk_statistics_lookup_client import ChunkStatisticsLookupClient

# Wrap an existing lookup client
stats_client = ChunkStatisticsLookupClient(
    actual_lookup_client=base_client,
    config=engine_config,
)

# Start collecting statistics
stats_client.start_statistics()

# Perform lookups (statistics are collected automatically)
result = stats_client.lookup(token_ids=[1, 2, 3], lookup_id="req-001")

# Retrieve statistics
stats_client.wait_for_async_processing(timeout=5.0)
stats = stats_client.get_statistics()
print(f"Reuse rate: {stats['reuse_rate']:.2%}")
print(f"Unique chunks: {stats['unique_chunks']}")
print(f"Overhead: {stats['timing']['overhead_percentage']:.1f}%")

# Stop and reset
stats_client.stop_statistics()
stats_client.reset_statistics()
stats_client.close()