[python]argparse 包在谈天呆板人中的应用

前言

在开发一个 AI 驱动的 IM 应用 Bot 时，某些场景用下令会更快更正确。我的假想是先按空格分割用户输入的文本，拿到第一段去匹配下令字典，假如匹配上了，分析用户想要实行下令，接着交给下令类处置惩罚即可；假如未匹配到，分析用户发的只是自然语言，那就必要交给 AI 相干的模块来处置惩罚。
我在之前一篇先容 __init_subclass__() 方法的博客中有提到过怎么处置惩罚下令，不外那内里只能处置惩罚简单格式的下令，下令文本只能按空格切片，不支持 --xx 如许的参数。在想着怎么处置惩罚这些差别情势的下令参数时，我忽然想起来 Python 标准库内里的 argparse。直接用 argparse 来剖析不更方便嘛！简单看了下 argparse 的文档和源码，感觉应该可行，说干就干！
流程逻辑

简单形貌下游程逻辑：

• 用户通过 HTTP API /api/chat 发送消息
  
• 后端应用吸取到消息后，按空格分割用户输入的文本，拿到第一段
  
• 匹配下令字典，假如没匹配到，则当成自然语言处置惩罚
  
• 匹配到下令字典后，交给下令类处置惩罚。下令类创建下令剖析器来剖析参数
  
• 返回效果给用户
  

按照风俗，具体下令类是动态加载的，不必要在代码中挨个引入。如许以后添加下令时，只要在指定目次添加代码文件，然后按照规范开发具体下令类即可。
本文紧张先容怎样用 argparse 在 web 应用中剖析用户下令，并不包罗 AI 处置惩罚自然语言的相干实现，以是本文用到的第三方依靠只有 FastAPI 充当 HTTP 框架，换成 Flask 或别的框架也是没题目标。
代码实现

代码结构：

1. ├── internal
2. │   └── cmd
3. │       ├── admin.py
4. │       ├── base.py
5. │       ├── demo.py
6. │       └── __init__.py
7. ├── main.py
8. ├── pyproject.toml
9. └── 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 缓冲区，存起来备用。
  

1. class ChatArgparser(argparse.ArgumentParser):
2.     def error(self, message):
3.         self.parse_error_triggered = True
4.         self.error_message = message
5.         raise argparse.ArgumentError(None, message)
7.     def exit(self, status=0, message=None):
8.         self.parse_error_triggered = True
9.         if self.help_text:
10.             self.error_message = f"Help requested:\n{self.help_text}"
11.         elif message:
12.             self.error_message = message
13.         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 是内置的资助下令，遍历全部已注册的可见下令，拼接出资助信息返回。

1. class HelpCommand(ChatCommand):
2.     main_name: str = "/help"
3.     description: str = "Show help message for all commands"
4.     is_visible: bool = True
6.     async def run(self) -> str:
7.         help_message = "Available commands:\n"
8.         for main_name, info in _loaded_chat_commands.items():
9.             if info["is_visible"]:
10.                 help_message += f"{main_name}: {info['description']}\n"
11.         return help_message

复制代码

具体下令示例

以 DemoCommand 为例，它继续 --name 和 --age 两个参数。在 run() 中，先用 shlex.split() 把用户消息按 shell 语法拆成列表，去掉第一个元素（即下令本身），然后把剩余参数交给 ChatArgparser 剖析。
这里用 shlex.split() 而不是直接 str.split()，是由于用户在 IM 中输入参数时大概会用引号包裹有空格的参数值，shlex.split() 能正确处置惩罚这种环境。

1. class DemoCommand(ChatCommand):
2.     main_name: str = "/demo"
3.     description: str = "Demo command for testing"
4.     is_enable: bool = True
5.     is_visible: bool = True
7.     async def run(self) -> str:
8.         cmd_args = shlex.split(self.user_message)[1:]
9.         parsed_args = self.arg_parser.parse_args(cmd_args)
10.         return f"Hello, {parsed_args.name}! You are {parsed_args.age} years old."
12.     def create_parser(self) -> ChatArgparser:
13.         parser = ChatArgparser(prog="demo", description=self.description)
14.         parser.add_argument("--name", type=str, help="Name of the user")
15.         parser.add_argument("--age", type=int, help="Age of the user")
16.         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，直接把资助文本返回；否则返回剖析错误提示。
  

