发布于2026-06-30 阅读(0)
扫一扫,手机访问
在开发一个AI驱动的IM应用Bot时,总会遇到一个经典问题:什么时候该让用户敲命令,什么时候交给AI处理自然语言?答案其实挺明确的——某些场景下,命令比对话要快得多,也精准得多。
核心思路是这样的:先把用户输入的文本按空格切分,取第一个词去匹配命令字典。匹配上了,说明用户想执行一条命令,交给对应的命令类处理即可;没匹配上,那就说明用户说的是自然语言,转给AI模块去理解。这种“先命令,后AI”的判断顺序,在实际使用中非常直观。

之前在一篇关于__init_subclass__()的博客中聊过怎么处理命令,但那时只支持最简单的格式——文本按空格切片,处理不了--xx这样的参数。后来琢磨这事的时候,突然想起Python标准库里的argparse,心想:用现成的解析器不就省事多了?翻了翻文档和源码,确认可行,那就动手吧。
整体流程走起来其实不复杂:
/api/chat发来消息按习惯,具体的命令类走动态加载路线,不需要在代码里一个一个手动引入。以后加新命令,只要在指定目录下新建文件,按规范写好具体命令类就行。这种做法的灵活度谁都懂。
需要明确一点:本文重点在怎么用argparse在Web应用里解析用户命令,不涉及AI处理自然语言的具体实现。所以依赖就只用了FastAPI做HTTP框架——换Flask或其他框架也完全没问题。
先看下代码结构:
├── internal
│ └── cmd
│ ├── admin.py
│ ├── base.py
│ ├── demo.py
│ └── __init__.py
├── main.py
├── pyproject.toml
└── README.md
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 = "A vailable 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的输出里,只有知道这个命令的管理员才能使用。这种设计对隐藏敏感命令来说非常实用。
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}
在实际应用中,输出可以稍微美化一下。
1.发送/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": "A vailable commands:\n/demo: Demo command for testing\n/help: Show help message for all commands\n"
}
2.用户发送 /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"
}
3.用户发送 /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"
}
/admin就有点危险了。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:
"""执行命令的异步方法, 返回命令执行结果"""
...
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
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
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 = "A vailable 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__ + "."):
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
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 lifep(app: FastAPI):
print("Starting up...")
try:
yield
finally:
print("Shutting down...")
app = FastAPI(lifep=lifep)
@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)
售后无忧
立即购买>office旗舰店
售后无忧
立即购买>office旗舰店
售后无忧
立即购买>office旗舰店
售后无忧
立即购买>office旗舰店
正版软件
正版软件
正版软件
正版软件
正版软件
1
2
3
7
8