要交给 AI 相关的模块来处理。

我在之前一篇介绍 init_subclass() 方法的博客中有提到过怎么处理命令，不过那里面只能处理简单格式的命令，命令文本只能按空格切片，不支持 --xx 这样的参数。在想着怎么处理这些不同形式的命令参数时，我突然想起来 Python 标准库里面的 argparse。直接用 argparse 来解析不更方便嘛！简单看了下 argparse 的文档和源码，感觉应该可行，说干就干！

流程逻辑
简单描述下流程逻辑：

用户通过 HTTP API /api/chat 发送消息
后端应用接收到消息后，按空格分割用户输入的文本，拿到第一段
匹配命令字典，如果没匹配到，则当成自然语言处理
匹配到命令字典后，交给命令类处理。命令类创建命令解析器来解析参数
返回结果给用户
按照习惯，具体命令类是动态加载的，不需要在代码中挨个引入。这样以后添加命令时，只要在指定目录添加代码文件，然后按照规范开发具体命令类即可。

本文主要介绍如何用 argparse 在 web 应用中解析用户命令，并不包含 AI 处理自然语言的相关实现，所以本文用到的第三方依赖只有 FastAPI 充当 HTTP 框架，换成 Flask 或其它框架也是没问题的。

代码实现
代码结构：

├── internal
│ └── cmd
│ ├── admin.py
│ ├── base.py
│ ├── demo.py
│ └── init.py
├── main.py
├── pyproject.toml
└── README.md
核心抽象：ChatArgparser 与 ChatCommand
argparse 是为命令行工具设计的，默认行为是解析出错时直接打印错误信息并退出进程，这显然不适合 web 应用。所以我们需要继承 argparse.ArgumentParser，重写它的 error()、exit() 和 print_help() 方法，把"退出进程"变成"抛出异常"。这样一来，异常被上层捕获后，就能以 HTTP 响应的形式返回给用户。

ChatArgparser 做了三件事：

重写 error()：不调用 sys.exit()，而是记录错误信息并抛出 argparse.ArgumentError。
重写 exit()：argparse 在用户输入 --help 时会调用 exit()，这里同样改为抛异常，同时把帮助文本附在异常信息里。
重写 print_help()：把帮助信息输出到 StringIO 缓冲区，存起来备用。
class ChatArgparser(argparse.ArgumentParser):
def error(self, message):
self.parse_error_triggered = True
self.error_message = message
raise argparse.ArgumentError(None, message)

def exit(self, status=0, message=None):
    self.parse_error_triggered = True
    if self.help_text:
        self.error_message = f"Help requested:\n{self.help_text}"
    elif message:
        self.error_message = message
    raise argparse.ArgumentError(None, self.error_message)

ChatCommand 是所有命令的抽象基类，定义了两个接口：create_parser() 返回一个 ChatArgparser 实例，声明该命令接受的参数；run() 是异步方法，执行实际的命令逻辑。

命令加载：自动发现与注册
load_chat_commands() 函数负责扫描 internal.cmd 包下的所有模块，找出继承自 ChatCommand 的类，然后根据类属性 main_name、is_enable、is_visible 来判断是否注册。

跳过 base 和 init 这两个模块，避免把基类和自己注册进去。每个命令类需要定义几个类属性：

main_name：命令名，以 / 开头，比如 /demo。
description：命令的简要说明。
is_enable：是否启用该命令，关闭后不会被注册。
is_visible：是否在帮助列表中显示，适合隐藏管理员命令。
HelpCommand 是内置的帮助命令，遍历所有已注册的可见命令，拼接出帮助信息返回。

class HelpCommand(ChatCommand):
main_name: str = “/help”
description: str = “Show help message for all commands”
is_visible: bool = True

async def run(self) -> str:
    help_message = "Available commands:\n"
    for main_name, info in _loaded_chat_commands.items():
        if info["is_visible"]:
            help_message += f"{main_name}: {info['description']}\n"
    return help_message

具体命令示例
以 DemoCommand 为例，它接受 --name 和 --age 两个参数。在 run() 中，先用 shlex.split() 把用户消息按 shell 语法拆成列表，去掉第一个元素（即命令本身），然后把剩余参数交给 ChatArgparser 解析。

这里用 shlex.split() 而不是直接 str.split()，是因为用户在 IM 中输入参数时可能会用引号包裹有空格的参数值，shlex.split() 能正确处理这种情况。