1. @app.post("/api/chat")
2. async def post_chat(req: RequestChat):
3.     msg_list = req.message.strip().split(" ")
4.     if not msg_list[0].startswith("/"):
5.         return {"info": "自然语言, 预期将由AI处理"}
7.     cmders = load_chat_commands()
8.     if msg_list[0] not in cmders:
9.         return {"info": "未知命令, 预期将由AI处理"}
11.     cmd_cls = cmders[msg_list[0]]["cmdcls"]
12.     cmd_instance = cmd_cls(req.message)
13.     rst = await cmd_instance.run()
14.     return {"result": rst}

复制代码

现实效果

现实应用中可以稍微美化下输出

• 发送/help, 获取可用下令。由于/admin设置不可见，以是不会输出出来
  

1. curl --request POST \
2.   --url http://127.0.0.1:10001/api/chat \
3.   --header 'content-type: application/json' \
4.   --data '{
5.   "session_id": "qwerasd",
6.   "message": "/help"
7. }'
9. # 响应
10. {
11.   "session_id": "qwerasd",
12.   "result": "Available commands:\n/demo: Demo command for testing\n/help: Show help message for all commands\n"
13. }

复制代码

• 用户发送 /demo --help
  

1. curl --request POST \
2.   --url http://127.0.0.1:10001/api/chat \
3.   --header 'content-type: application/json' \
4.   --data '{
5.   "session_id": "qwerasd",
6.   "message": "/demo --help"
7. }'
9. # 响应
10. {
11.   "session_id": "qwerasd",
12.   "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"
13. }

复制代码

• 用户发送 /admin --host 192.168.1.1 --port=12345
  

1. curl --request POST \
2.   --url http://127.0.0.1:10001/api/chat \
3.   --header 'content-type: application/json' \
4.   --data '{
5.   "session_id": "qwerasd",
6.   "message": "/admin --host 192.168.1.1 --port=12345"
7. }'
9. # 响应
10. {
11.   "session_id": "qwerasd",
12.   "result": "Admin command executed! Host: 192.168.1.1, Port: 12345"
13. }

复制代码

改进点

• 下令类是否启用和可见性应该设置在别处，大概支持动态设置。
  
• 现实应用中要思量添加权限控制。
  
• 动态加载下令类的方法简直有点黑箱，假如下令不多的话，也可以在代码中手动挨个导入。
  

完备示例代码

internal/cmd/base.py

1. import argparse
2. from abc import ABC, abstractmethod
3. from io import StringIO
5. class ChatArgparser(argparse.ArgumentParser):
6.     """自定义的ArgumentParser, 用于解析聊天命令的参数, 重写error和exit方法, 捕获解析错误并返回错误信息, 而不是直接退出程序"""
7.     def __init__(self, *args, **kwargs):
8.         super().__init__(*args, **kwargs)
9.         self.parse_error_triggered = False
10.         self.error_message = ""
11.         self.help_text = ""
13.     def print_help(self, file=None):
14.         """重写print_help方法, 捕获帮助信息, 以便在解析错误时返回给用户"""
15.         help_buffer = StringIO()
16.         super().print_help(help_buffer)
17.         self.help_text = help_buffer.getvalue()
19.     def error(self, message):
20.         """重写ArgumentParser的error方法: 不退出进程, 捕获解析错误并记录错误信息"""
21.         self.parse_error_triggered = True
22.         self.error_message = message
24.         # 抛出异常后, 中断后续的参数解析流程
25.         raise argparse.ArgumentError(None, message)
27.     def exit(self, status=0, message=None):
28.         """重写ArgumentParser的exit方法: 不退出进程, 捕获退出调用并记录错误信息"""
29.         self.parse_error_triggered = True
30.         if self.help_text:
31.             self.error_message = f"Help requested:\n{self.help_text}"
32.         elif message:
33.             self.error_message = message
34.         else:
35.             self.error_message = "Exit triggered without message"
37.         raise argparse.ArgumentError(None, self.error_message)
39. class ChatCommand(ABC):
40.     """聊天命令的抽象基类, 定义了命令的基本结构和接口"""
41.     def __init__(self, user_message: str):
42.         self.user_message = user_message
44.     @abstractmethod
45.     def create_parser(self) -> ChatArgparser:
46.         """创建并返回一个ChatArgparser实例, 定义命令的参数结构"""
47.         ...
49.     @abstractmethod
50.     async def run(self) -> str:
51.         """执行命令的异步方法, 返回命令执行结果"""
52.         ...

复制代码

internal/cmd/demo.py

