扩展 CLI#

LMCache 的 CLI 基于插件式架构构建，支持 N 级嵌套子命令，并具有零注册自动发现功能。本指南解释了如何在每个级别添加命令。

架构概述#

CLI 框架由两个核心类组成，位于 lmcache/cli/commands/base.py：

  • BaseCommand — 叶命令的抽象基类（执行实际工作的命令）。

  • CompositeCommand — 一个 BaseCommand 子类，用于仅将子子命令分组的命令。它通过扫描定义它的包自动发现子命令。

发现机制使用 pkgutil.iter_modules 扫描包的直接子模块，然后收集在这些模块中找到的所有具体的 BaseCommand 子类。这意味着：

  1. 每个命令都是一个单独的 .py 文件（或带有 __init__.py 的子包）。

  2. 无需手动注册 — 只需创建文件，它会自动被识别。

  3. 工具/辅助模块应以 _ 开头（例如 _helpers.py），以便在扫描时将其排除。

自动发现机制#

CLI 使用统一的 discover_subclasses() 工具（定义在 lmcache/v1/utils/subclass_discovery.py 中）来定位每个级别的命令类。理解这两个发现入口是扩展 CLI 的关键。

顶级命令发现#

When an invocation runs, main(argv) calls register_commands(subparsers, route), which calls discover_subclasses() on its own package (lmcache.cli.commands). This scans all direct submodules — both .py files and sub-packages (via their __init__.py) — and collects every concrete BaseCommand subclass found. Merely importing the package imports no concrete command wrapper modules.

# Simplified from lmcache/cli/commands/__init__.py (runs per invocation)
from lmcache.v1.utils.subclass_discovery import discover_subclasses

commands = [
    cls()
    for cls in discover_subclasses(
        __name__,                          # "lmcache.cli.commands"
        BaseCommand,
        module_filter=lambda name: name != "base",  # skip base.py itself
    )
]

The command matching the invocation route is then fully registered with the argparse parser; unselected commands contribute only name/help, so every listing stays complete. This means any new .py file (or sub-package) placed under lmcache/cli/commands/ is automatically available as a top-level command — no manual imports or registration needed.

顶级发现的过滤规则：

  • 名为 base 的模块被明确排除（它定义了抽象基类，而不是一个命令）。

  • 以 _ 开头的模块根据底层的 pkgutil.iter_modules 约定被排除（它们被视为私有）。

  • 仅收集 具体 子类 — 抽象类会被跳过。

  • 仅收集在扫描模块中**定义的**类（通过 require_defined_in_module=True 忽略重新导出）。

子命令发现（CompositeCommand）#

当 CompositeCommand 向解析器注册时，它的 register() 方法会在定义 CompositeCommand 子类的包上调用 discover_subclasses()``（即它自己的 ``__init__.py 所在的包）。

# Simplified from CompositeCommand.register() in base.py
package = self.__class__.__module__   # e.g. "lmcache.cli.commands.bench"

for cls in discover_subclasses(
    package,
    BaseCommand,
    module_filter=lambda name: not name.startswith("_"),
    require_defined_in_module=True,
):
    if cls is self.__class__:
        continue          # skip the CompositeCommand itself
    inst = cls()
    inst.register(inner)  # register as a nested subcommand

子命令发现的过滤规则：

  • 以 _ 开头的模块会被排除（使用这个前缀来命名像 _utils.py 这样的辅助文件）。

  • CompositeCommand 类本身被跳过以避免无限递归。

  • 仅收集在扫描模块中**定义的**具体 BaseCommand 子类。

  • 扫描是 非递归 的（仅深度 1）——每个 CompositeCommand 仅能看到其所在包的直接子项。

嵌套如何工作： 如果一个子命令本身是一个 CompositeCommand``（在子包的 ``__init__.py 中定义），它自己的 register() 方法将反过来扫描 它 的包以查找更多子命令。这在没有任何特殊配置的情况下创建了递归嵌套。

lmcache tool cache-simulator simulate ...      (one invocation)
└── register_commands(route) discovers the top-level wrappers
    ├── ping.py             → summary only (unselected)
    ├── bench/__init__.py   → summary: children NOT scanned
    └── tool/__init__.py    → selected: registers, scans children
        └── cache_simulator/__init__.py → selected: registers, scans children
            ├── sweep_command.py        → summary only
            ├── gen_dataset_command.py  → summary only
            └── simulate_command.py     → fully registered (add_arguments runs)

目录结构#

