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

Posted 2023-03-14

【中文标题】Discord.py 如何制作干净的对话树？

【英文标题】：Discord.py How to make clean dialog trees?

【发布时间】：2021-10-15 05:11:29

【问题描述】：

我的目标是清理我的代码，以便我可以更轻松地制作对话树，而无需不断复制不必要的片段。我可以在 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)
            ]
        
    

【参考方案1】：

您的总体想法是正确的：可以用与您描述的结构相似的结构来表示这样的系统。它被称为finite state machine。我已经编写了一个示例来说明如何实现其中之一——这个特定的结构使用类似于interactive fiction 的结构，如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")

这是一个示例运行：

【参考方案2】：

di = '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
        

    

# Modified the dictionary a little bit, so we can get the option values directly, and the starter keywords.

def make_check(options, message):
    def predicate(msg):
        return msg.author == message.author and msg.channel == message.channel and msg.content.lower() in options
    return predicate
# I noticed the check function in your code was repetitive, we use higher order functions to solve this

async def response(dialogues, number, message, client): 
    await message.channel.send(dialogues[number]['Text'])
    options = [x[0] for x in dialogues[number]['Options']]
    if options:
        msg = await client.wait_for("message", check=make_check(options, message), timeout=30.0)
        return await response(dialogues, dialogues[number]['Options'][msg], message, client)
    else:
        pass
        # end dialogues
# Use recursion to remove redundant code, we navigate through the dialogues with the numbers provided

async def on_message(message):
    # basic on_message for example
    starters = ['hallucinated']
    initial = [x for x in starters if x in message.content.lower()]
    if initial:
        initial_opening_conversation = initial[0]
        await response(di.get(initial_opening_conversation), 1, message, client)

此代码应该可以正常工作，但您可能需要处理 wait_for 中的 TimeoutError，如果您的选项值不正确，它可能会进入无限循环。