1. import argparse
2. import shlex
4. from internal.cmd.base import ChatArgparser, ChatCommand
6. class DemoCommand(ChatCommand):
7.     main_name: str = "/demo"
8.     description: str = "Demo command for testing"
9.     is_enable: bool = True
10.     is_visible: bool = True
12.     def __init__(self, user_message: str):
13.         super().__init__(user_message)
14.         self.arg_parser = self.create_parser()
16.     async def run(self) -> str:
17.         try:
18.             cmd_args = shlex.split(self.user_message)[1:]  # 去掉命令本身
19.         except ValueError as e:
20.             return f"shlex 参数解析错误: {str(e)}"
22.         try:
23.             parsed_args = self.arg_parser.parse_args(cmd_args)
24.             return f"Hello, {parsed_args.name}! You are {parsed_args.age} years old."
25.         except argparse.ArgumentError as e:
26.             error_msg = str(e)
27.             if error_msg.startswith("Help requested:"):
28.                 return error_msg
29.             return f"parser 参数解析错误: {str(e)}"
31.     def create_parser(self) -> ChatArgparser:
32.         parser = ChatArgparser(prog="demo", description=self.description)
34.         parser.add_argument(
35.             "--name",
36.             type=str,
37.             help="Name of the user",
38.         )
39.         parser.add_argument(
40.             "--age",
41.             type=int,
42.             help="Age of the user",
43.         )
44.         return parser

复制代码

internal/cmd/admin.py

1. import argparse
2. import shlex
4. from internal.cmd.base import ChatArgparser, ChatCommand
6. class AdminCommand(ChatCommand):
7.     main_name: str = "/admin"
8.     description: str = "Admin command"
9.     is_enable: bool = True
10.     is_visible: bool = False  # 管理命令默认不在/help中显示, 需要管理员知道具体命令才使用
12.     def __init__(self, user_message: str):
13.         super().__init__(user_message)
14.         self.arg_parser = self.create_parser()
16.     async def run(self) -> str:
17.         try:
18.             cmd_args = shlex.split(self.user_message)[1:]  # 去掉命令本身
19.         except ValueError as e:
20.             return f"shlex 参数解析错误: {str(e)}"
22.         try:
23.             parsed_args = self.arg_parser.parse_args(cmd_args)
24.             return f"Admin command executed! Host: {parsed_args.host}, Port: {parsed_args.port}"
25.         except argparse.ArgumentError as e:
26.             error_msg = str(e)
27.             if error_msg.startswith("Help requested:"):
28.                 return error_msg
29.             return f"parser 参数解析错误: {str(e)}"
31.     def create_parser(self) -> ChatArgparser:
32.         parser = ChatArgparser(prog="admin", description=self.description)
34.         parser.add_argument(
35.             "--host",
36.             type=str,
37.             help="Hostname or IP address of the server",
38.         )
39.         parser.add_argument(
40.             "--port",
41.             type=int,
42.             help="Port number of the server",
43.         )
44.         return parser

复制代码

internal/cmd/__init__.py

