65.9K
CodeProject 正在变化。 阅读更多。
Home

在 Python 中为 CodeProject.AI Server 模块提供长时操作支持

Chris Maunder
starIconstarIconstarIconstarIcon
emptyStarIcon
starIcon

4.67/5 (2投票s)

2024 年 4 月 4 日

CPOL

6分钟阅读

viewsIcon

5861

本文将向您展示如何为 CodeProject.AI Server 创建一个模块，该模块封装了一些需要长时间才能完成的代码。

引言

对于那些代码执行快速推理然后将结果返回给服务器的情况，将优秀的 AI 代码封装到 CodeProject.AI 模块中是 很简单的。而对于 AI 操作时间较长的情况——例如生成式 AI——由于超时和普遍糟糕的用户体验，这种流程将无法正常工作。

本文将向您展示如何为 CodeProject.AI Server 创建一个模块，该模块封装了一些需要长时间才能完成的代码。我们将只关注编写 AI 代码适配器所需的代码，而不是 AI 代码本身。关于后者以及在您的桌面上实现 LLM 的有趣示例，请阅读 Matthew Dennis 的后续文章 为 CodeProject.AI Server 创建 LLM 聊天模块。

入门

我们假设您已经阅读了 CodeProject.AI 模块创建：Python 中的完整演练。我们将以完全相同的方式创建一个模块，只是增加了一点，我们将展示如何处理长时间运行的进程。

首先，像往常一样，克隆 CodeProject.AI Server 仓库，然后在 /src/modules 文件夹中为您要创建的模块创建一个新文件夹。我们将它命名为 PythonLongProcess。对于我们这些简单的人来说，这是一个简单的名字。

我们还假设我们有一些代码想要通过 CodeProject.AI Server 暴露。我们将封装的神奇代码如下。

import time

cancelled = False

def a_long_process(callback):

    result    = ""
    step      = 0
    cancelled = False

    for i in range(1, 11):
       if cancelled: break
       
       time.sleep(1)
       step   = 1 if not step else step + 1
       result = str(step) if not result else f"{result} {step}"
       callback(result, step)


def cancel_process():
    global cancelled
    cancelled = True

所有代码所做的就是逐步构建一个包含数字 1-10 的字符串。在每一步，它都会检查进程是否已被取消，并调用一个回调函数以允许调用者检查进度。这没什么令人兴奋的，但足以作为演示。

创建适配器

我们想将这个长时进程代码封装到 CodeProject.AI Server 模块中，所以我们将创建一个适配器、一个 modulesettings.json 文件、安装脚本和一个测试页面。我们将从适配器开始。

我们的适配器将非常基础。我们不需要从调用者那里获取值，不需要太多错误检查，也不会记录任何信息。

我们需要创建一个派生自 ModuleRunner 的类，并重写 initialize 和 process 方法。为了提供对长时操作的支持，我们还需要重写 command_status、cancel_command_task，并提供一个实际调用我们要封装的长时进程的方法。正是这最后一部分为模块提供了长时操作的支持。

长时操作支持

为了让 CodeProject.AI Server 模块能够处理长时操作，我们做了三件事：

  1. 向调用者和 Server 本身发出信号，表明对某个方法的调用将导致一个长时操作。
  2. 在后台运行长时操作。
  3. 提供检查状态和必要时取消的机制。

为了做到这一点，我们从常规的 process 方法返回一个 Callable，而不是一个通常包含调用结果的 JSON 对象。返回一个 Callable 会向服务器发出信号，表明我们需要在后台运行一个方法。然后，调用者需要轮询模块状态 API 来检查进度，并在需要时调用取消任务 API 来取消正在运行的长时进程。

  • 要检查模块的状态，您可以向 /v1/<moduleid>/get_command_status 发起 API 调用。
  • To cancel a long process you make an API call to /v1/<moduleid>/cancel_command.

这些路由会自动添加到每个模块，无需在模块设置文件中定义。这些调用将分别映射到模块的 command_status 和 cancel_command_task 方法。

代码

这是我们适配器的（大部分）完整列表。请注意常规的 initialize 和 process 方法，以及 long_process 方法，该方法从 process 返回以表示长时操作正在开始。

在 long_process 中，我们除了调用我们要封装的代码（a_long_process）并报告结果之外，几乎没有做什么。

command_status 和 cancel_command_task 方法同样简单：返回我们目前为止所拥有的内容，并在请求时取消长时操作。

