扩展 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 的关键。

顶级命令发现#

当 CLI 启动时,lmcache/cli/commands/__init__.py 在其自身的包(lmcache.cli.commands)上调用 discover_subclasses()。这会扫描所有 直接子模块 — 包括 .py 文件和子包(通过它们的 __init__.py) — 并收集找到的每个具体的 BaseCommand 子类。

# Simplified from lmcache/cli/commands/__init__.py
from lmcache.v1.utils.subclass_discovery import discover_subclasses

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

生成的 ALL_COMMANDS 列表随后在 main.py 中注册到根 argparse 解析器。这意味着放置在 lmcache/cli/commands/ 下的任何新 .py 文件(或子包)会自动作为顶级命令可用 — 无需手动导入或注册。

顶级发现的过滤规则:

  • 名为 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() 方法将反过来扫描 的包以查找更多子命令。这在没有任何特殊配置的情况下创建了递归嵌套。

CLI startup
└── __init__.py discovers ALL top-level commands
    ├── ping.py          → PingCommand (leaf)
    ├── bench/__init__.py → BenchCommand (composite)
    │   └── BenchCommand.register() discovers:
    │       ├── server_bench/__init__.py → ServerBenchCommand (leaf)
    │       ├── l2_adapter_bench/__init__.py → L2AdapterBenchCommand (leaf)
    │       └── engine_bench/__init__.py → EngineBenchCommand (leaf)
    └── tool/__init__.py → ToolCommand (composite)
        └── ToolCommand.register() discovers:
            └── cache_simulator/__init__.py → CacheSimulatorCommand (composite)
                └── CacheSimulatorCommand.register() discovers:
                    ├── simulate_command.py     → SimulateCommand (leaf)
                    ├── sweep_command.py        → SweepCommand (leaf)
                    └── gen_dataset_command.py  → GenDatasetCommand (leaf)

目录结构#

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

备注

这之所以有效,是因为顶层的 lmcache/cli/commands/__init__.py 在导入时调用了 discover_subclasses。它使用 pkgutil.iter_modules 查找所有直接子模块(文件和子包),导入每一个,并收集每个具体的 BaseCommand 子类。结果列表存储在 ALL_COMMANDS 中,并在 main.py 中与参数解析器注册。因此,只需添加一个带有 BaseCommand 子类的新 .py 文件即可——不需要编辑其他任何文件。

级别 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 组(例如 benchquotatrace),您可以通过简单地添加 一个新文件 来扩展它 — 不需要其他更改。

例如,要在现有的 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)

步骤 2:完成!父级 CompositeCommand (BenchCommand) 在启动时会自动发现新的子命令。无需注册代码,无需添加导入,无需在父级中编辑 __init__.py

备注

这工作是因为 CompositeCommand.register() 每次 CLI 启动时都会扫描其包的所有直接子模块。只要满足以下条件,新文件(或子包)会被自动识别:

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

  • 它包含一个具体的 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)。