最新国产好看的视频,伊人天堂AV在线,国产Aaaaaa视频,蜜臀视频在线观看一区,人妻av色图,密臀久久久精品影片,青青视频免费观看毛片,久草在线观看视,国产三级精品色情在线

一文詳解python中argparse包在聊天機器人中的應(yīng)用

 更新時間：2026年06月01日 09:36:19   作者：花酒鋤作田  
在開發(fā)一個?AI?驅(qū)動的?IM?應(yīng)用?Bot?時,某些場景用命令會更快更準確,本文主要介紹如何用?argparse?在?web?應(yīng)用中解析用戶命令,并不包含?AI?處理自然語言的相關(guān)實現(xiàn),所以本文用到的第三方依賴只有?FastAPI?充當?HTTP?框架,換成?Flask?或其它框架也是沒問題的

前言

在開發(fā)一個 AI 驅(qū)動的 IM 應(yīng)用 Bot 時，某些場景用命令會更快更準確。我的設(shè)想是先按空格分割用戶輸入的文本，拿到第一段去匹配命令字典，如果匹配上了，說明用戶想要執(zhí)行命令，接著交給命令類處理即可；如果未匹配到，說明用戶發(fā)的只是自然語言，那就需要交給 AI 相關(guān)的模塊來處理。

我在之前一篇介紹 __init_subclass__() 方法的博客中有提到過怎么處理命令，不過那里面只能處理簡單格式的命令，命令文本只能按空格切片，不支持 --xx 這樣的參數(shù)。在想著怎么處理這些不同形式的命令參數(shù)時，我突然想起來 Python 標準庫里面的 argparse。直接用 argparse 來解析不更方便嘛！簡單看了下 argparse 的文檔和源碼，感覺應(yīng)該可行，說干就干！

流程邏輯

簡單描述下流程邏輯：

  1. 用戶通過 HTTP API /api/chat 發(fā)送消息
  2. 后端應(yīng)用接收到消息后，按空格分割用戶輸入的文本，拿到第一段
  3. 匹配命令字典，如果沒匹配到，則當成自然語言處理
  4. 匹配到命令字典后，交給命令類處理。命令類創(chuàng)建命令解析器來解析參數(shù)
  5. 返回結(jié)果給用戶

按照習慣，具體命令類是動態(tài)加載的，不需要在代碼中挨個引入。這樣以后添加命令時，只要在指定目錄添加代碼文件，然后按照規(guī)范開發(fā)具體命令類即可。

本文主要介紹如何用 argparse 在 web 應(yīng)用中解析用戶命令，并不包含 AI 處理自然語言的相關(guān)實現(xiàn)，所以本文用到的第三方依賴只有 FastAPI 充當 HTTP 框架，換成 Flask 或其它框架也是沒問題的。

代碼實現(xiàn)

代碼結(jié)構(gòu)：

├── internal
│   └── cmd
│       ├── admin.py
│       ├── base.py
│       ├── demo.py
│       └── __init__.py
├── main.py
├── pyproject.toml
└── README.md

核心抽象：ChatArgparser 與 ChatCommand

argparse 是為命令行工具設(shè)計的，默認行為是解析出錯時直接打印錯誤信息并退出進程，這顯然不適合 web 應(yīng)用。所以我們需要繼承 argparse.ArgumentParser，重寫它的 error()、exit() 和 print_help() 方法，把"退出進程"變成"拋出異常"。這樣一來，異常被上層捕獲后，就能以 HTTP 響應(yīng)的形式返回給用戶。

ChatArgparser 做了三件事：

  • 重寫 error()：不調(diào)用 sys.exit()，而是記錄錯誤信息并拋出 argparse.ArgumentError。
  • 重寫 exit()：argparse 在用戶輸入 --help 時會調(diào)用 exit()，這里同樣改為拋異常，同時把幫助文本附在異常信息里。
  • 重寫 print_help()：把幫助信息輸出到 StringIO 緩沖區(qū)，存起來備用。
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 實例，聲明該命令接受的參數(shù)；run() 是異步方法，執(zhí)行實際的命令邏輯。

命令加載：自動發(fā)現(xiàn)與注冊

load_chat_commands() 函數(shù)負責掃描 internal.cmd 包下的所有模塊，找出繼承自 ChatCommand 的類，然后根據(jù)類屬性 main_name、is_enable、is_visible 來判斷是否注冊。

跳過 base 和 __init__ 這兩個模塊，避免把基類和自己注冊進去。每個命令類需要定義幾個類屬性：

  • main_name：命令名，以 / 開頭，比如 /demo。
  • description：命令的簡要說明。
  • is_enable：是否啟用該命令，關(guān)閉后不會被注冊。
  • is_visible：是否在幫助列表中顯示，適合隱藏管理員命令。

HelpCommand 是內(nèi)置的幫助命令，遍歷所有已注冊的可見命令，拼接出幫助信息返回。

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 兩個參數(shù)。在 run() 中，先用 shlex.split() 把用戶消息按 shell 語法拆成列表，去掉第一個元素（即命令本身），然后把剩余參數(shù)交給 ChatArgparser 解析。

