python - 从 python 中运行交互式程序

Question

我想实现与 this 非常相似的东西.

我的实际目标是在 python 中运行 Rasa。摘自 Rasa 的网站:

Rasa is a framework for building conversational software: Messenger/Slack bots, Alexa skills, etc. We’ll abbreviate this as a bot in this documentation.

它基本上是一个在命令提示符下运行的聊天机器人。这是它在 cmd 上的工作方式:

现在我想从 python 运行 Rasa，以便我可以将它与基于 Django 的网站集成。即我想继续从用户那里获取输入，将其传递给 rasa，rasa 处理文本并给我一个输出，我将其显示给用户。

我已经试过了(现在从 cmd 运行它)

import sys
import subprocess
from threading import Thread
from queue import Queue, Empty  # python 3.x


def enqueue_output(out, queue):
    for line in iter(out.readline, b''):
        queue.put(line)
    out.close()


def getOutput(outQueue):
    outStr = ''
    try:
        while True: #Adds output from the Queue until it is empty
            outStr+=outQueue.get_nowait()
    except Empty:
        return outStr

p = subprocess.Popen('command_to_run_rasa', 
                    stdin=subprocess.PIPE, 
                    stdout=subprocess.PIPE, 
                    stderr=subprocess.PIPE, 
                    shell=False, 
                    universal_newlines=True,
                    )

outQueue = Queue()

outThread = Thread(target=enqueue_output, args=(p.stdout, outQueue))

outThread.daemon = True

outThread.start()

someInput = ""

while someInput != "stop":
    someInput = input("Input: ") # to take input from user
    p.stdin.write(someInput) # passing input to be processed by the rasa command
    p.stdin.flush()
    output = getOutput(outQueue)
    print("Output: " + output + "\n")
    p.stdout.flush()

但它只适用于第一行输出。不适用于连续的输入/输出周期。请参阅下面的输出。

如何让它在多个周期内工作？我提到了 this ，我想我从中理解了我的代码中的问题，但我不知道如何解决它。

编辑:我在 Windows 10 上使用 Python 3.6.2(64 位)

Answer 1

您需要继续与您的子进程进行交互 - 目前，一旦您从子进程中选择了输出，您就可以关闭它的 STDOUT 流。

下面是继续用户输入-> 处理输出循环的最基本的方法:

import subprocess
import sys
import time

if __name__ == "__main__":  # a guard from unintended usage
    input_buffer = sys.stdin  # a buffer to get the user input from
    output_buffer = sys.stdout  # a buffer to write rasa's output to
    proc = subprocess.Popen(["path/to/rasa", "arg1", "arg2", "etc."],  # start the process
                            stdin=subprocess.PIPE,  # pipe its STDIN so we can write to it
                            stdout=output_buffer, # pipe directly to the output_buffer
                            universal_newlines=True)
    while True:  # run a main loop
        time.sleep(0.5)  # give some time for `rasa` to forward its STDOUT
        print("Input: ", end="", file=output_buffer, flush=True)  # print the input prompt
        print(input_buffer.readline(), file=proc.stdin, flush=True)  # forward the user input

您可以将 input_buffer 替换为来自您的远程用户的缓冲区，并将 output_buffer 替换为将数据转发给您的用户的缓冲区，然后您您基本上会得到您正在寻找的东西 - 子流程将直接从用户那里获取输入 (input_buffer) 并将其输出打印给用户 (output_buffer) .

如果您需要在所有这些都在后台运行时执行其他任务，只需在单独的线程中运行 if __name__ == "__main__": 守卫下的所有内容，我建议添加一个 try..except block 以获取 KeyboardInterrupt 并正常退出。

但是......很快你就会注意到它并不是一直都能正常工作 - 如果等待 rasa 打印它的 的时间超过半秒STDOUT 并进入等待STDIN 阶段，输出将开始混合。这个问题比您想象的要复杂得多。主要问题是 STDOUT 和 STDIN(和 STDERR)是独立的缓冲区，您无法知道子进程何时真正期待其 标准输入。这意味着如果没有来自子进程的明确指示(例如，您在 Windows CMD 提示符的 STDOUT 中有 \r\n[path]>)，您只能将数据发送到子进程 STDIN 并希望它被拾取。

根据您的屏幕截图，它并没有真正给出可区分的 STDIN 请求提示，因为第一个提示是 ... :\n 然后等待 STDIN，但是一旦命令被发送，它就会列出选项，而不指示其结束 STDOUT 流(技术上使提示只是 ...\n 但这也将匹配它前面的任何行)。也许你可以聪明地逐行阅读 STDOUT，然后在每一行上测量自子进程写入以来已经过去了多少时间，一旦达到不活动阈值，假设 rasa 需要输入并提示用户输入。像这样的东西:

import subprocess
import sys
import threading

# we'll be using a separate thread and a timed event to request the user input
def timed_user_input(timer, wait, buffer_in, buffer_out, buffer_target):
    while True:  # user input loop
        timer.wait(wait)  # wait for the specified time...
        if not timer.is_set():  # if the timer was not stopped/restarted...
            print("Input: ", end="", file=buffer_out, flush=True)  # print the input prompt
            print(buffer_in.readline(), file=buffer_target, flush=True)  # forward the input
        timer.clear()  # reset the 'timer' event

if __name__ == "__main__":  # a guard from unintended usage
    input_buffer = sys.stdin  # a buffer to get the user input from
    output_buffer = sys.stdout  # a buffer to write rasa's output to
    proc = subprocess.Popen(["path/to/rasa", "arg1", "arg2", "etc."],  # start the process
                            stdin=subprocess.PIPE,  # pipe its STDIN so we can write to it
                            stdout=subprocess.PIPE,  # pipe its STDIN so we can process it
                            universal_newlines=True)
    # lets build a timer which will fire off if we don't reset it
    timer = threading.Event()  # a simple Event timer
    input_thread = threading.Thread(target=timed_user_input,
                                    args=(timer,  # pass the timer
                                          1.0,  # prompt after one second
                                          input_buffer, output_buffer, proc.stdin))
    input_thread.daemon = True  # no need to keep the input thread blocking...
    input_thread.start()  # start the timer thread
    # now we'll read the `rasa` STDOUT line by line, forward it to output_buffer and reset
    # the timer each time a new line is encountered
    for line in proc.stdout:
        output_buffer.write(line)  # forward the STDOUT line
        output_buffer.flush()  # flush the output buffer
        timer.set()  # reset the timer

您可以使用类似的技术来检查更复杂的“预期用户输入”模式。有一个完整的模块叫做 pexpect专为处理此类任务而设计，如果您愿意放弃一些灵 active ，我衷心推荐它。

现在...综上所述，您知道Rasa是用 Python 构建的，作为 Python 模块安装并具有 Python API，对吧？既然您已经在使用 Python，那么当您可以直接从 Python 代码运行它时，为什么还要将其称为子进程并处理所有这些 STDOUT/STDIN 恶作剧？只需导入它并直接与之交互，他们甚至有一个非常简单的示例可以完全满足您的要求:Rasa Core with minimal Python .