Workflow design

随着2023年大模型在国内的发展成熟，现在大家逐步开始真正关注到大模型应用的开发上。有人说，2024年是大模型应用的落地元年。的确，看似无所不能的大模型，也并不是直接丢到一个系统就能用的，如何开发大模型应用，实际上还处于一个刚起步的阶段。一些成熟的技术，经历过许多人摸索后，总能给出一个"best practice"；对于大模型应用开发来说，却仍还在探索阶段。 目前，一个看起来比较靠谱的模型，是这样定义大模型应用开发的几个level的：

最简单的Wrapper层级，实际就是把大模型的提示词封装成一个固定功能的”方法“，然后集成到业务系统中。KPP的Function（或者Prompt），就是一个Wrapper。Wrapper能做什么完全取决于大模型的能力，比如生成一段文字，甚至写一段代码，都没有问题。但是如果想做一个复杂的功能，恐怕就不行了，比如，先生成代码再执行得到代码运行结果，这样就没法直接通过大模型请求来做了，原因是大模型本身无法直接执行代码 。

显然，对于复杂的功能，我们可以尝试分步骤来解决。比如一键生成PPT就是很好的例子，第一步先生成一个outline；然后再根据outline，去填充内容；最后通过一个生成文档的API生成一个美化后的PPT。这样每一步都可以用最适合的技术来进行，比如生成outline和填充内容用LLM，可以选择最合适的模型；美化PPT则可以直接使用已有的API来处理。

这其实就是Flow了，或者更明确一点，Agentic Workflow。不是普通的工作流，是“智能”的工作流。通过流程的编排，我们可以扩充LLM的能力，通过一些组合来实现复杂的业务功能。相对于Wrapper而言，Workflow可以更好地实现复杂的AI能力。 因此，我们在KPP中也实现了Workflow的能力。这里就技术层面分享一下KPP在构建Workflow时的一些设计和实现。

为什么要自研，搞个开源的不行么？

实际上，在开始设计工作流之前，就已经有不少公司已经在做了。其中最具有参考价值的主要是Dify和扣子。其中，Dify是一个开源的系统，理论上可以直接拿过来用。 但是，考虑到WPS AI的实际情况，有几个比较重要的问题是Dify难以支撑的：

• 无法直接嵌入到WPS AI的现有体系，包括网关、KPP、AI Server等，一定会存在一个二次开发的成本
• Dify是用Python写的，如果大规模使用，很可能会遇到性能问题，包括工作流本身的异步多线程执行、以及水平扩展的性价比

实际上，字节的Coze是一个非常适合我们的开发平台。它的产品设计相当专业和完善，Workflow、图像流、Agent、多Agent协作等，是目前所有平台中定义最清晰的。相对而言，一些平台自己都不知道自己在做什么，比如百度的Agent Builder和App Builder，做着做着就成一样的了。 Dify在这方面也有明显的问题，看起来还没有搞清楚Agent和Workflow的关系。

然而很遗憾目前为止没有一个开源的系统可以达到Coze的高度。 正是由于种种原因，最终KPP决定完全自研Workflow系统。

理论基础

即便从零开始实现一个Workflow其实也比较简单，因为从原理上来说，Workflow无非就是一个DAG（有向无环图）而已。 它是工作流的核心。DAG相信大家都比较熟悉，之所以要用到DAG正式因为它有两个重要的特点：

• 节点之间的连接是有向的，否则我们将无法判断节点之间的依赖关系和执行顺序
• 整个图是无环的，这也限制了在Workflow中是不可以存在循环的。这点非常重要，如果是一个Agent，那么我们可以一直给Agent发消息聊下去，这是一个循环；但是Workflow一定不能存在这种不确定的循环，或者说，我们期望在Workflow能够在确定有限的时间内运行完成并获取结果，它是Stateless（无状态）的

使用拓扑排序确定执行顺序

执行DAG第一件事情就是确定节点的执行顺序，从哪里开始，又如何结束？

考虑如下的几种场景：

从图中我们可以看出这也一些信息：

• 严格而言，理论上DAG只要求有向无环，但是不必要“连通”的。也即是说，图中可以存在孤立的点，或者多个分散的DAG构成一个大的DAG
• 一个节点可能有多个后继节点，意味着某个任务完成后，可能触发多个任务的执行
• 一个节点也可能有多个依赖，那么只有依赖节点全部执行完成后，才能开始它的执行

我们可以通过拓扑排序（Topological sorting）来确定节点的执行顺序。比较著名的算法是Kahn's algorithm，原理也很简单，从图中选出入度为零的节点，然后删掉这些节点和边，再重复这一过程直到原图为空。这样选出的这些节点就是拓扑排序的结果。

