python - Discord.py 如何制作干净的对话树？

Question

我的目标是清理我的代码，以便我可以更轻松地制作对话树，而无需不断复制不必存在的片段。我可以在 python 中干净地做到这一点，但 discord.py 似乎有不同的要求。这是我当前非常冗余的代码的示例：

    if 'I need help' in message.content.lower():
        await message.channel.trigger_typing()
        await asyncio.sleep(2)
        response = 'Do you need help'
        await message.channel.send(response)
        await message.channel.send("yes or no?")

        def check(msg):
            return msg.author == message.author and msg.channel == message.channel and msg.content.lower() in ["yes", "no"]
        msg = await client.wait_for("message", check=check)

        if msg.content.lower() == "no":
            await message.channel.trigger_typing()
            await asyncio.sleep(2)
            response = 'okay'
            await message.channel.send(response)

        if msg.content.lower() == "yes":
            await message.channel.trigger_typing()
            await asyncio.sleep(2)
            response = 'I have something. Would you like to continue?'
            await message.channel.send(response)
            await message.channel.send("yes or no?")

            def check(msg):
                return msg.author == message.author and msg.channel == message.channel and msg.content.lower() in ["yes", "no"]
            msg = await client.wait_for("message", check=check)

            if msg.content.lower() == "no":
                await message.channel.trigger_typing()
                await asyncio.sleep(2)
                response = 'Okay'
                await message.channel.send(response)

我试图制作处理重复代码的函数，但没有成功。例如，使用：

async def respond(response, channel):
    await channel.trigger_typing()
    await asyncio.sleep(2)
    await channel.send(response)
...
await respond(response, message.channel)

理想情况下，我希望能够为树对话框本身做这样的事情，就像在 python 中一样：

if __name__=='__main__':
    hallucinated = {
        1: {
          'Text': [
                "It sounds like you may be hallucinating, would you like help with trying to disprove it?"
            ],
          'Options': [
              ("yes", 2),
              ("no", 3)
            ]
        },
        2: {    
            'Text': [
                "Is it auditory, visual, or tactile?"
            ],
            'Options': [
              ("auditory", 4),
              ("visual", 5),
              ("tactile", 6)
            ]
        }
    }

标签： pythondiscorddiscord.pychatbotcode-cleanup

Answer 1

您的总体想法是正确的：可以用与您描述的结构相似的结构来表示这样的系统。它被称为有限状态机。我写了一个例子来说明如何实现其中的一个——这个特定的例子使用类似于 Zork 这样的交互式小说的结构，但同样的原则也适用于对话树。

from typing import Tuple, Mapping, Callable, Optional, Any
import traceback
import discord
import logging
import asyncio
logging.basicConfig(level=logging.DEBUG)

client = discord.Client()

NodeId = str

ABORT_COMMAND = '!abort'

class BadFSMError(ValueError):
    """ Base class for exceptions that occur while evaluating the dialog FSM. """

class FSMAbortedError(BadFSMError):
    """ Raised when the user aborted the execution of a FSM. """

class LinkToNowhereError(BadFSMError):
    """ Raised when a node links to another node that doesn't exist. """

class NoEntryNodeError(BadFSMError):
    """ Raised when the entry node is unset. """

class Node:
    """ Node in the dialog FSM. """
    def __init__(self,
                 text_on_enter: Optional[str],
                 choices: Mapping[str, Tuple[NodeId, Callable[[Any], None]]],
                 delay_before_text: int = 2, is_exit_node: bool = False):
        self.text_on_enter = text_on_enter
        self.choices = choices
        self.delay_before_text = delay_before_text
        self.is_exit_node = is_exit_node

    async def walk_from(self, message) -> Optional[NodeId]:
        """ Get the user's input and return the next node in the FSM that the user went to. """
        async with message.channel.typing():
            await asyncio.sleep(self.delay_before_text)
        if self.text_on_enter:
            await message.channel.send(self.text_on_enter)

        if self.is_exit_node: return None

        def is_my_message(msg):
            return msg.author == message.author and msg.channel == message.channel
        user_message = await client.wait_for("message", check=is_my_message)
        choice = user_message.content
        while choice not in self.choices:
            if choice == ABORT_COMMAND: raise FSMAbortedError
            await message.channel.send("Please select one of the following: " + ', '.join(list(self.choices)))       
            user_message = await client.wait_for("message", check=is_my_message)
            choice = user_message.content

        result = self.choices[choice]
        if isinstance(result, tuple):
            next_id, mod_func = self.choices[choice]
            mod_func(self)
        else: next_id = result
        return next_id

class DialogFSM:
    """ Dialog finite state machine. """
    def __init__(self, nodes={}, entry_node=None):
        self.nodes: Mapping[NodeId, Node] = nodes
        self.entry_node: NodeId = entry_node

    def add_node(self, id: NodeId, node: Node):
        """ Add a node to the FSM. """
        if id in self.nodes: raise ValueError(f"Node with ID {id} already exists!")
        self.nodes[id] = node

    def set_entry(self, id: NodeId):
        """ Set entry node. """ 
        if id not in self.nodes: raise ValueError(f"Tried to set unknown node {id} as entry")
        self.entry_node = id

    async def evaluate(self, message):
        """ Evaluate the FSM, beginning from this message. """
        if not self.entry_node: raise NoEntryNodeError
        current_node = self.nodes[self.entry_node]
        while current_node is not None:
            next_node_id = await current_node.walk_from(message)
            if next_node_id is None: return
            if next_node_id not in self.nodes: raise LinkToNowhereError(f"A node links to {next_node_id}, which doesn't exist")
            current_node = self.nodes[next_node_id]


def break_glass(node):
    node.text_on_enter = "You are in a blue room. The remains of a shattered stained glass ceiling are scattered around. There is a step-ladder you can use to climb out."
    del node.choices['break']
    node.choices['u'] = 'exit'
nodes = {
    'central': Node("You are in a white room. There are doors leading east, north, and a ladder going up.", {'n': 'xroom', 'e': 'yroom', 'u': 'zroom'}),
    'xroom': Node("You are in a red room. There is a large 'X' on the wall in front of you. The only exit is south.", {'s': 'central'}),
    'yroom': Node("You are in a green room. There is a large 'Y' on the wall to the right. The only exit is west.", {'w': 'central'}),
    'zroom': Node("You are in a blue room. There is a large 'Z' on the stained glass ceiling. There is a step-ladder and a hammer.", {'d': 'central', 'break': ('zroom', break_glass)}),
    'exit': Node("You have climbed out into a forest. You see the remains of a glass ceiling next to you. You are safe now.", {}, is_exit_node=True)
}

fsm = DialogFSM(nodes, 'central')

@client.event
async def on_message(msg):
    if msg.content == '!begin':
       try:
           await fsm.evaluate(msg)
           await msg.channel.send("FSM terminated successfully")
       except:
           await msg.channel.send(traceback.format_exc())

client.run("token")

这是一个示例运行：