1. from __future__ import annotations
3. import importlib
4. import pkgutil
5. from typing import Dict, TypedDict
7. from .base import ChatArgparser, ChatCommand
9. class CommandInfo(TypedDict):
10.     description: str
11.     cmdcls: type[ChatCommand]
12.     is_visible: bool
14. _loaded_chat_commands: Dict[str, CommandInfo] = {}
16. class HelpCommand(ChatCommand):
17.     """内置的帮助命令, 用于展示所有可用命令的帮助信息"""
18.     main_name: str = "/help"
19.     description: str = "Show help message for all commands"
20.     is_visible: bool = True
22.     def create_parser(self) -> ChatArgparser:
23.         """HelpCommand 不需要参数, 直接返回一个空的ChatArgparser实例"""
24.         return ChatArgparser(
25.             prog="help", description="Show help message for all commands"
26.         )
28.     async def run(self) -> str:
29.         """执行帮助命令, 返回所有可用命令的帮助信息"""
30.         if not _loaded_chat_commands:
31.             load_chat_commands()
33.         help_message = "Available commands:\n"
34.         for main_name, info in _loaded_chat_commands.items():
35.             if info["is_visible"]:
36.                 help_message += f"{main_name}: {info['description']}\n"
37.         return help_message
39. def load_chat_commands() -> Dict[str, CommandInfo]:
40.     """加载所有命令类"""
41.     if _loaded_chat_commands:
42.         return _loaded_chat_commands
44.     pkg_path = "internal.cmd"
45.     pkg = importlib.import_module(pkg_path)
46.     print(f"Loading chat commands from package: {pkg_path}")
48.     for _, name, ispkg in pkgutil.iter_modules(pkg.__path__, pkg.__name__ + "."):
49.         # 如果以后各个命令类比较复杂, 可以把命令类放在一个单独的模块中, 加载的时候只加载模块
50.         # 目前命令类比较简单, 就直接放在internal.cmd包下, 加载的时候直接加载模块中的类
51.         # if not ispkg:
52.         #     continue
53.         if ispkg:
54.             continue
55.         skipped_modules = {"base", "__init__"}
56.         if any(name.endswith(skiped) for skiped in skipped_modules):
57.             continue
58.         module = importlib.import_module(name)
59.         for attr_name in dir(module):
60.             attr = getattr(module, attr_name)
61.             if (
62.                 isinstance(attr, type)
63.                 and issubclass(attr, ChatCommand)
64.                 and attr is not ChatCommand
65.             ):
66.                 main_name = getattr(attr, "main_name", None)
67.                 description = getattr(attr, "description", None)
68.                 is_enable = getattr(attr, "is_enable", False)
69.                 is_visible = getattr(attr, "is_visible", True)
70.                 if not main_name or not description:
71.                     continue
72.                 if not is_enable:
73.                     continue
74.                 main_name = main_name.strip()
75.                 description = description.strip()
76.                 if main_name.startswith("/") and main_name not in _loaded_chat_commands:
77.                     _loaded_chat_commands[main_name] = {
78.                         "description": description,
79.                         "cmdcls": attr,
80.                         "is_visible": is_visible,
81.                     }
83.     # 手动注册HelpCommand, 确保/help命令始终可用
84.     if "/help" not in _loaded_chat_commands:
85.         _loaded_chat_commands["/help"] = {
86.             "description": HelpCommand.description,
87.             "cmdcls": HelpCommand,
88.             "is_visible": True,
89.         }
91.     return _loaded_chat_commands

复制代码

main.py

1. import argparse
2. from contextlib import asynccontextmanager
4. import uvicorn
5. from fastapi import FastAPI
6. from pydantic import BaseModel, Field, ValidationInfo, field_validator
8. from internal.cmd import load_chat_commands
10. class RequestChat(BaseModel):
11.     session_id: str = Field(
12.         ..., min_length=1, description="Unique identifier for the chat session"
13.     )
14.     message: str = Field(
15.         ..., min_length=1, description="The chat message sent by the user"
16.     )
18.     @field_validator("session_id", "message")
19.     @classmethod
20.     def validate_fields(cls, v: str, info: ValidationInfo) -> str:
21.         if not v or not v.strip():
22.             raise ValueError(f"Field '{info.field_name}' cannot be empty")
23.         return v.strip()
25. @asynccontextmanager
26. async def lifespan(app: FastAPI):
27.     print("Starting up...")
28.     try:
29.         yield
30.     finally:
31.         print("Shutting down...")
33. app = FastAPI(lifespan=lifespan)
35. @app.post("/api/chat")
36. async def post_chat(req: RequestChat):
37.     try:
38.         msg_list = req.message.strip().split(" ")
39.         if not msg_list[0].startswith("/"):
40.             return {
41.                 "session_id": req.session_id,
42.                 "message": req.message,
43.                 "info": "自然语言, 预期将由AI处理",
44.             }
46.         cmders = load_chat_commands()
47.         if msg_list[0] not in cmders:
48.             return {
49.                 "session_id": req.session_id,
50.                 "message": req.message,
51.                 "info": "未知命令, 预期将由AI处理",
52.             }
54.         cmd_cls = cmders[msg_list[0]]["cmdcls"]
55.         cmd_instance = cmd_cls(req.message)
56.         rst = await cmd_instance.run()
57.         return {"session_id": req.session_id, "result": rst}
59.     except argparse.ArgumentError as e:
60.         error_msg = str(e)
61.         if error_msg.startswith("Help requested:"):
62.             return {"session_id": req.session_id, "result": error_msg}
63.         return {"session_id": req.session_id, "message": f"参数解析错误: {str(e)}"}
64.     except Exception as e:
65.         return {"session_id": req.session_id, "message": f"参数解析错误: {str(e)}"}
67. if __name__ == "__main__":
68.     uvicorn.run("main:app", host="127.0.0.1", port=10001, workers=1)

复制代码