运行 来自 python 的互动程序
Running interactive program from within python
我想实现与 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()
但它只适用于第一行输出。不适用于连续的 input/output 个周期。请参阅下面的输出。
如何让它工作多个周期?
我已经提到 this,我想我从中理解了我的代码中的问题,但我不知道如何解决它。
编辑:我在 Windows 10
上使用 Python 3.6.2(64 位)
您需要继续与您的子流程进行交互 - 目前,一旦您从您的子流程中选择了输出,您就可以关闭它的 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
打印其 [=12= 的时间超过半秒]并进入等待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
您可以使用类似的技术来检查更复杂的 'expected user input' 模式。有一个名为 pexpect
的完整模块旨在处理此类任务,如果您愿意放弃一些灵活性,我衷心推荐它。
现在...综上所述,您知道Rasa
is built in Python, installs as a Python module and has a Python API, right? Since you're already using Python why would you call it as a subprocess and deal with all this STDOUT/STDIN
shenanigans when you can directly run it from your Python code? Just import it and interact with it directly, they even have a very simple example that does exactly what you're trying to do: Rasa Core with minimal Python.
我想实现与 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()
但它只适用于第一行输出。不适用于连续的 input/output 个周期。请参阅下面的输出。
如何让它工作多个周期? 我已经提到 this,我想我从中理解了我的代码中的问题,但我不知道如何解决它。
编辑:我在 Windows 10
上使用 Python 3.6.2(64 位)您需要继续与您的子流程进行交互 - 目前,一旦您从您的子流程中选择了输出,您就可以关闭它的 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
打印其 [=12= 的时间超过半秒]并进入等待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
您可以使用类似的技术来检查更复杂的 'expected user input' 模式。有一个名为 pexpect
的完整模块旨在处理此类任务,如果您愿意放弃一些灵活性,我衷心推荐它。
现在...综上所述,您知道Rasa
is built in Python, installs as a Python module and has a Python API, right? Since you're already using Python why would you call it as a subprocess and deal with all this STDOUT/STDIN
shenanigans when you can directly run it from your Python code? Just import it and interact with it directly, they even have a very simple example that does exactly what you're trying to do: Rasa Core with minimal Python.