前言在开发一个 AI 驱动的 IM 应用 Bot 时某些场景用命令会更快更准确。我的设想是先按空格分割用户输入的文本拿到第一段去匹配命令字典如果匹配上了说明用户想要执行命令接着交给命令类处理即可如果未匹配到说明用户发的只是自然语言那就需要交给 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 与 ChatCommandargparse是为命令行工具设计的默认行为是解析出错时直接打印错误信息并退出进程这显然不适合 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, status0, messageNone): self.parse_error_triggered True if self.help_text: self.error_message fHelp 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 fHello, {parsed_args.name}! You are {parsed_args.age} years old. def create_parser(self) - ChatArgparser: parser ChatArgparser(progdemo, descriptionself.description) parser.add_argument(--name, typestr, helpName of the user) parser.add_argument(--age, typeint, helpAge of the user) return parserAdminCommand的结构类似不同之处在于is_visible False这样它不会出现在/help的输出中只有知道具体命令的管理员才能使用。HTTP 接口/api/chatmain.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 --helpcurl --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 --port12345curl --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 --port12345 } # 响应 { session_id: qwerasd, result: Admin command executed! Host: 192.168.1.1, Port: 12345 }改进点命令类是否启用和可见性应该配置在别处或者支持动态配置。实际应用中要考虑添加权限控制。动态加载命令类的方法的确有点黑箱如果命令不多的话也可以在代码中手动挨个导入。完整示例代码internal/cmd/base.pyimport 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, fileNone): 重写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, status0, messageNone): 重写ArgumentParser的exit方法: 不退出进程, 捕获退出调用并记录错误信息 self.parse_error_triggered True if self.help_text: self.error_message fHelp 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.pyimport 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 fshlex 参数解析错误: {str(e)} try: parsed_args self.arg_parser.parse_args(cmd_args) return fHello, {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 fparser 参数解析错误: {str(e)} def create_parser(self) - ChatArgparser: parser ChatArgparser(progdemo, descriptionself.description) parser.add_argument( --name, typestr, helpName of the user, ) parser.add_argument( --age, typeint, helpAge of the user, ) return parser