class DemoCommand(ChatCommand):
main_name: str = “/demo”
description: str = “Demo command for testing”
is_enable: bool = True
is_visible: bool = True

async def run(self) -> str:
    cmd_args = shlex.split(self.user_message)[1:]
    parsed_args = self.arg_parser.parse_args(cmd_args)
    return f"Hello, {parsed_args.name}! You are {parsed_args.age} years old."

def create_parser(self) -> ChatArgparser:
    parser = ChatArgparser(prog="demo", description=self.description)
    parser.add_argument("--name", type=str, help="Name of the user")
    parser.add_argument("--age", type=int, help="Age of the user")
    return parser

AdminCommand 的结构类似，不同之处在于 is_visible = False，这样它不会出现在 /help 的输出中，只有知道具体命令的管理员才能使用。

HTTP 接口：/api/chat
main.py 中的 /api/chat 端点接收用户消息，处理流程如下：

用 strip().split(" “) 取出第一个词，判断是否以 / 开头。
不以 / 开头，说明是自然语言，直接返回，交给 AI 模块处理（本文略过）。
以 / 开头，调用 load_chat_commands() 查找对应命令。找不到也按自然语言处理。
找到命令后，实例化命令类，调用 run() 执行。
整个流程用 try/except 包裹，捕获 argparse.ArgumentError——如果异常信息以 “Help requested:” 开头，说明用户输入了 --help，直接把帮助文本返回；否则返回解析错误提示。
@app.post(”/api/chat")
async def post_chat(req: RequestChat):
msg_list = req.message.strip().split(" “)
if not msg_list[0].startswith(”/"):
return {“info”: “自然语言, 预期将由AI处理”}

cmders = load_chat_commands()
if msg_list[0] not in cmders:
    return {"info": "未知命令, 预期将由AI处理"}

cmd_cls = cmders[msg_list[0]]["cmdcls"]
cmd_instance = cmd_cls(req.message)
rst = await cmd_instance.run()
return {"result": rst}

实际效果
实际应用中可以稍微美化下输出

发送/help, 获取可用命令。因为/admin设置不可见，所以不会输出出来
curl --request POST
–url http://127.0.0.1:10001/api/chat
–header ‘content-type: application/json’
–data ‘{
“session_id”: “qwerasd”,
“message”: “/help”
}’

响应

{
“session_id”: “qwerasd”,
“result”: “Available commands:\n/demo: Demo command for testing\n/help: Show help message for all commands\n”
}
用户发送 /demo --help
curl --request POST
–url http://127.0.0.1:10001/api/chat
–header ‘content-type: application/json’
–data ‘{
“session_id”: “qwerasd”,
“message”: “/demo --help”
}’

响应

{
“session_id”: “qwerasd”,
“result”: “Help requested:\nusage: demo [-h] [–name NAME] [–age AGE]\n\nDemo command for testing\n\noptions:\n -h, --help show this help message and exit\n --name NAME Name of the user\n --age AGE Age of the user\n”
}
用户发送 /admin --host 192.168.1.1 --port=12345
curl --request POST
–url http://127.0.0.1:10001/api/chat
–header ‘content-type: application/json’
–data ‘{
“session_id”: “qwerasd”,
“message”: “/admin --host 192.168.1.1 --port=12345”
}’

响应

{
“session_id”: “qwerasd”,
“result”: “Admin command executed! Host: 192.168.1.1, Port: 12345”
}
改进点
命令类是否启用和可见性应该配置在别处，或者支持动态配置。
实际应用中要考虑添加权限控制。
动态加载命令类的方法的确有点黑箱，如果命令不多的话，也可以在代码中手动挨个导入。
完整示例代码
internal/cmd/base.py
import argparse
from abc import ABC, abstractmethod
from io import StringIO

class ChatArgparser(argparse.ArgumentParser):
“”“自定义的ArgumentParser, 用于解析聊天命令的参数, 重写error和exit方法, 捕获解析错误并返回错误信息, 而不是直接退出程序”“”
def init(self, *args, **kwargs):
super().init(*args, **kwargs)
self.parse_error_triggered = False
self.error_message = “”
self.help_text = “”

def print_help(self, file=None):
    """重写print_help方法, 捕获帮助信息, 以便在解析错误时返回给用户"""
    help_buffer = StringIO()
    super().print_help(help_buffer)
    self.help_text = help_buffer.getvalue()

def error(self, message):
    """重写ArgumentParser的error方法: 不退出进程, 捕获解析错误并记录错误信息"""
    self.parse_error_triggered = True
    self.error_message = message

    # 抛出异常后, 中断后续的参数解析流程
    raise argparse.ArgumentError(None, message)

def exit(self, status=0, message=None):
    """重写ArgumentParser的exit方法: 不退出进程, 捕获退出调用并记录错误信息"""
    self.parse_error_triggered = True
    if self.help_text:
        self.error_message = f"Help requested:\n{self.help_text}"
    elif message:
        self.error_message = message
    else:
        self.error_message = "Exit triggered without message"

    raise argparse.ArgumentError(None, self.error_message)

class ChatCommand(ABC):
“”“聊天命令的抽象基类, 定义了命令的基本结构和接口”“”
def init(self, user_message: str):
self.user_message = user_message

@abstractmethod
def create_parser(self) -> ChatArgparser:
    """创建并返回一个ChatArgparser实例, 定义命令的参数结构"""
    ...

@abstractmethod
async def run(self) -> str:
    """执行命令的异步方法, 返回命令执行结果"""
    ...

internal/cmd/demo.py
import argparse
import shlex

from internal.cmd.base import ChatArgparser, ChatCommand

class DemoCommand(ChatCommand):
main_name: str = “/demo”
description: str = “Demo command for testing”
is_enable: bool = True
is_visible: bool = True

def __init__(self, user_message: str):
    super().__init__(user_message)
    self.arg_parser = self.create_parser()

async def run(self) -> str:
    try:
        cmd_args = shlex.split(self.user_message)[1:]  # 去掉命令本身
    except ValueError as e:
        return f"shlex 参数解析错误: {str(e)}"

    try:
        parsed_args = self.arg_parser.parse_args(cmd_args)
        return f"Hello, {parsed_args.name}! You are {parsed_args.age} years old."
    except argparse.ArgumentError as e:
        error_msg = str(e)
        if error_msg.startswith("Help requested:"):
            return error_msg
        return f"parser 参数解析错误: {str(e)}"

def create_parser(self) -> ChatArgparser:
    parser = ChatArgparser(prog="demo", description=self.description)

    parser.add_argument(
        "--name",
        type=str,
        help="Name of the user",
    )
    parser.add_argument(
        "--age",
        type=int,
        help="Age of the user",
    )
    return parser

internal/cmd/admin.py
import argparse
import shlex

from internal.cmd.base import ChatArgparser, ChatCommand

class AdminCommand(ChatCommand):
main_name: str = “/admin”
description: str = “Admin command”
is_enable: bool = True
is_visible: bool = False # 管理命令默认不在/help中显示, 需要管理员知道具体命令才使用

def __init__(self, user_message: str):
    super().__init__(user_message)
    self.arg_parser = self.create_parser()

async def run(self) -> str:
    try:
        cmd_args = shlex.split(self.user_message)[1:]  # 去掉命令本身
    except ValueError as e:
        return f"shlex 参数解析错误: {str(e)}"

    try:
        parsed_args = self.arg_parser.parse_args(cmd_args)
        return f"Admin command executed! Host: {parsed_args.host}, Port: {parsed_args.port}"
    except argparse.ArgumentError as e:
        error_msg = str(e)
        if error_msg.startswith("Help requested:"):
            return error_msg
        return f"parser 参数解析错误: {str(e)}"

def create_parser(self) -> ChatArgparser:
    parser = ChatArgparser(prog="admin", description=self.description)

    parser.add_argument(
        "--host",
        type=str,
        help="Hostname or IP address of the server",
    )
    parser.add_argument(
        "--port",
        type=int,
        help="Port number of the server",
    )
    return parser

internal/cmd/init.py
from future import annotations

import importlib
import pkgutil
from typing import Dict, TypedDict

from .base import ChatArgparser, ChatCommand

class CommandInfo(TypedDict):
description: str
cmdcls: type[ChatCommand]
is_visible: bool

_loaded_chat_commands: Dict[str, CommandInfo] = {}

class HelpCommand(ChatCommand):
“”“内置的帮助命令, 用于展示所有可用命令的帮助信息”“”
main_name: str = “/help”
description: str = “Show help message for all commands”
is_visible: bool = True

def create_parser(self) -> ChatArgparser:
    """HelpCommand 不需要参数, 直接返回一个空的ChatArgparser实例"""
    return ChatArgparser(
        prog="help", description="Show help message for all commands"
    )

async def run(self) -> str:
    """执行帮助命令, 返回所有可用命令的帮助信息"""
    if not _loaded_chat_commands:
        load_chat_commands()

    help_message = "Available commands:\n"
    for main_name, info in _loaded_chat_commands.items():
        if info["is_visible"]:
            help_message += f"{main_name}: {info['description']}\n"
    return help_message

def load_chat_commands() -> Dict[str, CommandInfo]:
“”“加载所有命令类”“”
if _loaded_chat_commands:
return _loaded_chat_commands

pkg_path = "internal.cmd"
pkg = importlib.import_module(pkg_path)
print(f"Loading chat commands from package: {pkg_path}")

for _, name, ispkg in pkgutil.iter_modules(pkg.__path__, pkg.__name__ + "."):
    # 如果以后各个命令类比较复杂, 可以把命令类放在一个单独的模块中, 加载的时候只加载模块
    # 目前命令类比较简单, 就直接放在internal.cmd包下, 加载的时候直接加载模块中的类
    # if not ispkg:
    #     continue
    if ispkg:
        continue
    skipped_modules = {"base", "__init__"}
    if any(name.endswith(skiped) for skiped in skipped_modules):
        continue
    module = importlib.import_module(name)
    for attr_name in dir(module):
        attr = getattr(module, attr_name)
        if (
            isinstance(attr, type)
            and issubclass(attr, ChatCommand)
            and attr is not ChatCommand
        ):
            main_name = getattr(attr, "main_name", None)
            description = getattr(attr, "description", None)
            is_enable = getattr(attr, "is_enable", False)
            is_visible = getattr(attr, "is_visible", True)
            if not main_name or not description:
                continue
            if not is_enable:
                continue
            main_name = main_name.strip()
            description = description.strip()
            if main_name.startswith("/") and main_name not in _loaded_chat_commands:
                _loaded_chat_commands[main_name] = {
                    "description": description,
                    "cmdcls": attr,
                    "is_visible": is_visible,
                }

# 手动注册HelpCommand, 确保/help命令始终可用
if "/help" not in _loaded_chat_commands:
    _loaded_chat_commands["/help"] = {
        "description": HelpCommand.description,
        "cmdcls": HelpCommand,
        "is_visible": True,
    }

return _loaded_chat_commands

main.py
import argparse
from contextlib import asynccontextmanager

import uvicorn
from fastapi import FastAPI
from pydantic import BaseModel, Field, ValidationInfo, field_validator

from internal.cmd import load_chat_commands

class RequestChat(BaseModel):
session_id: str = Field(
…, min_length=1, description=“Unique identifier for the chat session”
)
message: str = Field(
…, min_length=1, description=“The chat message sent by the user”
)

@field_validator("session_id", "message")
@classmethod
def validate_fields(cls, v: str, info: ValidationInfo) -> str:
    if not v or not v.strip():
        raise ValueError(f"Field '{info.field_name}' cannot be empty")
    return v.strip()

@asynccontextmanager
async def lifespan(app: FastAPI):
print(“Starting up…”)
try:
yield
finally:
print(“Shutting down…”)

app = FastAPI(lifespan=lifespan)

@app.post(“/api/chat”)
async def post_chat(req: RequestChat):
try:
msg_list = req.message.strip().split(" “)
if not msg_list[0].startswith(”/"):
return {
“session_id”: req.session_id,
“message”: req.message,
“info”: “自然语言, 预期将由AI处理”,
}

    cmders = load_chat_commands()
    if msg_list[0] not in cmders:
        return {
            "session_id": req.session_id,
            "message": req.message,
            "info": "未知命令, 预期将由AI处理",
        }

    cmd_cls = cmders[msg_list[0]]["cmdcls"]
    cmd_instance = cmd_cls(req.message)
    rst = await cmd_instance.run()
    return {"session_id": req.session_id, "result": rst}

except argparse.ArgumentError as e:
    error_msg = str(e)
    if error_msg.startswith("Help requested:"):
        return {"session_id": req.session_id, "result": error_msg}
    return {"session_id": req.session_id, "message": f"参数解析错误: {str(e)}"}
except Exception as e:
    return {"session_id": req.session_id, "message": f"参数解析错误: {str(e)}"}

if name == “main”:
uvicorn.run(“main:app”, host=“127.0.0.1”, port=10001, workers=1)