从 Graph 到 EDA，LlamaIndex 复合AI系统的实现方案

前些天 LlamaIndex 推出了 Workflow 的实现， 旨在解决构建复杂 Agent AI 中遇到的问题。

LlamaIndex 工作流

LlamaIndex 工作流是一种在用户构建的日益复杂的 AI 应用程序中协调操作的机制。

背景

随着 LLM 的出现，人工智能应用程序由多个任务组成， 这些任务由不同的组件实现，这已成为事实上的标准。

市场上的开源框架致力于通过提供易于使用的基础组件抽象 （如数据加载器、LLM、向量数据库和重新排序器）以及外部服务， 让人工智能工程师的工作更加轻松。 与此同时，所有这些框架也在寻求最佳抽象来协调这些组件， 研究对人工智能开发人员来说最直观、最有效的方法， 以实现将复合人工智能系统结合在一起的逻辑。

现有的解决方案

其中两种潜在的编排模式是链和管道，它们都是相同有向无环图 (DAG)抽象的实现。

年初发布的查询管道(Query Pipeline)[1]中也尝试了这一点。

查询管道查询管道是一个声明性 API， 可让您针对不同用例（如 QA、结构化提取和代理自动化）编排数据的从简单到高级的查询工作流程。

基于Graph（DAG）的局限性

DAG 的一个基本方面是 DAG 中的『A』： 它们是非循环的，这意味着没有循环。

但在一个越来越具有代理性的世界里， 人工智能应用程序的逻辑中无法执行循环是完全不可接受的。 例如，如果一个组件提供了不好的结果，人工智能开发人员应该有办法告诉系统自我纠正并重试。

即使没有向 DAG 添加循环和重复（无方向），查询管道也存在一些明显的问题：

• 出现问题时很难调试

• 它们掩盖了组件和模块的执行方式

• 我们的管道编排器变得越来越复杂，必须处理大量不同的边缘情况

• 对于复杂的管道来说，它们很难读取

但是，一旦我们在查询管道中添加了循环，这些围绕图表的开发人员用户体验问题就会被放大。 我们在以下领域亲身体验了开发人员的痛苦：

• 许多核心编排逻辑（如if-else语句和while循环）都嵌入到图的边缘中。定义这些边缘变得繁琐而冗长。

• 处理可选值和默认值的极端情况变得困难。作为一个框架，我们很难确定参数是否会从上游节点传递过来。

• 对于构建代理的开发人员来说，定义带有循环的图表并不总是那么自然。代理封装了一个由 LLM 驱动的通用实体，该实体可以接收观察并生成响应。在这里，图表 UX 强制“代理”节点明确定义传入边和传出边，迫使用户定义与其他节点的详细通信模式。

从图到EDA：走向事件驱动

EDAEDA 是 『Event-Driven Architecture』 的缩写，即事件驱动架构。 这种架构模式强调系统对事件的响应，而不是由一个中心化的控制流程驱动。

复合 AI 系统可通过 LlamaIndex工作流实现。 工作流通过一组称为步骤的 Python 函数来回调度事件。 每个步骤可视为系统的一个组件：

• 一个用于处理查询，

• 一个用于与 LLM 对话，

• 一个用于从矢量数据库加载数据，等等。

每个步骤都会接收一个或多个要处理的事件，并且可以选择性地发回将在需要时中继到其他组件的事件。

转向事件驱动架构会导致设计发生根本性转变。 在许多图形实现中，图形遍历算法负责确定接下来应运行哪个组件以及应传递哪些数据。 在事件驱动架构中，组件会订阅某些类型的事件，并最终负责根据收到的数据决定要做什么。

工作流入门

我们来看一个例子。最小的 LlamaIndex 工作流程如下所示（为了方便查看，我们隐藏包导入的部分）：

class OpenAIGenerator(Workflow):

@step()

async def generate(self, ev: StartEvent) -> StopEvent:

query = ev.get("query")

llm = OpenAI()

response = await llm.acomplete(query)

return StopEvent(result=str(response))

w = OpenAIGenerator(timeout=10, verbose=False)

result = await w.run(query="What's LlamaIndex?")

print(result)

使用装饰器@step标记generate函数为工作流的一个步骤， 并使用带有适当类型注释的方法签名声明它想要接收哪些事件以及它将发送回哪些事件。

为了运行工作流， 我们创建OpenAIGenerator类的实例并传递一些配置参数（例如所需的超时时间），

w = OpenAIGenerator(timeout=10, verbose=False)

然后调用该示例的run方法。

result = await w.run(query="What's LlamaIndex?")

我们在看generate方法的实现：

• 传递给run的任何关键字参数，都将被打包到StartEvent中；

• 通过ev.get("query")获取参数；

• 返回一个特殊类型的事件StopEvent，向工作流发出信号以正常停止其执行；

•StopEvent携带工作流结果返回给调用者，比如本例中的 LLM 响应。

复杂的用法可以参考 LlamaIndex 的文档：

• Introducing workflows[2]

• Workflow guides[3]

调试工作流程

工作流程的复杂性将随着应用程序逻辑的复杂性而增长， 有时仅通过查看 Python 代码可能很难理解事件在执行过程中的流动方式。

为了便于理解复杂的工作流程并支持工作流程执行的调试，LlamaIndex 提供了两个函数：

•draw_all_possible_flows生成一幅图，展示工作流程中的所有步骤以及事件可能如何流动

•draw_most_recent_execution生成类似的图片，仅显示上次工作流执行期间实际发送的事件

工作流的功能

工作流具有一组您通常会期望从更成熟的软件中获得的功能：

• 完全异步，支持流媒体

• 默认进行检测，通过支持的集成提供一键可观察性

• 逐步执行，更易于调试

• 事件驱动依赖关系的验证和可视化

• 事件被实现为 pydantic 模型，以简化新功能的定制和进一步开发

Pydantic模型Pydantic 是 是 Python 中使用最广泛的数据验证框架。他可以定义数据类型， 方便和各种工具集成。比如上期我们使用 FastAPI 编写 Dify 自定义工具时就是用 Pydantic 来定义参数，从而生成 OpenAPI 文档。 更多信息可以查看 Pydantic文档[4]。

引用链接