拓扑排序还有一个附带的好处，就是可以检测图中是否有环。如果图中没有入度为零的节点，但是选出的节点数目仍然小于图的节点数，那么拓扑排序失败，证明图中有环存在。

拓扑排序的结果并不一定是唯一的。比如场景一中，由于节点相互之间完全无依赖，因此任意一个组合都是正确的拓扑排序结果。

值得注意的是，尽管拓扑排序得到的是一个有序的列表，这并不意味着你需要“串行”执行。场景三的拓扑排序结果为[A,B,C,D]或者[A,B,D,C]，其中，C和D是可以并行执行的。

事件驱动架构

确定了节点的执行顺序之后，我们可以按照顺序来执行节点，看起来工作流的执行就这样完成了:

sorted_nodes <- dag.topological_sort()
for node in sorted_nodes:
    node.wait_for_dependencies()
    node.execute()

但是我们忽略了几个重要的问题：

• 节点之间是有交互的。考虑场景三中，节点B是可能需要使用A的运行结果的，也就是说，一个节点的运行结果应该对其后继节点可见
• 我们需要能够实时观测工作流的运行情况，比如当前执行到哪一个节点了？这对于界面调试、API调用等都十分有用。

使用事件驱动架构可以比较容易来解决这个问题。我们将工作流的运行抽象成一系列的事件，节点的执行产生事件（比如节点开始、节点结束），而这些事件具体如何使用则由消费者来决定。 这样，在UI上调试工作流时，我们只需要根据这些事件来更新页面上的节点状态；当使用API调用时，我们则可以过滤掉不感兴趣的事件，只监听结果事件并返回给调用方即可。 这样对于工作流执行引擎将有非常强的扩展性和可测试性。

传递闭包简化

对于使用DAG的系统，一个不可忽视的问题就是DAG的不确定性：用户很可能构建出奇奇怪怪的巨复杂的图出来，其中可能会有不必要的边。 这对于理解和执行工作流都可能造成麻烦。例如，下图中左边a→d, a→e 都是不必要的边。通过传递闭包简化（Transitive reduction）算法可以对无用的连接进行删除，从而得到一个唯一确定的简化工作流。

使用Go语言实现工作流

在有以上的一些理论基础后，实现思路已经非常清晰，尽管仍然有许多技术细节需要确定，就看如何具体设计和实现了。我们使用Go语言来实现，初步支持下面几种节点：

• 开始、结束
• 条件判断
• 大模型调用
• 插件：包括bing搜索、高召回、分流

概念模型

尽管前面我们认为工作流是一个DAG，但并不是直接搞一个DAG就能实现向Coze这样的工作流系统，这里面最大的问题在于，DAG中边只包含了方向， 但实际工作流产品的设计上，一条边还可能附带一些其他的信息。

例如，ComfyUI中一个节点到另一个节点连接是区分不同的点的：

当然它的业务场景跟我们差距比较大。但是，考虑到条件节点的表示，我们发现也存在类似的情况。

定义工作流Schema

在实现之前，首先需要解决的一个问题是，我们如何来表示一个工作流？比如条件判断的条件怎么存储？大模型调用的提示词怎么表示？

这是一个非常重要的问题，因为它是一切的基础，我们需要通过一个标准的协议来统一前端、后端等对于工作流的表示。不难发现，工作流的定义包括：

• DAG的表示，即节点和边
• 节点的自身属性，根据节点类型不同，其可用的属性也不尽相同

我们使用JSON Schema定义了一个工作流格式，大概长这样：

{
  "type": "object",
  "properties": {
    "name": {
      "type": "string"
    },
    "nodes": {
      "type": "array",
      "items": {
        "$ref": "#/definitions/node"
      }
    },
    "edges": {
      "type": "array",
      "items": {
        "$ref": "#/definitions/link"
      }
    },
    ...
  }
}

对于不同的节点，我们可以分别定义节点具体有哪些属性，格式是什么，是否必须等，例如：

{
  ...
  "bingSearchPluginProperties": {
    "properties": {
      "pluginType": {
        "type": "string"
      },
      "version": {
        "type": "string"
      },
      "query": {
        "$ref": "#/definitions/variableBinding"
      },
      "count": {
        "type": "number"
      }
    },
    "required": [
      "pluginType",
      "version",
      "query"
    ]
  }
}

完整的定义比较复杂，但通过这样一个Schema，我们可以解决很多问题：

• 这个工作流是否合法？至少格式上是否正确
• 前端开发时，如何清楚工作流支持多少个节点？每个节点属性是什么？

对于这种复杂的工作流来说，传统的API文档是无法精确到这样的程度的，不仅如此，它的最大好处是，它是真的可以运行的！

使用Go routine 实现异步工作流运行模型

前面我们有提到，尽管拓扑排序

未来展望

关于为什么要做Workflow，感兴趣可以从这里了解一些其他的信息。