lmcache/cli/commands/
├── __init__.py            # Top-level discovery (scans this package)
├── base.py                # BaseCommand & CompositeCommand
├── ping.py                # Level-1 leaf command
├── server.py              # Level-1 leaf command
├── quota/                 # Level-2 composite command
│   ├── __init__.py        # QuotaCommand(CompositeCommand)
│   ├── _helpers.py        # Utility (excluded from scan by _ prefix)
│   ├── get_command.py     # Level-2 leaf: ``lmcache quota get``
│   ├── set_command.py     # Level-2 leaf: ``lmcache quota set``
│   └── ...
└── tool/                  # Level-2 composite command
    ├── __init__.py         # ToolCommand(CompositeCommand)
    └── cache_simulator/    # Level-3 composite command
        ├── __init__.py     # CacheSimulatorCommand(CompositeCommand)
        ├── simulate_command.py   # Level-3 leaf: ``lmcache tool cache-simulator simulate``
        └── sweep_command.py      # Level-3 leaf: ``lmcache tool cache-simulator sweep``

关键规则#

  • 一个 CompositeCommand 子类 必须 在其包的 __init__.py 中定义。

  • 以 _ 开头的模块会被 排除 在自动发现之外（用于帮助器、工具、内部逻辑）。

  • 每个叶命令文件应包含一个具体的 BaseCommand 子类。

  • 扫描是 非递归 的——每个 CompositeCommand 仅扫描其自身包的直接子模块。

级别 1：添加顶级命令#

一个顶级命令直接出现在 lmcache <command> 之下。

步骤 1：在 lmcache/cli/commands/ 下创建一个新文件：

# lmcache/cli/commands/hello.py
"""``lmcache hello`` — a simple greeting command."""

import argparse

from lmcache.cli.commands.base import BaseCommand


class HelloCommand(BaseCommand):
    """Print a greeting message."""

    def name(self) -> str:
        return "hello"

    def help(self) -> str:
        return "Print a greeting message."

    def add_arguments(self, parser: argparse.ArgumentParser) -> None:
        parser.add_argument("--name", default="World", help="Who to greet.")

    def execute(self, args: argparse.Namespace) -> None:
        metrics = self.create_metrics("Hello", args)
        metrics.add("greeting", "Greeting", f"Hello, {args.name}!")
        metrics.emit()

步骤 2：完成！命令会自动被发现。测试一下：

lmcache hello --name LMCache

备注

This works because main() calls register_commands() per invocation, which runs discover_subclasses on the package. It uses pkgutil.iter_modules to find all direct submodules (files and sub-packages), imports each one, and collects every concrete BaseCommand subclass. The command matching the invocation route is fully registered; the others bind name/help summaries. So adding a new .py file with a BaseCommand subclass is all that is needed — no edits to any other file.

级别 2：添加子命令组#

子命令组的形式为 lmcache <group> <subcommand>。

步骤 1：创建一个包目录：

mkdir lmcache/cli/commands/mygroup/

第 2 步：在 __init__.py 中定义 CompositeCommand：

# lmcache/cli/commands/mygroup/__init__.py
"""``lmcache mygroup`` command group.

Sub-subcommands are auto-discovered from modules in this package.
"""

from lmcache.cli.commands.base import CompositeCommand


class MyGroupCommand(CompositeCommand):
    """Command group for my custom operations."""

    def name(self) -> str:
        return "mygroup"

    def help(self) -> str:
        return "My custom command group."

第 3 步：将叶子子命令作为单独的文件添加：

# lmcache/cli/commands/mygroup/foo_command.py
"""``lmcache mygroup foo`` — do something."""

import argparse

from lmcache.cli.commands.base import BaseCommand


class FooCommand(BaseCommand):
    """Execute the foo action."""

    def name(self) -> str:
        return "foo"

    def help(self) -> str:
        return "Execute the foo action."

    def add_arguments(self, parser: argparse.ArgumentParser) -> None:
        parser.add_argument("--value", type=int, required=True)

    def execute(self, args: argparse.Namespace) -> None:
        metrics = self.create_metrics("Foo Result", args)
        metrics.add("result", "Result", args.value * 2)
        metrics.emit()

步骤 4： （可选）添加以 _ 前缀的辅助模块：

# lmcache/cli/commands/mygroup/_utils.py
"""Internal utilities for mygroup commands (not auto-discovered)."""

def compute_something(x: int) -> int:
    return x * 42

结果:

lmcache mygroup foo --value 5

级别 2：向现有组添加子命令#

如果已经存在一个 CompositeCommand 组（例如 bench、quota、trace），您可以通过简单地添加 一个新文件 来扩展它 — 不需要其他更改。

例如，要在现有的 bench 组下添加一个新的 lmcache bench l2 子命令：

步骤 1：在现有组的包目录中创建一个单文件（或子包）：

# lmcache/cli/commands/bench/l2_adapter_bench/__init__.py
"""``lmcache bench l2`` subpackage."""

import argparse

from lmcache.cli.commands.base import BaseCommand


