扩展 CLI#
LMCache 的 CLI 基于插件式架构构建,支持 N 级嵌套子命令,并具有零注册自动发现功能。本指南解释了如何在每个级别添加命令。
架构概述#
CLI 框架由两个核心类组成,位于 lmcache/cli/commands/base.py:
BaseCommand — 叶命令的抽象基类(执行实际工作的命令)。
CompositeCommand — 一个
BaseCommand子类,用于仅将子子命令分组的命令。它通过扫描定义它的包自动发现子命令。
发现机制使用 pkgutil.iter_modules 扫描包的直接子模块,然后收集在这些模块中找到的所有具体的 BaseCommand 子类。这意味着:
每个命令都是一个单独的
.py文件(或带有__init__.py的子包)。无需手动注册 — 只需创建文件,它会自动被识别。
工具/辅助模块应以
_开头(例如_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 组(例如 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)
步骤 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 (顶部) |
单个 |
创建 |
2+ |
包目录 |
创建 |
N (任何) |
嵌套包 |
与级别 2 相同,但在现有复合命令的包内。每个 |
小技巧
将帮助/工具模块前缀加上
_以将其排除在自动发现之外。每个
CompositeCommand必须在其包的__init__.py中定义。name()方法确定 CLI 令牌(例如"foo"变为lmcache ... foo)。