在Python中运行交互式程序

Question

我希望实现的目标与 this 非常相似。

我的实际目标是从 Python 中运行 Rasa。 从 Rasa 的网站获取：

Rasa 是构建对话软件的框架：Messenger/Slack 机器人、Alexa 技能等。在本文档中，我们将其缩写为 bot。

它基本上是一个在命令提示符中运行的聊天机器人。这是它在 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位）

- deveshasha

---

只需使用shell=True... - undefined

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块以捕获KeyboardInterrupt并优雅地退出。

但是...很快你会注意到它并不总是正常工作 - 如果等待rasa打印其STDOUT并进入等待STDIN阶段的时间超过半秒钟，输出将开始混合。这个问题比你想象的要复杂得多。主要问题是STDOUT和STDIN(以及STDERR)是分开的缓冲区，你无法知道子进程实际上何时在其STDIN上等待某些东西。这意味着除非从子进程中清晰地指示(例如在Windows CMD提示符中的\r\n[path]>上的STDOUT)，否则你只能将数据发送到子进程的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 的整个模块专门处理这种类型的任务，如果你愿意放弃一些灵活性，我会全力推荐它。

现在...所有这些都说了，您知道 Rasa 是用Python构建的，作为Python模块安装，并具有Python API，对吧？既然您已经在使用Python，为什么不直接从您的Python代码中运行它，而不必调用它作为子进程并处理所有这些STDOUT/STDIN麻烦呢？只需导入它并直接与之交互，他们甚至有一个非常简单的示例，完全可以做到您正在尝试做的事情: 最小化 Python 的 Rasa Core。

- zwer