class L2AdapterBenchCommand(BaseCommand):
    """Benchmark an L2 adapter (store / lookup / load)."""

    def name(self) -> str:
        return "l2"

    def help(self) -> str:
        return "Benchmark an L2 adapter (store / lookup / load)."

    def add_arguments(self, parser: argparse.ArgumentParser) -> None:
        from lmcache.cli.commands.bench.l2_adapter_bench.command import (
            add_l2_arguments,
        )
        add_l2_arguments(parser)

    def execute(self, args: argparse.Namespace) -> None:
        from lmcache.cli.commands.bench.l2_adapter_bench.command import (
            run_l2_adapter_bench,
        )
        run_l2_adapter_bench(self, args)

Step 2: Done! The parent CompositeCommand (BenchCommand) auto-discovers the new subcommand whenever the route selects it. No imports to add, no __init__.py edits in the parent.

备注

This works because CompositeCommand.register() scans all direct submodules of its package whenever the route selects it. A new file (or sub-package) is automatically picked up as long as:

  • 它**不**以``_``开头。

  • 它包含一个具体的 BaseCommand 子类。

级别 N：任意嵌套#

该框架支持 无限嵌套深度。每个级别遵循相同的模式：包的 __init__.py 中的 CompositeCommand 自动发现其子命令。

示例：在 lmcache mygroup 下添加第 3 级：

mkdir lmcache/cli/commands/mygroup/nested/

# lmcache/cli/commands/mygroup/nested/__init__.py
"""``lmcache mygroup nested`` — a nested command group."""

from lmcache.cli.commands.base import CompositeCommand


class NestedCommand(CompositeCommand):
    """Nested subcommand group."""

    def name(self) -> str:
        return "nested"

    def help(self) -> str:
        return "A nested command group under mygroup."

# lmcache/cli/commands/mygroup/nested/bar_command.py
"""``lmcache mygroup nested bar`` — a deeply nested command."""

import argparse

from lmcache.cli.commands.base import BaseCommand


class BarCommand(BaseCommand):
    """Execute the bar action at level 3."""

    def name(self) -> str:
        return "bar"

    def help(self) -> str:
        return "Execute the bar action."

    def add_arguments(self, parser: argparse.ArgumentParser) -> None:
        parser.add_argument("--msg", default="deep")

    def execute(self, args: argparse.Namespace) -> None:
        metrics = self.create_metrics("Bar Result", args)
        metrics.add("message", "Message", args.msg)
        metrics.emit()

结果:

lmcache mygroup nested bar --msg "hello from level 3"

您可以通过重复此模式无限制地继续嵌套。

现实世界示例#

现有的 lmcache tool cache-simulator simulate 命令演示了 3 级嵌套：

lmcache tool cache-simulator simulate
│       │    │               └── Level-3 leaf (SimulateCommand in simulate_command.py)
│       │    └── Level-2 composite (CacheSimulatorCommand in cache_simulator/__init__.py)
│       └── Level-1 composite (ToolCommand in tool/__init__.py)
└── CLI entry point

使用指标系统#

指标系统使用 处理器 + 格式化器 架构：

  • 指标 — 收集器。包含部分和条目。

  • 处理器 — 目标（标准输出、文件等）。

  • 格式化器 — 渲染（ASCII 表格、JSON 等）。

BaseCommand.create_metrics() 会自动配置默认处理器，命令开发者只需构建指标数据并调用 emit() 即可：

def execute(self, args: argparse.Namespace) -> None:
    # create_metrics() auto-registers:
    #   - StreamHandler → stdout (formatter chosen by --format, default: terminal)
    #   - FileHandler   → if --output is set (same format as --format)
    metrics = self.create_metrics("Bench KV Cache Result", args)

    # Create named sections
    metrics.add_section("ops", "Operations (ops/s)")
    metrics["ops"].add("store", "Store", 41.3)
    metrics["ops"].add("retrieve", "Retrieve", 127.3)

    # Top-level metrics (no section header)
    metrics.add("status", "Status", "OK")

    # Trigger all handlers
    metrics.emit()

--format 和 --output 标志由 BaseCommand.register() 自动添加 — 子命令无需手动添加。

摘要#

级别

模式

如何添加

1 (顶部)

单个 .py 文件

创建 lmcache/cli/commands/<name>.py，并使用 BaseCommand 子类。

2+

包目录

创建 lmcache/cli/commands/<group>/__init__.py，并使用 CompositeCommand 子类，然后将叶命令作为同级 .py 文件添加。

N (任何)

嵌套包

与级别 2 相同，但在现有复合命令的包内。每个 CompositeCommand 仅扫描其直接子模块。

小技巧

  • 将帮助/工具模块前缀加上 _ 以将其排除在自动发现之外。

  • 每个 CompositeCommand 必须在其包的 __init__.py 中定义。

  • name() 方法确定 CLI 令牌（例如 "foo" 变为 lmcache ... foo）。