最后一部分是我们传递给 long_process 的 long_process_callback。它将接收来自 long_process 的更新，并让我们有机会收集中间结果。

... other imports go here

# Import the method of the module we're wrapping
from long_process import a_long_process, cancel_process

class PythonLongProcess_adapter(ModuleRunner):

    def initialise(self) -> None:
        # Results from the long process
        self.result      = None
        self.step        = 0
        # Process state
        self.cancelled   = False
        self.stop_reason = None

    def process(self, data: RequestData) -> JSON:
        # This is a long process module, so all we need to do here is return the
        # long process method that will be run
        return self.long_process

    def long_process(self, data: RequestData) -> JSON:
        """ This calls the actual long process code and returns the results """
        self.cancelled   = False
        self.stop_reason = None
        self.result      = None
        self.step        = 0

        start_time = time.perf_counter()
        a_long_process(self.long_process_callback)
        inferenceMs : int = int((time.perf_counter() - start_time) * 1000)

        if self.stop_reason is None:
            self.stop_reason = "completed"

        response = {
            "success":     True, 
            "result":      self.result,
            "stop_reason": self.stop_reason,
            "processMs":   inferenceMs,
            "inferenceMs": inferenceMs
        }

        return response

    def command_status(self) -> JSON:
        """ This method will be called regularly during the long process to provide updates """
        return {
            "success": True, 
            "result":  self.result or ""
        }

    def cancel_command_task(self):
        """ This process is called when the client requests the process to stop """
        cancel_process()
        self.stop_reason = "cancelled"
        self.force_shutdown = False  # Tell ModuleRunner we'll shut ourselves down



    def long_process_callback(self, result, step):
        """ We'll provide this method as the callback for the a_long_process() 
            method in long_process.py """
        self.result = result
        self.step   = step


if __name__ == "__main__":
    PythonLongProcess_adapter().start_loop()

创建 modulesettings.json 文件

再次，请确保您已查看 Python 中的完整演练 和 ModuleSettings 文件。我们的 modulesettings 文件非常基础，有趣的部分是：

  • 我们将用于启动模块的适配器路径是 long_process_demo_adapter.py。
  • 我们将使用 python3.9 运行。
  • 我们将定义一个路由 "pythonlongprocess/long-process"，它接受一个不接受任何输入值的命令 "command"，并返回一个字符串 "reply"。
  • 它可以在所有平台上运行。
{
  "Modules": {
 
    "PythonLongProcess": {
      "Name": "Python Long Process Demo",
      "Version": "1.0.0",
 
      "PublishingInfo" : {
         ... 
      },
 
      "LaunchSettings": {
        "FilePath":    "llama_chat_adapter.py",
        "Runtime":     "python3.8",
      },
 
      "EnvironmentVariables": {
         ...
      },
 
      "GpuOptions" : {
         ...
      },
      
      "InstallOptions" : {
        "Platforms": [ "all" ],
        ...
      },
  
      "RouteMaps": [
        {
          "Name": "Long Process",
          "Route": "pythonlongprocess/long-process",
          "Method": "POST",
          "Command": "command",
          "MeshEnabled": false,
          "Description": "Demos a long process.",
          
          "Inputs": [
          ],
          "Outputs": [
            {
              "Name": "success",
              "Type": "Boolean",
              "Description": "True if successful."
            },
            {
              "Name": "reply",
              "Type": "Text",
              "Description": "The reply from the model."
            },
            ...
          ]
        }
      ]
    }
  }
}

这个代码片段中省略了很多样板代码，所以请参考源代码来查看完整内容。

安装脚本。

我们的示例实际上没有需要安装的。当此模块下载时，服务器会将其解压，将文件移动到正确的文件夹，然后运行安装脚本，以便我们可以执行任何必要的模块设置操作。我们不需要做任何事，所以我们将包含空的脚本。不包含脚本将向服务器发出信号，表明此模块不应被安装。

@if "%1" NEQ "install" (
    echo This script is only called from ..\..\setup.bat
    @goto:eof
)
call "!sdkScriptsDirPath!\utils.bat" WriteLine "No custom setup steps for this module." "!color_info!"

if [ "$1" != "install" ]; then
    read -t 3 -p "This script is only called from: bash ../../setup.sh"
    echo
    exit 1 
fi
writeLine "No custom setup steps for this module" "$color_info"

创建 CodeProject.AI 测试页面（以及 Explorer UI）