這里用 shlex.split() 而不是直接 str.split()，是因為用戶在 IM 中輸入?yún)?shù)時可能會用引號包裹有空格的參數(shù)值，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 的結(jié)構(gòu)類似，不同之處在于 is_visible = False，這樣它不會出現(xiàn)在 /help 的輸出中，只有知道具體命令的管理員才能使用。

HTTP 接口：/api/chat

main.py 中的 /api/chat 端點接收用戶消息，處理流程如下：

  1. 用 strip().split(" ") 取出第一個詞，判斷是否以 / 開頭。
  2. 不以 / 開頭，說明是自然語言，直接返回，交給 AI 模塊處理（本文略過）。
  3. 以 / 開頭，調(diào)用 load_chat_commands() 查找對應(yīng)命令。找不到也按自然語言處理。
  4. 找到命令后，實例化命令類，調(diào)用 run() 執(zhí)行。
  5. 整個流程用 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}

實際效果

實際應(yīng)用中可以稍微美化下輸出

1.發(fā)送/help, 獲取可用命令。因為/admin設(shè)置不可見，所以不會輸出出來

curl --request POST \
  --url http://127.0.0.1:10001/api/chat \
  --header 'content-type: application/json' \
  --data '{
  "session_id": "qwerasd",
  "message": "/help"
}'

# 響應(yīng)
{
  "session_id": "qwerasd",
  "result": "Available commands:\n/demo: Demo command for testing\n/help: Show help message for all commands\n"
}

2.用戶發(fā)送 /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"
}'

# 響應(yīng)
{
  "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.用戶發(fā)送 /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"
}'

# 響應(yīng)
{
  "session_id": "qwerasd",
  "result": "Admin command executed! Host: 192.168.1.1, Port: 12345"
}

改進點

  • 命令類是否啟用和可見性應(yīng)該配置在別處，或者支持動態(tài)配置。
  • 實際應(yīng)用中要考慮添加權(quán)限控制。
  • 動態(tài)加載命令類的方法的確有點黑箱，如果命令不多的話，也可以在代碼中手動挨個導入。

完整示例代碼

internal/cmd/base.py

import argparse
from abc import ABC, abstractmethod
from io import StringIO


class ChatArgparser(argparse.ArgumentParser):
    """自定義的ArgumentParser, 用于解析聊天命令的參數(shù), 重寫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

        # 拋出異常后, 中斷后續(xù)的參數(shù)解析流程
        raise argparse.ArgumentError(None, message)

    def exit(self, status=0, message=None):
        """重寫ArgumentParser的exit方法: 不退出進程, 捕獲退出調(diào)用并記錄錯誤信息"""
        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):
    """聊天命令的抽象基類, 定義了命令的基本結(jié)構(gòu)和接口"""
    def __init__(self, user_message: str):
        self.user_message = user_message

    @abstractmethod
    def create_parser(self) -> ChatArgparser:
        """創(chuàng)建并返回一個ChatArgparser實例, 定義命令的參數(shù)結(jié)構(gòu)"""
        ...

    @abstractmethod
    async def run(self) -> str:
        """執(zhí)行命令的異步方法, 返回命令執(zhí)行結(jié)果"""
        ...

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 參數(shù)解析錯誤: {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 參數(shù)解析錯誤: {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 參數(shù)解析錯誤: {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 參數(shù)解析錯誤: {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):
    """內(nèi)置的幫助命令, 用于展示所有可用命令的幫助信息"""
    main_name: str = "/help"
    description: str = "Show help message for all commands"
    is_visible: bool = True

    def create_parser(self) -> ChatArgparser:
        """HelpCommand 不需要參數(shù), 直接返回一個空的ChatArgparser實例"""
        return ChatArgparser(
            prog="help", description="Show help message for all commands"
        )

    async def run(self) -> str:
        """執(zhí)行幫助命令, 返回所有可用命令的幫助信息"""
        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"參數(shù)解析錯誤: {str(e)}"}
    except Exception as e:
        return {"session_id": req.session_id, "message": f"參數(shù)解析錯誤: {str(e)}"}


if __name__ == "__main__":
    uvicorn.run("main:app", host="127.0.0.1", port=10001, workers=1)

以上就是一文詳解python中argparse包在聊天機器人中的應(yīng)用的詳細內(nèi)容，更多關(guān)于python argparse解析用戶命令的資料請關(guān)注腳本之家其它相關(guān)文章！

相關(guān)文章

最新評論

石城县| 浦东新区| 安丘市| 邮箱| 綦江县| 晋州市| 通河县| 金乡县| 安溪县| 万荣县| 姜堰市| 闽侯县| 海原县| 泽州县| 加查县| 湘潭县| 葵青区| 巴马| 吉木萨尔县| 波密县| 盐津县| 龙游县| 阳朔县| 睢宁县| 益阳市| 尼勒克县| 靖远县| 深泽县| 华宁县| 额尔古纳市| 蒙自县| 水城县| 泸西县| 公安县| 鱼台县| 惠来县| 澜沧| 门源| 英吉沙县| 改则县| 湘潭县|