我们拥有希望封装并向外界公开的代码、一个用于此目的的适配器、一个用于设置和启动适配器的 modulesettings.json 文件，以及我们的安装脚本。最后一部分是允许我们测试新模块的演示页面。

我们的演示页面（explore.html）尽可能地简单：一个启动长时操作的按钮，一个取消按钮，以及一个用于查看结果的输出窗格。

<!DOCTYPE html>
<html lang="en" xmlns="http://www.w3.org/1999/xhtml">
<head>
    <meta charset="utf-8" />
    <title>Python Long Process demo module</title>

    <link id="bootstrapCss" rel="stylesheet" type="text/css" href="https://:32168/assets/bootstrap-dark.min.css">
    <link rel="stylesheet" type="text/css" href="https://:32168/assets/server.css?v=2.6.1.0">
    <script type="text/javascript" src="https://:32168/assets/server.js"></script>
    <script type="text/javascript" src="https://:32168/assets/explorer.js"></script>

    <style>
/* START EXPLORER STYLE */
/* END EXPLORER STYLE */
    </style>

</head>
<body class="dark-mode">
<div class="mx-auto" style="max-width: 800px;">
    <h2 class="mb-3">Python Long Process demo module</h2>
    <form method="post" action="" enctype="multipart/form-data" id="myform">

<!-- START EXPLORER MARKUP -->
        <div class="form-group row g-0">
            <input id="_MID_things" class="form-control btn-success" type="button" value="Start long process"
                   style="width:9rem" onclick="_MID_onLongProcess()"/>
            <input id="_MID_cancel" class="form-control btn-warn" type="button" value="Cancel"
                   style="width:5rem" onclick="_MID_onCancel()"/>
        </div>
<!-- END EXPLORER MARKUP -->
        <div>
            <h2>Results</h2>
            <div id="results" name="results" class="bg-light p-3" style="min-height: 100px;"></div>
        </div>

    </form>

    <script type="text/javascript">
// START EXPLORER SCRIPT

        let _MID_params = null;

        async function _MID_onLongProcess() {

            if (_MID_params) {
                setResultsHtml("Process already running. Cancel first to start a new process");
                return;
            }

            setResultsHtml("Starting long process...");
            let data = await submitRequest('pythonlongprocess/long-process', 'command', null, null);
            if (data) {

                _MID_params = [['commandId', data.commandId], ['moduleId', data.moduleId]];

                let done = false;

                while (!done) {
                    
                    await delay(1000);

                    if (!_MID_params)    // may have been cancelled
                        break;

                    let results = await submitRequest('pythonlongprocess', 'get_command_status',
                                                        null, _MID_params);
                    if (results && results.success) {

                        if (results.commandStatus == "failed") {
                            done = true;
                            setResultsHtml(results?.error || "Unknown error");
                        } 
                        else {
                            let message = results.result;
                            if (results.commandStatus == "completed")
                                done = true;

                            setResultsHtml(message);
                        }
                    }
                    else {
                        done = true;
                        setResultsHtml(results?.error || "No response from server");
                    }
                }

                _MID_params = null;
            };
        }

        async function _MID_onCancel() {
            if (!_MID_params)
                return;
				
			let moduleId = _MID_params[1][1];
            let result = await submitRequest(moduleId, 'cancel_command', null, _MID_params);
            if (result.success) {
                _MID_params = null;
                setResultsHtml("Command stopped");
            }
        }
// END EXPLORER SCRIPT
    </script>
</div>
</body>
</html>

结论

由于服务器的帮助，将需要长时间执行的代码封装到 CodeProject.AI 模块中变得很直接。如果被封装的代码提供了定期查询其进度的机制，那将非常有帮助，但即使没有（尽管用户体验会稍差），也是可行的。

我们使用长时操作支持来封装一个使用 stable diffusion 的文本到图像模块，以及 Llama 大型语言模型，以便在您的桌面上提供 ChatGPT 功能。除了编写标准的 CodeProject.AI Server 模块之外，唯一的额外工作是向适配器添加用于检查状态和必要时取消的方法，以及在我们的测试 HTML 页面中实际调用这些方法所需的代码。

长时操作支持非常适合生成式 AI 解决方案，但对于希望在低配置硬件上支持 AI 操作的情况也很有用。例如，OCR 在性能不错的机器上可能只需要不到一秒钟，但在 Raspberry Pi 上对大量数据运行相同的文本检测和识别模型可能需要一段时间。通过长时操作模块提供该功能可以提供更好的用户体验并避免 HTTP 超时问题。

© . All rights reserved.